Legal Ocr

# ===== 统一设置 =====
LEGAL_OCR_BACKEND=auto
LEGAL_OCR_POST_PROCESS=true
# auto=检测到法律材料才启用；true/always=强制启用；false/never=关闭
LEGAL_OCR_LEGAL_TERMS=auto
LEGAL_OCR_LINE_MERGE=true
LEGAL_OCR_CUSTOM_TERMS_PATH=
LEGAL_OCR_LOG_LEVEL=medium
# 瞬态网络错误自动重试：尝试次数（含首次）/ 首次重试前秒数 / 单次重试等待上限秒数
LEGAL_OCR_RETRY_ATTEMPTS=3
LEGAL_OCR_RETRY_BASE_DELAY=1.0
LEGAL_OCR_RETRY_MAX_DELAY=30.0

# ===== PaddleOCR 后端 =====
# 从 https://www.paddleocr.com 对应模型 API 页面复制完整 layout-parsing 端点和 Access Token。
PADDLEOCR_DOC_PARSING_API_URL=https://your-endpoint.example.com/layout-parsing
PADDLEOCR_ACCESS_TOKEN=your_access_token_here
PADDLEOCR_API_PROTOCOL=auto
PADDLEOCR_MODEL=PaddleOCR-VL-1.5
PADDLEOCR_DOC_ORIENTATION=false
PADDLEOCR_DOC_UNWARP=false
PADDLEOCR_TEXTLINE_ORIENTATION=false
PADDLEOCR_CHART_RECOG=false
PADDLEOCR_VISUALIZE=false
PADDLEOCR_VL_LAYOUT_DETECTION=true
PADDLEOCR_VL_LAYOUT_SHAPE_MODE=rect
PADDLEOCR_API_EXTRA_JSON=
PADDLEOCR_DOC_PARSING_TIMEOUT=600
PADDLEOCR_POLL_INTERVAL=5
PADDLEOCR_POLL_TIMEOUT=1800
PADDLEOCR_BATCH_PAGES=40
PADDLEOCR_MAX_BASE64_MB=20
PADDLEOCR_LOG_LEVEL=medium
# 单后端覆盖；不设则取 LEGAL_OCR_RETRY_*
# PADDLEOCR_RETRY_ATTEMPTS=3
# PADDLEOCR_RETRY_BASE_DELAY=1.0
# PADDLEOCR_RETRY_MAX_DELAY=30.0

# PaddleOCR API 兼容别名（如来自 pdf-processor，可二选一使用）
# PADDLE_OCR_API_ENDPOINT=https://your-aistudio-app.example.com/api/v2/ocr/jobs
# PADDLE_OCR_API_KEY=your_paddle_token_here
# API_URL=https://your-aistudio-app.example.com/api/v2/ocr/jobs
# TOKEN=your_paddle_token_here

# ===== MinerU 后端 =====
# 不填 Token 时，本地/远程小文件默认走免登录轻量接口。
MINERU_API_BASE=https://mineru.net/api/v4
MINERU_API_TOKEN=
MINERU_ENABLE_OCR=true
MINERU_ENABLE_TABLE=true
MINERU_ENABLE_FORMULA=false
MINERU_LANGUAGE_CODE=ch
MINERU_MODEL_VERSION=pipeline
MINERU_PAGE_RANGES=
MINERU_POLL_MAX=20
MINERU_POLL_SLEEP=10
MINERU_LOG_LEVEL=medium
# 单后端覆盖；不设则取 LEGAL_OCR_RETRY_*
# MINERU_RETRY_ATTEMPTS=3
# MINERU_RETRY_BASE_DELAY=1.0
# MINERU_RETRY_MAX_DELAY=30.0

{
  "replacements": [
    {
      "source": "某OCR误词",
      "replacement": "正确法律词",
      "description": "示例：客户名称、专有术语或高频误识别"
    },
    {
      "pattern": "管[辖輖]权",
      "replacement": "管辖权",
      "description": "示例：使用正则修正常见 OCR 误识别"
    }
  ]
}

MIT License

Copyright (c) 2025 杨卫薪律师（微信ywxlaw）

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Protocol

from common import SourceInfo


@dataclass
class ConvertOptions:
    pages: str | None = None
    model: str | None = None
    paddle_model: str | None = None
    log_level: str = "medium"


@dataclass
class BackendResult:
    backend: str
    mode: str
    provider: str
    markdown: str
    images: list[dict[str, str]] = field(default_factory=list)
    batches: list[dict[str, Any]] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)
    backend_result_dir: Path | None = None


class OCRBackend(Protocol):
    name: str

    def convert(
        self,
        source: SourceInfo,
        options: ConvertOptions,
        work_dir: Path,
        assets_dir: Path,
    ) -> BackendResult:
        ...

from __future__ import annotations

import re
from dataclasses import dataclass, field


@dataclass
class PostprocessResult:
    text: str
    log: list[dict[str, str]] = field(default_factory=list)


CHAPTER_RE = re.compile(r"^(第[一二三四五六七八九十百千万\d]+章\s*.+)$")
SECTION_RE = re.compile(r"^(第[一二三四五六七八九十百千万\d]+节\s*.+)$")
ARTICLE_RE = re.compile(r"^(第[一二三四五六七八九十百千万\d]+条\s*.+)$")
COURT_HEADING_RE = re.compile(
    r"^(原告诉称|被告辩称|第三人述称|上诉人上诉称|被上诉人辩称|申请人称|被申请人辩称|"
    r"诉讼请求|事实和理由|审理经过|本院查明|经审理查明|本院认为|裁判结果|判决如下|"
    r"裁定如下|审理查明|法院认为)[:：]?$"
)


def _normalize_heading(line: str) -> tuple[str, str | None]:
    stripped = line.strip()
    if not stripped or stripped.startswith("#"):
        return stripped, None
    if CHAPTER_RE.match(stripped):
        return f"## {stripped}", "chapter"
    if SECTION_RE.match(stripped):
        return f"### {stripped}", "section"
    if COURT_HEADING_RE.match(stripped):
        return f"## {stripped}", "court_heading"
    if ARTICLE_RE.match(stripped) and not stripped.startswith("**"):
        return f"**{stripped}**", "article"
    return stripped, None


