bid-doctor/server/src/bid_doctor/project/extracted.py

"""把解析出的项目基本信息和分包写回项目（FR-M2-03、04）。

- AI 值和人工值分开存：用户改过的字段不覆盖；项目名称只在还是「未命名项目」时自动填。
- 补遗带来的新值先放进待确认（FR-M2-10），用户确认后才生效；首份招标文件直接生效。
- 分包：解析出包时替换掉建项目时的默认包；已经有包的项目不动（要改由用户在选包页处理）。
"""

from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from typing import Any
from uuid import UUID

from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession

from bid_doctor.checklist.extract import Extracted, ProjectInfo, amount_yuan
from bid_doctor.core.db import TenantContext
from bid_doctor.core.ids import uuid7
from bid_doctor.document import citations
from bid_doctor.document.models import SourceDocument
from bid_doctor.project import service as projects
from bid_doctor.project.models import Lot, Project, ProjectField

MONEY = ("budget_total", "ceiling_total")
COLUMNS = ("name", "code", "purchaser", "agency", "procurement_method", "legal_regime")
# 更正公告不改项目是谁：补遗、澄清里抽到的这几项（常把公告标题当成项目名称）不进待确认，免得一片噪音（FR-M2-10）
IDENTITY = ("name", "code", "purchaser", "agency", "legal_regime")


async def apply(
    s: AsyncSession,
    ctx: TenantContext,
    doc: SourceDocument,
    info: ProjectInfo,
    regime_by_rule: str,
    items: list[Extracted],
) -> dict[str, UUID]:
    """写入字段、分包和投标截止时间。返回 {包号: 分包编号}，给清单项挂所属包用。"""
    p = await projects.get(s, ctx, doc.project_id)
    values: dict[str, tuple[Any, Any]] = {}
    for key, f in info.fields.items():
        value = amount_yuan(f.value, f.unit) if key in MONEY else f.value
        if key == "legal_regime" and value not in ("gov_procurement", "tendering"):
            continue
        values[key] = (value, f.sentence)
    if "legal_regime" not in values and regime_by_rule != "unknown":
        values["legal_regime"] = (regime_by_rule, None)
    deadline = _deadline(items)
    if deadline:
        values["deadline_at"] = deadline
    for key, (value, sentence) in values.items():
        if doc.role in ("addendum", "clarification") and key in IDENTITY:
            continue
        cite = None
        if sentence is not None:
            (c,) = await citations.from_sentences(s, ctx, doc, [sentence], "ai", ["id", "quote", "basis"])
            cite = c.id
        await _set_field(s, ctx, p, doc, key, value, cite)
    return await _lots(s, ctx, p, info)


def _deadline(items: list[Extracted]) -> tuple[str, Any] | None:
    for e in items:
        sched = e.spec.get("schedule") or {}
        if e.zone == "formal" and sched.get("date_kind") == "bid_deadline" and sched.get("at"):
            return sched["at"], e.sentences[0]
    return None


async def _set_field(
    s: AsyncSession, ctx: TenantContext, p: Project, doc: SourceDocument, key: str, value: Any, cite: UUID | None
) -> None:
    f = (
        await s.execute(select(ProjectField).where(ProjectField.project_id == p.id, ProjectField.field_key == key))
    ).scalar_one_or_none()
    if f is None:
        f = ProjectField(tenant_id=ctx.tenant_id, project_id=p.id, field_key=key)
        s.add(f)
    if doc.role in ("addendum", "clarification") and f.ai_value is not None and f.ai_value != value:
        f.pending_value, f.pending_citation_id = value, cite  # 补遗的新值：等用户确认（FR-M2-10）
    else:
        f.ai_value, f.ai_citation_id, f.ai_document_id = value, cite, doc.id
    await s.flush()
    if f.human_value is not None or f.pending_value is not None:
        return
    if key == "name" and p.name != projects.UNNAMED:
        return  # 用户建项目时填了名称：以用户的为准
    if key in COLUMNS:
        setattr(p, key, value)
    elif key == "deadline_at":
        p.deadline_at = datetime.fromisoformat(value)


async def _lots(s: AsyncSession, ctx: TenantContext, p: Project, info: ProjectInfo) -> dict[str, UUID]:
    existing = (await s.execute(select(Lot).where(Lot.project_id == p.id).order_by(Lot.sort_order))).scalars().all()
    if info.lots and all(lot.is_default for lot in existing):
        await s.execute(delete(Lot).where(Lot.project_id == p.id, Lot.is_default))
        for i, lot in enumerate(info.lots):
            budget, ceiling = lot.get("budget") or {}, lot.get("ceiling") or {}
            b, c = (
                amount_yuan(budget.get("value"), budget.get("unit")),
                amount_yuan(ceiling.get("value"), ceiling.get("unit")),
            )
            s.add(
                Lot(
                    id=uuid7(),
                    tenant_id=ctx.tenant_id,
                    project_id=p.id,
                    lot_no=str(lot["lot_no"])[:40],
                    name=lot.get("name"),
                    origin="ai",
                    sort_order=i,
                    edited_fields=[],
                    budget_amount=Decimal(b) if b else None,
                    ceiling_amount=Decimal(c) if c else None,
                    ai_values={k: v for k, v in lot.items() if k != "sentence"},
                )
            )
        await s.flush()
        existing = (await s.execute(select(Lot).where(Lot.project_id == p.id))).scalars().all()
    return {lot.lot_no: lot.id for lot in existing}