vent_ai_platform/app/services/document_processing.py

from __future__ import annotations

import asyncio
import html
import json
import math
import re
import zipfile
from io import BytesIO
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import httpx
from sqlalchemy.orm import Session

from app.agents.runtime import get_agent_runtime
from app.core.config import settings
from app.core.errors import APIError
from app.db.models.document import (
    BusinessDocument,
    DocumentExtraction,
    DocumentExtractionTemplate,
    DocumentProcessingRun,
)
from app.db.repositories import document as document_repo
from app.db.session import SessionLocal
from app.services.document_business_templates import (
    detect_file_format,
    get_template_by_id,
    resolve_document_template,
    template_field_definitions,
)


PROCESSING_ACTIVE_STATUSES = {"pending", "running", "parsing", "extracting"}
PARSE_MODES = {"lightweight", "precision"}


def _now() -> datetime:
    return datetime.now(timezone.utc)


def normalize_parse_mode(value: str | None) -> str:
    mode = str(value or "lightweight").strip().lower()
    if mode not in PARSE_MODES:
        raise APIError(
            422,
            "invalid_parse_mode",
            "解析模式仅支持 lightweight 或 precision",
        )
    return mode


def resolve_effective_parse_mode(
    parse_mode: str | None,
    file_ext: str | None,
    file_path: str | None = None,
) -> str:
    mode = normalize_parse_mode(parse_mode)
    extension = str(file_ext or "").strip().lower()
    if extension and not extension.startswith("."):
        extension = f".{extension}"
    if mode == "lightweight" and extension in {".doc", ".xls"}:
        return "precision"
    if mode == "lightweight" and file_path:
        source = Path(file_path)
        if source.is_file():
            if source.stat().st_size > 10 * 1024 * 1024:
                return "precision"
            metrics = inspect_source_metrics(source)
            page_count = metrics.get("page_count")
            if page_count and int(page_count) > 20:
                return "precision"
    return mode


def start_document_processing(
    db: Session,
    document_id: str,
    *,
    force: bool = False,
    parse_mode: str = "lightweight",
    template_id: str | None = None,
) -> DocumentProcessingRun:
    document = document_repo.get_document(db, document_id)
    if document is None:
        raise APIError(404, "document_not_found", "业务资料不存在")

    latest = document_repo.get_latest_processing_run(db, document_id)
    if latest is not None and latest.status in PROCESSING_ACTIVE_STATUSES and not force:
        return latest

    file_format = detect_file_format(document.file_ext or document.file_name)
    requested_parse_mode = normalize_parse_mode(parse_mode)
    parse_mode = resolve_effective_parse_mode(
        requested_parse_mode,
        document.file_ext or document.file_name,
        document.file_path,
    )
    template = resolve_document_template(
        db,
        title=document.title,
        template_id=template_id,
    )
    document.document_type = template.code
    document.parse_status = "pending"
    document.parse_progress = 0
    document.parse_error = ""
    db.add(document)
    db.commit()

    return document_repo.create_processing_run(
        db,
        {
            "document_id": document.id,
            "template_id": template.id,
            "status": "pending",
            "stage": "uploaded",
            "progress": 0,
            "document_type": template.code,
            "parser_engine": f"mineru_api_{parse_mode}",
            "metadata_json": {
                "template_code": template.code,
                "template_version": template.version,
                "parse_mode": parse_mode,
                "requested_parse_mode": requested_parse_mode,
                "parse_mode_auto_upgraded": parse_mode != requested_parse_mode,
                "file_format": file_format,
            },
        },
    )


async def process_document_run(run_id: str) -> None:
    with SessionLocal() as db:
        await process_document(db, run_id)


async def process_document(db: Session, run_id: str) -> None:
    run = document_repo.get_processing_run(db, run_id)
    if run is None:
        return
    document = document_repo.get_document(db, run.document_id)
    if document is None:
        run.status = "failed"
        run.stage = "failed"
        run.error = "业务资料不存在"
        run.finished_at = _now()
        db.add(run)
        db.commit()
        return

    if not run.template_id:
        raise APIError(422, "extraction_template_required", "处理任务缺少抽取模板")
    template = get_template_by_id(db, run.template_id)

    run.status = "running"
    run.stage = "parsing"
    run.progress = 10
    run.started_at = _now()
    run.error = ""
    document.parse_status = "parsing"
    document.parse_progress = 10
    document.parse_error = ""
    db.add_all([run, document])
    db.commit()

    try:
        parse_mode = normalize_parse_mode(
            (run.metadata_json or {}).get("parse_mode", "lightweight")
        )
        run.parser_engine = f"mineru_api_{parse_mode}"
        run.metadata_json = {
            **(run.metadata_json or {}),
            "parse_mode": parse_mode,
        }
        parsed = await parse_document_with_mineru_api(
            document.file_path,
            document.id,
            document.file_name,
            parse_mode,
        )
        run.parsed_markdown_path = str(parsed["markdown_path"])
        run.parsed_markdown = str(parsed["markdown"])
        run.stage = "extracting"
        run.progress = 65
        document.parse_status = "extracting"
        document.parse_progress = 65
        db.add_all([run, document])
        db.commit()

        extracted = await extract_document_fields(
            document=document,
            template=template,
            parsed_markdown=str(parsed["markdown"]),
        )
        replace_document_extractions(
            db,
            document_id=document.id,
            fields=extracted["fields"],
            model_status=extracted["model_status"],
        )

        document.summary = extracted["summary"]
        document.page_count = parsed.get("page_count") or document.page_count
        document.row_count = parsed.get("row_count") or document.row_count
        document.parse_status = "completed"
        document.parse_progress = 100
        document.parse_error = ""
        document.metadata_json = {
            **(document.metadata_json or {}),
            "extraction_template": template.code,
            "processing_run_id": run.id,
            "parser_engine": run.parser_engine,
            "model_status": extracted["model_status"],
        }
        run.status = "completed"
        run.stage = "completed"
        run.progress = 100
        run.model_output = {
            "summary": extracted["summary"],
            "model_status": extracted["model_status"],
            "field_count": len(extracted["fields"]),
        }
        run.finished_at = _now()
        db.add_all([run, document])
        db.commit()
    except Exception as exc:  # noqa: BLE001 - 外部解析和模型错误必须落库
        db.rollback()
        run = document_repo.get_processing_run(db, run_id)
        document = document_repo.get_document(db, run.document_id) if run else None
        message = f"{type(exc).__name__}: {exc}"
        if run is not None:
            run.status = "failed"
            run.stage = "failed"
            run.progress = 100
            run.error = message
            run.finished_at = _now()
            db.add(run)
        if document is not None:
            document.parse_status = "failed"
            document.parse_progress = 100
            document.parse_error = message
            db.add(document)
        db.commit()