def run_basic_postprocess(markdown: str) -> PostprocessResult:
    text = markdown.replace("\r\n", "\n").replace("\r", "\n")
    log: list[dict[str, str]] = []
    normalized_lines: list[str] = []

    for line_no, raw_line in enumerate(text.split("\n"), start=1):
        line = raw_line.rstrip()
        normalized, action = _normalize_heading(line)
        if action:
            log.append({"line": str(line_no), "action": action, "text": normalized})
        normalized_lines.append(normalized)

    text = "\n".join(normalized_lines).strip()
    text = re.sub(r"\n{3,}", "\n\n", text)
    return PostprocessResult(text=text, log=log)

from __future__ import annotations

import hashlib
import math
import os
import re
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, TypeVar
from urllib.parse import unquote, urlparse

try:
    import httpx
except ImportError:  # pragma: no cover - httpx is required for OCR but optional for type-checkers
    httpx = None  # type: ignore[assignment]


SUPPORTED_IMAGE_SUFFIXES = (
    ".png",
    ".jpg",
    ".jpeg",
    ".bmp",
    ".tiff",
    ".tif",
    ".webp",
    ".gif",
    ".jp2",
)
PADDLE_LOCAL_SUFFIXES = (".pdf",) + SUPPORTED_IMAGE_SUFFIXES
MINERU_LOCAL_SUFFIXES = (
    ".pdf",
    ".doc",
    ".docx",
    ".ppt",
    ".pptx",
    ".png",
    ".jpg",
    ".jpeg",
    ".jp2",
    ".webp",
    ".gif",
    ".bmp",
    ".xls",
    ".xlsx",
)
DOCUMENT_URL_SUFFIXES = MINERU_LOCAL_SUFFIXES


@dataclass(frozen=True)
class SourceInfo:
    raw: str
    is_url: bool
    source_type: str
    suffix: str
    file_name: str
    base_name: str
    path: Path | None = None
    size_bytes: int | None = None
    page_count: int | None = None


def get_skill_root() -> Path:
    return Path(__file__).resolve().parent.parent


def get_config_path() -> Path:
    return get_skill_root() / "config" / ".env"


def read_env_file(path: Path) -> dict[str, str]:
    if not path.exists():
        return {}

    env: dict[str, str] = {}
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        env[key.strip()] = value.strip().strip('"').strip("'")
    return env


def load_env() -> dict[str, str]:
    file_env = read_env_file(get_config_path())
    return {**file_env, **os.environ}


def first_non_empty(env: dict[str, str], *keys: str) -> str:
    for key in keys:
        value = env.get(key, "").strip()
        if value:
            return value
    return ""


def parse_bool(value: str | bool | None, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    normalized = str(value).strip().lower()
    if not normalized:
        return default
    return normalized in {"1", "true", "yes", "on"}


def parse_positive_int(value: str | None, default: int) -> int:
    if not value or not str(value).strip():
        return default
    try:
        parsed = int(str(value).strip())
    except ValueError as error:
        raise ValueError(f"整数配置无效：{value}") from error
    if parsed <= 0:
        raise ValueError(f"整数配置必须大于 0：{value}")
    return parsed


def parse_positive_float(value: str | None, default: float) -> float:
    if not value or not str(value).strip():
        return default
    try:
        parsed = float(str(value).strip())
    except ValueError as error:
        raise ValueError(f"数字配置无效：{value}") from error
    if not math.isfinite(parsed) or parsed <= 0:
        raise ValueError(f"数字配置必须大于 0：{value}")
    return parsed


def sanitize_name(value: str) -> str:
    sanitized = re.sub(r"[^\w\u4e00-\u9fff.-]+", "_", value, flags=re.UNICODE)
    sanitized = sanitized.strip("._")
    return sanitized or "document"


def is_http_url(value: str) -> bool:
    return value.lower().startswith(("http://", "https://"))


def suffix_from_url(url: str) -> str:
    path = unquote(urlparse(url).path)
    return Path(path).suffix.lower()


def derive_name_from_url(url: str) -> str:
    parsed_path = unquote(urlparse(url).path).rstrip("/")
    filename = Path(parsed_path).name if parsed_path else ""
    if not filename:
        host = urlparse(url).hostname or "remote-document"
        filename = sanitize_name(host)
    return sanitize_name(filename)


def is_html_like_url(url: str) -> bool:
    suffix = suffix_from_url(url)
    return not suffix or suffix not in DOCUMENT_URL_SUFFIXES


def estimate_base64_mb(path: Path) -> float:
    return (path.stat().st_size * 4 / 3) / 1024 / 1024


def sha256_of_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def resolve_output_markdown_path(source: SourceInfo, output_arg: str | None) -> Path:
    if output_arg:
        output_path = Path(output_arg).expanduser()
        if output_path.suffix.lower() == ".md":
            return output_path.resolve()
        if output_arg.endswith("/") or output_path.is_dir() or not output_path.suffix:
            return (output_path / f"{source.base_name}.md").resolve()
        return output_path.resolve()

    if source.path:
        return source.path.with_suffix(".md").resolve()
    return (Path.cwd() / f"{source.base_name}.md").resolve()


def resolve_images_dir(markdown_path: Path) -> Path:
    return markdown_path.with_name(f"{markdown_path.stem}_images")


def build_source_info(raw_source: str, page_count: int | None = None) -> SourceInfo:
    raw = raw_source.strip()
    if is_http_url(raw):
        file_name = derive_name_from_url(raw)
        suffix = suffix_from_url(raw)
        base_name = Path(file_name).stem or "remote-document"
        source_type = "remote_html_url" if is_html_like_url(raw) else "remote_doc_url"
        return SourceInfo(
            raw=raw,
            is_url=True,
            source_type=source_type,
            suffix=suffix,
            file_name=file_name,
            base_name=sanitize_name(base_name),
            page_count=page_count,
        )

    path = Path(raw).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"文件不存在：{path}")
    if not path.is_file():
        raise ValueError(f"不是普通文件：{path}")
    if path.stat().st_size == 0:
        raise ValueError(f"文件为空：{path}")

    return SourceInfo(
        raw=str(path),
        is_url=False,
        source_type="local_file",
        suffix=path.suffix.lower(),
        file_name=path.name,
        base_name=sanitize_name(path.stem),
        path=path,
        size_bytes=path.stat().st_size,
        page_count=page_count,
    )


