<commit_before>"""
Mitmproxy Content Views
=======================

mitmproxy includes a set of content views which can be used to
format/decode/highlight data. While they are currently used for HTTP message
bodies only, the may be used in other contexts in the future, e.g. to decode
protobuf messages sent as WebSocket frames.

Thus, the View API is very minimalistic. The only arguments are `data` and
`**metadata`, where `data` is the actual content (as bytes). The contents on
metadata depend on the protocol in use. For HTTP, the message headers are
passed as the ``headers`` keyword argument. For HTTP requests, the query
parameters are passed as the ``query`` keyword argument.
"""
import traceback
from typing import Dict, Optional  # noqa
from typing import List  # noqa

from mitmproxy import exceptions
from mitmproxy.net import http
from mitmproxy.utils import strutils
from . import (
    auto, raw, hex, json, xml_html, wbxml, javascript, css,
    urlencoded, multipart, image, query, protobuf
)
from .base import View, KEY_MAX, format_text, format_dict, TViewResult

views: List[View] = []
content_types_map: Dict[str, List[View]] = {}


def get(name: str) -> Optional[View]:
    for i in views:
        if i.name.lower() == name.lower():
            return i
    return None


def add(view: View) -> None:
    # TODO: auto-select a different name (append an integer?)
    for i in views:
        if i.name == view.name:
            raise exceptions.ContentViewException("Duplicate view: " + view.name)

    views.append(view)

    for ct in view.content_types:
        l = content_types_map.setdefault(ct, [])
        l.append(view)


def remove(view: View) -> None:
    for ct in view.content_types:
        l = content_types_map.setdefault(ct, [])
        l.remove(view)

        if not len(l):
            del content_types_map[ct]

    views.remove(view)


def safe_to_print(lines, encoding="utf8"):
    """
    Wraps a content generator so that each text portion is a *safe to print* unicode string.
    """
    for line in lines:
        clean_line = []
        for (style, text) in line:
            if isinstance(text, bytes):
                text = text.decode(encoding, "replace")
            text = strutils.escape_control_characters(text)
            clean_line.append((style, text))
        yield clean_line


def get_message_content_view(viewname, message):
    """
    Like get_content_view, but also handles message encoding.
    """
    viewmode = get(viewname)
    if not viewmode:
        viewmode = get("auto")
    try:
        content = message.content
    except ValueError:
        content = message.raw_content
        enc = "[cannot decode]"
    else:
        if isinstance(message, http.Message) and content != message.raw_content:
            enc = "[decoded {}]".format(
                message.headers.get("content-encoding")
            )
        else:
            enc = None

    if content is None:
        return "", iter([[("error", "content missing")]]), None

    metadata = {}
    if isinstance(message, http.Request):
        metadata["query"] = message.query
    if isinstance(message, http.Message):
        metadata["headers"] = message.headers

    description, lines, error = get_content_view(
        viewmode, content, **metadata
    )

    if enc:
        description = "{} {}".format(enc, description)

    return description, lines, error

<commit_msg>Add message to ContentView metadata<commit_after>"""
Mitmproxy Content Views
=======================

mitmproxy includes a set of content views which can be used to
format/decode/highlight data. While they are currently used for HTTP message
bodies only, the may be used in other contexts in the future, e.g. to decode
protobuf messages sent as WebSocket frames.

Thus, the View API is very minimalistic. The only arguments are `data` and
`**metadata`, where `data` is the actual content (as bytes). The contents on
metadata depend on the protocol in use. For HTTP, the message headers are
passed as the ``headers`` keyword argument. For HTTP requests, the query
parameters are passed as the ``query`` keyword argument.
"""
import traceback
from typing import Dict, Optional  # noqa
from typing import List  # noqa

from mitmproxy import exceptions
from mitmproxy.net import http
from mitmproxy.utils import strutils
from . import (
    auto, raw, hex, json, xml_html, wbxml, javascript, css,
    urlencoded, multipart, image, query, protobuf
)
from .base import View, KEY_MAX, format_text, format_dict, TViewResult

views: List[View] = []
content_types_map: Dict[str, List[View]] = {}


def get(name: str) -> Optional[View]:
    for i in views:
        if i.name.lower() == name.lower():
            return i
    return None


def add(view: View) -> None:
    # TODO: auto-select a different name (append an integer?)
    for i in views:
        if i.name == view.name:
            raise exceptions.ContentViewException("Duplicate view: " + view.name)

    views.append(view)

    for ct in view.content_types:
        l = content_types_map.setdefault(ct, [])
        l.append(view)


def remove(view: View) -> None:
    for ct in view.content_types:
        l = content_types_map.setdefault(ct, [])
        l.remove(view)

        if not len(l):
            del content_types_map[ct]

    views.remove(view)


def safe_to_print(lines, encoding="utf8"):
    """
    Wraps a content generator so that each text portion is a *safe to print* unicode string.
    """
    for line in lines:
        clean_line = []
        for (style, text) in line:
            if isinstance(text, bytes):
                text = text.decode(encoding, "replace")
            text = strutils.escape_control_characters(text)
            clean_line.append((style, text))
        yield clean_line


def get_message_content_view(viewname, message):
    """
    Like get_content_view, but also handles message encoding.
    """
    viewmode = get(viewname)
    if not viewmode:
        viewmode = get("auto")
    try:
        content = message.content
    except ValueError:
        content = message.raw_content
        enc = "[cannot decode]"
    else:
        if isinstance(message, http.Message) and content != message.raw_content:
            enc = "[decoded {}]".format(
                message.headers.get("content-encoding")
            )
        else:
            enc = None

    if content is None:
        return "", iter([[("error", "content missing")]]), None

    metadata = {}
    if isinstance(message, http.Request):
        metadata["query"] = message.query
    if isinstance(message, http.Message):
        metadata["headers"] = message.headers
    metadata["message"] = message

    description, lines, error = get_content_view(
        viewmode, content, **metadata
    )

    if enc:
        description = "{} {}".format(enc, description)

    return description, lines, error


{'fraction_non_alphanumeric': 0.10638606676342525, 'fraction_numerical': 0.00029027576197387516, 'mean_word_length': 2.832591768631813, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4649849', 'n_tokens_mistral': 1968, 'n_tokens_neox': 1906, 'n_words': 684}