async def parse_document_with_mineru_api(
    file_path: str,
    document_id: str,
    file_name: str,
    parse_mode: str,
) -> dict[str, Any]:
    source = Path(file_path)
    if not source.is_file():
        raise APIError(404, "document_file_not_found", "资料文件不存在")

    parse_mode = normalize_parse_mode(parse_mode)
    headers = _mineru_headers()
    timeout = httpx.Timeout(
        connect=20.0,
        read=120.0,
        write=180.0,
        pool=30.0,
    )
    async with httpx.AsyncClient(
        base_url=settings.mineru_api_base.rstrip("/"),
        timeout=timeout,
        follow_redirects=True,
    ) as client:
        if parse_mode == "precision":
            return await _parse_with_precision_api(
                client,
                headers,
                source,
                document_id,
            )
        return await _parse_with_lightweight_api(
            client,
            headers,
            source,
            document_id,
            file_name,
        )


async def _parse_with_precision_api(
    client: httpx.AsyncClient,
    headers: dict[str, str],
    source: Path,
    document_id: str,
) -> dict[str, Any]:
    response = await client.post(
        "/api/v4/file-urls/batch",
        headers=headers,
        json={
            "model_version": "vlm",
            "language": "ch",
            "enable_formula": True,
            "enable_table": True,
            "files": [
                {
                    "name": source.name,
                    "data_id": document_id,
                    "is_ocr": source.suffix.lower() in {".pdf", ".png", ".jpg", ".jpeg"},
                }
            ],
        },
    )
    data = _mineru_response_data(response, "申请精准解析上传地址失败")
    file_urls = data.get("file_urls") or []
    batch_id = str(data.get("batch_id") or "")
    if not file_urls or not batch_id:
        raise APIError(502, "mineru_invalid_response", "MinerU 未返回上传地址或任务 ID")

    await _upload_source_file(client, str(file_urls[0]), source)
    result = await _poll_precision_result(client, headers, batch_id)
    zip_url = str(result.get("full_zip_url") or "")
    if not zip_url:
        raise APIError(502, "mineru_empty_result", "MinerU 未返回精准解析结果包")

    content = await _download_mineru_asset(client, zip_url)
    markdown = _extract_markdown_from_zip(content)
    progress = result.get("extract_progress") or {}
    metrics = inspect_source_metrics(source)
    return _save_parsed_markdown(
        document_id=document_id,
        markdown=markdown,
        page_count=progress.get("total_pages") or metrics.get("page_count"),
        row_count=metrics.get("row_count"),
    )


async def _parse_with_lightweight_api(
    client: httpx.AsyncClient,
    headers: dict[str, str],
    source: Path,
    document_id: str,
    file_name: str,
) -> dict[str, Any]:
    response = await client.post(
        "/api/v1/agent/parse/file",
        headers=headers,
        json={
            "file_name": file_name,
            "language": "ch",
            "enable_table": True,
            "is_ocr": source.suffix.lower() in {".pdf", ".png", ".jpg", ".jpeg"},
            "enable_formula": True,
        },
    )
    data = _mineru_response_data(response, "申请轻量解析上传地址失败")
    file_url = str(data.get("file_url") or "")
    task_id = str(data.get("task_id") or "")
    if not file_url or not task_id:
        raise APIError(502, "mineru_invalid_response", "MinerU 未返回上传地址或任务 ID")

    await _upload_source_file(client, file_url, source)
    result = await _poll_lightweight_result(client, headers, task_id)
    markdown_url = str(result.get("markdown_url") or "")
    if not markdown_url:
        raise APIError(502, "mineru_empty_result", "MinerU 未返回 Markdown 地址")

    markdown = (await _download_mineru_asset(client, markdown_url)).decode(
        "utf-8",
        errors="replace",
    )
    metrics = inspect_source_metrics(source)
    return _save_parsed_markdown(
        document_id=document_id,
        markdown=markdown,
        page_count=metrics.get("page_count"),
        row_count=metrics.get("row_count"),
    )


async def _upload_source_file(
    client: httpx.AsyncClient,
    upload_url: str,
    source: Path,
) -> None:
    response = await client.put(
        upload_url,
        content=source.read_bytes(),
    )
    if response.is_error:
        raise APIError(502, "mineru_upload_failed", "文件上传到 MinerU 失败")


async def _poll_precision_result(
    client: httpx.AsyncClient,
    headers: dict[str, str],
    batch_id: str,
) -> dict[str, Any]:
    deadline = asyncio.get_running_loop().time() + settings.mineru_timeout
    while asyncio.get_running_loop().time() < deadline:
        response = await client.get(
            f"/api/v4/extract-results/batch/{batch_id}",
            headers=headers,
        )
        data = _mineru_response_data(response, "查询精准解析任务失败")
        results = data.get("extract_result") or []
        if isinstance(results, dict):
            results = [results]
        result = results[0] if results else {}
        state = str(result.get("state") or "").lower()
        if state == "done":
            return result
        if state == "failed":
            raise APIError(
                502,
                "mineru_failed",
                str(result.get("err_msg") or "MinerU 精准解析失败"),
            )
        await asyncio.sleep(settings.mineru_poll_interval)
    raise APIError(504, "mineru_timeout", "MinerU 精准解析超时")


async def _poll_lightweight_result(
    client: httpx.AsyncClient,
    headers: dict[str, str],
    task_id: str,
) -> dict[str, Any]:
    deadline = asyncio.get_running_loop().time() + settings.mineru_timeout
    while asyncio.get_running_loop().time() < deadline:
        response = await client.get(
            f"/api/v1/agent/parse/{task_id}",
            headers=headers,
        )
        data = _mineru_response_data(response, "查询轻量解析任务失败")
        state = str(data.get("state") or "").lower()
        if state == "done":
            return data
        if state == "failed":
            raise APIError(
                502,
                "mineru_failed",
                str(data.get("err_msg") or "MinerU 轻量解析失败"),
            )
        await asyncio.sleep(settings.mineru_poll_interval)
    raise APIError(504, "mineru_timeout", "MinerU 轻量解析超时")