def read_official_mineru_token() -> str:
    yaml_path = Path.home() / ".mineru" / "config.yaml"
    if not yaml_path.exists():
        return ""
    try:
        content = yaml_path.read_text(encoding="utf-8")
    except OSError:
        return ""
    patterns = [
        r"^\s*token\s*:\s*[\"']?([^\"'#\r\n]+)[\"']?\s*$",
        r"^\s*api_token\s*:\s*[\"']?([^\"'#\r\n]+)[\"']?\s*$",
        r"^\s*mineru_token\s*:\s*[\"']?([^\"'#\r\n]+)[\"']?\s*$",
    ]
    for pattern in patterns:
        match = re.search(pattern, content, flags=re.MULTILINE)
        if match:
            return sanitize_config_value(match.group(1))
    return ""


def sanitize_config_value(value: str | None) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    lowered = text.lower()
    placeholders = {
        "your_token_here",
        "your_mineru_api_token_here",
        "your_access_token_here",
        "your_paddle_token_here",
        "your-endpoint.example.com",
    }
    if lowered in placeholders:
        return ""
    return text


def resolve_mineru_token(env: dict[str, str]) -> str:
    return (
        sanitize_config_value(first_non_empty(env, "MINERU_API_TOKEN"))
        or sanitize_config_value(os.environ.get("MINERU_API_TOKEN"))
        or sanitize_config_value(os.environ.get("MINERU_TOKEN"))
        or read_official_mineru_token()
    )


def is_transient_httpx_error(exc: BaseException) -> bool:
    """判断一个异常是否属于可安全重试的瞬态网络错误。

    覆盖：
    - DNS 解析失败（[Errno 8] nodename nor servname provided）
    - TCP 连接失败、连接超时、读取超时
    - 远端在写入响应中途关闭连接
    - 协议层错误
    """
    if httpx is None:
        return False
    return isinstance(exc, httpx.RequestError)


def is_transient_http_status(status_code: int) -> bool:
    """判断 HTTP 状态码是否属于可重试的瞬态错误。"""
    if status_code == 429:
        return True
    if 500 <= status_code < 600:
        return True
    return False


class PaddleOCRRateLimited(Exception):
    """PaddleOCR 服务端瞬态限流(HTTP 400 + code:10010 任务提交队列已满)。

    与普通 RuntimeError 的区别:这是可安全退避重试的瞬态错误,不应立即把段标 failed。
    retry_with_backoff 用 is_paddle_rate_limited 接住它。
    """

    # PaddleOCR 业务码:任务提交队列已满(瞬态,可重试)
    RATE_LIMITED_CODES = {"10010"}

    def __init__(self, message: str, *, trace_id: str | None = None) -> None:
        super().__init__(message)
        self.trace_id = trace_id


def is_paddle_rate_limited_response(status_code: int, body: str) -> bool:
    """判断一个 PaddleOCR 响应是否属于 code:10010 服务端限流。

    响应体形如:{"traceId":"...","code":10010,"msg":"任务提交队列已满,请稍后重试"}
    """
    if status_code not in {400, 429, 503}:
        return False
    for code in PaddleOCRRateLimited.RATE_LIMITED_CODES:
        if f'"code":{code}' in body or f'"code": {code}' in body:
            return True
    return False


def is_paddle_rate_limited(exc: BaseException) -> bool:
    """retry_with_backoff 的 is_transient 钩子:识别 PaddleOCRRateLimited。"""
    return isinstance(exc, PaddleOCRRateLimited)


T = TypeVar("T")


def retry_with_backoff(
    func: Callable[..., T],
    *args: Any,
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    is_transient: Callable[[BaseException], bool] | None = None,
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    **kwargs: Any,
) -> T:
    """以指数退避重试 ``func(*args, **kwargs)``，仅在 ``is_transient`` 判为瞬态时重试。

    参数:
    - ``max_attempts``: 最多尝试次数（含首次），默认 3。
    - ``base_delay``: 首次重试前的等待秒数，默认 1.0。
    - ``max_delay``: 单次重试等待上限，默认 30.0。
    - ``is_transient``: 自定义瞬态判定；默认 ``is_transient_httpx_error``。
    - ``on_retry``: 重试前回调，签名 ``(attempt, exc, delay) -> None``，用于记录日志。

    非瞬态异常或达到最大次数后透传给调用方；首参后的 ``max_attempts`` 计数从 1 开始。
    """
    if max_attempts < 1:
        raise ValueError("max_attempts 必须 >= 1")
    transient = is_transient or is_transient_httpx_error
    last_exc: BaseException | None = None
    for attempt in range(1, max_attempts + 1):
        try:
            return func(*args, **kwargs)
        except BaseException as exc:  # noqa: BLE001
            if not transient(exc) or attempt >= max_attempts:
                raise
            last_exc = exc
            delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
            if on_retry:
                on_retry(attempt, exc, delay)
            time.sleep(delay)
    if last_exc is not None:
        raise last_exc
    raise RuntimeError("retry_with_backoff: 不可达分支")


def has_paddle_config(env: dict[str, str]) -> bool:
    return bool(
        sanitize_config_value(
            first_non_empty(
                env,
                "PADDLEOCR_DOC_PARSING_API_URL",
                "PADDLE_OCR_API_ENDPOINT",
                "API_URL",
            )
        )
        and sanitize_config_value(
            first_non_empty(
                env,
                "PADDLEOCR_ACCESS_TOKEN",
                "PADDLE_OCR_API_KEY",
                "TOKEN",
            )
        )
    )

ObjC.import("Foundation");

