xiaoxia-saas/packages/application/viral_video/xml_parser.py

"""XML 标签式输出解析器（替代 json.loads）。

LLM 按 ``<tag attr="x">内容</tag>`` 输出，本模块解析，解析失败不抛异常，
由调用方走规则 fallback。采用栈式扫描，嵌套标签全部可提取（内外层都保留）。
"""

from __future__ import annotations

import re
from html import unescape
from typing import Optional

_OPEN_RE = re.compile(r"<(?P<tag>[\w-]+)(?P<attrs>(?:\s(?:[^>]*?\S)?)?)(?P<self>/?)>")
_CLOSE_RE = re.compile(r"</(?P<tag>[\w-]+)\s*>")
_ATTR_RE = re.compile(r"""([\w:-]+)\s*=\s*(?:"([^"]*)"|'([^']*)')""")


def parse_attributes(raw: str) -> dict[str, str]:
    """解析标签属性字符串。"""
    attrs: dict[str, str] = {}
    for match in _ATTR_RE.finditer(raw or ""):
        value = match.group(2) if match.group(2) is not None else match.group(3)
        attrs[match.group(1)] = value
    return attrs


def parse_tags(text: Optional[str]) -> list[dict]:
    """提取全部标签（含嵌套内外层），返回 [{tag, attrs, text}]，按开标签出现顺序。"""
    if not text:
        return []
    results: list[dict] = []
    stack: list[dict] = []
    token_re = re.compile(r"<[^>]+>")
    for token in token_re.finditer(text):
        raw_token = token.group(0)
        # 先按开/闭标签匹配
        open_match = _OPEN_RE.match(raw_token)
        close_match = _CLOSE_RE.match(raw_token)
        is_close_tag = raw_token.startswith("</")
        if not is_close_tag and open_match:
            is_self_close = open_match.group("self") == "/"
            node = {
                "tag": open_match.group("tag"),
                "attrs": parse_attributes(open_match.group("attrs")),
                "text": "",
                "_start": token.end(),
            }
            if is_self_close:
                node.pop("_start")
                results.append(node)
            else:
                stack.append(node)
                results.append(node)
        elif is_close_tag and close_match:
            tag = close_match.group("tag")
            # 弹出到最近同名开标签
            for idx in range(len(stack) - 1, -1, -1):
                if stack[idx]["tag"] == tag:
                    node = stack[idx]
                    node["text"] = unescape(text[node["_start"] : token.start()].strip())
                    node.pop("_start", None)
                    del stack[idx:]
                    break
    # 未闭合标签：给剩余部分作为文本
    for node in stack:
        if "_start" in node:
            node["text"] = unescape(text[node["_start"] :].strip())
            node.pop("_start", None)
    return results


def find_all(text: Optional[str], tag: str) -> list[dict]:
    """提取指定标签的全部节点。"""
    return [n for n in parse_tags(text) if n["tag"] == tag]


def find_first(text: Optional[str], tag: str) -> Optional[dict]:
    nodes = find_all(text, tag)
    return nodes[0] if nodes else None


def text_of(text: Optional[str], tag: str, default: str = "") -> str:
    node = find_first(text, tag)
    return node["text"] if node else default


def attr_bool(value: Optional[str], default: bool = False) -> bool:
    if value is None:
        return default
    return value.strip().lower() in {"true", "1", "yes", "是"}


def attr_float(value: Optional[str], default: float = 0.0) -> float:
    try:
        return float(value) if value is not None and value.strip() else default
    except (TypeError, ValueError):
        return default


def attr_int(value: Optional[str], default: int = 0) -> int:
    try:
        return int(float(value)) if value is not None and value.strip() else default
    except (TypeError, ValueError):
        return default