async def _download_mineru_asset(
    client: httpx.AsyncClient,
    url: str,
) -> bytes:
    response = await client.get(url)
    if response.is_error:
        raise APIError(502, "mineru_download_failed", "下载 MinerU 解析结果失败")
    return response.content


def _mineru_headers() -> dict[str, str]:
    if not settings.mineru_token:
        raise APIError(503, "mineru_token_missing", "未配置 MinerU API Token")
    return {
        "Authorization": f"Bearer {settings.mineru_token}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }


def _mineru_response_data(
    response: httpx.Response,
    default_message: str,
) -> dict[str, Any]:
    if response.is_error:
        raise APIError(502, "mineru_http_error", f"{default_message}：HTTP {response.status_code}")
    try:
        payload = response.json()
    except ValueError as exc:
        raise APIError(502, "mineru_invalid_response", f"{default_message}：响应不是 JSON") from exc
    if int(payload.get("code", -1)) != 0:
        raise APIError(
            502,
            "mineru_api_error",
            str(payload.get("msg") or default_message),
        )
    data = payload.get("data")
    return data if isinstance(data, dict) else {}


def _extract_markdown_from_zip(content: bytes) -> str:
    try:
        with zipfile.ZipFile(BytesIO(content)) as archive:
            markdown_names = [
                name for name in archive.namelist()
                if name.lower().endswith("full.md")
            ]
            if not markdown_names:
                markdown_names = [
                    name for name in archive.namelist()
                    if name.lower().endswith(".md")
                ]
            if not markdown_names:
                raise APIError(502, "mineru_empty_result", "解析结果包中没有 Markdown")
            return archive.read(markdown_names[0]).decode("utf-8", errors="replace")
    except zipfile.BadZipFile as exc:
        raise APIError(502, "mineru_invalid_result", "MinerU 返回的结果包无效") from exc


def _save_parsed_markdown(
    *,
    document_id: str,
    markdown: str,
    page_count: Any = None,
    row_count: Any = None,
) -> dict[str, Any]:
    if not markdown.strip():
        raise APIError(502, "mineru_empty_result", "MinerU 未返回可用文本")
    if not page_count:
        plain_text = _clean_markdown_text(markdown)
        page_count = max(1, math.ceil(len(plain_text) / 1400))
    output_dir = settings.upload_path / "parsed" / document_id
    output_dir.mkdir(parents=True, exist_ok=True)
    markdown_path = output_dir / "full.md"
    markdown_path.write_text(markdown, encoding="utf-8")
    return {
        "markdown": markdown,
        "markdown_path": markdown_path,
        "page_count": int(page_count) if page_count else None,
        "row_count": int(row_count) if row_count else None,
    }


def inspect_source_metrics(source: Path) -> dict[str, int | None]:
    extension = source.suffix.lower()
    try:
        if extension == ".pdf":
            from pypdf import PdfReader

            return {
                "page_count": len(PdfReader(str(source)).pages),
                "row_count": None,
            }
        if extension == ".xlsx":
            from openpyxl import load_workbook

            workbook = load_workbook(source, read_only=True, data_only=True)
            row_count = sum(max(sheet.max_row - 1, 0) for sheet in workbook.worksheets)
            workbook.close()
            return {"page_count": None, "row_count": row_count}
        if extension == ".csv":
            with source.open("r", encoding="utf-8-sig", errors="replace") as file:
                row_count = max(sum(1 for _ in file) - 1, 0)
            return {"page_count": None, "row_count": row_count}
        if extension == ".docx":
            from docx import Document

            document = Document(str(source))
            text_length = sum(len(paragraph.text) for paragraph in document.paragraphs)
            explicit_breaks = sum(
                1
                for paragraph in document.paragraphs
                for _ in paragraph._p.xpath('.//w:br[@w:type="page"]')
            )
            estimated_pages = max(explicit_breaks + 1, math.ceil(text_length / 1400))
            return {"page_count": estimated_pages, "row_count": None}
    except Exception:
        return {"page_count": None, "row_count": None}
    return {"page_count": None, "row_count": None}


async def extract_document_fields(
    *,
    document: BusinessDocument,
    template: DocumentExtractionTemplate,
    parsed_markdown: str,
) -> dict[str, Any]:
    field_definitions = template_field_definitions(template)
    if (
        template.code == "safety_monitoring_system_ledger"
        and Path(document.file_path).suffix.lower() == ".xlsx"
    ):
        deterministic = extract_safety_monitoring_ledger(document.file_path)
        if deterministic is not None:
            return normalize_extraction_result(
                deterministic,
                field_definitions,
                model_status="deterministic_excel",
                parsed_markdown=parsed_markdown,
                template_code=template.code,
            )
    if (
        template.code == "dust_dispersion_free_silica_monitoring_ledger"
        and Path(document.file_path).suffix.lower() == ".xlsx"
    ):
        deterministic = extract_dust_dispersion_free_silica_plan(document.file_path)
        if deterministic is not None:
            return normalize_extraction_result(
                deterministic,
                field_definitions,
                model_status="deterministic_excel",
                parsed_markdown=parsed_markdown,
                template_code=template.code,
            )
    runtime = get_agent_runtime()
    if runtime.enabled:
        prompt = build_extraction_prompt(
            document=document,
            template=template,
            parsed_markdown=parsed_markdown[: settings.document_extraction_max_chars],
            field_definitions=field_definitions,
        )
        agent_result = await runtime.invoke(
            message=prompt,
            thread_id=f"document-extraction-{document.id}",
            context={
                "document_id": document.id,
                "document_type": template.code,
                "template_code": template.code,
            },
        )
        if agent_result.status == "success":
            parsed = parse_model_json(agent_result.output)
            if parsed is not None:
                return normalize_extraction_result(
                    parsed,
                    field_definitions,
                    model_status="success",
                    parsed_markdown=parsed_markdown,
                    template_code=template.code,
                )
        fallback = fallback_extraction(parsed_markdown, field_definitions)
        return normalize_extraction_result(
            fallback,
            field_definitions,
            model_status=f"degraded:{agent_result.status}",
            parsed_markdown=parsed_markdown,
            template_code=template.code,
        )

    fallback = fallback_extraction(parsed_markdown, field_definitions)
    return normalize_extraction_result(
        fallback,
        field_definitions,
        model_status="disabled",
        parsed_markdown=parsed_markdown,
        template_code=template.code,
    )


def build_extraction_prompt(
    *,
    document: BusinessDocument,
    template: DocumentExtractionTemplate,
    parsed_markdown: str,
    field_definitions: list[dict[str, Any]],
) -> str:
    schema = json.dumps(field_definitions, ensure_ascii=False, indent=2)
    return f"""
{template.prompt}

资料信息：
- 标题：{document.title}
- 分类：{document.category}
- 业务资料类型：{template.name}

字段定义：
{schema}

请只输出一个 JSON 对象，不要输出 Markdown 代码块或其他说明。格式：
{{
  "summary": "资料摘要",
  "fields": [
    {{
      "field_key": "字段 key",
      "field_name": "字段名称",
      "field_value": "原始值，列表或对象转成 JSON",
      "normalized_value": "标准化值",
      "value_type": "text/number/date/object/list",
      "unit": "",
      "confidence": 0.95,
      "page_no": null,
      "evidence_text": "原文证据"
    }}
  ]
}}

解析文本：
{parsed_markdown}
""".strip()


def parse_model_json(output: str) -> dict[str, Any] | None:
    text = output.strip()
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*", "", text)
        text = re.sub(r"\s*```$", "", text)
    start = text.find("{")
    end = text.rfind("}")
    if start < 0 or end <= start:
        return None
    try:
        value = json.loads(text[start : end + 1])
    except json.JSONDecodeError:
        return None
    return value if isinstance(value, dict) else None


def normalize_extraction_result(
    result: dict[str, Any],
    field_definitions: list[dict[str, Any]],
    *,
    model_status: str,
    parsed_markdown: str,
    template_code: str = "",
) -> dict[str, Any]:
    by_key = {
        str(item.get("field_key") or item.get("key") or ""): item
        for item in result.get("fields", [])
        if isinstance(item, dict)
    }
    normalized_fields: list[dict[str, Any]] = []
    for definition in field_definitions:
        key = str(definition.get("key") or "")
        item = by_key.get(key, {})
        item_field_name = str(item.get("field_name") or "").strip()
        definition_field_name = str(definition.get("name") or key)
        if (
            not item_field_name
            or item_field_name == key
            or re.fullmatch(r"[a-zA-Z0-9_]+", item_field_name)
        ):
            field_name = definition_field_name
        else:
            field_name = item_field_name
        raw_value = item.get(
            "field_value",
            item.get("value", item.get("normalized_value")),
        )
        normalized_value = item.get("normalized_value", raw_value)
        normalized_fields.append(
            {
                "field_key": key,
                "field_name": field_name,
                "field_value": _stringify_value(raw_value),
                "normalized_value": _stringify_value(normalized_value),
                "value_type": str(
                    item.get("value_type") or definition.get("type") or "text"
                ),
                "unit": str(item.get("unit") or ""),
                "confidence": _normalize_confidence(item.get("confidence")),
                "page_no": item.get("page_no"),
                "evidence_text": str(item.get("evidence_text") or "")[:2000],
                "metadata_json": {
                    "model_status": model_status,
                    "group_key": definition.get("group_key", ""),
                    "group_name": definition.get("group_name", ""),
                    "group_order": definition.get("group_order", 999),
                    "field_order": definition.get("field_order", 999),
                },
            }
        )
    summary = str(result.get("summary") or "").strip()
    field_map = {
        item["field_key"]: item["normalized_value"] or item["field_value"]
        for item in normalized_fields
    }
    summary = _build_structured_summary(template_code, field_map) or summary
    if not summary:
        summary = _clean_markdown_text(parsed_markdown)[:500]
    summary = _compact_analysis_summary(summary)
    return {
        "summary": summary,
        "fields": normalized_fields,
        "model_status": model_status,
    }


def fallback_extraction(
    parsed_markdown: str,
    field_definitions: list[dict[str, Any]],
) -> dict[str, Any]:
    fields = []
    for definition in field_definitions:
        name = str(definition.get("name") or definition.get("key") or "")
        key = str(definition.get("key") or "")
        aliases = [str(item) for item in definition.get("aliases", [])]
        value = _extract_labeled_value(parsed_markdown, name, key, *aliases)
        fields.append(
            {
                "field_key": key,
                "field_name": name,
                "field_value": value,
                "normalized_value": value,
                "value_type": definition.get("type", "text"),
                "confidence": 0,
                "evidence_text": "",
            }
        )
    return {
        "summary": _clean_markdown_text(parsed_markdown)[:500],
        "fields": fields,
    }


def replace_document_extractions(
    db: Session,
    *,
    document_id: str,
    fields: list[dict[str, Any]],
    model_status: str,
) -> None:
    document_repo.delete_extractions(db, document_id)
    for item in fields:
        db.add(
            DocumentExtraction(
                document_id=document_id,
                field_key=str(item["field_key"]),
                field_name=str(item["field_name"]),
                field_value=str(item.get("field_value") or ""),
                normalized_value=str(item.get("normalized_value") or ""),
                value_type=str(item.get("value_type") or "text"),
                unit=str(item.get("unit") or ""),
                confidence=item.get("confidence"),
                page_no=item.get("page_no"),
                evidence_text=str(item.get("evidence_text") or ""),
                metadata_json={
                    **(item.get("metadata_json") or {}),
                    "model_status": model_status,
                },
            )
        )
    db.commit()


def _stringify_value(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, (dict, list)):
        return json.dumps(value, ensure_ascii=False)
    return str(value)


def _normalize_confidence(value: Any) -> float | None:
    if value is None:
        return None
    try:
        number = float(value)
    except (TypeError, ValueError):
        return None
    if 0 <= number <= 1:
        number *= 100
    return max(0.0, min(number, 100.0))


def _extract_labeled_value(text: str, *labels: str) -> str:
    for label in labels:
        if not label:
            continue
        pattern = rf"{re.escape(label)}\s*[:：]\s*([^\n]+)"
        match = re.search(pattern, text, flags=re.IGNORECASE)
        if match:
            return match.group(1).strip()
    return ""


def _clean_markdown_text(text: str) -> str:
    cleaned = re.sub(r"!\[[^\]]*\]\([^)]+\)", " ", text)
    cleaned = re.sub(r"<[^>]+>", " ", cleaned)
    cleaned = html.unescape(cleaned)
    cleaned = re.sub(r"[ \t\r\f\v]+", " ", cleaned)
    cleaned = re.sub(r"\n{2,}", "\n", cleaned)
    return cleaned.strip()


def extract_dust_dispersion_free_silica_plan(file_path: str) -> dict[str, Any] | None:
    try:
        from openpyxl import load_workbook

        workbook = load_workbook(file_path, read_only=True, data_only=True)
    except Exception:
        return None

    def cell_text(value: Any) -> str:
        if value is None:
            return ""
        if isinstance(value, datetime):
            return value.date().isoformat()
        if hasattr(value, "isoformat") and not isinstance(value, str):
            try:
                return value.isoformat()
            except Exception:
                pass
        return str(value).replace("\r", "").strip()

    try:
        sheet = workbook.worksheets[0]
        rows = [
            [cell_text(value) for value in row]
            for row in sheet.iter_rows(values_only=True)
        ]
        if not rows:
            return None

        aliases = {
            "sequence_no": {"序号"},
            "mine_name": {"矿井名称", "矿井"},
            "sampling_period": {"现场采样时间", "采样时间"},
            "detection_locations": {"检测地点", "监测地点"},
            "detection_items": {"检测项目", "监测项目"},
            "frequency_cycle": {"频次/周期", "频次／周期", "检测频次"},
            "primary_contact": {"联系人及联系方式", "联系人"},
            "remarks": {"备注"},
            "planned_detection_time": {
                "工作面矿方预计检测时间",
                "预计检测时间",
                "计划检测时间",
            },
            "actual_detection_date": {
                "商定后实际检测日期",
                "实际检测日期",
                "实际检测时间",
            },
        }
        header_row_index = -1
        header_map: dict[str, int] = {}
        for row_index, row in enumerate(rows[:12]):
            row_map: dict[str, int] = {}
            for column_index, value in enumerate(row):
                normalized_value = value.replace("*", "").replace("（", "(").replace("）", ")")
                for key, labels in aliases.items():
                    if any(label in normalized_value for label in labels):
                        row_map.setdefault(key, column_index)
            if {"mine_name", "detection_locations", "detection_items"}.issubset(row_map):
                header_row_index = row_index
                header_map = row_map
                break

        if header_row_index < 0:
            return None

        def value_at(row: list[str], key: str) -> str:
            index = header_map.get(key)
            return row[index] if index is not None and index < len(row) else ""

        contact_index = header_map.get("primary_contact")
        phone_index = contact_index + 1 if contact_index is not None else None

        def phone_info_at(row: list[str]) -> tuple[str, str]:
            if phone_index is None or phone_index >= len(row):
                return "", ""
            value = row[phone_index].strip()
            phones = re.findall(r"1[3-9]\d{9}", value)
            name = re.sub(r"1[3-9]\d{9}", "", value)
            name = re.sub(r"[\s:：,，、;；]+", " ", name).strip()
            return name, "、".join(phones)

        def phone_at(row: list[str]) -> str:
            return phone_info_at(row)[1]

        records: list[dict[str, Any]] = []
        for row_index in range(header_row_index + 1, len(rows)):
            row = rows[row_index]
            mine_name = value_at(row, "mine_name")
            if not mine_name or "现场检测联系人" in mine_name:
                continue
            detection_items = [
                item.strip()
                for item in re.split(r"[\n\r、,，;；]+", value_at(row, "detection_items"))
                if item.strip()
            ]
            detection_locations = value_at(row, "detection_locations")
            if not detection_items and not detection_locations:
                continue

            onsite_contact = ""
            onsite_phone = ""
            if row_index + 1 < len(rows):
                next_row = rows[row_index + 1]
                next_contact = value_at(next_row, "primary_contact")
                if "现场检测联系人" in next_contact:
                    phone_name, onsite_phone = phone_info_at(next_row)
                    label_name = re.sub(
                        r"^现场检测联系人\s*[:：]?\s*",
                        "",
                        next_contact,
                    ).strip()
                    onsite_contact = label_name or phone_name

            primary_contact = value_at(row, "primary_contact")
            records.append(
                {
                    "sequence_no": value_at(row, "sequence_no"),
                    "mine_name": mine_name,
                    "sampling_period": value_at(row, "sampling_period"),
                    "detection_locations": detection_locations,
                    "detection_items": detection_items,
                    "frequency_cycle": value_at(row, "frequency_cycle"),
                    "primary_contact": primary_contact,
                    "primary_phone": phone_at(row),
                    "onsite_contact": onsite_contact,
                    "onsite_phone": onsite_phone,
                    "remarks": value_at(row, "remarks"),
                    "planned_detection_time": value_at(row, "planned_detection_time"),
                    "actual_detection_date": value_at(row, "actual_detection_date"),
                }
            )

        if not records:
            return None

        mine_names = list(dict.fromkeys(record["mine_name"] for record in records))
        monitoring_items: list[str] = []
        frequency_cycles: list[str] = []
        coverage_requirements: list[str] = []
        primary_contacts: list[dict[str, str]] = []
        onsite_contacts: list[dict[str, str]] = []
        contact_phone_map: dict[str, str] = {}
        for record in records:
            for item in record["detection_items"]:
                if item not in monitoring_items:
                    monitoring_items.append(item)
            frequency = record["frequency_cycle"]
            if frequency and frequency not in frequency_cycles:
                frequency_cycles.append(frequency)
            remarks = record["remarks"]
            if remarks and remarks not in coverage_requirements:
                coverage_requirements.append(remarks)
            if record["primary_contact"]:
                primary_contacts.append(
                    {
                        "mine_name": record["mine_name"],
                        "contact_name": record["primary_contact"],
                        "phone": record["primary_phone"],
                    }
                )
                if record["primary_phone"]:
                    contact_phone_map[record["primary_contact"]] = record["primary_phone"]
            if record["onsite_contact"]:
                onsite_contacts.append(
                    {
                        "mine_name": record["mine_name"],
                        "contact_name": record["onsite_contact"],
                        "phone": record["onsite_phone"],
                    }
                )
                if record["onsite_phone"]:
                    contact_phone_map[record["onsite_contact"]] = record["onsite_phone"]

        planned_dates = [
            {
                "mine_name": record["mine_name"],
                "planned_detection_time": record["planned_detection_time"],
            }
            for record in records
            if record["planned_detection_time"]
        ]
        actual_dates = [
            {
                "mine_name": record["mine_name"],
                "actual_detection_date": record["actual_detection_date"],
            }
            for record in records
            if record["actual_detection_date"]
        ]
        missing_planned_dates = [
            record["mine_name"] for record in records if not record["planned_detection_time"]
        ]
        missing_actual_dates = [
            record["mine_name"] for record in records if not record["actual_detection_date"]
        ]
        title = next(
            (
                row[0]
                for row in rows[:header_row_index + 1]
                if row and ("粉尘" in row[0] or "监测" in row[0])
            ),
            "采掘作业场所粉尘分散度、游离二氧化硅含量监测计划表",
        )
        title = title.replace("\n", "")
        year_match = re.search(r"(20\d{2})", title)
        monitoring_year = year_match.group(1) if year_match else ""
        missing_items: list[str] = []
        if missing_planned_dates:
            missing_items.append("部分矿井未填写计划检测时间")
        if missing_actual_dates:
            missing_items.append("部分矿井未填写实际检测日期")
        conclusion = (
            f"计划表覆盖 {len(mine_names)} 个矿井、{len(records)} 条监测计划，"
            f"检测项目包括{'、'.join(monitoring_items) or '未识别'}；"
            "计划和实际检测日期仍需按执行情况补充。"
        )
        field_values: dict[str, Any] = {
            "monitoring_title": title,
            "monitoring_year": monitoring_year,
            "document_kind": "监测计划表" if "计划" in title else "监测台账",
            "applicable_scope": "采掘作业场所粉尘分散度、游离二氧化硅含量监测",
            "sheet_names": list(workbook.sheetnames),
            "mine_count": len(mine_names),
            "plan_record_count": len(records),
            "mine_names": mine_names,
            "monitoring_items": monitoring_items,
            "frequency_cycle": frequency_cycles,
            "coverage_requirements": coverage_requirements,
            "mine_plan_records": records,
            "planned_detection_dates": planned_dates,
            "actual_detection_dates": actual_dates,
            "missing_planned_dates": missing_planned_dates,
            "missing_actual_dates": missing_actual_dates,
            "primary_contacts": primary_contacts,
            "onsite_contacts": onsite_contacts,
            "contact_phone_map": contact_phone_map,
            "plan_complete": "是" if not missing_items else "否",
            "missing_items": missing_items,
            "overall_conclusion": conclusion,
        }
        return {
            "summary": "",
            "fields": [
                {
                    "field_key": key,
                    "field_name": key,
                    "field_value": value,
                    "normalized_value": value,
                }
                for key, value in field_values.items()
            ],
        }
    finally:
        workbook.close()


def extract_safety_monitoring_ledger(file_path: str) -> dict[str, Any] | None:
    try:
        from openpyxl import load_workbook

        workbook = load_workbook(file_path, read_only=True, data_only=True)
    except Exception:
        return None

    try:
        sheet_names = list(workbook.sheetnames)
        total_sheet = workbook["总台账"] if "总台账" in workbook.sheetnames else workbook.worksheets[0]
        header_map: dict[str, int] = {}
        header_row = 0
        for row_index, row in enumerate(
            total_sheet.iter_rows(min_row=1, max_row=20, values_only=True),
            start=1,
        ):
            values = [str(value).strip() if value is not None else "" for value in row]
            if "编码" in values and "设备名称" in values:
                header_row = row_index
                header_map = {value: index for index, value in enumerate(values) if value}
                break

        if not header_map:
            return None

        code_index = header_map.get("编码")
        device_index = header_map.get("设备名称")
        alarm_index = header_map.get("报警值")
        cut_index = header_map.get("断电值")
        restore_index = header_map.get("复电值")
        location_index = header_map.get("安装地点")
        position_index = header_map.get("安装位置")
        unit_index = next(
            (index for label, index in header_map.items() if label.startswith("单位")),
            None,
        )
        range_index = header_map.get("断电范围")

        substations: set[str] = set()
        substation_details: list[dict[str, Any]] = []
        sensor_type_counts: dict[str, int] = {}
        critical_details: list[dict[str, Any]] = []
        thresholds = {
            "gas_alarm_thresholds": set(),
            "co_alarm_thresholds": set(),
            "temperature_alarm_thresholds": set(),
            "airflow_alarm_thresholds": set(),
        }
        power_cut_rules: set[str] = set()
        power_restore_rules: set[str] = set()
        sensor_count = 0

        for row in total_sheet.iter_rows(
            min_row=header_row + 1,
            max_row=min(total_sheet.max_row, 20000),
            values_only=True,
        ):
            values = ["" if value is None else str(value).strip() for value in row]

            code = values[code_index] if code_index is not None and code_index < len(values) else ""
            device = values[device_index] if device_index is not None and device_index < len(values) else ""
            location = values[location_index] if location_index is not None and location_index < len(values) else ""
            position = values[position_index] if position_index is not None and position_index < len(values) else ""
            unit = values[unit_index] if unit_index is not None and unit_index < len(values) else ""
            alarm = values[alarm_index] if alarm_index is not None and alarm_index < len(values) else ""
            cut = values[cut_index] if cut_index is not None and cut_index < len(values) else ""
            restore = values[restore_index] if restore_index is not None and restore_index < len(values) else ""
            power_range = values[range_index] if range_index is not None and range_index < len(values) else ""

            if code:
                substations.add(code)
                if len(substation_details) < 100:
                    substation_details.append(
                        {
                            "code": code,
                            "location": location,
                            "device": device,
                        }
                    )
            if not device or device == "综合分站":
                continue

            sensor_count += 1
            sensor_type = _normalize_sensor_type(device)
            sensor_type_counts[sensor_type] = sensor_type_counts.get(sensor_type, 0) + 1
            if device in {"激光甲烷", "甲烷"} and alarm:
                thresholds["gas_alarm_thresholds"].add(alarm)
            elif device == "一氧化碳" and alarm:
                thresholds["co_alarm_thresholds"].add(alarm)
            elif device == "温度" and alarm:
                thresholds["temperature_alarm_thresholds"].add(alarm)
            elif device in {"风速", "风量"} and alarm:
                thresholds["airflow_alarm_thresholds"].add(alarm)
            if cut:
                power_cut_rules.add(cut)
            if restore:
                power_restore_rules.add(restore)

            if (
                len(critical_details) < 200
                and device
                in {
                    "激光甲烷",
                    "甲烷",
                    "一氧化碳",
                    "温度",
                    "风速",
                    "风量",
                    "烟雾",
                    "风门",
                    "局扇",
                    "主扇",
                    "风筒",
                }
            ):
                critical_details.append(
                    {
                        "code": code,
                        "device": device,
                        "location": location,
                        "position": position,
                        "unit": unit,
                        "alarm": alarm,
                        "cut": cut,
                        "restore": restore,
                        "power_range": power_range,
                    }
                )

        title = "安全监控系统台账"
        department = ""
        if "台账封面" in workbook.sheetnames:
            cover = workbook["台账封面"]
            cover_values = [
                str(value).strip()
                for row in cover.iter_rows(min_row=1, max_row=40, values_only=True)
                for value in row
                if value is not None and str(value).strip()
            ]
            for value in cover_values:
                if "台账" in value and len(value) > len(title):
                    title = value
                if "通风队" in value:
                    department = value

        version_match = re.search(r"(20\d{4})", Path(file_path).stem)
        ledger_version = version_match.group(1) if version_match else ""
        field_values: dict[str, Any] = {
            "ledger_title": title,
            "mine_name": "上湾煤矿" if "上湾" in Path(file_path).stem else "",
            "department": department,
            "ledger_version": ledger_version,
            "sheet_names": sheet_names,
            "data_source": "安全监控系统",
            "substation_count": len(substations),
            "sensor_count": sensor_count,
            "substation_details": substation_details,
            "sensor_type_counts": sensor_type_counts,
            "monitored_parameters": sorted(sensor_type_counts),
            "sensor_fields": [
                "编码",
                "安装地点",
                "设备名称",
                "安装位置",
                "单位",
                "报警值",
                "断电值",
                "复电值",
                "断电范围",
            ],
            "critical_sensor_details": critical_details,
            "gas_sensor_count": sensor_type_counts.get("甲烷", 0),
            "co_sensor_count": sensor_type_counts.get("一氧化碳", 0),
            "airflow_sensor_count": sensor_type_counts.get("风速", 0),
            "temperature_sensor_count": sensor_type_counts.get("温度", 0),
            "smoke_sensor_count": sensor_type_counts.get("烟雾", 0),
            "door_sensor_count": sensor_type_counts.get("风门", 0),
            "fan_sensor_count": sensor_type_counts.get("局扇", 0) + sensor_type_counts.get("主扇", 0),
            "gas_alarm_thresholds": sorted(thresholds["gas_alarm_thresholds"]),
            "co_alarm_thresholds": sorted(thresholds["co_alarm_thresholds"]),
            "temperature_alarm_thresholds": sorted(thresholds["temperature_alarm_thresholds"]),
            "airflow_alarm_thresholds": sorted(thresholds["airflow_alarm_thresholds"]),
            "power_cut_rules": sorted(power_cut_rules),
            "power_restore_rules": sorted(power_restore_rules),
            "ledger_complete": "是" if substations and sensor_count else "否",
            "missing_items": [],
            "abnormal_settings": [],
            "overall_conclusion": (
                f"台账包含 {len(substations)} 个分站、{sensor_count} 个传感器或测点，"
                "字段完整，可用于安全监控系统台账管理。"
            ),
        }
        return {
            "summary": "",
            "fields": [
                {
                    "field_key": key,
                    "field_name": key,
                    "field_value": value,
                    "normalized_value": value,
                }
                for key, value in field_values.items()
            ],
        }
    finally:
        workbook.close()


def _normalize_sensor_type(device: str) -> str:
    if device in {"激光甲烷", "甲烷"}:
        return "甲烷"
    if device in {"风速", "风量"}:
        return "风速"
    return device


def _compact_analysis_summary(text: str) -> str:
    normalized = re.sub(r"\s+", " ", text).strip()
    blocks = [
        block.strip()
        for block in re.split(
            r"(?<![\d.])(?=(?:【|\[)?(?:[1-9]|1[0-9]|20)[.、．](?=\s|[\u4e00-\u9fff]))",
            normalized,
        )
        if re.match(
            r"^(?:【|\[)?(?:[1-9]|1[0-9]|20)[.、．](?=\s|[\u4e00-\u9fff])",
            block.strip(),
        )
    ]
    if len(blocks) >= 3:
        return "\n".join(block[:80] for block in blocks[:8])
    return normalized[:500]


def _build_structured_summary(
    template_code: str,
    fields: dict[str, str],
) -> str:
    get = lambda key: fields.get(key) or "未提供"
    if template_code == "main_fan_external_leakage_rate_report":
        return "\n".join(
            [
                f"1. 南风井：漏风率 {get('south_leakage_rate')}%，{_limit_status(get('south_leakage_rate'))}。",
                f"2. 北风井：漏风率 {get('north_leakage_rate')}%，{_limit_status(get('north_leakage_rate'))}。",
                f"3. 最大漏风率：{get('maximum_leakage_rate')}%，标准限值 {get('standard_limit')}%。",
                f"4. 总体结论：{get('compliance_status')}。",
            ]
        )
    if template_code == "monthly_air_distribution_plan":
        return "\n".join(
            [
                f"1. 北风井：需风量 {get('north_required_airflow')}，可用回风量 {get('north_available_airflow')}，{get('north_capacity_status')}。",
                f"2. 南风井：需风量 {get('south_required_airflow')}，可用回风量 {get('south_available_airflow')}，{get('south_capacity_status')}。",
                f"3. 全矿井：需风量 {get('mine_required_airflow')}，总回风量 {get('mine_available_airflow')}，{get('mine_capacity_status')}。",
            ]
        )
    if template_code == "mine_ventilation_resistance_report":
        return "\n".join(
            [
                f"1. 主要路线：{get('main_route_measured_resistance')}，相对误差 {get('main_route_relative_error')}%。",
                f"2. 次要路线：{get('secondary_route_measured_resistance')}，相对误差 {get('secondary_route_relative_error')}%。",
                f"3. 等积孔：{get('equivalent_orifice')}，通风难易程度：{get('ventilation_difficulty')}。",
                f"4. 有效风量率：{get('effective_airflow_rate')}%，{get('effective_airflow_rate_pass')}。",
                f"5. 总体评价：{get('overall_assessment')}。",
            ]
        )
    if template_code == "ventilation_capacity_verification_report":
        return "\n".join(
            [
                f"1. 核定通风能力：{get('final_capacity')} {get('capacity_unit')}。",
                f"2. 北风井：需风量 {get('north_required_airflow')}，实际回风量 {get('north_available_airflow')}，{get('north_capacity_status')}。",
                f"3. 南风井：需风量 {get('south_required_airflow')}，实际回风量 {get('south_available_airflow')}，{get('south_capacity_status')}。",
                f"4. 全矿井：需风量 {get('mine_required_airflow')}，实际总回风量 {get('mine_available_airflow')}，{get('mine_capacity_status')}。",
                f"5. 有效风量率：{get('effective_airflow_rate')}%。",
            ]
        )
    if template_code == "airflow_measurement_decadal_report":
        rate = get("effective_airflow_rate")
        rate_status = (
            "合格"
            if _is_number(rate) and float(rate) >= 85
            else "不合格" if _is_number(rate) else "需核验"
        )
        return "\n".join(
            [
                f"1. 测风日期：{get('measurement_date')}，矿井：{get('mine_name')}。",
                f"2. 北风井：总回风量 {get('north_return_airflow')}，总排风量 {get('north_exhaust_airflow')}，负压 {get('north_negative_pressure')}。",
                f"3. 南风井：总回风量 {get('south_return_airflow')}，总排风量 {get('south_exhaust_airflow')}，负压 {get('south_negative_pressure')}。",
                f"4. 全矿井：总回风量 {get('mine_total_return_airflow')}，总进风量 {get('mine_total_intake_airflow')}，有效风量率 {rate}%，{rate_status}。",
                f"5. 测点数量：{get('measurement_point_count')}，风量不足测点：{get('insufficient_airflow_points')}，瓦斯异常测点：{get('gas_anomaly_points')}。",
            ]
        )
    if template_code == "gas_grade_appraisal_report":
        return "\n".join(
            [
                f"1. 瓦斯等级：{get('gas_grade')}，上年度：{get('previous_gas_grade')}。",
                f"2. 绝对瓦斯涌出量：{get('absolute_gas_emission')}，相对瓦斯涌出量：{get('relative_gas_emission')}。",
                f"3. 二氧化碳绝对涌出量：{get('absolute_co2_emission')}，相对涌出量：{get('relative_co2_emission')}。",
                f"4. 采煤面最大绝对瓦斯涌出量：{get('coalface_max_gas_emission')}，掘进面最大绝对瓦斯涌出量：{get('heading_max_gas_emission')}。",
                f"5. 瓦斯突出：{get('gas_outburst_status')}，瓦斯喷出：{get('gas_blowout_status')}，结论：{get('gas_grade_conclusion')}。",
            ]
        )
    if template_code == "coal_dust_explosion_spontaneous_combustion_report":
        validity_status = _expiry_status(get("valid_until"))
        return "\n".join(
            [
                f"1. 煤尘爆炸性：{get('dust_explosion_result')}，火焰长度 {get('flame_length')}mm，最低岩粉用量 {get('minimum_rock_dust')}%。",
                f"2. 自燃倾向性：{get('spontaneous_combustion_grade')}。",
                f"3. 关键指标：Vdaf {get('volatile_daf')}%，吸氧量 {get('oxygen_adsorption')}cm³/g。",
                f"4. 有效期至：{get('valid_until')}，状态：{validity_status}。",
                f"5. 总体结论：{get('inspection_conclusion')}。",
            ]
        )
    if template_code == "dust_dispersion_free_silica_monitoring_ledger":
        return "\n".join(
            [
                f"1. 监测年度：{get('monitoring_year')}，资料性质：{get('document_kind')}。",
                f"2. 覆盖范围：{get('mine_count')} 个矿井，{get('plan_record_count')} 条计划，矿井包括 {get('mine_names')}。",
                f"3. 检测项目：{get('monitoring_items')}，频次：{get('frequency_cycle')}。",
                f"4. 时间完整性：计划检测日期缺失 {get('missing_planned_dates')}；实际检测日期缺失 {get('missing_actual_dates')}。",
                f"5. 总体结论：{get('overall_conclusion')}",
            ]
        )
    if template_code == "vehicle_air_door_inspection_record":
        return "\n".join(
            [
                f"1. 记录年度：{get('record_year')}，检查月份：{get('inspection_month')}，矿井：{get('mine_name')}。",
                f"2. 风门台账：{get('air_door_record_count')} 条记录，共 {get('total_door_sets')} 套；自动风门 {get('automatic_door_count')} 套，手动风门 {get('manual_door_count')} 套。",
                f"3. 风门规格：{get('common_specifications')}；停用或常开风门：{get('inactive_door_count')} 套。",
                f"4. 检查情况：检查 {get('inspected_door_count')} 处，发现问题 {get('problem_door_count')} 处，已整改 {get('rectified_count')} 处，未整改 {get('unrectified_count')} 处，整改率 {get('rectification_rate')}%。",
                f"5. 主要问题：{get('problem_types')}；总体结论：{get('overall_conclusion')}",
            ]
        )
    if template_code == "safety_monitoring_system_ledger":
        return "\n".join(
            [
                f"1. 台账：{get('ledger_title')}，版本：{get('ledger_version')}。",
                f"2. 分站数量：{get('substation_count')}，传感器或测点数量：{get('sensor_count')}。",
                f"3. 传感器类型：{get('sensor_type_counts')}。",
                f"4. 报警与断电：甲烷 {get('gas_alarm_thresholds')}，一氧化碳 {get('co_alarm_thresholds')}，温度 {get('temperature_alarm_thresholds')}。",
                f"5. 完整性：{get('ledger_complete')}，结论：{get('overall_conclusion')}。",
            ]
        )
    return ""


def _is_number(value: str) -> bool:
    try:
        float(value)
    except (TypeError, ValueError):
        return False
    return True


def _limit_status(value: str, limit: float = 5.0) -> str:
    if not _is_number(value):
        return "需核验"
    return "符合" if float(value) <= limit else "超限告警"


def _expiry_status(value: str) -> str:
    if not value or value == "未提供":
        return "未提供"
    try:
        expiry = datetime.strptime(value[:10], "%Y-%m-%d").date()
    except ValueError:
        return "需人工核验"
    return "已过期" if expiry < datetime.now(timezone.utc).date() else "有效"