function shellQuote(value) {
  return "'" + String(value).replace(/'/g, "'\\''") + "'";
}

function getScriptPath() {
  const args = $.NSProcessInfo.processInfo.arguments;
  if (args.count < 4) {
    throw new Error("无法定位当前脚本路径");
  }
  return ObjC.unwrap(args.objectAtIndex(3));
}

function run(argv) {
  const app = Application.currentApplication();
  app.includeStandardAdditions = true;
  const sh = (cmd) => app.doShellScript(cmd);

  try {
    const scriptPath = getScriptPath();
    const scriptDir = sh(`/usr/bin/dirname ${shellQuote(scriptPath)}`).trim();
    const pythonScript = `${scriptDir}/convert.py`;
    const skillRoot = sh(`/usr/bin/dirname ${shellQuote(scriptDir)}`).trim();
    const uvPath = sh(`/bin/zsh -lc 'command -v uv || true'`).trim();
    const runner = uvPath ? `${shellQuote(uvPath)} run` : "/usr/bin/python3";
    const argList = (argv || []).map(shellQuote).join(" ");
    const command =
      `cd ${shellQuote(skillRoot)} && ${runner} ${shellQuote(pythonScript)}` +
      (argList ? ` ${argList}` : "");

    const output = sh(command);
    console.log(output);
    return output;
  } catch (error) {
    const message =
      "legal-ocr 转换失败\n" +
      "===============================================\n" +
      `${error.message}\n\n` +
      "建议：\n" +
      "1. 先运行 `uv run scripts/smoke_test.py --skip-api-test`\n" +
      "2. 检查 `config/.env` 是否已配置\n" +
      "3. 直接运行 `uv run scripts/convert.py \"文件路径或URL\"` 查看详细日志";
    console.log(message);
    return message;
  }
}

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#   "httpx>=0.27.0",
#   "pypdfium2>=4.30.0",
# ]
# ///

"""legal-ocr 统一入口：通用 OCR 自动路由到 PaddleOCR 或 MinerU，输出 Markdown 与 archive。"""

from __future__ import annotations

import argparse
import json
import shutil
import sys
import tempfile
import traceback
from dataclasses import replace
from datetime import datetime
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))

from base import BackendResult, ConvertOptions  # noqa: E402
from mineru_ocr import MinerUBackend  # noqa: E402
from paddle_ocr import PaddleOCRBackend  # noqa: E402
from common import (  # noqa: E402
    PADDLE_LOCAL_SUFFIXES,
    SourceInfo,
    build_source_info,
    first_non_empty,
    get_skill_root,
    load_env,
    parse_bool,
    resolve_images_dir,
    resolve_output_markdown_path,
    sanitize_name,
    sha256_of_file,
)
from pdf_tools import get_pdf_page_count  # noqa: E402
from basic_postprocess import run_basic_postprocess  # noqa: E402
from legal_terms import detect_legal_context, run_legal_terms_postprocess  # noqa: E402
from linebreaks import run_linebreak_postprocess  # noqa: E402
from router import RouteDecision, choose_backend  # noqa: E402


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="将 PDF、图片、Office 文档或 URL 转换为 Markdown，并在法律材料中自动启用保守增强"
    )
    parser.add_argument("input", help='本地文件、远程 URL，或 "checktoken"')
    parser.add_argument("--backend", choices=["auto", "paddle", "mineru"], default=None)
    parser.add_argument("--output", help="输出 Markdown 路径；如果不是 .md，则按目录处理")
    parser.add_argument("--pages", help='页码范围，例如 "1-20" 或 "1-5,8,10-12"')
    parser.add_argument("--archive-name", help="自定义 archive 目录名后缀，默认使用输入文件名")
    parser.add_argument("--no-archive", action="store_true", help="不写入 archive，仅输出 Markdown")
    parser.add_argument("--no-post-process", action="store_true", help="跳过全部后处理")
    parser.add_argument("--no-legal-terms", action="store_true", help="跳过法律术语优化")
    parser.add_argument(
        "--legal-terms",
        choices=["auto", "always", "never"],
        help="法律术语优化模式；auto 仅在检测到法律材料时启用",
    )
    parser.add_argument("--no-line-merge", action="store_true", help="跳过 OCR 硬换行整理")
    parser.add_argument("--model", choices=["pipeline", "vlm"], help="MinerU Token API 模型")
    parser.add_argument(
        "--paddle-model",
        choices=["PP-OCRv5", "PaddleOCR-VL-1.5"],
        help="PaddleOCR 异步任务模型；Markdown 输出建议 PaddleOCR-VL-1.5",
    )
    parser.add_argument(
        "--paddle-api-protocol",
        choices=["auto", "sync", "async"],
        help="PaddleOCR API 协议；layout-parsing 用 sync，/api/v2/ocr/jobs 用 async",
    )
    parser.add_argument("--paddle-api-extra-json", help="额外 PaddleOCR optionalPayload JSON 文件")
    return parser.parse_args(argv)


def write_text(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def copytree_if_exists(source: Path | None, target: Path) -> None:
    if source and source.exists():
        shutil.copytree(source, target, dirs_exist_ok=True)


def build_source(raw_input: str) -> SourceInfo:
    source = build_source_info(raw_input)
    if source.path and source.suffix == ".pdf":
        source = replace(source, page_count=get_pdf_page_count(source.path))
    return source


def should_postprocess(env: dict[str, str], no_post_process: bool) -> bool:
    if no_post_process:
        return False
    return parse_bool(env.get("LEGAL_OCR_POST_PROCESS"), default=True)


def legal_terms_mode(env: dict[str, str], no_legal_terms: bool, cli_value: str | None) -> str:
    if no_legal_terms:
        return "never"
    configured = (cli_value or env.get("LEGAL_OCR_LEGAL_TERMS") or "auto").strip().lower()
    aliases = {
        "true": "always",
        "1": "always",
        "yes": "always",
        "on": "always",
        "always": "always",
        "force": "always",
        "false": "never",
        "0": "never",
        "no": "never",
        "off": "never",
        "never": "never",
        "auto": "auto",
    }
    if configured not in aliases:
        raise ValueError("LEGAL_OCR_LEGAL_TERMS 仅支持 auto/true/false，或命令行 auto/always/never")
    return aliases[configured]


def should_line_merge(env: dict[str, str], no_line_merge: bool) -> bool:
    if no_line_merge:
        return False
    return parse_bool(env.get("LEGAL_OCR_LINE_MERGE"), default=True)


def make_backend(name: str, env: dict[str, str]) -> Any:
    if name == "paddle":
        return PaddleOCRBackend(env)
    if name == "mineru":
        return MinerUBackend(env)
    raise ValueError(f"未知后端：{name}")


def source_json(source: SourceInfo) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "raw": source.raw,
        "type": source.source_type,
        "name": source.file_name,
        "suffix": source.suffix,
        "page_count": source.page_count,
    }
    if source.path:
        payload["path"] = str(source.path)
        payload["sha256"] = sha256_of_file(source.path)
        payload["size_bytes"] = source.size_bytes
    return payload


