PayrollManagementSystem/backend/app/services/tplus/master_import.py

"""主数据先预览再执行，按源ID关联，只更新允许同步的字段。"""

import json
from collections import Counter
from datetime import datetime
from decimal import Decimal

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from app.models import BusinessRole, Customer, Personnel, Product, TPlusSourceLink, TPlusWorkLine
from app.services.tplus.snapshots import source_records
from app.services.tplus.workbench import RULE_VERSION, WorkbenchError, digest, get_run, latest_facts

MODELS = {"personnel": Personnel, "products": Product, "customers": Customer}
ORDER = {"personnel": 0, "products": 1, "customers": 2}
# 恢复库 eap_EnumItem：211=客户，228=客户/供应商，226=供应商。
CUSTOMER_ONLY = 211


def _identity(row):
    keys = {
        "personnel": ("personnel_name",),
        "products": ("product_name", "spec_model"),
        "customers": ("company_name", "owner_name"),
    }[row["entity_type"]]
    return row["entity_type"], tuple(str(row["values"][key]).rstrip().casefold() for key in keys)


def _target_snapshot(target):
    if target is None:
        return None
    return {c.name: getattr(target, c.name) for c in target.__table__.columns}


def _master_values(entity, record, facts, session, trades=None):
    row = record.payload_json
    if entity == "personnel":
        code = str(row.get("code") or "").strip()
        fact = facts.get(("PERSONNEL", code))
        if fact is None:
            return None, "NOT_IN_PERSONNEL_ROSTER"
        person = fact.data_json
        if person["code"] != code or person["name"] != str(row.get("name") or "").strip():
            return None, "PERSONNEL_IDENTITY_CONFLICT"
        role = session.scalar(
            select(BusinessRole).where(BusinessRole.role_code == person["role_code"])
        )
        if role is None:
            return None, "BUSINESS_ROLE_MISSING"
        return {
            "personnel_name": person["name"],
            "business_role_id": role.id,
            "status": person["status"],
        }, None
    if entity == "products":
        fact = facts.get(("PRODUCT", record.source_id))
        if fact is None:
            return None, "PRODUCT_CATEGORY_PENDING"
        unit = row.get("unit_name")
        if not unit or unit != fact.data_json["unit"]:
            return None, "PRODUCT_UNIT_PENDING"
        trade = (trades or {}).get(record.source_id)
        return {
            "product_name": row["name"],
            "spec_model": row.get("specification") or "",
            "unit": unit,
            "product_category": fact.data_json["category"],
            "source_trade_json": trade,
            "stock_quantity": Decimal(trade["quantity"]) if trade else None,
            "sale_price": Decimal(trade["price"]) if trade else None,
        }, None
    if row.get("partnerType") != CUSTOMER_ONLY:
        return None, "NOT_CUSTOMER_ONLY"
    fact = facts.get(("CUSTOMER", record.source_id))
    if fact is None:
        return None, "CUSTOMER_ATTRIBUTE_PENDING"
    return {
        "company_name": row["name"],
        "owner_name": fact.data_json["owner_name"],
        "customer_type": fact.data_json["customer_type"],
    }, None


def _protected_update(entity, values, target=None):
    allowed = {
        "personnel": {"personnel_name"},
        "products": {"product_name", "spec_model", "unit", "source_trade_json"},
        "customers": {"company_name"},
    }[entity]
    if entity == "products" and target is not None and not target.quantity_price_manual:
        allowed = allowed | {"stock_quantity", "sale_price"}
    return {k: v for k, v in values.items() if k in allowed}


def _conflict(session, entity, values, target_id):
    model = MODELS[entity]
    if entity == "personnel":
        conditions = [model.personnel_name == values["personnel_name"]]
    elif entity == "products":
        conditions = [
            model.product_name == values["product_name"],
            model.spec_model == values["spec_model"],
        ]
    else:
        conditions = [
            model.company_name == values["company_name"],
            model.owner_name == values["owner_name"],
        ]
    if target_id:
        conditions.append(model.id != target_id)
    return session.scalar(select(model.id).where(*conditions).limit(1)) is not None


