name: "http-framework-test"
command: "python3"
args:
  - "-c"
  - |
    import argparse
    import json
    import os
    import re
    import shlex
    import socket
    import ssl
    import sys
    import time
    import urllib.parse
    from typing import Dict, List, Tuple

    try:
        import httpx
    except ImportError:
        print("Missing dependency: httpx. Install it with `pip install httpx`.", file=sys.stderr)
        sys.exit(1)

    try:
        from charset_normalizer import from_bytes as charset_from_bytes
    except ImportError:
        charset_from_bytes = None

    try:
        import chardet
    except ImportError:
        chardet = None

    METRIC_KEYS = [
        "dns_lookup",
        "tcp_connect",
        "tls_handshake",
        "pretransfer",
        "ttfb",
        "total",
        "speed_download",
        "size_download",
        "http_code",
        "redirects",
    ]


    def parse_headers(raw: str) -> List[Tuple[str, str]]:
        if not raw:
            return []
        raw = raw.strip()
        if not raw:
            return []
        headers: List[Tuple[str, str]] = []
        parsed = None
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError:
            parsed = None
        if isinstance(parsed, dict):
            for key, value in parsed.items():
                headers.append((str(key).strip(), str(value).strip()))
            return headers
        if isinstance(parsed, list):
            for item in parsed:
                if isinstance(item, str) and ":" in item:
                    key, value = item.split(":", 1)
                    headers.append((key.strip(), value.strip()))
            if headers:
                return headers
        for line in raw.replace(";", "\n").splitlines():
            if ":" not in line:
                continue
            key, value = line.split(":", 1)
            headers.append((key.strip(), value.strip()))
        return headers


    def parse_cookies(raw: str) -> Dict[str, str]:
        cookies: Dict[str, str] = {}
        if not raw:
            return cookies
        for part in raw.split(";"):
            if "=" not in part:
                continue
            name, value = part.split("=", 1)
            name = name.strip()
            if not name:
                continue
            cookies[name] = value.strip()
        return cookies


    def parse_additional_options(raw: str) -> Dict[str, str]:
        options: Dict[str, str] = {}
        if not raw:
            return options
        try:
            tokens = shlex.split(raw)
        except ValueError:
            tokens = raw.split()
        for token in tokens:
            if "=" in token:
                key, value = token.split("=", 1)
                options[key.strip()] = value.strip()
            else:
                options[token.strip()] = "true"
        return options


    def str_to_bool(value) -> bool:
        if isinstance(value, bool):
            return value
        if value is None:
            return False
        return str(value).strip().lower() in {"1", "true", "yes", "on"}


    def smart_encode_url(url: str, safe_path="/:@&=%+,$-~", safe_query="/:@&=%+,$-~"):
        try:
            parts = urllib.parse.urlsplit(url)
        except ValueError:
            return url
        path = urllib.parse.quote(parts.path or "/", safe=safe_path)
        query = urllib.parse.quote(parts.query, safe=safe_query)
        fragment = urllib.parse.quote(parts.fragment, safe=safe_query)
        return urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, query, fragment))


    def quote_form_component_preserving_pct(value: str) -> str:
        normalized = urllib.parse.unquote(value)
        return urllib.parse.quote_plus(normalized, safe="")


    def encode_form_data(data: str) -> str:
        if not data:
            return data

        def find_key_value_pairs(text):
            pairs = []
            i = 0
            text_len = len(text)

            while i < text_len:
                while i < text_len and text[i] in " \t\n\r":
                    i += 1
                if i >= text_len:
                    break

                key_start = i
                while i < text_len and text[i] != "=":
                    i += 1

                if i >= text_len:
                    remaining = text[key_start:].strip()
                    if remaining:
                        pairs.append((None, remaining))
                    break

                key = text[key_start:i].strip()
                i += 1

                if not key:
                    continue

                value_start = i
                value_end = text_len

                j = value_start
                while j < text_len:
                    if text[j] == "&":
                        k = j + 1
                        while k < text_len and text[k] in " \t\n\r":
                            k += 1
                        m = k
                        while m < text_len and text[m] not in "=&":
                            m += 1
                        if m < text_len and text[m] == "=":
                            value_end = j
                            i = j + 1
                            break
                    j += 1

                value = text[value_start:value_end]
                pairs.append((key, value))

                if value_end < text_len:
                    i = value_end + 1
                else:
                    break

            return pairs

        pairs = find_key_value_pairs(data)
        parts = []

        for key, value in pairs:
            if key is None:
                parts.append(quote_form_component_preserving_pct(value))
            else:
                encoded_key = quote_form_component_preserving_pct(key)
                encoded_value = quote_form_component_preserving_pct(value)
                parts.append(f"{encoded_key}={encoded_value}")

        return "&".join(parts)


    def should_encode_form(headers: httpx.Headers, data: str) -> bool:
        if not data:
            return False
        if not headers:
            return False
        content_type = headers.get("Content-Type")
        if not content_type:
            return False
        return "application/x-www-form-urlencoded" in content_type.lower()


    def extract_charset_from_content_type(content_type: str) -> str:
        if not content_type:
            return ""
        parts = content_type.split(";")
        for part in parts[1:]:
            lowered = part.strip().lower()
            if lowered.startswith("charset="):
                return part.split("=", 1)[1].strip().strip('"').strip("'")
        return ""


    def extract_declared_charset_from_body(data: bytes) -> str:
        if not data:
            return ""
        sample = data[:16384]
        try:
            sample_text = sample.decode("iso-8859-1", errors="ignore")
        except UnicodeDecodeError:
            return ""
        for line in sample_text.splitlines():
            lowered = line.lower()
            if "content-type:" in lowered and "charset=" in lowered:
                charset_index = lowered.index("charset=") + len("charset=")
                remainder = line[charset_index:]
                for separator in [";", " ", "\t"]:
                    remainder = remainder.split(separator)[0]
                return remainder.strip().strip('"').strip("'")
        meta_match = re.search(r'charset=["\']?([a-zA-Z0-9_\-.:]+)', sample_text, re.IGNORECASE)
        if meta_match:
            return meta_match.group(1)
        return ""


    def decode_body_bytes(data: bytes, headers: httpx.Headers, user_encoding: str = ""):
        attempts = []
        if user_encoding:
            attempts.append(("user", user_encoding))
        header_declared = extract_charset_from_content_type(headers.get("Content-Type", "")) if headers else ""
        if header_declared:
            attempts.append(("header", header_declared))
        body_declared = extract_declared_charset_from_body(data) if not header_declared else ""
        if body_declared:
            attempts.append(("body", body_declared))
        for source, encoding in attempts:
            enc = (encoding or "").strip()
            if not enc:
                continue
            try:
                return data.decode(enc), enc, source
            except (LookupError, UnicodeDecodeError):
                continue
        if charset_from_bytes is not None and data:
            best = charset_from_bytes(data).best()
            if best and best.encoding:
                try:
                    return data.decode(best.encoding), best.encoding, "detected"
                except (LookupError, UnicodeDecodeError):
                    pass
        if chardet is not None and data:
            detection = chardet.detect(data)
            encoding = detection.get("encoding")
            if encoding:
                try:
                    return data.decode(encoding), encoding, "detected"
                except (LookupError, UnicodeDecodeError):
                    pass
        try:
            return data.decode("utf-8"), "utf-8", "fallback"
        except UnicodeDecodeError:
            return data.decode("utf-8", errors="replace"), "utf-8", "fallback"


    def prepare_body(data: str, headers: httpx.Headers, debug: bool = False):
        meta = {
            "source": "inline",
            "mode": "none",
            "length": 0,
            "charset": None,
            "encoded": False,
        }
        if not data:
            return None, meta
        if data.startswith("@"):
            path = data[1:]
            if path == "-":
                payload = sys.stdin.buffer.read()
                meta["source"] = "stdin"
            else:
                path = os.path.expanduser(path)
                if not os.path.isfile(path):
                    raise FileNotFoundError(f"Body file not found: {path}")
                with open(path, "rb") as fh:
                    payload = fh.read()
                meta["source"] = path
            meta["mode"] = "binary"
            meta["length"] = len(payload)
            return payload, meta
        stripped = data.strip()
        content_type = headers.get("Content-Type")
        if not content_type:
            guessed = ""
            if stripped.startswith("{") or stripped.startswith("["):
                guessed = "application/json"
            elif "=" in stripped and "&" in stripped:
                guessed = "application/x-www-form-urlencoded"
            elif stripped.startswith("<"):
                guessed = "application/xml"
            if guessed:
                headers["Content-Type"] = guessed
                content_type = guessed
        processed = data
        if should_encode_form(headers, data):
            processed = encode_form_data(data)
            meta["mode"] = "form-urlencoded"
            meta["encoded"] = True
        else:
            normalized = (headers.get("Content-Type") or "").lower()
            if normalized.startswith("application/json"):
                meta["mode"] = "json"
            elif normalized.startswith("multipart/form-data"):
                meta["mode"] = "multipart"
            elif normalized.startswith("text/") or "xml" in normalized:
                meta["mode"] = "text"
            else:
                meta["mode"] = "raw"
        content_type = headers.get("Content-Type")
        charset = extract_charset_from_content_type(content_type) if content_type else ""
        if not charset and meta["mode"] in ("json", "text", "form-urlencoded"):
            charset = "utf-8"
            base = (content_type or "text/plain").split(";")[0].strip()
            headers["Content-Type"] = f"{base}; charset={charset}"
        elif not charset:
            charset = "utf-8"
        body_bytes = processed.encode(charset, errors="surrogatepass")
        meta["charset"] = charset
        meta["length"] = len(body_bytes)
        if debug:
            print("\n===== Debug: Body Encoding =====")
            print(f"Mode: {meta['mode']}")
            print(f"Charset: {charset}")
            print(f"Length: {meta['length']} bytes")
            print(f"Source: {meta['source']}")
            if meta["encoded"]:
                print("Form data was URL-encoded prior to sending.")
        return body_bytes, meta


    def probe_connection(url: str, timeout: float, verify_tls: bool, skip: bool):
        metrics = {}
        if skip:
            return metrics
        parsed = urllib.parse.urlsplit(url)
        host = parsed.hostname
        if not host:
            return metrics
        port = parsed.port or (443 if parsed.scheme == "https" else 80)
        dns_start = time.perf_counter()
        try:
            addr_info = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
        except socket.gaierror:
            return metrics
        dns_time = time.perf_counter() - dns_start
        family, socktype, proto, _, sockaddr = addr_info[0]
        sock = socket.socket(family, socktype, proto)
        sock.settimeout(timeout or 30.0)
        try:
            connect_start = time.perf_counter()
            sock.connect(sockaddr)
            connect_time = time.perf_counter() - connect_start
            tls_time = 0.0
            if parsed.scheme == "https":
                ctx = ssl.create_default_context()
                if not verify_tls:
                    ctx.check_hostname = False
                    ctx.verify_mode = ssl.CERT_NONE
                try:
                    tls_start = time.perf_counter()
                    tls_sock = ctx.wrap_socket(sock, server_hostname=host, do_handshake_on_connect=False)
                    try:
                        tls_sock.do_handshake()
                        tls_time = time.perf_counter() - tls_start
                    finally:
                        tls_sock.close()
                except ssl.SSLError:
                    tls_time = 0.0
            else:
                sock.close()
        except OSError:
            sock.close()
            return {"dns_lookup": dns_time}
        metrics["dns_lookup"] = dns_time
        metrics["tcp_connect"] = connect_time
        metrics["tls_handshake"] = tls_time
        metrics["pretransfer"] = dns_time + connect_time + tls_time
        return metrics


    def format_metric_value(key: str, value):
        if value is None:
            return "n/a"
        if key in {"http_code", "redirects", "size_download"}:
            return str(int(value))
        if key == "speed_download":
            return f"{value:.2f} B/s"
        return f"{value:.6f}s"


    def summarize(values):
        if not values:
            return None
        count = len(values)
        total = sum(values)
        return min(values), total / count, max(values)


    def render_request_overview(method: str, url: str, headers: httpx.Headers, body_meta: Dict[str, str]):
        items = list(headers.items())
        print("\n===== Prepared Request =====")
        print(f"Method: {method}")
        print(f"URL: {url}")
        print(f"Headers ({len(items)} total):")
        for key, value in items:
            print(f"  {key}: {value}")
        if body_meta["length"]:
            charset = body_meta.get("charset") or "n/a"
            print(f"Body: {body_meta['length']} bytes ({body_meta['mode']}, charset={charset}, source={body_meta['source']})")
        else:
            print("Body: <empty>")


    def compile_response_filter(pattern: str, ignore_case: bool):
        flags = 0
        if ignore_case:
            flags |= re.IGNORECASE
        try:
            return re.compile(pattern, flags)
        except re.error as exc:
            print(f"Invalid response_filter regex: {exc}", file=sys.stderr)
            sys.exit(2)


    def truncate_utf8(text: str, max_bytes: int) -> Tuple[str, bool]:
        if max_bytes <= 0 or not text:
            return text, False
        encoded = text.encode("utf-8", errors="replace")
        if len(encoded) <= max_bytes:
            return text, False
        truncated = encoded[:max_bytes].decode("utf-8", errors="ignore")
        return truncated, True


    def cap_line_entries(entries: List[Tuple[int, str]], max_lines: int) -> Tuple[List[Tuple[int, str]], bool]:
        if max_lines <= 0 or len(entries) <= max_lines:
            return entries, False
        return entries[:max_lines], True


    def expand_line_context(line_numbers: List[int], total_lines: int, context: int) -> List[int]:
        if context <= 0:
            return sorted(set(line_numbers))
        included = set()
        for num in line_numbers:
            start = max(1, num - context)
            end = min(total_lines, num + context)
            for i in range(start, end + 1):
                included.add(i)
        return sorted(included)


    def format_line_entries(lines: List[str], indices: List[int], ellipsis_gaps: bool = True) -> str:
        if not indices:
            return ""
        chunks = []
        prev = None
        for num in indices:
            if ellipsis_gaps and prev is not None and num > prev + 1:
                chunks.append("  ...")
            chunks.append(f"  L{num}: {lines[num - 1]}")
            prev = num
        return "\n".join(chunks)


    def filter_body_by_lines(
        lines: List[str],
        compiled: "re.Pattern",
        invert: bool,
        context_lines: int,
        max_lines: int,
    ) -> Tuple[str, Dict[str, object]]:
        matched_nums = []
        for idx, line in enumerate(lines, start=1):
            hit = compiled.search(line) is not None
            if invert:
                hit = not hit
            if hit:
                matched_nums.append(idx)
        total = len(lines)
        meta = {
            "mode": "line",
            "total_lines": total,
            "matched_lines": len(matched_nums),
            "invert": invert,
            "truncated": False,
            "byte_truncated": False,
        }
        if not matched_nums:
            return "", meta
        display_nums = expand_line_context(matched_nums, total, context_lines)
        entries = [(n, lines[n - 1]) for n in display_nums]
        entries, line_capped = cap_line_entries(entries, max_lines)
        meta["truncated"] = line_capped
        meta["display_lines"] = len(entries)
        return format_line_entries(lines, [n for n, _ in entries], ellipsis_gaps=context_lines > 0), meta


    def filter_body_multiline(
        text: str,
        compiled: "re.Pattern",
        invert: bool,
        max_lines: int,
        dotall: bool,
    ) -> Tuple[str, Dict[str, object]]:
        flags = compiled.flags
        if dotall:
            pattern = re.compile(compiled.pattern, flags | re.DOTALL | re.MULTILINE)
        else:
            pattern = re.compile(compiled.pattern, flags | re.MULTILINE)
        matches = list(pattern.finditer(text))
        if invert:
            if matches:
                return "", {"mode": "multiline" if not dotall else "full", "total_lines": text.count("\n") + (1 if text else 0), "matched_lines": 0, "invert": True, "truncated": False, "byte_truncated": False}
            output = text
            meta = {"mode": "multiline" if not dotall else "full", "matched_lines": 1, "invert": True, "truncated": False, "byte_truncated": False}
            lines = text.splitlines()
            if max_lines > 0 and len(lines) > max_lines:
                output = "\n".join(lines[:max_lines])
                meta["truncated"] = True
            meta["total_lines"] = len(lines)
            meta["display_lines"] = min(len(lines), max_lines) if max_lines > 0 else len(lines)
            return output, meta
        chunks = []
        for match in matches:
            snippet = match.group(0)
            if "\n" in snippet:
                snippet = snippet.replace("\n", "\\n")
            start_line = text.count("\n", 0, match.start()) + 1
            chunks.append((start_line, f"  @{start_line}: {snippet}"))
        entries, line_capped = cap_line_entries(chunks, max_lines if max_lines > 0 else len(chunks))
        meta = {
            "mode": "multiline" if not dotall else "full",
            "total_lines": text.count("\n") + (1 if text else 0),
            "matched_lines": len(matches),
            "invert": False,
            "truncated": line_capped,
            "byte_truncated": False,
            "display_lines": len(entries),
        }
        return "\n".join(line for _, line in entries), meta


    def apply_body_limits_plain(text: str, max_lines: int) -> Tuple[str, Dict[str, object]]:
        lines = text.splitlines()
        meta = {
            "mode": "plain",
            "total_lines": len(lines),
            "matched_lines": len(lines),
            "invert": False,
            "truncated": False,
            "byte_truncated": False,
            "display_lines": len(lines),
        }
        output = text
        if max_lines > 0 and len(lines) > max_lines:
            output = "\n".join(lines[:max_lines])
            meta["truncated"] = True
            meta["display_lines"] = max_lines
        return output, meta


    def format_response_body_output(
        decoded_body: str,
        filter_pattern: str,
        filter_mode: str,
        filter_invert: bool,
        filter_ignore_case: bool,
        max_lines: int,
        max_bytes: int,
        preview_lines: int,
        context_lines: int,
        compiled_filter=None,
    ) -> Tuple[str, Dict[str, object]]:
        text = decoded_body.rstrip("\r\n")
        if not text:
            return "", {"mode": "empty", "total_lines": 0, "matched_lines": 0, "invert": filter_invert, "truncated": False, "byte_truncated": False, "display_lines": 0}

        lines = text.splitlines()
        mode = (filter_mode or "line").strip().lower()
        if mode not in {"line", "multiline", "full"}:
            mode = "line"

        if filter_pattern:
            compiled = compiled_filter or compile_response_filter(filter_pattern, filter_ignore_case)
            if mode == "line":
                output, meta = filter_body_by_lines(lines, compiled, filter_invert, context_lines, max_lines)
            else:
                output, meta = filter_body_multiline(text, compiled, filter_invert, max_lines, dotall=(mode == "full"))
            meta["filter_pattern"] = filter_pattern
            if not output and not filter_invert:
                preview = min(max(preview_lines, 0), len(lines))
                if preview > 0:
                    preview_text = format_line_entries(lines, list(range(1, preview + 1)), ellipsis_gaps=False)
                    preview_text, byte_truncated = truncate_utf8(preview_text, max_bytes)
                    return preview_text, {
                        **meta,
                        "preview": True,
                        "matched_lines": 0,
                        "display_lines": preview,
                        "byte_truncated": byte_truncated,
                    }
                return "", {**meta, "preview": False, "matched_lines": 0, "display_lines": 0}
        else:
            output, meta = apply_body_limits_plain(text, max_lines)

        output, byte_truncated = truncate_utf8(output, max_bytes)
        if byte_truncated:
            meta["byte_truncated"] = True
        return output, meta


    def print_response_body_summary(meta: Dict[str, object]):
        mode = meta.get("mode")
        if mode == "empty":
            return
        parts = [f"mode={mode}"]
        if meta.get("filter_pattern"):
            parts.append(f"pattern={meta['filter_pattern']!r}")
        if meta.get("invert"):
            parts.append("invert=true")
        total = meta.get("total_lines")
        matched = meta.get("matched_lines")
        displayed = meta.get("display_lines")
        if total is not None and matched is not None:
            parts.append(f"matched {matched}/{total} lines")
        if displayed is not None:
            parts.append(f"showing {displayed}")
        if meta.get("preview"):
            parts.append("preview on zero match")
        if meta.get("truncated"):
            parts.append("line cap applied")
        if meta.get("byte_truncated"):
            parts.append("byte cap applied")
        print(f"[body] {' | '.join(parts)}")


    def main():
        parser = argparse.ArgumentParser(description="Pure Python HTTP testing helper powered by httpx")
        parser.add_argument("--url", required=True)
        parser.add_argument("--method", default="GET")
        parser.add_argument("--data", default="")
        parser.add_argument("--headers", default="", type=str)
        parser.add_argument("--cookies", default="")
        parser.add_argument("--user-agent", dest="user_agent", default="")
        parser.add_argument("--proxy", default="")
        parser.add_argument("--timeout", default="")
        parser.add_argument("--repeat", type=int, default=1)
        parser.add_argument("--delay", default="0")
        parser.add_argument("--additional-args", dest="additional_args", default="")
        parser.add_argument("--action", default="")
        parser.add_argument("--include-headers", dest="include_headers", action="store_true")
        parser.add_argument("--no-include-headers", dest="include_headers", action="store_false")
        parser.add_argument("--auto-encode-url", dest="auto_encode_url", action="store_true")
        parser.add_argument("--no-auto-encode-url", dest="auto_encode_url", action="store_false")
        parser.add_argument("--follow-redirects", dest="follow_redirects", action="store_true")
        parser.add_argument("--allow-insecure", dest="allow_insecure", action="store_true")
        parser.add_argument("--verbose-output", dest="verbose_output", action="store_true")
        parser.add_argument("--show-command", dest="show_command", action="store_true")
        parser.add_argument("--show-summary", dest="show_summary", action="store_true")
        parser.add_argument("--debug", dest="debug", action="store_true")
        parser.add_argument("--response-encoding", dest="response_encoding", default="")
        parser.add_argument("--download", dest="download", default="")
        parser.add_argument("--response-filter", dest="response_filter", default="")
        parser.add_argument("--response-filter-mode", dest="response_filter_mode", default="line")
        parser.add_argument("--response-filter-invert", dest="response_filter_invert", action="store_true")
        parser.add_argument("--no-response-filter-invert", dest="response_filter_invert", action="store_false")
        parser.add_argument("--response-filter-ignore-case", dest="response_filter_ignore_case", action="store_true")
        parser.add_argument("--no-response-filter-ignore-case", dest="response_filter_ignore_case", action="store_false")
        parser.add_argument("--response-max-lines", dest="response_max_lines", type=int, default=0)
        parser.add_argument("--response-max-bytes", dest="response_max_bytes", type=int, default=0)
        parser.add_argument("--response-preview-lines", dest="response_preview_lines", type=int, default=5)
        parser.add_argument("--response-context-lines", dest="response_context_lines", type=int, default=0)
        parser.set_defaults(
            include_headers=False,
            auto_encode_url=False,
            follow_redirects=False,
            allow_insecure=False,
            verbose_output=False,
            show_command=False,
            show_summary=False,
            debug=False,
            response_filter_invert=False,
            response_filter_ignore_case=False,
        )
        args = parser.parse_args()

        response_filter = (args.response_filter or "").strip()
        response_max_lines = max(0, args.response_max_lines or 0)
        response_max_bytes = max(0, args.response_max_bytes or 0)
        response_preview_lines = max(0, args.response_preview_lines if args.response_preview_lines is not None else 5)
        response_context_lines = max(0, args.response_context_lines or 0)
        compiled_response_filter = None
        if response_filter:
            compiled_response_filter = compile_response_filter(
                response_filter, args.response_filter_ignore_case
            )

        repeat = max(1, args.repeat)
        try:
            delay_between = float(args.delay or "0")
            if delay_between < 0:
                delay_between = 0.0
        except ValueError:
            delay_between = 0.0

        prepared_url = smart_encode_url(args.url) if args.auto_encode_url else args.url
        method = (args.method or "GET").upper()

        # 处理 headers：支持字典（JSON字符串）和字符串格式
        # 框架会将 object 类型序列化为 JSON 字符串传递
        headers_list = []
        if args.headers:
            headers_str = args.headers.strip()
            # 优先尝试解析为 JSON（框架传递的字典会被序列化为 JSON）
            if headers_str.startswith("{") or headers_str.startswith("["):
                try:
                    parsed = json.loads(headers_str)
                    if isinstance(parsed, dict):
                        # 字典格式：直接转换为 (key, value) 元组列表
                        headers_list = [(str(k).strip(), str(v).strip()) for k, v in parsed.items()]
                    elif isinstance(parsed, list):
                        # 数组格式：使用原有的 parse_headers 函数处理
                        headers_list = parse_headers(headers_str)
                    else:
                        headers_list = parse_headers(headers_str)
                except (json.JSONDecodeError, ValueError):
                    # JSON 解析失败，回退到原有的字符串解析逻辑
                    headers_list = parse_headers(headers_str)
            else:
                # 非 JSON 格式，使用原有的字符串解析逻辑（向后兼容）
                headers_list = parse_headers(headers_str)
        headers = httpx.Headers(headers_list)
        if args.user_agent:
            headers["User-Agent"] = args.user_agent

        body_bytes = None
        body_meta = {
            "source": "inline",
            "mode": "none",
            "length": 0,
            "charset": None,
            "encoded": False,
        }
        try:
            body_bytes, body_meta = prepare_body(args.data, headers, args.debug)
        except FileNotFoundError as exc:
            print(f"Body preparation failed: {exc}", file=sys.stderr)
            sys.exit(1)

        cookie_dict = parse_cookies(args.cookies)
        cookie_jar = httpx.Cookies()
        for name, value in cookie_dict.items():
            cookie_jar.set(name, value)

        additional_options = parse_additional_options(args.additional_args)

        timeout_value = None
        if args.timeout:
            try:
                timeout_value = float(args.timeout)
            except ValueError:
                timeout_value = None
        timeout = httpx.Timeout(timeout_value or 60.0)

        client_kwargs = {
            "timeout": timeout,
            "verify": not args.allow_insecure,
            "follow_redirects": args.follow_redirects,
            "cookies": cookie_jar,
        }
        if args.proxy:
            client_kwargs["proxies"] = args.proxy
        if "http2" in additional_options:
            client_kwargs["http2"] = str_to_bool(additional_options["http2"])
        if "cert" in additional_options:
            client_kwargs["cert"] = additional_options["cert"]
        if "verify" in additional_options:
            value = additional_options["verify"]
            lowered = value.strip().lower()
            client_kwargs["verify"] = str_to_bool(value) if lowered in {"true", "false", "1", "0", "yes", "no", "on", "off"} else value
        if "max_redirects" in additional_options:
            try:
                client_kwargs["max_redirects"] = int(additional_options["max_redirects"])
            except ValueError:
                pass
        if "trust_env" in additional_options:
            client_kwargs["trust_env"] = str_to_bool(additional_options["trust_env"])

        if args.show_command:
            render_request_overview(method, prepared_url, headers, body_meta)

        aggregate = None
        if args.show_summary:
            aggregate = {key: [] for key in METRIC_KEYS}
            aggregate["wall_time"] = []

        exit_code = 0
        verify_option = client_kwargs.get("verify", True)
        verify_tls = verify_option if isinstance(verify_option, bool) else True
        skip_probe = bool(args.proxy)

        client = httpx.Client(**client_kwargs)
        try:
            for run_index in range(repeat):
                if run_index > 0 and delay_between > 0:
                    time.sleep(delay_between)

                metrics = {key: None for key in METRIC_KEYS}
                probe_metrics = probe_connection(prepared_url, timeout_value or 60.0, verify_tls, skip_probe)
                metrics.update(probe_metrics)

                start = time.perf_counter()
                first_byte_time = None
                body_buffer = bytearray()

                try:
                    stream_kwargs = {
                        "method": method,
                        "url": prepared_url,
                        "headers": headers,
                    }
                    if body_bytes is not None:
                        stream_kwargs["content"] = body_bytes

                    with client.stream(**stream_kwargs) as response:
                        status_code = response.status_code
                        metrics["http_code"] = status_code
                        metrics["redirects"] = len(response.history)
                        http_version = response.http_version or "HTTP/1.1"

                        for chunk in response.iter_bytes():
                            if first_byte_time is None:
                                first_byte_time = time.perf_counter()
                            body_buffer.extend(chunk)

                        total_elapsed = time.perf_counter() - start
                        metrics["total"] = total_elapsed
                        metrics["ttfb"] = (first_byte_time - start) if first_byte_time else total_elapsed
                        metrics["size_download"] = len(body_buffer)
                        metrics["speed_download"] = (len(body_buffer) / total_elapsed) if total_elapsed > 0 else 0.0
                        if metrics.get("pretransfer") is None:
                            metrics["pretransfer"] = (metrics.get("dns_lookup") or 0.0) + (metrics.get("tcp_connect") or 0.0) + (metrics.get("tls_handshake") or 0.0)

                        decoded_body, used_encoding, encoding_source = decode_body_bytes(bytes(body_buffer), response.headers, args.response_encoding)

                        print(f"\n===== Response #{run_index + 1} =====")
                        download_target = (args.download or "").strip()
                        if download_target:
                            # 支持多次请求时通过 {i} 占位符区分文件名
                            if repeat > 1 and "{i}" in download_target:
                                filename = download_target.format(i=run_index + 1)
                            else:
                                filename = download_target
                            try:
                                if os.path.dirname(filename):
                                    os.makedirs(os.path.dirname(filename), exist_ok=True)
                                with open(filename, "wb") as fh:
                                    fh.write(body_buffer)
                                print(f"[saved response body to file: {filename}]")
                            except OSError as exc:
                                print(f"[failed to save response body to {filename}: {exc}]", file=sys.stderr)
                        if args.include_headers:
                            status_line = f"{http_version} {status_code} {response.reason_phrase}"
                            print(status_line)
                            for key, value in response.headers.items():
                                print(f"{key}: {value}")
                            print("")
                        output_body, body_output_meta = format_response_body_output(
                            decoded_body,
                            response_filter,
                            args.response_filter_mode,
                            args.response_filter_invert,
                            args.response_filter_ignore_case,
                            response_max_lines,
                            response_max_bytes,
                            response_preview_lines,
                            response_context_lines,
                            compiled_filter=compiled_response_filter,
                        )
                        has_filter_or_cap = bool(
                            response_filter or response_max_lines > 0 or response_max_bytes > 0
                        )
                        if has_filter_or_cap and body_output_meta.get("mode") != "empty":
                            print_response_body_summary(body_output_meta)
                            if body_output_meta.get("preview") and not body_output_meta.get("matched_lines"):
                                print("[body] no regex match; showing preview:")
                        if output_body:
                            print(output_body)
                            if body_output_meta.get("truncated") or body_output_meta.get("byte_truncated"):
                                omitted = (body_output_meta.get("total_lines") or 0) - (
                                    body_output_meta.get("display_lines") or 0
                                )
                                if omitted > 0:
                                    print(f"[body] ... {omitted} more line(s) omitted (use --download for full body)")
                        elif body_output_meta.get("mode") == "empty":
                            print("[no body]")
                        elif response_filter and not body_output_meta.get("preview"):
                            print("[body] no lines matched filter")
                        else:
                            print("[no body]")

                        print(f"\n----- Meta #{run_index + 1} -----")
                        if args.show_summary:
                            for key in METRIC_KEYS:
                                label = key.replace("_", " ").title()
                                print(f"{label}: {format_metric_value(key, metrics.get(key))}")
                                value = metrics.get(key)
                                if aggregate is not None and isinstance(value, (int, float)):
                                    aggregate[key].append(float(value))
                            print(f"Wall Time (client): {total_elapsed:.6f}s")
                            if aggregate is not None:
                                aggregate["wall_time"].append(total_elapsed)
                        else:
                            print("Summary disabled (--show-summary=false).")
                            print(f"Wall Time (client): {total_elapsed:.6f}s")

                        print(f"Encoding Used: {used_encoding} ({encoding_source})")

                        if args.verbose_output:
                            sent_bytes = len(body_bytes) if body_bytes else 0
                            print("\nVerbose diagnostics:")
                            print(f"  Sent body bytes: {sent_bytes}")
                            if probe_metrics:
                                for key, value in probe_metrics.items():
                                    print(f"  Probe {key}: {value:.6f}s")
                            print(f"  History length: {len(response.history)}")

                except httpx.HTTPError as exc:
                    exit_code = 1
                    elapsed = time.perf_counter() - start
                    print(f"\n===== Response #{run_index + 1} =====")
                    print(f"Request failed after {elapsed:.6f}s: {exc}")
                    continue

            if aggregate and repeat > 1:
                print("\n===== Aggregate Timing =====")
                for key, values in aggregate.items():
                    summary = summarize(values)
                    if not summary:
                        continue
                    label = key.replace("_", " ").title()
                    min_v, avg_v, max_v = summary
                    if key == "speed_download":
                        print(f"{label}: min {min_v:.2f} B/s | avg {avg_v:.2f} B/s | max {max_v:.2f} B/s")
                    elif key == "size_download":
                        print(f"{label}: min {min_v:.0f}B | avg {avg_v:.0f}B | max {max_v:.0f}B")
                    elif key in {"http_code", "redirects"}:
                        print(f"{label}: min {min_v:.0f} | avg {avg_v:.2f} | max {max_v:.0f}")
                    else:
                        print(f"{label}: min {min_v:.6f}s | avg {avg_v:.6f}s | max {max_v:.6f}s")

        finally:
            client.close()

        sys.exit(exit_code)


    if __name__ == "__main__":
        try:
            main()
        except KeyboardInterrupt:
            print("\nInterrupted.", file=sys.stderr)
            sys.exit(130)
