Hermes CI Fix
9986f38faa
- 后端: kpis/bsc_layers/cash/predict/growth_quality/tax_compliance/data 的 entity_id 参数统一改为 Depends(get_entity_id) - 前端: 拦截器删除自动附加X-Entity-Id/entity_id; 登录页公司选择器(按用户名授权过滤); 切换器改POST /auth/switch-entity重新签发token+整页刷新; ReportCenter同步改造 - 修复前后端不匹配: login-entities路由→/auth/entities; login响应entity_id取user.entity_id
886 lines
34 KiB
Python
886 lines
34 KiB
Python
"""税务合规智能体 API — 管理会计OS
|
||
① 税负监控: 税务记录CRUD + 税负率计算(实缴/收入×100%) + 行业基准预警(增值税3.5%/所得税2.5%, 超±20%预警)
|
||
② 发票校验: 发票录入 + 批量校验(发票号33位数字 / 金额与报销单匹配 / 供应商与合同匹配) + 异常查询
|
||
③ 社保比对: 缴费记录CRUD + 比对(基数与工资匹配60%~300% / 单位缴纳比例24.5% / 漏缴提醒) + 异常查询
|
||
看板: /tax/dashboard 聚合税负趋势+行业对比 + 发票异常列表 + 社保异常列表
|
||
"""
|
||
import json
|
||
from datetime import datetime, date
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.database import get_db
|
||
from app.deps import get_entity_id
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.models import TaxRecord, InvoiceCheck, SocialSecurity, ExpenseReimbursement
|
||
|
||
router = APIRouter(prefix="/api/cma/tax", tags=["税务合规"])
|
||
|
||
# ── 行业平均税负率参考值 (%) ──
|
||
TAX_BENCHMARKS = {
|
||
"vat": 3.5, # 增值税平均税负率
|
||
"income": 2.5, # 企业所得税平均税负率
|
||
"surtax": 0.5, # 附加税平均税负率
|
||
}
|
||
TAX_TYPE_LABELS = {"vat": "增值税", "income": "所得税", "surtax": "附加税"}
|
||
INVOICE_TYPE_LABELS = {"vat": "增值税专用发票", "vat_normal": "增值税普通发票", "electronic": "电子发票", "other": "其他"}
|
||
SS_COMPANY_RATE = 24.5 # 单位缴纳比例: 养老16% + 医疗8% + 失业0.5%
|
||
SS_RATE_TOLERANCE = 1.0 # 比例允许偏差 ±1%
|
||
SS_BASE_LOW = 0.6 # 基数下限 = 工资60%
|
||
SS_BASE_HIGH = 3.0 # 基数上限 = 工资300%
|
||
INVOICE_NO_LEN = 33 # 发票号位数（全数字）
|
||
|
||
|
||
# ── 序列化 ──
|
||
def _tax_to_dict(t: TaxRecord) -> dict:
|
||
return {
|
||
"id": t.id,
|
||
"entity_id": t.entity_id,
|
||
"period": t.period,
|
||
"tax_type": t.tax_type,
|
||
"tax_type_label": TAX_TYPE_LABELS.get(t.tax_type, t.tax_type),
|
||
"tax_payable": t.tax_payable,
|
||
"tax_paid": t.tax_paid,
|
||
"tax_rate": t.tax_rate,
|
||
"income": t.income,
|
||
"tax_burden_rate": t.tax_burden_rate,
|
||
"benchmark": TAX_BENCHMARKS.get(t.tax_type),
|
||
"burden_status": t.burden_status,
|
||
"warning_msg": t.warning_msg,
|
||
"remark": t.remark,
|
||
"created_at": t.created_at.isoformat() if t.created_at else None,
|
||
}
|
||
|
||
|
||
def _invoice_to_dict(i: InvoiceCheck) -> dict:
|
||
return {
|
||
"id": i.id,
|
||
"entity_id": i.entity_id,
|
||
"invoice_no": i.invoice_no,
|
||
"amount": i.amount,
|
||
"invoice_type": i.invoice_type,
|
||
"invoice_type_label": INVOICE_TYPE_LABELS.get(i.invoice_type, i.invoice_type),
|
||
"invoice_date": i.invoice_date.isoformat() if i.invoice_date else None,
|
||
"supplier": i.supplier,
|
||
"reimb_no": i.reimb_no,
|
||
"contract_no": i.contract_no,
|
||
"check_status": i.check_status,
|
||
"check_result": i.check_result,
|
||
"check_reason": i.check_reason,
|
||
"checked_at": i.checked_at.isoformat() if i.checked_at else None,
|
||
"created_at": i.created_at.isoformat() if i.created_at else None,
|
||
}
|
||
|
||
|
||
def _ss_to_dict(s: SocialSecurity) -> dict:
|
||
return {
|
||
"id": s.id,
|
||
"entity_id": s.entity_id,
|
||
"employee": s.employee,
|
||
"period": s.period,
|
||
"base_amount": s.base_amount,
|
||
"salary": s.salary,
|
||
"company_amount": s.company_amount,
|
||
"personal_amount": s.personal_amount,
|
||
"company_rate": s.company_rate,
|
||
"check_status": s.check_status,
|
||
"warning_msg": s.warning_msg,
|
||
"remark": s.remark,
|
||
"created_at": s.created_at.isoformat() if s.created_at else None,
|
||
}
|
||
|
||
|
||
def _parse_date(v) -> datetime | None:
|
||
if not v:
|
||
return None
|
||
try:
|
||
if isinstance(v, datetime):
|
||
return v
|
||
if isinstance(v, date):
|
||
return datetime(v.year, v.month, v.day)
|
||
return datetime.fromisoformat(str(v)[:10])
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# ============================================================
|
||
# ① 税负监控 — 税务记录 CRUD + 税负率计算 + 行业基准预警
|
||
# ============================================================
|
||
|
||
def _calc_burden(t: TaxRecord):
|
||
"""计算单条税务记录的税负率 + 行业基准预警"""
|
||
rate = None
|
||
if t.income and t.income > 0:
|
||
rate = round(t.tax_paid / t.income * 100, 2)
|
||
t.tax_burden_rate = rate
|
||
bench = TAX_BENCHMARKS.get(t.tax_type)
|
||
t.burden_status = "normal"
|
||
t.warning_msg = None
|
||
if rate is None or bench is None:
|
||
return
|
||
low, high = bench * 0.8, bench * 1.2
|
||
if rate < low or rate > high:
|
||
t.burden_status = "alert"
|
||
t.warning_msg = (
|
||
f"税负率{rate}%超出行业均值{bench}%的±20%区间({low}%~{high}%)，"
|
||
f"{'偏高需核查进项/优惠' if rate > high else '偏低需核查申报完整性'}"
|
||
)
|
||
|
||
|
||
def _apply_burden_to_all(db: Session, entity_id: int):
|
||
"""重算某企业全部税务记录的税负率与预警"""
|
||
records = db.query(TaxRecord).filter(TaxRecord.entity_id == entity_id).all()
|
||
for t in records:
|
||
_calc_burden(t)
|
||
db.commit()
|
||
return records
|
||
|
||
|
||
@router.get("/records")
|
||
def list_tax_records(
|
||
period: str = Query(None),
|
||
tax_type: str = Query(None),
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_auth),
|
||
):
|
||
"""查询税务记录列表"""
|
||
q = db.query(TaxRecord)
|
||
if entity_id:
|
||
q = q.filter(TaxRecord.entity_id == entity_id)
|
||
if period:
|
||
q = q.filter(TaxRecord.period == period)
|
||
if tax_type:
|
||
q = q.filter(TaxRecord.tax_type == tax_type)
|
||
records = q.order_by(TaxRecord.period.desc(), TaxRecord.id.asc()).all()
|
||
return {"data": [_tax_to_dict(t) for t in records], "total": len(records)}
|
||
|
||
|
||
@router.post("/records")
|
||
def create_tax_record(
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""新增税务记录"""
|
||
period = data.get("period")
|
||
tax_type = data.get("tax_type")
|
||
if not period or not tax_type:
|
||
raise HTTPException(400, "缺少必要参数: period, tax_type")
|
||
if tax_type not in TAX_TYPE_LABELS:
|
||
raise HTTPException(400, f"无效税种: {tax_type}，可选 vat/income/surtax")
|
||
t = TaxRecord(
|
||
entity_id=data.get("entity_id") or 1,
|
||
period=period,
|
||
tax_type=tax_type,
|
||
tax_payable=data.get("tax_payable") or 0,
|
||
tax_paid=data.get("tax_paid") or 0,
|
||
tax_rate=data.get("tax_rate"),
|
||
income=data.get("income") or 0,
|
||
remark=data.get("remark"),
|
||
)
|
||
_calc_burden(t)
|
||
db.add(t)
|
||
db.commit()
|
||
db.refresh(t)
|
||
return {"message": "税务记录已创建", "data": _tax_to_dict(t)}
|
||
|
||
|
||
@router.put("/records/{record_id}")
|
||
def update_tax_record(
|
||
record_id: int,
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""更新税务记录（自动重算税负率与预警）"""
|
||
t = db.query(TaxRecord).filter(TaxRecord.id == record_id).first()
|
||
if not t:
|
||
raise HTTPException(404, "税务记录不存在")
|
||
if "period" in data:
|
||
t.period = data["period"]
|
||
if "tax_type" in data:
|
||
if data["tax_type"] not in TAX_TYPE_LABELS:
|
||
raise HTTPException(400, f"无效税种: {data['tax_type']}")
|
||
t.tax_type = data["tax_type"]
|
||
if "tax_payable" in data:
|
||
t.tax_payable = data["tax_payable"] or 0
|
||
if "tax_paid" in data:
|
||
t.tax_paid = data["tax_paid"] or 0
|
||
if "tax_rate" in data:
|
||
t.tax_rate = data.get("tax_rate")
|
||
if "income" in data:
|
||
t.income = data["income"] or 0
|
||
if "remark" in data:
|
||
t.remark = data.get("remark")
|
||
_calc_burden(t)
|
||
db.commit()
|
||
db.refresh(t)
|
||
return {"message": "税务记录已更新", "data": _tax_to_dict(t)}
|
||
|
||
|
||
@router.delete("/records/{record_id}")
|
||
def delete_tax_record(
|
||
record_id: int,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""删除税务记录"""
|
||
t = db.query(TaxRecord).filter(TaxRecord.id == record_id).first()
|
||
if not t:
|
||
raise HTTPException(404, "税务记录不存在")
|
||
db.delete(t)
|
||
db.commit()
|
||
return {"message": "税务记录已删除"}
|
||
|
||
|
||
@router.get("/burden")
|
||
def burden_analysis(
|
||
entity_id: int = Depends(get_entity_id),
|
||
tax_type: str = Query(None),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_auth),
|
||
):
|
||
"""税负率计算 + 行业对比 + 预警列表（前端趋势图数据源）"""
|
||
q = db.query(TaxRecord)
|
||
if entity_id:
|
||
q = q.filter(TaxRecord.entity_id == entity_id)
|
||
if tax_type:
|
||
q = q.filter(TaxRecord.tax_type == tax_type)
|
||
records = q.order_by(TaxRecord.period.asc(), TaxRecord.id.asc()).all()
|
||
for t in records:
|
||
_calc_burden(t)
|
||
db.commit()
|
||
|
||
# 按期间聚合税负率（每种税一个序列）
|
||
trend_map: dict[str, dict] = {}
|
||
for t in records:
|
||
if t.tax_burden_rate is None:
|
||
continue
|
||
entry = trend_map.setdefault(t.period, {"period": t.period})
|
||
entry[f"{t.tax_type}_rate"] = t.tax_burden_rate
|
||
entry[f"{t.tax_type}_benchmark"] = TAX_BENCHMARKS.get(t.tax_type)
|
||
trend = sorted(trend_map.values(), key=lambda x: x["period"])
|
||
|
||
alerts = [t for t in records if t.burden_status == "alert"]
|
||
return {
|
||
"trend": trend,
|
||
"benchmarks": TAX_BENCHMARKS,
|
||
"records": [_tax_to_dict(t) for t in records],
|
||
"alerts": [_tax_to_dict(t) for t in alerts],
|
||
"alert_count": len(alerts),
|
||
}
|
||
|
||
|
||
@router.post("/check")
|
||
def run_tax_check(
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""手动触发税负率重算与预警检查"""
|
||
records = _apply_burden_to_all(db, entity_id or 1)
|
||
alerts = [t for t in records if t.burden_status == "alert"]
|
||
return {"message": f"税负检查完成，共{len(records)}条记录，{len(alerts)}条预警", "alert_count": len(alerts)}
|
||
|
||
|
||
# ============================================================
|
||
# ② 发票校验 — 录入 + 批量校验 + 异常查询
|
||
# ============================================================
|
||
|
||
def _check_invoice(db: Session, inv: InvoiceCheck):
|
||
"""发票校验规则：号码格式 / 金额与报销单匹配 / 供应商与合同匹配"""
|
||
results: list[dict] = []
|
||
status = "valid"
|
||
|
||
# 规则1: 发票号格式（33位数字）
|
||
no = (inv.invoice_no or "").strip()
|
||
if len(no) != INVOICE_NO_LEN or not no.isdigit():
|
||
status = "invalid"
|
||
results.append({
|
||
"rule": "发票号格式",
|
||
"passed": False,
|
||
"message": f"发票号格式错误：应为{INVOICE_NO_LEN}位纯数字，当前{len(no)}位",
|
||
})
|
||
else:
|
||
results.append({"rule": "发票号格式", "passed": True, "message": "33位数字格式正确"})
|
||
|
||
# 规则2: 金额与报销单匹配
|
||
if inv.reimb_no:
|
||
reimb = db.query(ExpenseReimbursement).filter(ExpenseReimbursement.reimb_no == inv.reimb_no).first()
|
||
if not reimb:
|
||
status = "invalid"
|
||
results.append({"rule": "报销单匹配", "passed": False, "message": f"报销单{inv.reimb_no}不存在"})
|
||
elif abs((inv.amount or 0) - (reimb.amount or 0)) > 0.01:
|
||
status = "invalid"
|
||
results.append({
|
||
"rule": "报销单匹配",
|
||
"passed": False,
|
||
"message": f"发票金额{inv.amount}元与报销单{inv.reimb_no}金额{reimb.amount}元不符",
|
||
})
|
||
else:
|
||
results.append({"rule": "报销单匹配", "passed": True, "message": f"与报销单{inv.reimb_no}金额一致"})
|
||
|
||
# 规则3: 供应商与合同匹配
|
||
if inv.contract_no:
|
||
if not (inv.supplier or "").strip():
|
||
status = "invalid"
|
||
results.append({"rule": "供应商匹配", "passed": False, "message": f"合同{inv.contract_no}未关联供应商"})
|
||
else:
|
||
# 同合同下其他发票的供应商一致性
|
||
others = (
|
||
db.query(InvoiceCheck)
|
||
.filter(
|
||
InvoiceCheck.contract_no == inv.contract_no,
|
||
InvoiceCheck.id != inv.id,
|
||
InvoiceCheck.supplier.isnot(None),
|
||
)
|
||
.all()
|
||
)
|
||
mismatch = [o.supplier for o in others if o.supplier != inv.supplier]
|
||
if mismatch:
|
||
status = "invalid"
|
||
results.append({
|
||
"rule": "供应商匹配",
|
||
"passed": False,
|
||
"message": f"供应商{inv.supplier}与合同{inv.contract_no}下其他发票供应商{mismatch[0]}不一致",
|
||
})
|
||
else:
|
||
results.append({"rule": "供应商匹配", "passed": True, "message": f"供应商与合同{inv.contract_no}匹配"})
|
||
|
||
inv.check_result = results
|
||
inv.check_status = status
|
||
inv.check_reason = "；".join(r["message"] for r in results if not r["passed"]) or None
|
||
inv.checked_at = datetime.now()
|
||
|
||
|
||
@router.get("/invoices")
|
||
def list_invoices(
|
||
status: str = Query(None),
|
||
keyword: str = Query(None),
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_auth),
|
||
):
|
||
"""发票列表"""
|
||
q = db.query(InvoiceCheck)
|
||
if entity_id:
|
||
q = q.filter(InvoiceCheck.entity_id == entity_id)
|
||
if status:
|
||
q = q.filter(InvoiceCheck.check_status == status)
|
||
if keyword:
|
||
kw = f"%{keyword}%"
|
||
q = q.filter(
|
||
(InvoiceCheck.invoice_no.like(kw))
|
||
| (InvoiceCheck.supplier.like(kw))
|
||
| (InvoiceCheck.reimb_no.like(kw))
|
||
)
|
||
invoices = q.order_by(InvoiceCheck.id.desc()).all()
|
||
return {"data": [_invoice_to_dict(i) for i in invoices], "total": len(invoices)}
|
||
|
||
|
||
@router.post("/invoices")
|
||
def create_invoice(
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""录入发票（自动执行校验）"""
|
||
invoice_no = data.get("invoice_no")
|
||
if not invoice_no:
|
||
raise HTTPException(400, "缺少必要参数: invoice_no")
|
||
if data.get("amount") is None:
|
||
raise HTTPException(400, "缺少必要参数: amount")
|
||
inv = InvoiceCheck(
|
||
entity_id=data.get("entity_id") or 1,
|
||
invoice_no=str(invoice_no).strip(),
|
||
amount=data.get("amount"),
|
||
invoice_type=data.get("invoice_type") or "vat",
|
||
invoice_date=_parse_date(data.get("invoice_date")),
|
||
supplier=data.get("supplier"),
|
||
reimb_no=data.get("reimb_no"),
|
||
contract_no=data.get("contract_no"),
|
||
)
|
||
_check_invoice(db, inv)
|
||
db.add(inv)
|
||
db.commit()
|
||
db.refresh(inv)
|
||
return {"message": "发票已录入并校验", "data": _invoice_to_dict(inv)}
|
||
|
||
|
||
@router.put("/invoices/{invoice_id}")
|
||
def update_invoice(
|
||
invoice_id: int,
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""更新发票（自动重新校验）"""
|
||
inv = db.query(InvoiceCheck).filter(InvoiceCheck.id == invoice_id).first()
|
||
if not inv:
|
||
raise HTTPException(404, "发票不存在")
|
||
if "invoice_no" in data:
|
||
inv.invoice_no = str(data["invoice_no"]).strip()
|
||
if "amount" in data:
|
||
inv.amount = data["amount"]
|
||
if "invoice_type" in data:
|
||
inv.invoice_type = data["invoice_type"]
|
||
if "invoice_date" in data:
|
||
inv.invoice_date = _parse_date(data.get("invoice_date"))
|
||
if "supplier" in data:
|
||
inv.supplier = data.get("supplier")
|
||
if "reimb_no" in data:
|
||
inv.reimb_no = data.get("reimb_no")
|
||
if "contract_no" in data:
|
||
inv.contract_no = data.get("contract_no")
|
||
_check_invoice(db, inv)
|
||
db.commit()
|
||
db.refresh(inv)
|
||
return {"message": "发票已更新并重新校验", "data": _invoice_to_dict(inv)}
|
||
|
||
|
||
@router.delete("/invoices/{invoice_id}")
|
||
def delete_invoice(
|
||
invoice_id: int,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""删除发票"""
|
||
inv = db.query(InvoiceCheck).filter(InvoiceCheck.id == invoice_id).first()
|
||
if not inv:
|
||
raise HTTPException(404, "发票不存在")
|
||
db.delete(inv)
|
||
db.commit()
|
||
return {"message": "发票已删除"}
|
||
|
||
|
||
@router.post("/invoices/check")
|
||
def batch_check_invoices(
|
||
invoice_id: int = Query(None),
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""批量校验发票（默认校验全部 pending，可指定单张）"""
|
||
q = db.query(InvoiceCheck)
|
||
if invoice_id:
|
||
q = q.filter(InvoiceCheck.id == invoice_id)
|
||
elif entity_id:
|
||
q = q.filter(InvoiceCheck.entity_id == entity_id)
|
||
invoices = q.all()
|
||
for inv in invoices:
|
||
_check_invoice(db, inv)
|
||
db.commit()
|
||
abnormal = [i for i in invoices if i.check_status in ("invalid", "warning")]
|
||
return {
|
||
"message": f"批量校验完成：{len(invoices)}张，异常{len(abnormal)}张",
|
||
"total": len(invoices),
|
||
"abnormal_count": len(abnormal),
|
||
}
|
||
|
||
|
||
@router.get("/invoices/abnormal")
|
||
def list_abnormal_invoices(
|
||
entity_id: int = Depends(get_entity_id),
|
||
limit: int = Query(50),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_auth),
|
||
):
|
||
"""查询异常发票（invalid/warning）"""
|
||
q = db.query(InvoiceCheck).filter(InvoiceCheck.check_status.in_(["invalid", "warning"]))
|
||
if entity_id:
|
||
q = q.filter(InvoiceCheck.entity_id == entity_id)
|
||
invoices = q.order_by(InvoiceCheck.id.desc()).limit(limit).all()
|
||
return {"data": [_invoice_to_dict(i) for i in invoices], "total": len(invoices)}
|
||
|
||
|
||
# ============================================================
|
||
# ③ 社保比对 — 缴费记录 CRUD + 比对 + 异常查询
|
||
# ============================================================
|
||
|
||
def _check_ss(db: Session, s: SocialSecurity, all_records: list | None = None):
|
||
"""社保比对规则：基数与工资匹配 / 单位缴纳比例 / 漏缴（含月份断层检测）"""
|
||
warnings: list[str] = []
|
||
alerts: list[str] = []
|
||
|
||
# 规则1: 缴费基数与工资匹配（60%~300%区间）
|
||
if s.salary and s.salary > 0 and s.base_amount and s.base_amount > 0:
|
||
low, high = s.salary * SS_BASE_LOW, s.salary * SS_BASE_HIGH
|
||
if s.base_amount < low or s.base_amount > high:
|
||
warnings.append(
|
||
f"缴费基数{s.base_amount}元超出工资{s.salary}元的{int(SS_BASE_LOW*100)}%~{int(SS_BASE_HIGH*100)}%区间({low:.0f}~{high:.0f})"
|
||
)
|
||
|
||
# 规则2: 单位缴纳比例（养老16%+医疗8%+失业0.5%≈24.5%）
|
||
rate = None
|
||
if s.base_amount and s.base_amount > 0 and s.company_amount is not None:
|
||
rate = round(s.company_amount / s.base_amount * 100, 2)
|
||
s.company_rate = rate
|
||
if rate is not None and abs(rate - SS_COMPANY_RATE) > SS_RATE_TOLERANCE:
|
||
warnings.append(f"单位缴纳比例{rate}%与标准{SS_COMPANY_RATE}%±{SS_RATE_TOLERANCE}%不符")
|
||
|
||
# 规则3: 漏缴检测
|
||
if not s.base_amount or s.base_amount <= 0 or (not s.company_amount and not s.personal_amount):
|
||
alerts.append(f"{s.employee}本期缴费基数为0或未缴费（疑似漏缴）")
|
||
|
||
# 规则3b: 月份断层检测（同人相邻记录期间间隔>1个月 → 漏缴）
|
||
if all_records is not None:
|
||
periods = sorted(
|
||
r.period for r in all_records
|
||
if r.employee == s.employee and r.period != s.period
|
||
)
|
||
prev = None
|
||
for p in periods:
|
||
if prev is not None:
|
||
try:
|
||
py, pm = map(int, prev.split("-"))
|
||
cy, cm = map(int, p.split("-"))
|
||
gap = (cy - py) * 12 + (cm - pm)
|
||
if gap > 1:
|
||
alerts.append(f"{s.employee}在{prev}至{p}之间漏缴{max(0, gap-1)}个月")
|
||
except Exception:
|
||
pass
|
||
prev = p
|
||
|
||
if alerts:
|
||
s.check_status = "alert"
|
||
elif warnings:
|
||
s.check_status = "warning"
|
||
else:
|
||
s.check_status = "normal"
|
||
s.warning_msg = "；".join(alerts + warnings) or None
|
||
|
||
|
||
@router.get("/ss")
|
||
def list_ss_records(
|
||
period: str = Query(None),
|
||
employee: str = Query(None),
|
||
status: str = Query(None),
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_auth),
|
||
):
|
||
"""社保缴费记录列表"""
|
||
q = db.query(SocialSecurity)
|
||
if entity_id:
|
||
q = q.filter(SocialSecurity.entity_id == entity_id)
|
||
if period:
|
||
q = q.filter(SocialSecurity.period == period)
|
||
if employee:
|
||
q = q.filter(SocialSecurity.employee.like(f"%{employee}%"))
|
||
if status:
|
||
q = q.filter(SocialSecurity.check_status == status)
|
||
records = q.order_by(SocialSecurity.period.desc(), SocialSecurity.id.asc()).all()
|
||
return {"data": [_ss_to_dict(s) for s in records], "total": len(records)}
|
||
|
||
|
||
@router.post("/ss")
|
||
def create_ss_record(
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""新增社保缴费记录（自动比对）"""
|
||
employee = data.get("employee")
|
||
period = data.get("period")
|
||
if not employee or not period:
|
||
raise HTTPException(400, "缺少必要参数: employee, period")
|
||
all_records = db.query(SocialSecurity).filter(SocialSecurity.entity_id == (data.get("entity_id") or 1)).all()
|
||
s = SocialSecurity(
|
||
entity_id=data.get("entity_id") or 1,
|
||
employee=employee,
|
||
period=period,
|
||
base_amount=data.get("base_amount") or 0,
|
||
salary=data.get("salary"),
|
||
company_amount=data.get("company_amount") or 0,
|
||
personal_amount=data.get("personal_amount") or 0,
|
||
remark=data.get("remark"),
|
||
)
|
||
_check_ss(db, s, all_records)
|
||
db.add(s)
|
||
db.commit()
|
||
db.refresh(s)
|
||
return {"message": "社保记录已创建", "data": _ss_to_dict(s)}
|
||
|
||
|
||
@router.put("/ss/{ss_id}")
|
||
def update_ss_record(
|
||
ss_id: int,
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""更新社保记录（自动重新比对）"""
|
||
s = db.query(SocialSecurity).filter(SocialSecurity.id == ss_id).first()
|
||
if not s:
|
||
raise HTTPException(404, "社保记录不存在")
|
||
if "employee" in data:
|
||
s.employee = data["employee"]
|
||
if "period" in data:
|
||
s.period = data["period"]
|
||
if "base_amount" in data:
|
||
s.base_amount = data["base_amount"] or 0
|
||
if "salary" in data:
|
||
s.salary = data.get("salary")
|
||
if "company_amount" in data:
|
||
s.company_amount = data["company_amount"] or 0
|
||
if "personal_amount" in data:
|
||
s.personal_amount = data["personal_amount"] or 0
|
||
if "remark" in data:
|
||
s.remark = data.get("remark")
|
||
all_records = db.query(SocialSecurity).filter(SocialSecurity.entity_id == s.entity_id).all()
|
||
_check_ss(db, s, all_records)
|
||
db.commit()
|
||
db.refresh(s)
|
||
return {"message": "社保记录已更新", "data": _ss_to_dict(s)}
|
||
|
||
|
||
@router.delete("/ss/{ss_id}")
|
||
def delete_ss_record(
|
||
ss_id: int,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""删除社保记录"""
|
||
s = db.query(SocialSecurity).filter(SocialSecurity.id == ss_id).first()
|
||
if not s:
|
||
raise HTTPException(404, "社保记录不存在")
|
||
db.delete(s)
|
||
db.commit()
|
||
return {"message": "社保记录已删除"}
|
||
|
||
|
||
@router.post("/ss/check")
|
||
def batch_check_ss(
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""批量社保比对"""
|
||
q = db.query(SocialSecurity)
|
||
if entity_id:
|
||
q = q.filter(SocialSecurity.entity_id == entity_id)
|
||
records = q.all()
|
||
for s in records:
|
||
_check_ss(db, s, records)
|
||
db.commit()
|
||
abnormal = [s for s in records if s.check_status in ("alert", "warning")]
|
||
return {
|
||
"message": f"社保比对完成：{len(records)}条，异常{len(abnormal)}条",
|
||
"total": len(records),
|
||
"abnormal_count": len(abnormal),
|
||
}
|
||
|
||
|
||
@router.get("/ss/abnormal")
|
||
def list_abnormal_ss(
|
||
entity_id: int = Depends(get_entity_id),
|
||
limit: int = Query(50),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_auth),
|
||
):
|
||
"""查询社保异常（alert/warning）"""
|
||
q = db.query(SocialSecurity).filter(SocialSecurity.check_status.in_(["alert", "warning"]))
|
||
if entity_id:
|
||
q = q.filter(SocialSecurity.entity_id == entity_id)
|
||
records = q.order_by(SocialSecurity.period.desc(), SocialSecurity.id.desc()).limit(limit).all()
|
||
return {"data": [_ss_to_dict(s) for s in records], "total": len(records)}
|
||
|
||
|
||
# ============================================================
|
||
# ④ 税务看板聚合
|
||
# ============================================================
|
||
|
||
@router.get("/dashboard")
|
||
def tax_dashboard(
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_auth),
|
||
):
|
||
"""税务合规看板聚合：税负趋势+行业对比 / 发票异常 / 社保异常"""
|
||
eid = entity_id or 1
|
||
|
||
# 税负监控
|
||
records = db.query(TaxRecord).filter(TaxRecord.entity_id == eid).order_by(TaxRecord.period.asc()).all()
|
||
for t in records:
|
||
_calc_burden(t)
|
||
db.commit()
|
||
trend_map: dict[str, dict] = {}
|
||
for t in records:
|
||
if t.tax_burden_rate is None:
|
||
continue
|
||
entry = trend_map.setdefault(t.period, {"period": t.period})
|
||
entry[f"{t.tax_type}_rate"] = t.tax_burden_rate
|
||
entry[f"{t.tax_type}_benchmark"] = TAX_BENCHMARKS.get(t.tax_type)
|
||
burden_trend = sorted(trend_map.values(), key=lambda x: x["period"])
|
||
latest_period = max((t.period for t in records), default=None)
|
||
latest = next((x for x in burden_trend if x["period"] == latest_period), None)
|
||
tax_alerts = [t for t in records if t.burden_status == "alert"]
|
||
|
||
# 发票
|
||
invoices = db.query(InvoiceCheck).filter(InvoiceCheck.entity_id == eid).all()
|
||
inv_status_count = {"pending": 0, "valid": 0, "invalid": 0, "warning": 0}
|
||
for i in invoices:
|
||
inv_status_count[i.check_status] = inv_status_count.get(i.check_status, 0) + 1
|
||
inv_abnormal = (
|
||
db.query(InvoiceCheck)
|
||
.filter(InvoiceCheck.entity_id == eid, InvoiceCheck.check_status.in_(["invalid", "warning"]))
|
||
.order_by(InvoiceCheck.id.desc())
|
||
.limit(10)
|
||
.all()
|
||
)
|
||
|
||
# 社保
|
||
ss_records = db.query(SocialSecurity).filter(SocialSecurity.entity_id == eid).all()
|
||
ss_status_count = {"normal": 0, "warning": 0, "alert": 0}
|
||
for s in ss_records:
|
||
ss_status_count[s.check_status] = ss_status_count.get(s.check_status, 0) + 1
|
||
ss_abnormal = (
|
||
db.query(SocialSecurity)
|
||
.filter(SocialSecurity.entity_id == eid, SocialSecurity.check_status.in_(["alert", "warning"]))
|
||
.order_by(SocialSecurity.period.desc(), SocialSecurity.id.desc())
|
||
.limit(10)
|
||
.all()
|
||
)
|
||
|
||
return {
|
||
"entity_id": eid,
|
||
"burden": {
|
||
"trend": burden_trend,
|
||
"latest": latest,
|
||
"benchmarks": TAX_BENCHMARKS,
|
||
"alert_count": len(tax_alerts),
|
||
"alerts": [_tax_to_dict(t) for t in tax_alerts[:10]],
|
||
},
|
||
"invoice": {
|
||
"total": len(invoices),
|
||
"status_count": inv_status_count,
|
||
"abnormal": [_invoice_to_dict(i) for i in inv_abnormal],
|
||
},
|
||
"ss": {
|
||
"total": len(ss_records),
|
||
"status_count": ss_status_count,
|
||
"abnormal": [_ss_to_dict(s) for s in ss_abnormal],
|
||
},
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# 演示数据（幂等）
|
||
# ============================================================
|
||
|
||
@router.post("/demo-data")
|
||
def seed_demo_data(
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""生成税务合规演示数据（幂等：已存在期间+税种则跳过）"""
|
||
eid = entity_id or 1
|
||
created = {"tax": 0, "invoice": 0, "ss": 0}
|
||
|
||
# 税负记录: 2026-01 ~ 2026-07，其中02月增值税故意偏高触发预警
|
||
tax_demo = [
|
||
("2026-01", "vat", 52000, 48800, 1380000, 13),
|
||
("2026-02", "vat", 69000, 66000, 1460000, 13),
|
||
("2026-03", "vat", 41000, 39800, 1280000, 13),
|
||
("2026-04", "vat", 47500, 46200, 1350000, 13),
|
||
("2026-05", "vat", 53000, 51000, 1420000, 13),
|
||
("2026-06", "vat", 49800, 48200, 1400000, 13),
|
||
("2026-01", "income", 36000, 33500, 1380000, 25),
|
||
("2026-02", "income", 38000, 35500, 1460000, 25),
|
||
("2026-03", "income", 33000, 31000, 1280000, 25),
|
||
("2026-04", "income", 35000, 32800, 1350000, 25),
|
||
("2026-05", "income", 37000, 34800, 1420000, 25),
|
||
("2026-06", "income", 36500, 34000, 1400000, 25),
|
||
("2026-01", "surtax", 6240, 5900, 1380000, 12),
|
||
("2026-02", "surtax", 8280, 7900, 1460000, 12),
|
||
("2026-03", "surtax", 4920, 4700, 1280000, 12),
|
||
("2026-04", "surtax", 5700, 5500, 1350000, 12),
|
||
("2026-05", "surtax", 6360, 6100, 1420000, 12),
|
||
("2026-06", "surtax", 5976, 5800, 1400000, 12),
|
||
]
|
||
for period, ttype, payable, paid, income, rate in tax_demo:
|
||
exists = (
|
||
db.query(TaxRecord)
|
||
.filter(TaxRecord.entity_id == eid, TaxRecord.period == period, TaxRecord.tax_type == ttype)
|
||
.first()
|
||
)
|
||
if exists:
|
||
continue
|
||
t = TaxRecord(entity_id=eid, period=period, tax_type=ttype, tax_payable=payable,
|
||
tax_paid=paid, tax_rate=rate, income=income)
|
||
_calc_burden(t)
|
||
db.add(t)
|
||
created["tax"] += 1
|
||
|
||
# 报销单（供发票"金额与报销单匹配"规则使用）
|
||
reimb_demo = [
|
||
{"reimb_no": "BX202606150001", "applicant": "张伟", "department": "采购部", "expense_type": "office",
|
||
"title": "6月办公用品采购", "amount": 5600, "status": "approved"},
|
||
{"reimb_no": "BX202606200002", "applicant": "李娜", "department": "市场部", "expense_type": "management",
|
||
"title": "6月市场信息服务费", "amount": 3200, "status": "approved"},
|
||
{"reimb_no": "BX202607050003", "applicant": "王强", "department": "供应链部", "expense_type": "office",
|
||
"title": "7月供应链物流服务", "amount": 14000, "status": "pending"},
|
||
]
|
||
for d in reimb_demo:
|
||
exists = db.query(ExpenseReimbursement).filter(ExpenseReimbursement.reimb_no == d["reimb_no"]).first()
|
||
if exists:
|
||
continue
|
||
db.add(ExpenseReimbursement(**d))
|
||
|
||
# 发票: 正常2张(金额匹配) + 格式错误1张 + 金额不符1张 + 供应商与合同不符1张
|
||
invoice_demo = [
|
||
{"invoice_no": "9" * 33, "amount": 5600, "invoice_type": "vat", "invoice_date": "2026-06-15",
|
||
"supplier": "北京云启科技有限公司", "reimb_no": "BX202606150001", "contract_no": "HT-2026-018"},
|
||
{"invoice_no": "8" * 33, "amount": 3200, "invoice_type": "electronic", "invoice_date": "2026-06-20",
|
||
"supplier": "上海数联信息服务有限公司", "reimb_no": "BX202606200002", "contract_no": "HT-2026-021"},
|
||
{"invoice_no": "12345ABC", "amount": 1800, "invoice_type": "vat", "invoice_date": "2026-07-02",
|
||
"supplier": "广州锐思咨询有限公司", "reimb_no": None, "contract_no": None},
|
||
{"invoice_no": "7" * 33, "amount": 15000, "invoice_type": "vat", "invoice_date": "2026-07-05",
|
||
"supplier": "深圳恒达供应链有限公司", "reimb_no": "BX202607050003", "contract_no": "HT-2026-030"},
|
||
{"invoice_no": "6" * 33, "amount": 9800, "invoice_type": "vat", "invoice_date": "2026-07-08",
|
||
"supplier": None, "reimb_no": None, "contract_no": "HT-2026-033"},
|
||
]
|
||
for d in invoice_demo:
|
||
exists = (
|
||
db.query(InvoiceCheck)
|
||
.filter(InvoiceCheck.entity_id == eid, InvoiceCheck.invoice_no == d["invoice_no"])
|
||
.first()
|
||
)
|
||
if exists:
|
||
continue
|
||
inv = InvoiceCheck(entity_id=eid, **d)
|
||
_check_invoice(db, inv)
|
||
db.add(inv)
|
||
created["invoice"] += 1
|
||
|
||
# 社保: 正常3人×2月 + 基数不符1条 + 漏缴1条
|
||
ss_demo = [
|
||
{"employee": "张伟", "period": "2026-06", "base_amount": 12000, "salary": 12000, "company_amount": 2940, "personal_amount": 1248},
|
||
{"employee": "张伟", "period": "2026-07", "base_amount": 12000, "salary": 12000, "company_amount": 2940, "personal_amount": 1248},
|
||
{"employee": "李娜", "period": "2026-06", "base_amount": 18000, "salary": 18000, "company_amount": 4410, "personal_amount": 1872},
|
||
{"employee": "李娜", "period": "2026-07", "base_amount": 18000, "salary": 18000, "company_amount": 4410, "personal_amount": 1872},
|
||
{"employee": "王强", "period": "2026-06", "base_amount": 9000, "salary": 9000, "company_amount": 2205, "personal_amount": 936},
|
||
{"employee": "王强", "period": "2026-07", "base_amount": 9000, "salary": 9000, "company_amount": 2205, "personal_amount": 936},
|
||
{"employee": "赵敏", "period": "2026-06", "base_amount": 5000, "salary": 20000, "company_amount": 1225, "personal_amount": 520},
|
||
{"employee": "孙磊", "period": "2026-05", "base_amount": 0, "salary": 15000, "company_amount": 0, "personal_amount": 0},
|
||
]
|
||
for d in ss_demo:
|
||
exists = (
|
||
db.query(SocialSecurity)
|
||
.filter(SocialSecurity.entity_id == eid, SocialSecurity.employee == d["employee"],
|
||
SocialSecurity.period == d["period"])
|
||
.first()
|
||
)
|
||
if exists:
|
||
continue
|
||
s = SocialSecurity(entity_id=eid, **d)
|
||
all_records = db.query(SocialSecurity).filter(SocialSecurity.entity_id == eid).all()
|
||
_check_ss(db, s, all_records)
|
||
db.add(s)
|
||
created["ss"] += 1
|
||
db.commit()
|
||
return {"message": "演示数据生成完成", "created": created}
|