def build_archive(
    *,
    source: SourceInfo,
    archive_name: str,
    output_md_path: Path,
    output_images_dir: Path,
    raw_markdown: str,
    result_json: dict[str, Any],
    metadata: dict[str, Any],
    backend_result_dir: Path | None,
) -> Path:
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    archive_root = get_skill_root() / "archive" / f"{timestamp}_{sanitize_name(archive_name)}"
    output_dir = archive_root / "output"
    batches_dir = archive_root / "batches"
    backend_dir = archive_root / "backend_result"

    output_dir.mkdir(parents=True, exist_ok=True)

    shutil.copy2(output_md_path, output_dir / "result.md")
    write_text(output_dir / "result_raw.md", raw_markdown)
    write_json(output_dir / "result.json", result_json)

    if output_images_dir.exists():
        shutil.copytree(output_images_dir, output_dir / "images", dirs_exist_ok=True)

    if backend_result_dir and backend_result_dir.exists():
        shutil.copytree(backend_result_dir, backend_dir, dirs_exist_ok=True)
        batch_files = sorted(backend_result_dir.glob("batch_*.json"))
        if batch_files:
            batches_dir.mkdir(parents=True, exist_ok=True)
            for batch_file in batch_files:
                shutil.copy2(batch_file, batches_dir / batch_file.name)

    write_json(archive_root / "metadata.json", metadata)
    if metadata.get("postprocess_log"):
        write_json(archive_root / "postprocess_log.json", metadata["postprocess_log"])
    return archive_root


def copy_attempt_assets(attempt_assets_dir: Path, output_images_dir: Path) -> bool:
    if not attempt_assets_dir.exists() or not any(attempt_assets_dir.iterdir()):
        return False
    if output_images_dir.exists():
        shutil.rmtree(output_images_dir)
    shutil.copytree(attempt_assets_dir, output_images_dir)
    return True


def update_image_paths(images: list[dict[str, str]], output_images_dir: Path) -> None:
    for image in images:
        filename = image.get("filename")
        if filename:
            image["path"] = str(output_images_dir / filename)


def build_result_payload(
    *,
    source: SourceInfo,
    backend_result: BackendResult,
    route: RouteDecision,
    attempts: list[dict[str, str]],
    raw_markdown: str,
    final_markdown: str,
    postprocess_log: list[dict[str, str]],
    postprocess_enabled: bool,
    legal_context: dict[str, Any],
) -> dict[str, Any]:
    metadata = backend_result.metadata or {}
    return {
        "ok": True,
        "source": source_json(source),
        "backend": {
            "name": backend_result.backend,
            "mode": backend_result.mode,
            "provider": backend_result.provider,
        },
        "route": {
            "preferred": route.preferred,
            "candidates": route.candidates,
            "reason": route.reason,
            "notes": route.notes,
            "attempts": attempts,
        },
        "processing": metadata.get("processing", {}),
        "text": final_markdown,
        "raw_text": raw_markdown,
        "images": backend_result.images,
        "batches": backend_result.batches,
        "postprocess": {
            "enabled": postprocess_enabled,
            "log_count": len(postprocess_log),
            "legal_context": legal_context,
        },
    }


def classify_backend_error(error: Exception) -> str:
    message = str(error)
    lowered = message.lower()

    quota_terms = (
        "429",
        "quota",
        "rate limit",
        "rate-limit",
        "rate_limited",
        "too many requests",
        "limit exceeded",
        "insufficient",
        "配额",
        "额度",
        "余额",
        "用量",
        "频率",
        "资源包",
        "点数",
    )
    if any(term in lowered or term in message for term in quota_terms):
        return "quota_or_rate_limit"

    auth_terms = ("401", "403", "unauthorized", "forbidden", "access denied", "鉴权", "认证失败")
    if any(term in lowered or term in message for term in auth_terms):
        return "auth"

    if "免登录轻量接口限制" in message:
        return "light_limit"

    unsupported_terms = ("unsupported", "不支持", "仅由", "只能走", "暂不支持")
    if any(term in lowered or term in message for term in unsupported_terms):
        return "unsupported"

    timeout_terms = ("timeout", "timed out", "超时")
    if any(term in lowered or term in message for term in timeout_terms):
        return "timeout"

    network_terms = ("network", "connection", "connect", "网络", "连接")
    if any(term in lowered or term in message for term in network_terms):
        return "network"

    return "error"


def run_checktoken(env: dict[str, str]) -> int:
    print("legal-ocr 配置自检")
    print("===============================================")
    try:
        mineru = MinerUBackend(env)
        print(f"MinerU: {mineru.verify_token()}")
    except Exception as error:  # noqa: BLE001
        print(f"MinerU: {error}")

    try:
        PaddleOCRBackend(env)
        print("PaddleOCR: 已检测到必要配置。")
    except Exception as error:  # noqa: BLE001
        print(f"PaddleOCR: {error}")
    return 0


