15006a05a7
- 新增数字员工财务报告生成、邮件投递与渲染调度器 - 引入员工画像扫描调度与定时提醒任务 - 完善财务看板快照、排行口径与部门人员占比计算 - 优化数字员工工作看板仪表盘与技能目录 - 增强前端总览页图表、工作台摘要与顶部导航栏交互 - 新增差旅申请规划推动提醒与报销创建会话状态管理 - 补充财务报告、看板调度、数字员工工作记录测试覆盖
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
|
|
SERVER_DIR = Path(__file__).resolve().parents[1]
|
|
SRC_DIR = SERVER_DIR / "src"
|
|
if str(SRC_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SRC_DIR))
|
|
|
|
from app.db.session import get_session_factory # noqa: E402
|
|
from app.models.financial_record import ExpenseClaim # noqa: E402
|
|
from app.services.expense_claim_status_registry import ( # noqa: E402
|
|
is_known_approval_stage,
|
|
is_known_claim_status,
|
|
normalize_expense_claim_state,
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Audit expense claim status consistency.")
|
|
parser.add_argument("--sample-limit", type=int, default=20)
|
|
args = parser.parse_args()
|
|
|
|
session_factory = get_session_factory()
|
|
with session_factory() as db:
|
|
claims = list(
|
|
db.scalars(
|
|
select(ExpenseClaim).order_by(
|
|
ExpenseClaim.claim_no.asc(),
|
|
ExpenseClaim.created_at.asc(),
|
|
)
|
|
).all()
|
|
)
|
|
payload = audit_claims(claims, sample_limit=max(args.sample_limit, 0))
|
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
|
|
|
|
def audit_claims(claims: list[ExpenseClaim], *, sample_limit: int) -> dict[str, Any]:
|
|
status_counts: Counter[str] = Counter()
|
|
stage_counts: Counter[str] = Counter()
|
|
status_stage_counts: Counter[str] = Counter()
|
|
doc_type_counts: Counter[str] = Counter()
|
|
unknown_statuses: Counter[str] = Counter()
|
|
unknown_stages: Counter[str] = Counter()
|
|
normalization_counts: Counter[str] = Counter()
|
|
samples: list[dict[str, Any]] = []
|
|
|
|
for claim in claims:
|
|
status = str(claim.status or "").strip()
|
|
stage = str(claim.approval_stage or "").strip()
|
|
doc_type = _doc_type(claim)
|
|
status_counts[status or "<empty>"] += 1
|
|
stage_counts[stage or "<empty>"] += 1
|
|
status_stage_counts[f"{status or '<empty>'} | {stage or '<empty>'}"] += 1
|
|
doc_type_counts[doc_type] += 1
|
|
|
|
if not is_known_claim_status(status):
|
|
unknown_statuses[status or "<empty>"] += 1
|
|
if not is_known_approval_stage(stage):
|
|
unknown_stages[stage or "<empty>"] += 1
|
|
|
|
normalized = normalize_expense_claim_state(
|
|
status,
|
|
stage,
|
|
claim_no=claim.claim_no,
|
|
expense_type=claim.expense_type,
|
|
)
|
|
if normalized.changed:
|
|
key = (
|
|
f"{status or '<empty>'}/{stage or '<empty>'}"
|
|
f" -> {normalized.status}/{normalized.approval_stage}"
|
|
)
|
|
normalization_counts[key] += 1
|
|
if len(samples) < sample_limit:
|
|
samples.append(
|
|
{
|
|
"claim_no": claim.claim_no,
|
|
"doc_type": doc_type,
|
|
"status": status,
|
|
"approval_stage": stage,
|
|
"normalized_status": normalized.status,
|
|
"normalized_approval_stage": normalized.approval_stage,
|
|
"status_code": normalized.status_code,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"claim_count": len(claims),
|
|
"doc_type_counts": dict(doc_type_counts),
|
|
"status_counts": dict(status_counts),
|
|
"approval_stage_counts": dict(stage_counts),
|
|
"status_stage_counts": dict(status_stage_counts),
|
|
"unknown_statuses": dict(unknown_statuses),
|
|
"unknown_approval_stages": dict(unknown_stages),
|
|
"normalization_needed": sum(normalization_counts.values()),
|
|
"normalization_counts": dict(normalization_counts),
|
|
"normalization_samples": samples,
|
|
}
|
|
|
|
|
|
def _doc_type(claim: ExpenseClaim) -> str:
|
|
claim_no = str(claim.claim_no or "").strip().upper()
|
|
expense_type = str(claim.expense_type or "").strip().lower()
|
|
if claim_no.startswith(("AP-", "APP-")) or expense_type.endswith("_application"):
|
|
return "application"
|
|
if str(claim.project_code or "").strip().upper() == "SIM-DEMO":
|
|
return "sim_reimbursement"
|
|
return "reimbursement"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|