Blame for apps/documents/services/external_template/analysis_service.py

"""
外部模板分析服务

负责模板上传校验、文件保存、版本管理。
结构提取和 LLM 分析方法将在后续任务 (6.2, 6.3) 中实现。

Requirements: 1.1, 1.2, 1.3, 1.6, 1.7, 1.8, 9.1, 9.2, 9.3, 9.4
"""

from __future__ import annotations

import json
import logging
import re
import uuid
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar
from xml.etree import ElementTree as ET

from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import transaction
from django.utils.translation import gettext_lazy as _

from .fingerprint_service import FingerprintService

if TYPE_CHECKING:
    from django.core.files.uploadedfile import UploadedFile

    from apps.documents.models.external_template import ExternalTemplate
    from apps.documents.services.placeholders.registry import PlaceholderRegistry

logger: logging.Logger = logging.getLogger(__name__)


class AnalysisService:
    """外部模板分析服务：上传校验 + 结构提取 + LLM 字段映射"""

    MAX_FILE_SIZE: ClassVar[int] = 20 * 1024 * 1024  # 20MB

    def __init__(
        self,
        fingerprint_service: FingerprintService,
        llm_service: Any,
        placeholder_registry: PlaceholderRegistry,
    ) -> None:
        self._fingerprint_service = fingerprint_service
        self._llm_service = llm_service
        self._placeholder_registry = placeholder_registry

    # ------------------------------------------------------------------
    # 上传与校验
    # ------------------------------------------------------------------

    def upload_template(
        self,
        file: UploadedFile,
        name: str,
        source_name: str,
        uploaded_by: Any,
    ) -> ExternalTemplate:
        """
        上传外部模板：
        1. 校验 .docx 格式、文件大小 ≤ 20MB
        2. 保存文件（UUID 重命名）
        3. 验证 python-docx 可解析
        4. 处理版本管理（同机构 + 类别组合自增版本号）
        5. 创建 ExternalTemplate 记录
        """
        from apps.documents.models.external_template import ExternalTemplate

        self._validate_file(file)

        law_firm_id: int = uploaded_by.law_firm_id
        abs_path, rel_path = self._save_file(file, law_firm_id)

        try:
            self._validate_parseable(abs_path)
        except ValidationError:
            # 解析失败时删除已保存的文件
            if abs_path.exists():
                abs_path.unlink()
                logger.info("已删除无法解析的文件: %s", abs_path)
            raise

        file_size: int = file.size if file.size is not None else 0
        original_filename: str = file.name if file.name else ""

        with transaction.atomic():
            version, deactivated = self._handle_versioning(
                law_firm_id=law_firm_id,
                source_name=source_name,
            )

            template: ExternalTemplate = ExternalTemplate.objects.create(
                name=name,
                source_name=source_name,
                file_path=rel_path,
                original_filename=original_filename,
                file_size=file_size,
                version=version,
                is_active=True,
                uploaded_by=uploaded_by,
                law_firm_id=law_firm_id,
            )

        logger.info(
            "模板上传成功: id=%d, name=%s, version=%d, deactivated=%d",
            template.id,
            template.name,
            version,
            deactivated,
        )
        return template

    def _validate_file(self, file: UploadedFile) -> None:
        """校验 .docx 格式和文件大小"""
        filename: str = file.name if file.name else ""
        if not filename.lower().endswith(".docx"):
            logger.info("文件格式校验失败: %s", filename)
            raise ValidationError(_("仅支持 .docx 格式"))

        file_size: int = file.size if file.size is not None else 0
        if file_size > self.MAX_FILE_SIZE:
            logger.info(
                "文件大小超出限制: %s, size=%d",
                filename,
                file_size,
            )
            raise ValidationError(_("文件大小超出限制"))

    def _validate_parseable(self, file_path: Path) -> None:
        """尝试 python-docx 打开文件，验证可解析性"""
        try:
            from docx import Document as DocxDocument

            DocxDocument(str(file_path))
        except Exception as exc:
            logger.info(
                "文件无法解析: %s, error=%s",
                file_path.name,
                str(exc),
            )
            raise ValidationError(_("文件无法解析，请检查文件是否损坏或加密")) from exc

    def _save_file(self, file: UploadedFile, law_firm_id: int) -> tuple[Path, str]:
        """
        UUID 重命名保存文件

        Returns:
            (绝对路径, 相对于 MEDIA_ROOT 的路径)
        """
        media_root = Path(settings.MEDIA_ROOT)
        rel_dir = Path("documents") / "external_templates" / str(law_firm_id)
        abs_dir = media_root / rel_dir
        abs_dir.mkdir(parents=True, exist_ok=True)

        new_filename = f"{uuid.uuid4()}.docx"
        rel_path = rel_dir / new_filename
        abs_path = media_root / rel_path

        with abs_path.open("wb") as dest:
            for chunk in file.chunks():
                dest.write(chunk)

        logger.info("文件已保存: %s", abs_path)
        return abs_path, str(rel_path)

    # ------------------------------------------------------------------
    # 版本管理
    # ------------------------------------------------------------------

    def _handle_versioning(
        self,
        *,
        law_firm_id: int,
        source_name: str,
    ) -> tuple[int, int]:
        """
        处理版本管理：同一来源自增版本号，旧版本 is_active=False

        Returns:
            (新版本号, 被停用的旧版本数量)
        """
        from apps.documents.models.external_template import ExternalTemplate

        existing = ExternalTemplate.objects.filter(
            law_firm_id=law_firm_id,
            source_name=source_name,
        )

        max_version: int | None = existing.order_by("-version").values_list("version", flat=True).first()

        new_version: int = (max_version or 0) + 1

        deactivated: int = existing.filter(is_active=True).update(is_active=False)

        return new_version, deactivated

    # ------------------------------------------------------------------
    # 结构提取 (Requirements: 2.1, 2.2, 2.3, 2.4, 2.5, 2.6, 2.7, 2.8)
    # ------------------------------------------------------------------

    def extract_structure(self, template_id: int) -> dict[str, Any]:
        """
        从 .docx 提取文档结构：段落、表格（含嵌套）、复选框、删除不适用项。

        1. 从数据库加载模板记录，获取 file_path
        2. 使用 python-docx 打开文件
        3. 调用 _extract_paragraphs / _extract_tables / _extract_checkboxes
        4. 将结构 JSON 保存到模板记录的 structure_json 字段
        5. 返回结构 dict
        """
        from apps.documents.models.external_template import ExternalTemplate

        template: ExternalTemplate = ExternalTemplate.objects.get(pk=template_id)
        abs_path = Path(settings.MEDIA_ROOT) / template.file_path

        from docx import Document as DocxDocument

        doc: Any = DocxDocument(str(abs_path))

        paragraphs = self._extract_paragraphs(doc)
        tables = self._extract_tables(doc)
        checkboxes = self._extract_checkboxes(doc)

        structure: dict[str, Any] = {
            "paragraphs": paragraphs,
            "tables": tables,
            "checkboxes": checkboxes,
        }

        template.structure_json = structure
        template.save(update_fields=["structure_json", "updated_at"])

        logger.info(
            "结构提取完成: template_id=%d, paragraphs=%d, tables=%d, checkboxes=%d",
            template_id,
            len(paragraphs),
            len(tables),
            len(checkboxes),
        )
        return structure

    def _extract_paragraphs(self, doc: Any) -> list[dict[str, Any]]:
        """
        提取段落文本和位置索引。

        对每个段落记录：paragraph_index、text、position_locator。
        同时检测"删除不适用项"模式。
        """
        paragraphs: list[dict[str, Any]] = []

        for idx, para in enumerate(doc.paragraphs):
            text: str = para.text.strip()
            if not text:
                continue

            entry: dict[str, Any] = {
                "paragraph_index": idx,
                "text": text,
                "position_locator": {
                    "type": "paragraph",
                    "paragraph_index": idx,
                },
            }

            options = self._detect_delete_inapplicable(text)
            if options is not None:
                entry["delete_inapplicable"] = options
                entry["position_locator"]["type"] = "delete_inapplicable"

            paragraphs.append(entry)

        return paragraphs

    def _extract_tables(self, doc: Any) -> list[dict[str, Any]]:
        """
        提取表格结构（递归嵌套、合并单元格）。

        对每个表格遍历行和单元格，处理合并单元格（通过 _tc 检测），
        递归处理嵌套表格。
        """
        tables: list[dict[str, Any]] = []

        for table_idx, table in enumerate(doc.tables):
            table_data = self._extract_single_table(table, table_idx, [])
            tables.append(table_data)

        return tables

    def _extract_single_table(
        self,
        table: Any,
        table_index: int,
        nested_path: list[int],
    ) -> dict[str, Any]:
        """递归提取单个表格的结构，通过 XML 正确处理合并单元格。"""
        ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
        rows_data: list[dict[str, Any]] = []
        tbl = table._tbl
        tr_elements = tbl.findall("w:tr", ns)

        for row_idx, tr in enumerate(tr_elements):
            cells_data: list[dict[str, Any]] = []
            col_cursor = 0

            for tc in tr.findall("w:tc", ns):
                tc_pr = tc.find("w:tcPr", ns)
                grid_span = 1
                is_v_merge_continue = False

                if tc_pr is not None:
                    gs = tc_pr.find("w:gridSpan", ns)
                    if gs is not None:
                        grid_span = int(gs.get(f"{{{ns['w']}}}val", "1"))
                    vm = tc_pr.find("w:vMerge", ns)
                    if vm is not None:
                        val = vm.get(f"{{{ns['w']}}}val", "continue")
                        is_v_merge_continue = val != "restart"

                if is_v_merge_continue:
                    col_cursor += grid_span
                    continue

                texts = [t.text or "" for t in tc.findall(".//w:t", ns)]
                text = "".join(texts).strip()

                cell_entry: dict[str, Any] = {
                    "row": row_idx,
                    "col": col_cursor,
                    "text": text,
                    "position_locator": {
                        "type": "table_cell",
                        "table_index": table_index,
                        "row": row_idx,
                        "col": col_cursor,
                        "nested_table_path": nested_path,
                    },
                }
                if grid_span > 1:
                    cell_entry["col_span"] = grid_span

                options = self._detect_delete_inapplicable(text)
                if options is not None:
                    cell_entry["delete_inapplicable"] = options
                    cell_entry["position_locator"]["type"] = "delete_inapplicable"

                # 递归处理嵌套表格（通过 python-docx cell 对象）
                # 找到对应的 python-docx cell
                if col_cursor < len(table.rows[row_idx].cells):
                    py_cell = table.rows[row_idx].cells[col_cursor]
                    nested_tables: list[dict[str, Any]] = []
                    for nested_idx, nested_table in enumerate(py_cell.tables):
                        nested_data = self._extract_single_table(
                            nested_table,
                            nested_idx,
                            [*nested_path, table_index],
                        )
                        nested_tables.append(nested_data)
                    if nested_tables:
                        cell_entry["nested_tables"] = nested_tables

                cells_data.append(cell_entry)
                col_cursor += grid_span

            rows_data.append({"row_index": row_idx, "cells": cells_data})

        return {
            "table_index": table_index,
            "nested_table_path": nested_path,
            "rows": rows_data,
        }

    def _extract_checkboxes(self, doc: Any) -> list[dict[str, Any]]:
        """
        识别复选框控件。

        解析文档 XML，查找 w14:checkbox 或 w:sdt 中包含复选框的结构化文档标签。
        """
        checkboxes: list[dict[str, Any]] = []
        ns: dict[str, str] = {
            "w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main",
            "w14": "http://schemas.microsoft.com/office/word/2010/wordml",
        }

        body_xml: str = doc.element.xml
        try:
            root: ET.Element = ET.fromstring(body_xml)
        except ET.ParseError:
            logger.info("复选框提取: XML 解析失败，跳过")
            return checkboxes

        # 查找所有 sdt（结构化文档标签）
        sdt_elements = root.iter(f"{{{ns['w']}}}sdt")
        checkbox_index = 0

        for sdt in sdt_elements:
            sdt_pr = sdt.find(f"{{{ns['w']}}}sdtPr", ns)
            if sdt_pr is None:
                continue

            # 检查是否为复选框：w14:checkbox 或 w:checkbox
            is_checkbox = (
                sdt_pr.find(f"{{{ns['w14']}}}checkbox", ns) is not None
                or sdt_pr.find(f"{{{ns['w']}}}checkbox", ns) is not None
            )
            if not is_checkbox:
                continue

            # 提取关联文本标签
            sdt_content = sdt.find(f"{{{ns['w']}}}sdtContent", ns)
            label_text = ""
            if sdt_content is not None:
                runs = sdt_content.iter(f"{{{ns['w']}}}t")
                label_parts: list[str] = [r.text for r in runs if r.text]
                label_text = "".join(label_parts)

            # 尝试获取 checked 状态
            checked = False
            checkbox_el = sdt_pr.find(f"{{{ns['w14']}}}checkbox", ns)
            if checkbox_el is None:
                checkbox_el = sdt_pr.find(f"{{{ns['w']}}}checkbox", ns)
            if checkbox_el is not None:
                checked_el = checkbox_el.find(f"{{{ns['w14']}}}checked", ns)
                if checked_el is None:
                    checked_el = checkbox_el.find(f"{{{ns['w']}}}checked", ns)
                if checked_el is not None:
                    val = checked_el.get(
                        f"{{{ns['w14']}}}val",
                        checked_el.get(f"{{{ns['w']}}}val", "0"),
                    )
                    checked = val in ("1", "true")

            checkboxes.append(
                {
                    "checkbox_index": checkbox_index,
                    "label": label_text,
                    "checked": checked,
                    "position_locator": {
                        "type": "checkbox",
                        "checkbox_index": checkbox_index,
                    },
                }
            )
            checkbox_index += 1

        logger.info("复选框提取完成: 共 %d 个", len(checkboxes))
        return checkboxes

    def _detect_delete_inapplicable(self, text: str) -> list[str] | None:
        """
        检测"删除不适用项"格式（如"自然人/法人/非法人组织"）。

        使用正则匹配以 / 或 ／ 分隔的选项模式。
        返回选项列表，若未检测到则返回 None。
        """
        # 匹配 "A/B/C" 或 "A／B／C" 模式，至少 2 个选项
        pattern = r"([\u4e00-\u9fff\w]+(?:[/／][\u4e00-\u9fff\w]+){1,})"
        match = re.search(pattern, text)
        if match is None:
            return None

        matched_text: str = match.group(1)
        options: list[str] = re.split(r"[/／]", matched_text)

        # 过滤：每个选项至少 1 个中文字符，才认为是"删除不适用项"
        has_chinese = all(re.search(r"[\u4e00-\u9fff]", opt) for opt in options)
        if not has_chinese:
            return None

        return options

    # ------------------------------------------------------------------
    # LLM 分析 (Requirements: 3.3, 3.4, 3.5, 4.1–4.8, 5.6, 5.7, 11.2, 11.6)
    # ------------------------------------------------------------------

    def analyze_template(self, template_id: int) -> list[Any]:
        """
        分析模板并生成字段映射：
        1. 提取结构 → 计算指纹
        2. 指纹匹配 → 命中则复用映射（标注 mapping_source）
        3. 未命中 → 构建 LLM 提示词 → 调用 LLM → 解析结果
        4. 创建 FieldMapping 记录
        5. 更新模板状态为 ready
        """
        from apps.documents.models.choices import TemplateStatus
        from apps.documents.models.external_template import ExternalTemplate, ExternalTemplateFieldMapping

        template: ExternalTemplate = ExternalTemplate.objects.get(pk=template_id)

        # 更新状态为 analyzing
        template.status = TemplateStatus.ANALYZING
        template.save(update_fields=["status", "status_changed_at", "updated_at"])

        try:
            # 1. 提取结构
            structure_json: dict[str, Any] = self.extract_structure(template_id)

            # 2. 计算指纹
            abs_path = Path(settings.MEDIA_ROOT) / template.file_path
            fingerprint: str = self._fingerprint_service.compute_fingerprint(abs_path)
            template.structure_fingerprint = fingerprint
            template.save(update_fields=["structure_fingerprint", "updated_at"])

            # 3. 查找匹配模板
            matched: ExternalTemplate | None = self._fingerprint_service.find_matching_template(
                fingerprint, template.law_firm_id
            )

            # 排除自身
            if matched is not None and matched.pk == template.pk:
                matched = None

            created_mappings: list[ExternalTemplateFieldMapping]

            if matched is not None:
                # 复用已有映射
                created_mappings = self._copy_mappings_from(
                    source_template=matched,
                    target_template=template,
                )
                template.mapping_source = matched
                logger.info(
                    "复用映射: template_id=%d, source_id=%d, count=%d",
                    template_id,
                    matched.pk,
                    len(created_mappings),
                )
            else:
                # LLM 分析
                prompt: str = self._build_llm_prompt(structure_json)
                response = self._llm_service.complete(
                    prompt=prompt,
                    system_prompt=("你是一个法律文书模板分析专家。请分析模板结构并返回字段映射的 JSON 数组。"),
                    temperature=0.1,
                    max_tokens=4096,
                )
                raw_mappings: list[dict[str, Any]] = self._parse_llm_response(response.content)
                created_mappings = self._create_field_mappings(template, raw_mappings)
                logger.info(
                    "LLM 分析完成: template_id=%d, mappings=%d",
                    template_id,
                    len(created_mappings),
                )

            # 更新状态为 ready
            template.status = TemplateStatus.READY
            template.save(
                update_fields=[
                    "status",
                    "mapping_source",
                    "status_changed_at",
                    "updated_at",
                ]
            )

            return created_mappings

        except Exception:
            # 分析失败：设置状态并重新抛出
            logger.exception("模板分析失败: template_id=%d", template_id)
            template.refresh_from_db()
            template.status = TemplateStatus.ANALYSIS_FAILED
            template.save(update_fields=["status", "status_changed_at", "updated_at"])
            raise

    def _copy_mappings_from(
        self,
        source_template: Any,
        target_template: Any,
    ) -> list[Any]:
        """从源模板复制映射到目标模板"""
        from apps.documents.models.external_template import ExternalTemplateFieldMapping

        source_mappings = ExternalTemplateFieldMapping.objects.filter(template=source_template)
        created: list[ExternalTemplateFieldMapping] = []
        for m in source_mappings:
            new_mapping = ExternalTemplateFieldMapping.objects.create(
                template=target_template,
                position_locator=m.position_locator,
                position_description=m.position_description,
                semantic_label=m.semantic_label,
                fill_type=m.fill_type,
                sort_order=m.sort_order,
            )
            created.append(new_mapping)
        return created

    def _build_llm_prompt(self, structure_json: dict[str, Any]) -> str:
        """
        构建 LLM 提示词：结构 JSON + fill_type 说明
        """
        structure_text: str = json.dumps(structure_json, ensure_ascii=False, indent=2)

        prompt: str = (
            f"请分析以下法律文书模板的结构，识别所有可填充位置。\n\n"
            f"## 文档结构 JSON\n```json\n{structure_text}\n```\n\n"
            f"## 填充类型说明\n"
            f"- text: 文本替换（替换段落或单元格中的文本）\n"
            f"- checkbox: 勾选复选框\n"
            f"- delete_inapplicable: 删除不适用项（保留匹配项，删除其余选项）\n\n"
            f"## 输出要求\n"
            f"请返回一个 JSON 数组，每个元素包含以下字段：\n"
            f"- position_locator: 位置定位器（直接使用结构 JSON 中的 position_locator）\n"
            f"- semantic_label: 语义标签（中文描述该位置应填写的内容，如'被申请人姓名'、'住所地'）\n"
            f"- fill_type: 填充类型（text / checkbox / delete_inapplicable）\n\n"
            f"仅返回 JSON 数组，不要包含其他文字说明。"
        )
        return prompt

    def _parse_llm_response(self, response: str) -> list[dict[str, Any]]:
        """
        解析 LLM 返回的 JSON 映射结果。
        处理 markdown 代码块包裹的情况。
        """
        text: str = response.strip()

        # 去除 markdown 代码块
        if text.startswith("```"):
            # 去除首行 ```json 或 ```
            lines: list[str] = text.split("\n")
            # 找到第一个 ``` 之后的内容
            start_idx = 1
            end_idx = len(lines)
            for i in range(len(lines) - 1, 0, -1):
                if lines[i].strip().startswith("```"):
                    end_idx = i
                    break
            text = "\n".join(lines[start_idx:end_idx])

        try:
            parsed: Any = json.loads(text)
        except json.JSONDecodeError:
            logger.error("LLM 返回的 JSON 解析失败: %s", text[:500])
            raise

        if not isinstance(parsed, list):
            logger.error("LLM 返回的不是 JSON 数组: type=%s", type(parsed).__name__)
            raise ValueError(_("LLM 返回格式异常，期望 JSON 数组"))

        result: list[dict[str, Any]] = []
        for item in parsed:
            if not isinstance(item, dict):
                continue
            result.append(
                {
                    "position_locator": item.get("position_locator", {}),
                    "semantic_label": str(item.get("semantic_label", "")),
                    "fill_type": str(item.get("fill_type", "text")),
                }
            )

        return result

    def _create_field_mappings(
        self,
        template: Any,
        mappings: list[dict[str, Any]],
    ) -> list[Any]:
        """根据解析后的映射数据创建 FieldMapping 记录"""
        from apps.documents.models.choices import FillType
        from apps.documents.models.external_template import ExternalTemplateFieldMapping

        valid_fill_types: set[str] = {ft.value for ft in FillType}
        created: list[ExternalTemplateFieldMapping] = []

        for idx, m in enumerate(mappings):
            fill_type: str = m.get("fill_type", FillType.TEXT)
            if fill_type not in valid_fill_types:
                fill_type = FillType.TEXT

            position_locator: dict[str, Any] = m.get("position_locator", {})
            pos_type: str = str(position_locator.get("type", ""))
            position_description: str = ""
            if pos_type == "paragraph":
                p_idx = position_locator.get("paragraph_index", "")
                position_description = f"段落 {p_idx}"
            elif pos_type == "table_cell":
                t_idx = position_locator.get("table_index", "")
                row = position_locator.get("row", "")
                col = position_locator.get("col", "")
                position_description = f"表格{t_idx} 行{row} 列{col}"
            elif pos_type == "checkbox":
                cb_idx = position_locator.get("checkbox_index", "")
                position_description = f"复选框 {cb_idx}"
            elif pos_type == "delete_inapplicable":
                position_description = "删除不适用项"

            new_mapping = ExternalTemplateFieldMapping.objects.create(
                template=template,
                position_locator=position_locator,
                position_description=position_description,
                semantic_label=m.get("semantic_label", ""),
                fill_type=fill_type,
                sort_order=idx,
            )
            created.append(new_mapping)

        return created

    # ------------------------------------------------------------------
    # 重新分析
    # ------------------------------------------------------------------

    def retry_analysis(self, template_id: int) -> list[Any]:
        """重新触发 LLM 分析：删除旧映射后重新分析。"""
        from apps.documents.models.external_template import ExternalTemplateFieldMapping

        deleted_count, _ = ExternalTemplateFieldMapping.objects.filter(template_id=template_id).delete()
        logger.info(
            "重新分析: template_id=%d, 已删除旧映射 %d 条",
            template_id,
            deleted_count,
        )
        return self.analyze_template(template_id)

    def create_manual_mapping(
        self,
        *,
        template_id: int,
        position_locator: dict[str, Any],
        position_description: str,
        semantic_label: str,
        fill_type: str = "text",
    ) -> Any:
        """手动新增单条映射。"""
        from apps.documents.models.external_template import ExternalTemplate, ExternalTemplateFieldMapping

        template = ExternalTemplate.objects.get(pk=template_id)
        max_order: int = (
            ExternalTemplateFieldMapping.objects.filter(template=template)
            .order_by("-sort_order")
            .values_list("sort_order", flat=True)
            .first()
            or 0
        )
        return ExternalTemplateFieldMapping.objects.create(
            template=template,
            position_locator=position_locator,
            position_description=position_description,
            semantic_label=semantic_label,
            fill_type=fill_type,
            sort_order=max_order + 1,
        )

    def delete_mapping(self, mapping_id: int) -> None:
        """删除单条映射。"""
        from apps.documents.models.external_template import ExternalTemplateFieldMapping

        ExternalTemplateFieldMapping.objects.filter(pk=mapping_id).delete()