def build_master_plan(session, run, *, lock_targets=False):
    """重新计算源、事实、目标指纹；执行前复核所有输入。"""
    facts = latest_facts(session, run.source_database)
    links = {
        (v.entity_type, v.source_id): v
        for v in session.scalars(
            select(TPlusSourceLink).where(TPlusSourceLink.source_database == run.source_database)
        )
    }
    rows = []
    snapshots = []
    target_ids = []
    for entity in sorted(run.input_json["entity_types"], key=ORDER.__getitem__):
        records = source_records(
            session, run.source_database, entity, run.input_json.get("source_ids") or None
        )
        from app.services.product_completion import recent_trades

        trades = recent_trades(session, run.source_database, [r.source_id for r in records]) if entity == "products" else {}
        for record in records:
            link = links.get((entity, record.source_id))
            target = (
                session.get(
                    MODELS[entity],
                    link.target_id,
                    populate_existing=True,
                    with_for_update=lock_targets,
                )
                if link
                else None
            )
            values, reason = _master_values(entity, record, facts, session, trades)
            action = "PENDING" if reason else "CREATE"
            if reason in ("NOT_IN_PERSONNEL_ROSTER", "NOT_CUSTOMER_ONLY"):
                action = "SKIP"
            if link and target is None:
                action, reason = "CONFLICT", "TARGET_MISSING"
            elif values:
                if target:
                    values = _protected_update(entity, values, target)
                    action = (
                        "UPDATE"
                        if any(getattr(target, k) != v for k, v in values.items())
                        else "SKIP"
                    )
                if _conflict(
                    session,
                    entity,
                    {**(_target_snapshot(target) or {}), **values},
                    target.id if target else None,
                ):
                    action, reason = "CONFLICT", "TARGET_UNIQUE_CONFLICT"
            snapshots.append((entity, record.source_id, record.row_hash, record.id))
            target_ids.append((entity, record.source_id, _target_snapshot(target)))
            rows.append(
                {
                    "entity_type": entity,
                    "source_id": record.source_id,
                    "action": action,
                    "reason": reason,
                    "values": values or {},
                    "record_id": record.id,
                    "row_hash": record.row_hash,
                    "target_id": target.id if target else None,
                    "source_no": record.source_no,
                    "source_name": record.payload_json.get("name"),
                }
            )
    identities = Counter(_identity(r) for r in rows if r["action"] == "CREATE")
    for row in rows:
        if row["action"] == "CREATE" and identities[_identity(row)] > 1:
            row["action"], row["reason"] = "CONFLICT", "DUPLICATE_SOURCE_IDENTITY"
    # 以代码名单检查漏读人员，不能让缺失人员静默成为成功。
    if "personnel" in run.input_json["entity_types"]:
        seen = {r["source_no"] for r in rows if r["entity_type"] == "personnel"}
        for kind, key in facts:
            if kind == "PERSONNEL" and key not in seen and not run.input_json.get("source_ids"):
                rows.append(
                    {
                        "entity_type": "personnel",
                        "source_id": key,
                        "action": "PENDING",
                        "reason": "ROSTER_SOURCE_MISSING",
                        "values": {},
                    }
                )
    frozen = {
        "rule_version": RULE_VERSION,
        "source": snapshots,
        "targets": target_ids,
        "plan": rows,
        "facts": [
            {
                "id": f.id,
                "kind": f.kind,
                "key": f.fact_key,
                "version": f.version,
                "data": f.data_json,
                "evidence": f.evidence,
            }
            for f in facts.values()
        ],
    }
    frozen = json.loads(json.dumps(frozen, ensure_ascii=False, default=str))
    return rows, frozen, digest(frozen)


def _line(session, run, i, row, action=None, reason=None):
    session.add(
        TPlusWorkLine(
            run_no=run.run_no,
            line_no=i,
            entity_type=row["entity_type"],
            source_id=row["source_id"],
            action=action or row["action"],
            reason=reason or row.get("reason"),
            # 真实发货参照带 Decimal 数量、单价；JSON 留存使用十进制原文，
            # 与上方预览快照一致，不能转 float 或让预览任务在落明细时失败。
            data_json=json.loads(json.dumps(
                {k: v for k, v in row.items() if k not in ("entity_type", "action", "reason")},
                ensure_ascii=False, default=str,
            )),
        )
    )


def execute_master(session, run):
    """执行预览或原子导入；正式核算表永远不参与此服务。"""
    if run.kind == "MASTER_PREVIEW":
        rows, snapshot, fingerprint = build_master_plan(session, run)
        run.snapshot_json, run.snapshot_hash = snapshot, fingerprint
        for i, row in enumerate(rows, 1):
            _line(session, run, i, row)
        counts = Counter(r["action"] for r in rows)
        run.summary_json = {
            "counts": dict(counts),
            "rule_version": RULE_VERSION,
            "total": len(rows),
            "formal_payroll_written": False,
        }
        run.status = "PARTIAL" if counts["PENDING"] or counts["CONFLICT"] else "SUCCEEDED"
        return
    preview = get_run(session, run.input_json["preview_no"], run.source_database)
    if preview.kind != "MASTER_PREVIEW" or preview.status not in ("SUCCEEDED", "PARTIAL"):
        raise WorkbenchError("PREVIEW_REQUIRED")
    rows, frozen, fingerprint = build_master_plan(session, preview, lock_targets=True)
    if preview.snapshot_hash != fingerprint:
        raise WorkbenchError("PREVIEW_STALE")
    run.snapshot_hash, run.snapshot_json = fingerprint, frozen
    counts = Counter()
    for i, row in enumerate(rows, 1):
        action, reason = row["action"], row.get("reason")
        if action in ("CREATE", "UPDATE"):
            try:
                with session.begin_nested():
                    model = MODELS[row["entity_type"]]
                    if action == "CREATE":
                        target = model(
                            **row["values"],
                            t_plus_id=row["source_id"],
                            source_no=row.get("source_no"),
                            sync_batch_no=run.run_no,
                            synced_at=datetime.now(),
                        )
                        session.add(target)
                        session.flush()
                        session.add(
                            TPlusSourceLink(
                                source_database=run.source_database,
                                entity_type=row["entity_type"],
                                source_id=row["source_id"],
                                target_id=target.id,
                            )
                        )
                    else:
                        target = session.get(model, row["target_id"], with_for_update=True)
                        for field, value in row["values"].items():
                            setattr(target, field, value)
                        target.sync_batch_no, target.synced_at = run.run_no, datetime.now()
                    session.flush()
                    row["target_id"] = target.id
            except IntegrityError:
                action, reason = "CONFLICT", "TARGET_UNIQUE_CONFLICT"
        counts[action] += 1
        _line(session, run, i, row, action, reason)
    run.summary_json = {"counts": dict(counts), "total": len(rows), "formal_payroll_written": False}
    run.status = "PARTIAL" if counts["PENDING"] or counts["CONFLICT"] else "SUCCEEDED"