enabled: true
short_description: "纯Python HTTP测试框架（httpx，会话复用+编码守护）"
description: |
  纯 Python 构建的 HTTP 测试框架：基于 httpx 会话，结合 socket 探针补齐 DNS/TCP/TLS 观测，
  提供全流程的请求体编码守护、重复播放与可观测能力，适配盲注延时、复杂 payload 调优等场景。

  **亮点：**
  - 纯 Python 实现：httpx 会话重用、HTTP/2/代理/certs 直接在脚本内配置，无外部二进制依赖
  - 智能 Body 编码：支持 application/x-www-form-urlencoded 规范化编码、JSON/文本 charset 推断、
    `@file`/`@-` 注入二进制、可视化调试
  - 连接探针：在无代理场景下额外进行 DNS/TCP/TLS 探测，粗粒度复刻 curl -w 指标
  - 可重复观测：repeat/delay + TTFB/total/speed_download 统计，便于盲注/时序测试
  - 扩展开关：additional_args 解析 http2/cert/verify/trust_env/max_redirects 等 httpx 选项
  - 响应体瘦身：response_filter 按行/块正则提取，配合 max_lines/max_bytes 限制 stdout，降低 Agent token 消耗

  **响应过滤最佳实践：**
  - 大页面/HTML：用 `response_filter` 抓 error|exception|password|token|uid 等关键字行
  - 无 filter 时：设 `response_max_lines=80` 或 `response_max_bytes=8192` 防止整页灌入上下文
  - 0 命中：自动预览前 `response_preview_lines` 行，避免误判「空响应」
  - 完整留存：大 body 用 `download` 落盘，stdout 只保留摘要行
