Auto-Finance/src/mcp/tools/materials.py

"""会话与材料工具

阶段不落地存储，全部由 arb_case_status 从产物推导，宿主 Agent 负责循环与重试。
"""

from __future__ import annotations

import shutil
from pathlib import Path
from typing import Any

from ...cases import ARTIFACTS_DIR_NAME, CACHE_DIR_NAME, PAGES_DIR_NAME, SCREENSHOT_DIR_NAME, CaseStore, upload_page_url
from ..envelope import fail, ok

_store = CaseStore()


def case_create(name: str, invoice_type: str | None = None) -> dict[str, Any]:
    """新建报销会话目录，返回 case_id 与目录路径。

    name 是本次报销的人类可读名称（如「6月毕业采集」）；
    invoice_type 可选 travel|normal，未定时留空，抽取后由发票分类决定。
    """
    if not name or not name.strip():
        return fail("name 不能为空", next_action="提供一个可读的报销名称")
    d = _store.create(name, invoice_type)
    return ok(
        case_id=d.name,
        case_dir=str(d),
        upload_url=_store.upload_url(d),
        upload_page=upload_page_url(),
        upload_note="把 upload_url 发给用户：手机点开即上传（收件器需在运行：python -m src.upload.server）。"
        "这条链接含本会话令牌，只在这里给一次，别在其它回答里重复",
    )


def case_upload_link(case_id: str) -> dict[str, Any]:
    """重新取该会话的手机上传直链（用户把上一条弄丢了，或换了设备时用）。

    链接含令牌：只在用户明确要链接时调用，别在常规流程里反复调，也别把令牌抄进别处。
    """
    d = _store.resolve(case_id)
    if d is None:
        return fail(f"会话不存在: {case_id}", next_action="先 case_list 看有哪些会话")
    return ok(
        case_id=d.name,
        case_name=_store.meta(d).get("name") or d.name,
        upload_url=_store.upload_url(d),
        upload_page=upload_page_url(),
        pending_inbox=len(_store.inbox(d)),
    )


def case_list() -> dict[str, Any]:
    """列出所有报销会话及其推导阶段、计数与待收文件数。"""
    return ok(cases=[_store.summary(d) for d in _store.list_cases()])


def case_status(case_id: str) -> dict[str, Any]:
    """单个会话的完整态势：材料数、抽取完成/缺失/陈旧、匹配、信息、产物、收件箱、下一步建议。"""
    d = _store.resolve(case_id)
    if d is None:
        return fail(f"会话不存在: {case_id}", next_action="先调用 case_create 或从 case_list 里选一个")
    s = _store.summary(d)
    ex = s["extractions"]
    if s["pending_inbox"]:
        nxt = f"收件箱有 {s['pending_inbox']} 个新文件，先 case_add_files(case_id, source='inbox') 收取"
    elif s["stage"] == "filled":
        nxt = (
            "填报已完成（screenshots 里有步骤完成截图）：把单据编号与会话目录告诉用户，"
            "由他本人在财务系统里核对并提交；不要再 browser_new_form，也别重复填"
        )
    elif ex["missing"]:
        nxt = f"有 {len(ex['missing'])} 个材料未抽取：case_render_pages 后按 schema 抽取并 case_save_extraction"
    elif ex["stale"]:
        nxt = f"有 {len(ex['stale'])} 个材料在抽取后被改动，需要重新渲染并抽取"
    elif not s["matched"]:
        nxt = "调用 match_payments 做发票与支付记录的金额匹配"
    elif not s["info"]:
        nxt = "综合缓存与匹配结果生成 travel_info 或 normal_info，再 validate_info"
    else:
        nxt = "信息已就绪：普通发票可先生成申购单，然后 browser_open 开始填报；填完即终点，提交由本人做"
    return ok(**s, next_action=nxt)


def case_add_files(
    case_id: str,
    paths: list[str] | None = None,
    source: str | None = None,
) -> dict[str, Any]:
    """把材料拷入会话目录。

    paths 传绝对路径列表（从手机下载目录、微信文件目录等任意位置取文件）；
    source='inbox' 表示收取手机扫码上传落在 inbox 的文件。
    """
    d = _store.resolve(case_id)
    if d is None:
        return fail(f"会话不存在: {case_id}", next_action="先 case_create")
    results: list[dict[str, Any]] = []
    if source == "inbox":
        results = _store.drain_inbox(d)
    elif paths:
        for p in paths:
            results.append(_store.add_file(d, Path(p)))
    else:
        return fail("需要 paths 或 source='inbox'", next_action="paths 传绝对路径列表，或 source='inbox' 收手机上传")
    added = [r["file"] for r in results if r["status"] == "added"]
    skipped = [r for r in results if r["status"] != "added"]
    return ok(added=added, skipped=skipped, materials=len(_store.materials(d)))


def case_inbox(case_id: str) -> dict[str, Any]:
    """列出手机/PC 上传进来但尚未进入流程的文件。"""
    d = _store.resolve(case_id)
    if d is None:
        return fail(f"会话不存在: {case_id}", next_action="先 case_create")
    files = [{"file": f.name, "bytes": f.stat().st_size} for f in _store.inbox(d)]
    return ok(pending=len(files), files=files)


def case_reset(case_id: str, scope: str = "extractions") -> dict[str, Any]:
    """清理派生产物以便重跑。scope=extractions|match|info|pages|artifacts|screenshots|all。"""
    d = _store.resolve(case_id)
    if d is None:
        return fail(f"会话不存在: {case_id}", next_action="先 case_create")
    cache = d / CACHE_DIR_NAME
    cleared: list[str] = []
    if scope == "extractions":
        for p in list(cache.glob("*.json")):
            if p.name not in ("travel_info.json", "normal_info.json", "match_result.json"):
                p.unlink()
                cleared.append(p.name)
    elif scope == "match":
        p = cache / "match_result.json"
        if p.exists():
            p.unlink()
            cleared.append(p.name)
    elif scope == "info":
        for name in ("travel_info.json", "normal_info.json"):
            p = cache / name
            if p.exists():
                p.unlink()
                cleared.append(name)
    elif scope == "pages":
        pages = d / PAGES_DIR_NAME
        if pages.is_dir():
            shutil.rmtree(pages)
            pages.mkdir(parents=True, exist_ok=True)
            cleared.append(PAGES_DIR_NAME)
    elif scope == "artifacts":
        art = d / ARTIFACTS_DIR_NAME
        for p in art.glob("*"):
            p.unlink()
            cleared.append(p.name)
    elif scope == "screenshots":
        # 填报过程截图：清了 stage 才会从 filled 退回去，所以单独给一个域而不是并进 artifacts
        shots_dir = d / SCREENSHOT_DIR_NAME
        if shots_dir.is_dir():
            for shot in shots_dir.glob("*.png"):
                shot.unlink()
                cleared.append(shot.name)
    elif scope == "all":
        for sub in (CACHE_DIR_NAME, PAGES_DIR_NAME, SCREENSHOT_DIR_NAME, ARTIFACTS_DIR_NAME):
            target = d / sub
            if target.is_dir():
                shutil.rmtree(target)
            target.mkdir(parents=True, exist_ok=True)
            cleared.append(sub)
        for name in ("payment_records.csv", "invoice_summary.csv", "invoice_groups.json"):
            f = d / name
            if f.exists():
                f.unlink()
                cleared.append(name)
    else:
        return fail(f"未知 scope: {scope}", next_action="可选 extractions|match|info|pages|artifacts|all")
    return ok(scope=scope, cleared=cleared, case=_store.summary(d))