Hermes CI Fix
9d9a79545a
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.
580 lines
25 KiB
Python
580 lines
25 KiB
Python
"""外部输入台账录入/查询 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)}
|