def convert_once(
    *,
    backend_name: str,
    env: dict[str, str],
    source: SourceInfo,
    options: ConvertOptions,
    temp_root: Path,
) -> tuple[BackendResult, Path]:
    backend = make_backend(backend_name, env)
    work_dir = temp_root / backend_name
    assets_dir = temp_root / f"{backend_name}_assets"
    work_dir.mkdir(parents=True, exist_ok=True)
    return backend.convert(source, options, work_dir, assets_dir), assets_dir


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    env = load_env()
    if args.paddle_model:
        env["PADDLEOCR_MODEL"] = args.paddle_model
    if args.paddle_api_protocol:
        env["PADDLEOCR_API_PROTOCOL"] = args.paddle_api_protocol
    if args.paddle_api_extra_json:
        env["PADDLEOCR_API_EXTRA_JSON"] = args.paddle_api_extra_json

    if args.input in {"checktoken", "verify-token", "--verify-token"}:
        return run_checktoken(env)

    source = build_source(args.input)
    configured_backend = args.backend or env.get("LEGAL_OCR_BACKEND", "auto")
    route = choose_backend(source, env, configured_backend)
    output_md_path = resolve_output_markdown_path(source, args.output)
    output_images_dir = resolve_images_dir(output_md_path)
    archive_name = args.archive_name or source.base_name
    options = ConvertOptions(
        pages=args.pages,
        model=args.model,
        paddle_model=args.paddle_model,
        log_level=first_non_empty(env, "LEGAL_OCR_LOG_LEVEL", "PADDLEOCR_LOG_LEVEL", "MINERU_LOG_LEVEL") or "medium",
    )
    try:
        postprocess_enabled = should_postprocess(env, args.no_post_process)
        configured_legal_terms_mode = (
            legal_terms_mode(env, args.no_legal_terms, args.legal_terms) if postprocess_enabled else "disabled"
        )
    except ValueError as error:
        print(error, file=sys.stderr)
        return 2

    temp_root = Path(tempfile.mkdtemp(prefix="legal_ocr_"))
    attempts: list[dict[str, str]] = []
    try:
        last_error: Exception | None = None
        for attempt_index, backend_name in enumerate(route.candidates):
            try:
                backend_result, attempt_assets_dir = convert_once(
                    backend_name=backend_name,
                    env=env,
                    source=source,
                    options=options,
                    temp_root=temp_root,
                )
                attempts.append({"backend": backend_name, "status": "success"})
                raw_markdown = backend_result.markdown.strip()
                if postprocess_enabled:
                    working_markdown = raw_markdown
                    postprocess_log: list[dict[str, str]] = []
                    mode = configured_legal_terms_mode
                    legal_context = {
                        "mode": mode,
                        "enabled": False,
                        "detection": None,
                    }
                    if mode == "auto":
                        detection = detect_legal_context(working_markdown, source_name=source.file_name)
                        legal_context["detection"] = detection
                        legal_context["enabled"] = bool(detection["is_legal"])
                        if not detection["is_legal"]:
                            postprocess_log.append(
                                {
                                    "action": "legal_terms_skip",
                                    "category": "legal_context_auto",
                                    "description": "未检测到足够法律文书信号，跳过法律术语优化",
                                    "score": str(detection["score"]),
                                }
                            )
                    elif mode == "always":
                        legal_context["enabled"] = True

                    if legal_context["enabled"]:
                        legal_terms_result = run_legal_terms_postprocess(
                            working_markdown,
                            custom_terms_path=env.get("LEGAL_OCR_CUSTOM_TERMS_PATH"),
                        )
                        working_markdown = legal_terms_result.text
                        postprocess_log.extend(legal_terms_result.log)

                    if should_line_merge(env, args.no_line_merge):
                        linebreak_result = run_linebreak_postprocess(working_markdown)
                        working_markdown = linebreak_result.text
                        postprocess_log.extend(linebreak_result.log)

                    postprocessed = run_basic_postprocess(working_markdown)
                    final_markdown = postprocessed.text
                    postprocess_log.extend(postprocessed.log)
                else:
                    final_markdown = raw_markdown
                    postprocess_log = []
                    legal_context = {
                        "mode": "disabled",
                        "enabled": False,
                        "detection": None,
                    }

                write_text(output_md_path, final_markdown)
                if copy_attempt_assets(attempt_assets_dir, output_images_dir):
                    update_image_paths(backend_result.images, output_images_dir)

                result_json = build_result_payload(
                    source=source,
                    backend_result=backend_result,
                    route=route,
                    attempts=attempts,
                    raw_markdown=raw_markdown,
                    final_markdown=final_markdown,
                    postprocess_log=postprocess_log,
                    postprocess_enabled=postprocess_enabled,
                    legal_context=legal_context,
                )
                metadata = {
                    "created_at": datetime.now().isoformat(timespec="seconds"),
                    "source": source_json(source),
                    "backend": result_json["backend"],
                    "route": result_json["route"],
                    "output_markdown": str(output_md_path),
                    "image_output_dir": str(output_images_dir) if output_images_dir.exists() else None,
                    "postprocess_enabled": postprocess_enabled,
                    "legal_context": result_json["postprocess"]["legal_context"],
                    "postprocess_log": postprocess_log,
                    "backend_metadata": backend_result.metadata,
                }

                archive_path: Path | None = None
                if not args.no_archive:
                    archive_path = build_archive(
                        source=source,
                        archive_name=archive_name,
                        output_md_path=output_md_path,
                        output_images_dir=output_images_dir,
                        raw_markdown=raw_markdown,
                        result_json=result_json,
                        metadata=metadata,
                        backend_result_dir=backend_result.backend_result_dir,
                    )

                print("转换完成")
                print(f"Markdown: {output_md_path}")
                if output_images_dir.exists():
                    print(f"图片目录: {output_images_dir}")
                if archive_path:
                    print(f"Archive: {archive_path}")
                print(f"后端: {backend_result.backend} ({backend_result.mode})")
                # 输出实际使用的模型名，供上游 run_ocr.py 解析
                _model = (backend_result.metadata or {}).get("config", {}).get("model")
                if _model:
                    print(f"OCR_MODEL: {_model}")
                return 0
            except Exception as error:  # noqa: BLE001
                category = classify_backend_error(error)
                has_next_backend = attempt_index < len(route.candidates) - 1
                attempt: dict[str, str] = {
                    "backend": backend_name,
                    "status": "failed",
                    "category": category,
                    "error": str(error),
                }
                if route.fallback_allowed and has_next_backend:
                    attempt["fallback"] = "next_backend"
                attempts.append(attempt)
                last_error = error
                if not route.fallback_allowed or not has_next_backend:
                    break

        print("转换失败", file=sys.stderr)
        for attempt in attempts:
            detail = f"- {attempt['backend']}: {attempt['status']}"
            if "error" in attempt:
                detail += f" - {attempt['error']}"
            print(detail, file=sys.stderr)
        if last_error:
            # 2026-06-14 加诊断性 logging:
            # 旧实现 `print(f"最后错误:{last_error}")` 把异常 str() 化掉,丢了 type 和 traceback;
            # 上游 run_ocr.py:326 只抓 stderr 最后一行,导致 segments.last_error 短得无法定位真因
            # (例:`Invalid port: ':1]'` 是 httpx.InvalidURL 的 message,但看不出哪个 url 哪个 client 抛的)。
            # 现在最后一行带 type 名,traceback 在 stderr 全文里供 archive / cron 日志回溯。
            tb_lines = traceback.format_exception(type(last_error), last_error, last_error.__traceback__)
            print("最后错误 traceback:", file=sys.stderr)
            for ln in tb_lines:
                print(ln, file=sys.stderr, end="")
            # 末行(保留旧前缀让上游 splitlines()[-1] 兼容,但补上 type 名定位异常类)
            print(f"最后错误:{type(last_error).__name__}: {last_error}", file=sys.stderr)
        return 1
    finally:
        shutil.rmtree(temp_root, ignore_errors=True)


