ai_site/ai-service/url_capture.py

"""解析首页抓包 url.config：key/value 成对，对应菜单/图表/统计等接口回包。"""

from __future__ import annotations

import json
import re
from html.parser import HTMLParser
from typing import Any


class _MenuHTMLParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.items: list[str] = []
        self._in_a = False
        self._buf = ""

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag.lower() == "a":
            self._in_a = True
            self._buf = ""

    def handle_endtag(self, tag: str) -> None:
        if tag.lower() == "a" and self._in_a:
            text = re.sub(r"\s+", " ", self._buf).strip()
            if text:
                self.items.append(text)
            self._in_a = False
            self._buf = ""

    def handle_data(self, data: str) -> None:
        if self._in_a:
            self._buf += data


def parse_url_config(content: bytes | str) -> dict[str, Any]:
    """
    格式示例：
      key  https://.../left!newleftmenu.action?...
      value  <ul>...</ul>

      key https://.../hightchartCjlAndJianDu.action?...
      value  { ...json... }
    """
    if isinstance(content, bytes):
        text = content.decode("utf-8-sig", errors="ignore")
        if not text.strip():
            for enc in ("gb18030", "utf-16", "utf-16-le"):
                try:
                    text = content.decode(enc)
                    if text.strip():
                        break
                except Exception:  # noqa: BLE001
                    continue
    else:
        text = content

    text = text.replace("\r\n", "\n").replace("\r", "\n")
    # 按「行首 key 」切开
    parts = re.split(r"(?m)^key\s+", text)
    pairs: list[dict[str, Any]] = []
    for part in parts:
        part = part.strip()
        if not part:
            continue
        # 第一行 URL，其后 value 段
        if "\n" in part:
            first, rest = part.split("\n", 1)
        else:
            first, rest = part, ""
        url = first.strip()
        m = re.match(r"(?is)^value\s*(.*)\Z", rest.strip(), flags=re.S)
        body = (m.group(1) if m else rest).strip()
        # 去掉仅有的空白
        if not body:
            body = ""
        action = ""
        am = re.search(r"!([a-zA-Z0-9_]+)\.action", url)
        if am:
            action = am.group(1)
        else:
            am = re.search(r"/([a-zA-Z0-9_!]+)\.action", url)
            if am:
                action = am.group(1).split("!")[-1]
        kind = "empty"
        parsed: Any = None
        if not body:
            kind = "empty"
        elif body.lstrip().startswith("{") or body.lstrip().startswith("["):
            kind = "json"
            try:
                parsed = json.loads(body)
            except json.JSONDecodeError:
                kind = "json_invalid"
                parsed = None
        elif "<" in body and ">" in body:
            kind = "html"
            parsed = body
        else:
            kind = "text"
            parsed = body
        pairs.append(
            {
                "url": url,
                "action": action,
                "kind": kind,
                "body": body,
                "parsed": parsed,
                "empty": kind == "empty",
            }
        )

    nav_items: list[str] = []
    chart_json: dict[str, Any] | None = None
    stats_json: dict[str, Any] | None = None
    exceed_json: Any = None
    warnings: list[str] = []

    for p in pairs:
        act = (p.get("action") or "").lower()
        url = p.get("url") or ""
        if p["kind"] == "html" and ("leftmenu" in act or "left" in url or "menu" in act):
            parser = _MenuHTMLParser()
            try:
                parser.feed(str(p["parsed"] or ""))
                parser.close()
            except Exception:  # noqa: BLE001
                pass
            if parser.items:
                nav_items = parser.items
        elif p["kind"] == "json" and isinstance(p["parsed"], dict):
            data = p["parsed"]
            if "mcljChartData" in data or "cljdChartData" in data:
                chart_json = data
            elif any(k in data for k in ("yqwccds", "xzcxcds", "nodisposecount", "bdlength")):
                stats_json = data
            elif "gonghou" in act.lower() or "cjlcx" in act.lower() or "cx" in act:
                exceed_json = data
            elif chart_json is None and any(
                isinstance(v, (list, dict)) for v in data.values()
            ):
                # 兜底：第一个复杂 JSON 当图表
                chart_json = data
        elif p["empty"] and ("gonghou" in act.lower() or "cjlcx" in act.lower() or "CX" in url):
            warnings.append(
                f"接口 {p.get('action') or url} 返回为空："
                "超限测点表无抓包数据，将由主图表 JSON 按超限条件推导"
            )
        elif p["empty"]:
            warnings.append(f"接口 {p.get('action') or url} 返回为空，已跳过")

    return {
        "ok": True,
        "pairs": [
            {
                "url": p["url"],
                "action": p["action"],
                "kind": p["kind"],
                "empty": p["empty"],
                "body_len": len(p.get("body") or ""),
            }
            for p in pairs
        ],
        "nav_items": nav_items,
        "chart_json": chart_json,
        "stats_json": stats_json,
        "exceed_json": exceed_json,
        "warnings": warnings,
        "source": "url_config",
    }