inquiry_robot/inquiry-agent/agent/jobs/quote_fill.py

"""
按主账 mappingJson 向报价 Excel 写入已冻结事实。

本文件职责：复制模板 → 跳过黄区 → 写 fields / fee_rows → 产出工作副本路径。
调用方：Worker LibreOffice 槽之前；禁止在 HTTP 回调线程执行。
禁止：发明价格；直连 TMS；把本模块当六态 Owner。
"""

from __future__ import annotations

import json
import logging
import re
import shutil
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Optional

logger = logging.getLogger(__name__)

# 黄区近似 RGB（与 Java QuoteTemplateExcelSupport 对齐）
_YELLOW_RGB = {
    "FFFFFF00",
    "FFFF00",
    "FFFFEB9C",
    "FFFFC000",
    "FFFFFF99",
    "FFFFCC00",
    "FFFFD966",
    "FFFFF2CC",
    "FFFFE699",
}


@dataclass(frozen=True)
class QuoteFillResult:
    """一次填表结果。"""

    ok: bool
    output_xlsx: str = ""
    error: str = ""
    fields_written: int = 0


def _col_letter_to_index(letter: str) -> int:
    letter = (letter or "").strip().upper()
    n = 0
    for ch in letter:
        if not ("A" <= ch <= "Z"):
            continue
        n = n * 26 + (ord(ch) - ord("A") + 1)
    return max(n - 1, 0)


def _parse_cell_ref(cell_ref: str) -> tuple[str, int, int]:
    """
    解析 Sheet!A1 → (sheet, row0, col0)。
    """
    sheet = "Sheet1"
    addr = cell_ref
    if "!" in cell_ref:
        sheet, addr = cell_ref.split("!", 1)
    m = re.match(r"^\$?([A-Za-z]+)\$?(\d+)$", addr.strip())
    if not m:
        raise ValueError(f"非法单元格引用: {cell_ref}")
    col = _col_letter_to_index(m.group(1))
    row = int(m.group(2)) - 1
    return sheet, row, col


def fill_quote_workbook(
    *,
    source_xlsx: str | Path,
    output_xlsx: str | Path,
    mapping: Mapping[str, Any] | str,
    facts: Mapping[str, Any],
) -> QuoteFillResult:
    """
    按 mapping 写入冻结事实。

    参数：
      source_xlsx：主账版本对象落地的原模板
      output_xlsx：工作副本输出路径
      mapping：mappingJson 对象或 JSON 字符串
      facts：已冻结字段（含可选 fee_rows 列表）
    副作用：写磁盘 xlsx；不调 LibreOffice、不打主账。
    线程：必须在 Worker 槽内调用。
    """
    try:
        from openpyxl import load_workbook
    except ImportError:
        return QuoteFillResult(ok=False, error="openpyxl_not_installed")

    src = Path(source_xlsx)
    out = Path(output_xlsx)
    if not src.is_file():
        return QuoteFillResult(ok=False, error="source_not_found")

    if isinstance(mapping, str):
        try:
            mapping_obj = json.loads(mapping)
        except json.JSONDecodeError as exc:
            return QuoteFillResult(ok=False, error=f"mapping_json_invalid:{exc}")
    else:
        mapping_obj = dict(mapping)

    yellow = set(mapping_obj.get("fixedYellowCells") or [])
    fields = mapping_obj.get("fields") or {}
    fee_rows_cfg = mapping_obj.get("fee_rows")

    out.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(src, out)

    try:
        wb = load_workbook(out)
    except Exception as exc:
        return QuoteFillResult(ok=False, error=f"open_failed:{exc}")

    written = 0
    try:
        for code, cell_ref in fields.items():
            if not cell_ref or cell_ref in yellow:
                continue
            if code == "fee_rows":
                continue
            val = facts.get(code)
            if val is None:
                _write_cell(wb, str(cell_ref), "")
                continue
            _write_cell(wb, str(cell_ref), str(val))
            written += 1

        if fee_rows_cfg and facts.get("fee_rows"):
            written += _write_fee_rows(wb, fee_rows_cfg, facts.get("fee_rows"), yellow)

        wb.save(out)
    except Exception as exc:
        logger.exception("quote_fill 写入失败")
        return QuoteFillResult(ok=False, error=f"write_failed:{exc}")
    finally:
        wb.close()

    logger.info("quote_fill ok fields=%s out=%s", written, out)
    return QuoteFillResult(ok=True, output_xlsx=str(out), fields_written=written)


def _write_cell(wb, cell_ref: str, value: str) -> None:
    sheet_name, row, col = _parse_cell_ref(cell_ref)
    ws = wb[sheet_name] if sheet_name in wb.sheetnames else wb.worksheets[0]
    ws.cell(row=row + 1, column=col + 1, value=value)


def _write_fee_rows(wb, fee_cfg: Mapping[str, Any], rows: Any, yellow: set[str]) -> int:
    if not isinstance(rows, list) or not rows:
        return 0
    anchor = fee_cfg.get("anchor")
    columns = fee_cfg.get("columns") or {}
    if not anchor:
        return 0
    sheet_name, start_row, _ = _parse_cell_ref(str(anchor))
    ws = wb[sheet_name] if sheet_name in wb.sheetnames else wb.worksheets[0]
    n = 0
    for i, row_obj in enumerate(rows):
        if not isinstance(row_obj, Mapping):
            continue
        r = start_row + i
        for key in ("name", "amount", "remark"):
            col_letter = columns.get(key)
            if not col_letter:
                continue
            col = _col_letter_to_index(str(col_letter))
            ref = f"{ws.title}!{ _index_to_col(col) }{r + 1}"
            if ref in yellow:
                continue
            ws.cell(row=r + 1, column=col + 1, value=str(row_obj.get(key) or ""))
            n += 1
    return n


def _index_to_col(idx: int) -> str:
    n = idx + 1
    s = ""
    while n:
        n, rem = divmod(n - 1, 26)
        s = chr(65 + rem) + s
    return s


def fill_then_convert_pdf(
    *,
    source_xlsx: str | Path,
    work_dir: str | Path,
    mapping: Mapping[str, Any] | str,
    facts: Mapping[str, Any],
    converter: Any,
) -> tuple[QuoteFillResult, Optional[Any]]:
    """
    填表后在调用方已占用的 LibreOffice 单槽内转 PDF。

    converter：LibreOfficeConverter 实例；本函数不抢槽。
    返回：(FillResult, LibreOfficeResult|None)
    """
    work = Path(work_dir)
    work.mkdir(parents=True, exist_ok=True)
    out_xlsx = work / "filled.xlsx"
    fill = fill_quote_workbook(
        source_xlsx=source_xlsx,
        output_xlsx=out_xlsx,
        mapping=mapping,
        facts=facts,
    )
    if not fill.ok:
        return fill, None
    pdf_result = converter.convert_xlsx_to_pdf(str(out_xlsx))
    return fill, pdf_result