if __name__ == "__main__":
    # 2026-06-14 入口级 catch:之前 1.4.1 加的 try/except 在 main() 内 backend 循环里,
    # 但有些异常更早(如 PaddleOCRBackend.__init__ / MinerUBackend.__init__ 调
    # httpx.URL / httpx.Client 校验,或 route 解析阶段)就抛出,绕过内部 catch。
    # 在入口层兜底 catch BaseException + 立即 traceback.print_exc 到 stderr,
    # 保证不管哪行抛错,完整 stack 必进 stderr,上游 run_ocr.py 扩信道后能完整读到。
    try:
        raise SystemExit(main())
    except SystemExit:
        raise
    except BaseException as _entry_exc:
        print(f"convert.py 入口异常:{type(_entry_exc).__name__}: {_entry_exc}", file=sys.stderr)
        traceback.print_exc(file=sys.stderr)
        raise SystemExit(1)

#!/usr/bin/env python3
"""Fix broken image src paths in legal-ocr markdown output.

When the PaddleOCR backend returns images, the markdown text keeps the
original `imgs/img_in_image_box_*.jpg` references, but the script saves
images with batch-based names like `1-40_001.jpg`. This script reads the
result.json from the archive, builds a source→filename map, and replaces
the broken references in the markdown.
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path


def build_map(result_json_path: Path) -> dict[str, str]:
    data = json.loads(result_json_path.read_text(encoding="utf-8"))
    mapping: dict[str, str] = {}
    for image in data.get("images", []):
        source = image.get("source")
        filename = image.get("filename")
        if source and filename:
            mapping[source] = filename
    return mapping


def fix_markdown(md_path: Path, mapping: dict[str, str]) -> tuple[int, int]:
    text = md_path.read_text(encoding="utf-8")
    pattern = re.compile(r"imgs/[A-Za-z0-9_\-]+\.(?:jpg|jpeg|png|gif|webp)")
    replaced = 0
    missing = 0

    def sub(match: re.Match[str]) -> str:
        nonlocal replaced, missing
        key = match.group(0)
        if key in mapping:
            replaced += 1
            return mapping[key]
        missing += 1
        return key

    new_text = pattern.sub(sub, text)
    if new_text != text:
        md_path.write_text(new_text, encoding="utf-8")
    return replaced, missing


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("markdown", type=Path, help="Path to the markdown file to fix")
    parser.add_argument(
        "--result-json",
        type=Path,
        required=True,
        help="Path to result.json from the legal-ocr archive",
    )
    args = parser.parse_args()

    if not args.markdown.exists():
        print(f"Markdown not found: {args.markdown}", file=sys.stderr)
        return 1
    if not args.result_json.exists():
        print(f"result.json not found: {args.result_json}", file=sys.stderr)
        return 1

    mapping = build_map(args.result_json)
    if not mapping:
        print("No image mapping found in result.json", file=sys.stderr)
        return 1

    replaced, missing = fix_markdown(args.markdown, mapping)
    print(f"Image references replaced: {replaced}")
    if missing:
        print(f"Unresolved references: {missing}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

from __future__ import annotations

import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from basic_postprocess import PostprocessResult


@dataclass(frozen=True)
class LegalTermRule:
    pattern: str
    replacement: str
    category: str
    description: str
    flags: int = re.MULTILINE


@dataclass(frozen=True)
class LegalContextSignal:
    pattern: str
    label: str
    weight: int
    flags: int = re.MULTILINE


def _spaced_term_rule(term: str, category: str = "spaced_legal_term") -> LegalTermRule:
    chars = [re.escape(char) for char in term]
    optional_gap = r"(?:[ \t]*\n[ \t]*|[ \t]*)"
    required_gap = r"(?:[ \t]+|[ \t]*\n[ \t]*)"
    variants: list[str] = []
    for required_index in range(len(chars) - 1):
        parts: list[str] = []
        for index, char in enumerate(chars):
            parts.append(char)
            if index < len(chars) - 1:
                parts.append(required_gap if index == required_index else optional_gap)
        variants.append("".join(parts))
    pattern = "(?:" + "|".join(variants) + ")"
    return LegalTermRule(
        pattern=pattern,
        replacement=term,
        category=category,
        description=f"合并法律术语断字/断行：{term}",
    )


def _loose_term_pattern(term: str) -> str:
    chars = [re.escape(char) for char in term]
    return r"[ \t\n]*".join(chars)


LITERAL_CONFUSIONS = [
    ("亊", "事", "ocr_variant", "将 OCR 识别出的异体字“亊”规范为“事”"),
    ("入民法院", "人民法院", "ocr_confusion", "修正常见 OCR 误识别：入民法院"),
    ("入民检察院", "人民检察院", "ocr_confusion", "修正常见 OCR 误识别：入民检察院"),
    ("入民陪审员", "人民陪审员", "ocr_confusion", "修正常见 OCR 误识别：入民陪审员"),
    ("民亊", "民事", "ocr_variant", "修正常见 OCR 误识别：民亊"),
    ("刑亊", "刑事", "ocr_variant", "修正常见 OCR 误识别：刑亊"),
]


SPACED_TERMS = [
    "人民法院",
    "人民检察院",
    "人民陪审员",
    "民事判决书",
    "民事裁定书",
    "民事调解书",
    "刑事判决书",
    "行政判决书",
    "执行裁定书",
    "民事起诉状",
    "民事上诉状",
    "答辩状",
    "代理意见",
    "质证意见",
    "法定代表人",
    "委托诉讼代理人",
    "委托代理人",
    "诉讼请求",
    "事实和理由",
    "审理经过",
    "审理查明",
    "经审理查明",
    "基本事实",
    "争议焦点",
    "本院查明",
    "本院认为",
    "裁判结果",
    "判决如下",
    "裁定如下",
    "审判长",
    "审判员",
    "书记员",
    "原告",
    "被告",
    "第三人",
    "上诉人",
    "被上诉人",
    "申请人",
    "被申请人",
    "再审申请人",
    "申请执行人",
    "被执行人",
    "执行依据",
    "执行标的",
    "管辖权异议",
    "举证期限",
    "证据目录",
    "庭审笔录",
    "开庭传票",
    "送达地址确认书",
    "营业执照",
    "统一社会信用代码",
    "身份证号码",
    "案件受理费",
    "保全费",
    "公告费",
    "鉴定费",
    "律师费",
    "违约金",
    "滞纳金",
    "迟延履行期间",
    "加倍支付迟延履行期间的债务利息",
]


LABEL_RULES = [
    LegalTermRule(
        pattern=r"^(\s*)(原告|被告|第三人|上诉人|被上诉人|申请人|被申请人|再审申请人|申请执行人|被执行人|法定代表人|委托诉讼代理人|委托代理人)\s*[;；:：]\s*",
        replacement=r"\1\2：",
        category="legal_label",
        description="统一常见法律主体标签冒号",
    ),
    LegalTermRule(
        pattern=r"^(\s*)(案号|审理法院|裁判日期|书记员|审判长|审判员|人民陪审员)\s*[;；:：]\s*",
        replacement=r"\1\2：",
        category="legal_label",
        description="统一常见文书信息标签冒号",
    ),
]


DEFAULT_RULES = [
    *[
        LegalTermRule(
            pattern=re.escape(source),
            replacement=target,
            category=category,
            description=description,
        )
        for source, target, category, description in LITERAL_CONFUSIONS
    ],
    *[_spaced_term_rule(term) for term in SPACED_TERMS],
    *LABEL_RULES,
]


LEGAL_CONTEXT_SIGNALS = [
    *[
        LegalContextSignal(
            pattern=_loose_term_pattern(term),
            label=term,
            weight=weight,
        )
        for term, weight in [
            ("人民法院", 3),
            ("人民检察院", 3),
            ("民事判决书", 5),
            ("刑事判决书", 5),
            ("行政判决书", 5),
            ("民事裁定书", 5),
            ("执行裁定书", 5),
            ("民事起诉状", 4),
            ("民事上诉状", 4),
            ("委托诉讼代理人", 3),
            ("法定代表人", 2),
            ("诉讼请求", 3),
            ("事实和理由", 2),
            ("本院查明", 4),
            ("本院认为", 5),
            ("判决如下", 5),
            ("裁定如下", 5),
            ("审判长", 3),
            ("书记员", 3),
            ("证据目录", 3),
        ]
    ],
    LegalContextSignal(
        pattern=r"^\s*(原告|被告|第三人|上诉人|被上诉人|申请人|被申请人|申请执行人|被执行人)\s*[;；:：]",
        label="法律主体标签",
        weight=4,
    ),
    LegalContextSignal(pattern=r"^\s*(案号|审理法院|裁判日期)\s*[;；:：]", label="文书信息标签", weight=3),
    LegalContextSignal(pattern=r"[（(]\d{4}[）)][\u4e00-\u9fffA-Za-z0-9（）()第\-]+号", label="裁判文书案号", weight=4),
]


LEGAL_FILENAME_TERMS = [
    "判决书",
    "裁定书",
    "调解书",
    "起诉状",
    "上诉状",
    "答辩状",
    "证据目录",
    "庭审笔录",
    "法院",
    "卷宗",
    "案卷",
]


def detect_legal_context(markdown: str, *, source_name: str | None = None) -> dict[str, Any]:
    """Conservatively detect whether OCR text looks like a legal document."""

    sample = markdown[:50000]
    hits: list[dict[str, Any]] = []
    score = 0
    strong_signal_count = 0

    for signal in LEGAL_CONTEXT_SIGNALS:
        matches = re.findall(signal.pattern, sample, flags=signal.flags)
        if not matches:
            continue
        count = len(matches)
        capped_count = min(count, 3)
        score += signal.weight * capped_count
        if signal.weight >= 4:
            strong_signal_count += 1
        hits.append(
            {
                "label": signal.label,
                "count": count,
                "weight": signal.weight,
            }
        )

    filename_hits: list[str] = []
    if source_name:
        filename_hits = [term for term in LEGAL_FILENAME_TERMS if term in source_name]
        if filename_hits:
            score += min(len(filename_hits), 2) * 4

    is_legal = score >= 6 or strong_signal_count >= 2
    if filename_hits and score >= 4:
        is_legal = True

    return {
        "is_legal": is_legal,
        "score": score,
        "strong_signal_count": strong_signal_count,
        "hits": hits[:20],
        "filename_hits": filename_hits,
        "threshold": "score>=6 or strong_signal_count>=2",
    }


def _load_custom_rules(path_value: str | None) -> list[LegalTermRule]:
    if not path_value or not str(path_value).strip():
        return []
    path = Path(str(path_value).strip()).expanduser()
    if not path.exists():
        raise FileNotFoundError(f"LEGAL_OCR_CUSTOM_TERMS_PATH 不存在：{path}")

    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except ValueError as error:
        raise ValueError(f"LEGAL_OCR_CUSTOM_TERMS_PATH 不是合法 JSON：{path}") from error

    raw_rules: Any
    if isinstance(payload, dict) and isinstance(payload.get("replacements"), list):
        raw_rules = payload["replacements"]
    elif isinstance(payload, dict):
        raw_rules = [{"source": key, "replacement": value} for key, value in payload.items()]
    elif isinstance(payload, list):
        raw_rules = payload
    else:
        raise ValueError("LEGAL_OCR_CUSTOM_TERMS_PATH 顶层必须是对象、数组或包含 replacements 的对象")

    rules: list[LegalTermRule] = []
    for index, item in enumerate(raw_rules, start=1):
        if not isinstance(item, dict):
            raise ValueError(f"自定义术语第 {index} 项必须是对象")
        replacement = str(item.get("replacement") or item.get("target") or "").strip()
        if not replacement:
            raise ValueError(f"自定义术语第 {index} 项缺少 replacement")

        if item.get("pattern"):
            pattern = str(item["pattern"])
        else:
            source = str(item.get("source") or item.get("text") or "").strip()
            if not source:
                raise ValueError(f"自定义术语第 {index} 项缺少 source 或 pattern")
            pattern = re.escape(source)

        rules.append(
            LegalTermRule(
                pattern=pattern,
                replacement=replacement,
                category=str(item.get("category") or "custom_legal_term"),
                description=str(item.get("description") or "自定义法律术语替换"),
            )
        )
    return rules


def run_legal_terms_postprocess(
    markdown: str,
    *,
    custom_terms_path: str | None = None,
) -> PostprocessResult:
    text = markdown
    log: list[dict[str, str]] = []
    rules = [*DEFAULT_RULES, *_load_custom_rules(custom_terms_path)]

    for rule in rules:
        text, count = re.subn(rule.pattern, rule.replacement, text, flags=rule.flags)
        if count:
            log.append(
                {
                    "action": "legal_term_replace",
                    "category": rule.category,
                    "pattern": rule.pattern,
                    "replacement": rule.replacement,
                    "count": str(count),
                    "description": rule.description,
                }
            )
    return PostprocessResult(text=text, log=log)