feat(cma): external input ledgers for M2 capability eval (tax/hr/licence)

3 new tables + idempotent DDL migration (prod guard: refuses DB name cma without approval): tax_filing_log (P_TAX_TIMELY), hr_monthly_ledger (L_EMPLOYEE_TURNOVER), licence_ledger (P_LICENCE_OK). Entry API /api/cma/ext-ledger: create / CSV batch import / query by entity+period / derive trigger. on_time is server-derived only; HR rows rejected when end != begin + hires - separations. Deriver writes kpi_values only (source_type=derived, idempotent upsert); denominator 0 returns None and writes nothing; never creates or changes kpi_definitions. Tests: 22 new pytest cases (22 passed); scratch-DB e2e battery 28 assertions PASS with zero production write.
2026-09-15 18:12:05 +08:00
parent 5d0884cc3b
commit 9d9a79545a
9 changed files with 1956 additions and 1 deletions
@@ -0,0 +1,579 @@
+"""外部输入台账录入/查询 API（能力评价 M2 准入3 派生，2026-09-15）
+
+三张台账的单向录入接口（新增 / CSV 批量导入 / 按实体与期次查询）+ 台账→kpi_values 派生触发。
+
+设计口径：
+- **单向**：只提供「新增 + 批量导入 + 查询 + 派生」，不提供修改/删除接口
+  （历史登记行不可改；错登由人工在业务侧作废后重新登记一行，保持审计链）
+- **on_time 由服务端派生**：请求里带 on_time 一律忽略并在响应回报 on_time_ignored
+- **人事台账等式校验**：期末 ≠ 期初 + 入职 − 离职 → 拒绝写入并回报（HTTP 400 / 批量逐行 errors）
+- **多租户**：所有端点经 app.deps.assert_entity_scope 收敛账套（显式 entity_id 与会话账套不一致 → 403）
+- **鉴权**：router 级 require_auth（未认证 401）；写端点额外 require_role(ceo/finance/it)（角色不足 403）
+- 派生必须**幂等**（同 (entity_id, kpi_id, period) 重复运行只更新不新增），分母为 0 → 不写值
+
+CSV 列头（UTF-8，可带 BOM；列序无关，多余列忽略）：
+  申报台账  period,tax_type,legal_deadline,filed_date,evidence_no,registered_by,remark
+  人事台账  period,headcount_begin,hires,separations,headcount_end,registered_by,remark
+  证照台账  licence_name,licence_no,valid_from,valid_to,annual_check_status,issuer,owner_role,attachment_no,registered_by,remark
+"""
+from __future__ import annotations
+
+import csv
+import io
+import logging
+import re
+from datetime import date, datetime
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from sqlalchemy.orm import Session
+
+from app.auth_middleware import require_auth, require_role
+from app.database import get_db
+from app.deps import assert_entity_scope
+from app.models import HrMonthlyLedger, KPIDefinition, LicenceLedger, TaxFilingLog
+from app.models.ext_input_ledger import ANNUAL_CHECK_STATUSES, MONTH_PERIOD_RE, QUARTER_PERIOD_RE
+from app.services.ext_input_derive import (
+    ALL_KPI_CODES,
+    LICENCE_OK_CODE,
+    check_headcount_equation,
+    compute_employee_turnover,
+    derive_on_time,
+    explain_licence_ok,
+    explain_tax_timely_rate,
+    is_licence_valid,
+    month_to_quarter,
+    quarter_end_date,
+    run_derive,
+)
+
+logger = logging.getLogger("cma.ext_input_api")
+
+router = APIRouter(
+    prefix="/api/cma/ext-ledger",
+    tags=["外部输入台账"],
+    dependencies=[Depends(require_auth)],
+)
+
+# 写权限（录入/导入/派生）：ceo / finance / it（与 value_sources 等既有写端点同口径）
+write_guard = Depends(require_role("ceo", "finance", "it"))
+
+SUGGESTED_TAX_TYPES = ("增值税", "附加税", "企业所得税", "印花税", "个人所得税", "房产税", "土地使用税")
+
+# CSV 全量列（文档口径）+ 必需列（列缺失才拒绝；可选列缺省即留空）
+TAX_HEADERS = ("period", "tax_type", "legal_deadline", "filed_date", "evidence_no", "registered_by", "remark")
+HR_HEADERS = ("period", "headcount_begin", "hires", "separations", "headcount_end", "registered_by", "remark")
+LICENCE_HEADERS = ("licence_name", "licence_no", "valid_from", "valid_to", "annual_check_status",
+                   "issuer", "owner_role", "attachment_no", "registered_by", "remark")
+TAX_REQUIRED = ("period", "tax_type")
+HR_REQUIRED = ("period", "headcount_begin", "hires", "separations", "headcount_end")
+LICENCE_REQUIRED = ("licence_name", "licence_no", "annual_check_status")
+
+
+# ============================================================
+# 公共工具：账套收敛 / 字段校验
+# ============================================================
+
+def _resolve_entity(request: Request, db: Session, explicit: Optional[int]) -> int:
+    """账套收敛：会话 token 绑定实体优先；显式 entity_id 与会话不一致 → 403（文案同 deps.get_entity_id）"""
+    session_entity = assert_entity_scope(request, explicit, db)
+    if explicit is not None:
+        return explicit
+    if session_entity is not None:
+        return session_entity
+    return 1
+
+
+def _body_entity(payload: dict, query_entity: Optional[int]) -> Optional[int]:
+    raw = payload.get("entity_id") if isinstance(payload, dict) else None
+    if raw in (None, ""):
+        return query_entity
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        raise HTTPException(400, f"entity_id 非法: {raw}")
+
+
+def _req_str(value, name: str, max_len: int) -> str:
+    text = "" if value is None else str(value).strip()
+    if not text:
+        raise HTTPException(400, f"缺少必填字段: {name}")
+    if len(text) > max_len:
+        raise HTTPException(400, f"字段 {name} 超长（{len(text)} > {max_len}）")
+    return text
+
+
+def _opt_str(value, max_len: int) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    return text[:max_len]
+
+
+def _parse_date(value, name: str) -> Optional[date]:
+    if value in (None, ""):
+        return None
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    text = str(value).strip().replace("/", "-")
+    for fmt in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y%m%d"):
+        try:
+            return datetime.strptime(text, fmt).date()
+        except ValueError:
+            continue
+    raise HTTPException(400, f"字段 {name} 日期格式非法: {value}（应为 YYYY-MM-DD）")
+
+
+def _parse_int(value, name: str, required: bool = True) -> Optional[int]:
+    if value in (None, ""):
+        if required:
+            raise HTTPException(400, f"缺少必填字段: {name}")
+        return None
+    try:
+        return int(str(value).strip())
+    except (TypeError, ValueError):
+        raise HTTPException(400, f"字段 {name} 必须为整数: {value}")
+
+
+def _month_period(value, name: str = "period") -> str:
+    text = _req_str(value, name, 20)
+    if not re.match(MONTH_PERIOD_RE, text):
+        raise HTTPException(400, f"字段 {name} 期次格式非法: {text}（应为 YYYY-MM）")
+    return text
+
+
+def _target_period(kpi_code: str, period: str) -> str:
+    """指标实际落库期间：P_LICENCE_OK 为季频（传月期间时归到所在季度）"""
+    if kpi_code == LICENCE_OK_CODE and re.match(MONTH_PERIOD_RE, period):
+        return month_to_quarter(period)
+    return period
+
+
+# ============================================================
+# 字段选项（前端下拉动态加载用；禁前端硬编码）
+# ============================================================
+
+@router.get("/options")
+def get_options():
+    """录入表单选项：年检状态枚举 + 建议税种 + 期次格式"""
+    return {
+        "annual_check_status": list(ANNUAL_CHECK_STATUSES),
+        "tax_type_suggested": list(SUGGESTED_TAX_TYPES),
+        "period_format": {"monthly": "YYYY-MM 例:2026-08", "quarterly": "YYYY-Qn 例:2026-Q2"},
+        "derivable_kpi_codes": list(ALL_KPI_CODES),
+    }
+
+
+# ============================================================
+# 1. 申报行为台账 tax_filing_log
+# ============================================================
+
+def _tax_values(payload: dict) -> dict:
+    period = _month_period(payload.get("period"))
+    tax_type = _req_str(payload.get("tax_type"), "tax_type", 50)
+    legal_deadline = _parse_date(payload.get("legal_deadline"), "legal_deadline")
+    filed_date = _parse_date(payload.get("filed_date"), "filed_date")
+    return {
+        "period": period,
+        "tax_type": tax_type,
+        "legal_deadline": legal_deadline,
+        "filed_date": filed_date,
+        # on_time 唯一来源 = 服务端派生（请求里的同名字段一律忽略）
+        "on_time": derive_on_time(filed_date, legal_deadline),
+        "evidence_no": _opt_str(payload.get("evidence_no"), 100),
+        "registered_by": _opt_str(payload.get("registered_by"), 100),
+        "remark": _opt_str(payload.get("remark"), 500),
+    }
+
+
+def _tax_json(row: TaxFilingLog) -> dict:
+    return {
+        "id": row.id,
+        "entity_id": row.entity_id,
+        "period": row.period,
+        "tax_type": row.tax_type,
+        "legal_deadline": row.legal_deadline.isoformat() if row.legal_deadline else None,
+        "filed_date": row.filed_date.isoformat() if row.filed_date else None,
+        "on_time": row.on_time,
+        "evidence_no": row.evidence_no,
+        "registered_by": row.registered_by,
+        "remark": row.remark,
+    }
+
+
+@router.get("/tax-filings")
+def list_tax_filings(
+    request: Request,
+    period: Optional[str] = Query(None, description="属期 YYYY-MM"),
+    tax_type: Optional[str] = Query(None),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """按 (entity_id, period) 查询申报台账 + 及时率口径摘要"""
+    eid = _resolve_entity(request, db, entity_id)
+    q = db.query(TaxFilingLog).filter(TaxFilingLog.entity_id == eid)
+    if period:
+        q = q.filter(TaxFilingLog.period == period)
+    if tax_type:
+        q = q.filter(TaxFilingLog.tax_type == tax_type)
+    rows = q.order_by(TaxFilingLog.period.desc(), TaxFilingLog.id.asc()).all()
+    summary = explain_tax_timely_rate([r for r in rows if not period or r.period == period])
+    return {"entity_id": eid, "data": [_tax_json(r) for r in rows], "total": len(rows),
+            "summary": {"period": period, "tax_timely_rate": summary["value"],
+                        "denominator": summary["denominator"], "timely": summary["timely"],
+                        "no_definition": summary["no_definition"]}}
+
+
+@router.post("/tax-filings", status_code=201, dependencies=[write_guard])
+def create_tax_filing(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """新增一条申报行为台账（on_time 服务端派生；唯一键冲突 → 409）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    vals = _tax_values(payload)
+    exists = db.query(TaxFilingLog).filter(
+        TaxFilingLog.entity_id == eid, TaxFilingLog.period == vals["period"],
+        TaxFilingLog.tax_type == vals["tax_type"]).first()
+    if exists:
+        raise HTTPException(409, f"已存在同 (entity_id={eid}, period={vals['period']}, "
+                                 f"tax_type={vals['tax_type']}) 的登记行 id={exists.id}")
+    row = TaxFilingLog(entity_id=eid, **vals)
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return {"data": _tax_json(row),
+            "on_time_derived": row.on_time,
+            "on_time_ignored": payload.get("on_time") not in (None, "")}
+
+
+@router.post("/tax-filings/import", dependencies=[write_guard])
+def import_tax_filings(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """CSV 批量导入申报台账。on_conflict: skip(默认)/update；dry_run=true 只校验不落库"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    on_conflict = str(payload.get("on_conflict") or "skip").lower()
+    if on_conflict not in ("skip", "update"):
+        raise HTTPException(400, f"on_conflict 非法: {on_conflict}（skip/update）")
+    dry_run = bool(payload.get("dry_run"))
+    records, errors = _read_csv(payload.get("csv"), TAX_REQUIRED, _tax_values)
+    result = _apply_import(db, TaxFilingLog, eid, records, errors, on_conflict, dry_run,
+                           key_fields=("period", "tax_type"))
+    return result
+
+
+# ============================================================
+# 2. 人事月度台账 hr_monthly_ledger
+# ============================================================
+
+def _hr_values(payload: dict) -> dict:
+    period = _month_period(payload.get("period"))
+    begin = _parse_int(payload.get("headcount_begin"), "headcount_begin")
+    hires = _parse_int(payload.get("hires"), "hires")
+    separations = _parse_int(payload.get("separations"), "separations")
+    end = _parse_int(payload.get("headcount_end"), "headcount_end")
+    err = check_headcount_equation(begin, hires, separations, end)
+    if err:
+        raise HTTPException(400, err)
+    return {
+        "period": period,
+        "headcount_begin": begin,
+        "hires": hires,
+        "separations": separations,
+        "headcount_end": end,
+        "registered_by": _opt_str(payload.get("registered_by"), 100),
+        "remark": _opt_str(payload.get("remark"), 500),
+    }
+
+
+def _hr_json(row: HrMonthlyLedger) -> dict:
+    return {
+        "id": row.id, "entity_id": row.entity_id, "period": row.period,
+        "headcount_begin": row.headcount_begin, "hires": row.hires,
+        "separations": row.separations, "headcount_end": row.headcount_end,
+        "turnover_rate": compute_employee_turnover(row),
+        "registered_by": row.registered_by, "remark": row.remark,
+    }
+
+
+@router.get("/hr-monthly")
+def list_hr_monthly(
+    request: Request,
+    period: Optional[str] = Query(None, description="期间 YYYY-MM"),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """按 (entity_id, period) 查询人事月度台账（含流失率派生值）"""
+    eid = _resolve_entity(request, db, entity_id)
+    q = db.query(HrMonthlyLedger).filter(HrMonthlyLedger.entity_id == eid)
+    if period:
+        q = q.filter(HrMonthlyLedger.period == period)
+    rows = q.order_by(HrMonthlyLedger.period.desc(), HrMonthlyLedger.id.asc()).all()
+    return {"entity_id": eid, "data": [_hr_json(r) for r in rows], "total": len(rows)}
+
+
+@router.post("/hr-monthly", status_code=201, dependencies=[write_guard])
+def create_hr_monthly(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """新增人事月度台账（等式校验不通过 → 400 拒绝写入）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    vals = _hr_values(payload)
+    exists = db.query(HrMonthlyLedger).filter(
+        HrMonthlyLedger.entity_id == eid, HrMonthlyLedger.period == vals["period"]).first()
+    if exists:
+        raise HTTPException(409, f"已存在同 (entity_id={eid}, period={vals['period']}) 的登记行 id={exists.id}")
+    row = HrMonthlyLedger(entity_id=eid, **vals)
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return {"data": _hr_json(row)}
+
+
+@router.post("/hr-monthly/import", dependencies=[write_guard])
+def import_hr_monthly(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """CSV 批量导入人事月度台账（等式不成立的行被拒绝并在 errors 回报）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    on_conflict = str(payload.get("on_conflict") or "skip").lower()
+    if on_conflict not in ("skip", "update"):
+        raise HTTPException(400, f"on_conflict 非法: {on_conflict}（skip/update）")
+    dry_run = bool(payload.get("dry_run"))
+    records, errors = _read_csv(payload.get("csv"), HR_REQUIRED, _hr_values)
+    return _apply_import(db, HrMonthlyLedger, eid, records, errors, on_conflict, dry_run,
+                         key_fields=("period",))
+
+
+# ============================================================
+# 3. 证照台账 licence_ledger
+# ============================================================
+
+def _licence_values(payload: dict) -> dict:
+    name = _req_str(payload.get("licence_name"), "licence_name", 200)
+    no = _req_str(payload.get("licence_no"), "licence_no", 100)
+    status = _req_str(payload.get("annual_check_status"), "annual_check_status", 20)
+    if status not in ANNUAL_CHECK_STATUSES:
+        raise HTTPException(400, f"字段 annual_check_status 非法: {status}（合法值: {'/'.join(ANNUAL_CHECK_STATUSES)}）")
+    return {
+        "licence_name": name,
+        "licence_no": no,
+        "valid_from": _parse_date(payload.get("valid_from"), "valid_from"),
+        "valid_to": _parse_date(payload.get("valid_to"), "valid_to"),
+        "annual_check_status": status,
+        "issuer": _opt_str(payload.get("issuer"), 200),
+        "owner_role": _opt_str(payload.get("owner_role"), 100),
+        "attachment_no": _opt_str(payload.get("attachment_no"), 100),
+        "registered_by": _opt_str(payload.get("registered_by"), 100),
+        "remark": _opt_str(payload.get("remark"), 500),
+    }
+
+
+def _licence_json(row: LicenceLedger, as_of: Optional[date] = None) -> dict:
+    return {
+        "id": row.id, "entity_id": row.entity_id, "licence_name": row.licence_name,
+        "licence_no": row.licence_no,
+        "valid_from": row.valid_from.isoformat() if row.valid_from else None,
+        "valid_to": row.valid_to.isoformat() if row.valid_to else None,
+        "annual_check_status": row.annual_check_status, "issuer": row.issuer,
+        "owner_role": row.owner_role, "attachment_no": row.attachment_no,
+        "valid_as_of": is_licence_valid(row, as_of) if as_of else None,
+        "registered_by": row.registered_by, "remark": row.remark,
+    }
+
+
+@router.get("/licences")
+def list_licences(
+    request: Request,
+    as_of: Optional[str] = Query(None, description="有效性基准日 YYYY-MM-DD（默认今天）"),
+    licence_name: Optional[str] = Query(None),
+    valid_only: bool = Query(False, description="仅返回基准日有效的证照"),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """证照台账查询（有效/过期按基准日判定）"""
+    eid = _resolve_entity(request, db, entity_id)
+    base = _parse_date(as_of, "as_of") or date.today()
+    q = db.query(LicenceLedger).filter(LicenceLedger.entity_id == eid)
+    if licence_name:
+        q = q.filter(LicenceLedger.licence_name == licence_name)
+    rows = q.order_by(LicenceLedger.id.asc()).all()
+    # 合规率分母 = 应持证照数（不受 valid_only 过滤影响）
+    summary = explain_licence_ok(rows, base)
+    shown = [r for r in rows if is_licence_valid(r, base)] if valid_only else rows
+    return {"entity_id": eid, "as_of": base.isoformat(),
+            "data": [_licence_json(r, base) for r in shown], "total": len(shown),
+            "summary": {"licence_ok": summary["value"], "denominator": summary["denominator"],
+                        "valid": summary["valid"], "no_definition": summary["no_definition"]}}
+
+
+@router.post("/licences", status_code=201, dependencies=[write_guard])
+def create_licence(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """新增证照台账（状态白名单校验；唯一键冲突 → 409）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    vals = _licence_values(payload)
+    exists = db.query(LicenceLedger).filter(
+        LicenceLedger.entity_id == eid, LicenceLedger.licence_name == vals["licence_name"],
+        LicenceLedger.licence_no == vals["licence_no"]).first()
+    if exists:
+        raise HTTPException(409, f"已存在同 (entity_id={eid}, licence_name={vals['licence_name']}, "
+                                 f"licence_no={vals['licence_no']}) 的登记行 id={exists.id}")
+    row = LicenceLedger(entity_id=eid, **vals)
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return {"data": _licence_json(row, date.today())}
+
+
+@router.post("/licences/import", dependencies=[write_guard])
+def import_licences(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """CSV 批量导入证照台账"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    on_conflict = str(payload.get("on_conflict") or "skip").lower()
+    if on_conflict not in ("skip", "update"):
+        raise HTTPException(400, f"on_conflict 非法: {on_conflict}（skip/update）")
+    dry_run = bool(payload.get("dry_run"))
+    records, errors = _read_csv(payload.get("csv"), LICENCE_REQUIRED, _licence_values)
+    return _apply_import(db, LicenceLedger, eid, records, errors, on_conflict, dry_run,
+                         key_fields=("licence_name", "licence_no"))
+
+
+# ============================================================
+# 4. 台账 → kpi_values 派生
+# ============================================================
+
+@router.post("/derive", dependencies=[write_guard])
+def derive_kpis(
+    request: Request,
+    payload: dict,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """触发台账→kpi_values 派生（幂等）。period: 月频 YYYY-MM；P_LICENCE_OK 接受 YYYY-Qn。
+    dry_run=true 只算不写。分母为 0 的指标 → 不写值，回报 skipped_denominator_zero。"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    period = str(payload.get("period") or "").strip()
+    if not period:
+        raise HTTPException(400, "缺少必填字段: period")
+    if not (re.match(MONTH_PERIOD_RE, period) or re.match(QUARTER_PERIOD_RE, period)):
+        raise HTTPException(400, f"period 非法: {period}（应为 YYYY-MM 或 YYYY-Qn）")
+    kpi_codes = payload.get("kpi_codes") or payload.get("kpis") or None
+    if kpi_codes is not None and not isinstance(kpi_codes, list):
+        raise HTTPException(400, "kpi_codes 必须为数组")
+    try:
+        result = run_derive(db, eid, period, kpis=kpi_codes, write=not bool(payload.get("dry_run")))
+    except ValueError as e:
+        raise HTTPException(400, str(e))
+    return result
+
+
+@router.get("/derive/status")
+def derive_status(
+    request: Request,
+    period: str = Query(..., description="月期间 YYYY-MM 或季期间 YYYY-Qn"),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """只读：查看该期派生指标的定义/现有 kpi_values 行（不写入）"""
+    eid = _resolve_entity(request, db, entity_id)
+    if not (re.match(MONTH_PERIOD_RE, period) or re.match(QUARTER_PERIOD_RE, period)):
+        raise HTTPException(400, f"period 非法: {period}（应为 YYYY-MM 或 YYYY-Qn）")
+    out = {}
+    from app.models import KPIValue
+    for code in ALL_KPI_CODES:
+        kpi = (db.query(KPIDefinition)
+               .filter(KPIDefinition.entity_id == eid, KPIDefinition.kpi_code == code,
+                       KPIDefinition.status == "active")
+               .order_by(KPIDefinition.id.asc()).first())
+        item = {"kpi_defined": kpi is not None, "kpi_id": kpi.id if kpi else None,
+                "target_period": _target_period(code, period)}
+        if kpi is not None:
+            row = (db.query(KPIValue)
+                   .filter(KPIValue.kpi_id == kpi.id, KPIValue.source_type == "derived",
+                           KPIValue.period == item["target_period"])
+                   .order_by(KPIValue.id.desc()).first())
+            item["latest_derived"] = ({"period": row.period, "actual_value": row.actual_value,
+                                       "data_status": row.data_status} if row else None)
+        out[code] = item
+    return {"entity_id": eid, "period": period, "items": out}
+
+
+# ============================================================
+# CSV 解析 / 批量落库（三张表共用；校验规则与单条新增完全同源）
+# ============================================================
+
+def _read_csv(csv_text, required_headers, builder):
+    """解析 CSV：返回 (records, errors)。required_headers 为必需列（缺列即 400）；其余列可选。"""
+    if not csv_text or not str(csv_text).strip():
+        raise HTTPException(400, "缺少必填字段: csv")
+    text = str(csv_text).lstrip("\ufeff")
+    reader = csv.DictReader(io.StringIO(text))
+    fieldnames = [(f or "").strip().lower() for f in (reader.fieldnames or [])]
+    if not fieldnames:
+        raise HTTPException(400, "CSV 缺少表头行")
+    missing = [h for h in required_headers if h not in fieldnames]
+    if missing:
+        raise HTTPException(400, f"CSV 缺少必需列: {missing}（必需列: {list(required_headers)}）")
+    records, errors = [], []
+    for offset, raw in enumerate(reader, start=2):  # 表头占第 1 行
+        row = {(k or "").strip().lower(): v for k, v in raw.items() if k}
+        if not any(str(v or "").strip() for v in row.values()):
+            continue  # 跳过空行
+        try:
+            records.append({"line": offset, "values": builder(row)})
+        except HTTPException as e:
+            errors.append({"line": offset, "reason": e.detail})
+    return records, errors
+
+
+def _apply_import(db: Session, model, entity_id: int, records, errors, on_conflict: str,
+                  dry_run: bool, key_fields) -> dict:
+    """逐行 upsert；返回 {inserted, updated, skipped, errors, dry_run}"""
+    inserted = updated = skipped = 0
+    for rec in records:
+        vals = rec["values"]
+        filters = [getattr(model, f) == vals[f] for f in key_fields] + [model.entity_id == entity_id]
+        existing = db.query(model).filter(*filters).first()
+        if existing is not None:
+            if on_conflict == "update":
+                for k, v in vals.items():
+                    setattr(existing, k, v)
+                updated += 1
+            else:
+                skipped += 1
+        else:
+            db.add(model(entity_id=entity_id, **vals))
+            inserted += 1
+    if dry_run:
+        db.rollback()
+    else:
+        db.commit()
+    return {"entity_id": entity_id, "dry_run": dry_run, "on_conflict": on_conflict,
+            "inserted": inserted, "updated": updated, "skipped": skipped,
+            "errors": errors, "error_count": len(errors)}