parameters:
  - name: "url"
    type: "string"
    description: "目标URL（可自动编码路径/参数，避免特殊字符导致的请求失败如：https://www.target.com/s?wd=test）"
    required: true
    flag: "--url"
  - name: "method"
    type: "string"
    description: "HTTP方法（GET, POST, PUT, DELETE等）"
    required: false
    default: "GET"
    flag: "--method"
  - name: "data"
    type: "string"
    description: "请求数据/参数（支持JSON、表单、原始payload，前缀@file或@-可加载文件/STDIN）"
    required: false
    flag: "--data"
  - name: "headers"
    type: "object"
    description: "自定义请求头（字典格式，如 {\"X-Custom\": \"value\"}）"
    required: false
    flag: "--headers"
  - name: "cookies"
    type: "string"
    description: "自定义Cookie（格式：name1=value1; name2=value2）"
    required: false
    flag: "--cookies"
  - name: "user_agent"
    type: "string"
    description: "自定义User-Agent"
    required: false
    flag: "--user-agent"
  - name: "proxy"
    type: "string"
    description: "代理（http(s)://或socks5://地址，直接传递给httpx Client）"
    required: false
    flag: "--proxy"
  - name: "timeout"
    type: "string"
    description: "最大超时时间（秒，支持小数）"
    required: false
    flag: "--timeout"
  - name: "repeat"
    type: "int"
    description: "重复请求次数，用于盲注/稳定性观测（>=1）"
    required: false
    default: 1
    flag: "--repeat"
  - name: "delay"
    type: "string"
    description: "重复请求之间的延迟（秒，可为小数）"
    required: false
    default: "0"
    flag: "--delay"
  - name: "include_headers"
    type: "bool"
    description: "输出响应头（默认开启，可用 --no-include-headers 关闭）"
    required: false
    default: true
    flag: "--include-headers"
  - name: "auto_encode_url"
    type: "bool"
    description: "自动URL编码（默认开启，可通过 --no-auto-encode-url 关闭）"
    required: false
    default: true
    flag: "--auto-encode-url"
  - name: "follow_redirects"
    type: "bool"
    description: "跟随重定向（httpx follow_redirects）"
    required: false
    default: false
    flag: "--follow-redirects"
  - name: "allow_insecure"
    type: "bool"
    description: "忽略TLS证书错误（verify=False）"
    required: false
    default: false
    flag: "--allow-insecure"
  - name: "verbose_output"
    type: "bool"
    description: "输出额外调试信息（请求体字节数、探针延时、历史长度等）"
    required: false
    default: false
    flag: "--verbose-output"
  - name: "show_command"
    type: "bool"
    description: "打印请求概览（方法、URL、所有头、Body概况）"
    required: false
    default: true
    flag: "--show-command"
  - name: "show_summary"
    type: "bool"
    description: "打印指标摘要（默认开启，含DNS/TCP/TLS/TTFB/Total）"
    required: false
    default: true
    flag: "--show-summary"
  - name: "debug"
    type: "bool"
    description: "调试模式：显示Body编码处理细节（模式/charset/长度）"
    required: false
    default: false
    flag: "--debug"
  - name: "response_encoding"
    type: "string"
    description: "强制响应解码使用的编码（如GBK），覆盖自动探测"
    required: false
    flag: "--response-encoding"
  - name: "response_filter"
    type: "string"
    description: |
      响应体正则过滤（仅影响 stdout，不影响 --download 与指标）。
      默认 line 模式按行匹配；示例：'(error|exception|SQL|password|token|uid)'。
      与 response_max_lines/response_max_bytes 配合可显著减少 token 消耗。
    required: false
    flag: "--response-filter"
  - name: "response_filter_mode"
    type: "string"
    description: "过滤模式：line（按行，默认）、multiline（跨行块）、full（整段 DOTALL 匹配）"
    required: false
    default: "line"
    flag: "--response-filter-mode"
  - name: "response_filter_invert"
    type: "bool"
    description: "反向过滤：输出不匹配 regex 的行（用于剔除 HTML 噪音）"
    required: false
    default: false
    flag: "--response-filter-invert"
  - name: "response_filter_ignore_case"
    type: "bool"
    description: "正则忽略大小写"
    required: false
    default: false
    flag: "--response-filter-ignore-case"
  - name: "response_max_lines"
    type: "int"
    description: "stdout 最多输出行数（0=不限制）；有 filter 时限制命中行数，无 filter 时截断全文"
    required: false
    default: 0
    flag: "--response-max-lines"
  - name: "response_max_bytes"
    type: "int"
    description: "stdout 响应体 UTF-8 字节上限（0=不限制），超出部分截断"
    required: false
    default: 0
    flag: "--response-max-bytes"
  - name: "response_preview_lines"
    type: "int"
    description: "filter 零命中时预览的前 N 行（默认 5，0=不预览）"
    required: false
    default: 5
    flag: "--response-preview-lines"
  - name: "response_context_lines"
    type: "int"
    description: "line 模式下命中行上下各保留 N 行上下文（类似 grep -C）"
    required: false
    default: 0
    flag: "--response-context-lines"
  - name: "action"
    type: "string"
    description: "保留字段：标识调用意图（request, spider等），脚本内部不使用"
    required: false
    default: "request"
    flag: "--action"
  - name: "download"
    type: "string"
    description: |
      将响应体保存到本地文件，写入原始字节数据：
      - "output.bin"：单次请求直接保存到指定文件
      - "outputs/run-{i}.bin"：repeat>1 时按序号区分文件（{i} 将被替换为 1,2,...）
    required: false
    flag: "--download"
  - name: "additional_args"
    type: "string"
    description: |
      额外的 httpx 选项，需按 httpx.Client 的关键字参数名与类型填写，支持 "key=value" 或单词形式（等价于 key=true）：
      - "http2=true"
      - "cert=/path/to/client.pem"
      - "verify=/path/to/ca.pem" 或 "verify=false"
      - "trust_env=false"、"max_redirects=5" 等
    required: false
    flag: "--additional-args"
