437 lines
15 KiB
Python
437 lines
15 KiB
Python
"""预算管理 API — 管理会计OS
|
||
预算值的CRUD、自动分解、版本管理
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import func
|
||
from typing import Optional
|
||
from datetime import datetime
|
||
|
||
from app.database import get_db
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.models import BudgetPlan, KPIDefinition, OperationLog
|
||
|
||
router = APIRouter(prefix="/api/cma/budget", tags=["预算管理"],
|
||
dependencies=[Depends(require_role("ceo", "finance", "it"))],
|
||
)
|
||
|
||
|
||
@router.get("/plans")
|
||
def list_budget_plans(
|
||
kpi_id: Optional[int] = Query(None),
|
||
period: Optional[str] = Query(None),
|
||
year: Optional[int] = Query(None),
|
||
version: Optional[str] = Query(None),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""查询预算计划列表"""
|
||
query = db.query(BudgetPlan).join(
|
||
KPIDefinition, BudgetPlan.kpi_id == KPIDefinition.id
|
||
)
|
||
|
||
if kpi_id:
|
||
query = query.filter(BudgetPlan.kpi_id == kpi_id)
|
||
if period:
|
||
query = query.filter(BudgetPlan.period == period)
|
||
if year:
|
||
query = query.filter(BudgetPlan.budget_year == year)
|
||
if version:
|
||
query = query.filter(BudgetPlan.version == version)
|
||
|
||
plans = query.order_by(BudgetPlan.budget_year.desc(), BudgetPlan.budget_month.asc()).all()
|
||
|
||
result = []
|
||
for p in plans:
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == p.kpi_id).first()
|
||
result.append({
|
||
"id": p.id,
|
||
"kpi_id": p.kpi_id,
|
||
"kpi_code": kpi.kpi_code if kpi else "",
|
||
"kpi_name": kpi.kpi_name if kpi else "",
|
||
"period": p.period,
|
||
"budget_value": p.budget_value,
|
||
"budget_year": p.budget_year,
|
||
"budget_month": p.budget_month,
|
||
"version": p.version,
|
||
"status": p.status,
|
||
"remark": p.remark,
|
||
"created_at": p.created_at.isoformat() if p.created_at else None,
|
||
})
|
||
return {"data": result, "total": len(result)}
|
||
|
||
|
||
@router.post("/plans")
|
||
def create_budget_plan(
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(require_auth),
|
||
):
|
||
"""创建或更新单条预算计划"""
|
||
kpi_id = data.get("kpi_id")
|
||
period = data.get("period")
|
||
budget_value = data.get("budget_value")
|
||
|
||
if not all([kpi_id, period, budget_value is not None]):
|
||
raise HTTPException(400, "缺少必要参数: kpi_id, period, budget_value")
|
||
|
||
# 检查KPI是否存在
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
if not kpi:
|
||
raise HTTPException(404, "KPI不存在")
|
||
|
||
year, month = period.split("-")
|
||
version = data.get("version", "v1.0")
|
||
|
||
# 检查是否已有记录（去重）
|
||
existing = db.query(BudgetPlan).filter(
|
||
BudgetPlan.kpi_id == kpi_id,
|
||
BudgetPlan.period == period,
|
||
BudgetPlan.version == version,
|
||
BudgetPlan.status == "active",
|
||
).first()
|
||
|
||
if existing:
|
||
existing.budget_value = budget_value
|
||
existing.remark = data.get("remark", existing.remark)
|
||
db.commit()
|
||
db.refresh(existing)
|
||
return {"message": "预算已更新", "id": existing.id}
|
||
else:
|
||
plan = BudgetPlan(
|
||
kpi_id=kpi_id,
|
||
period=period,
|
||
budget_value=budget_value,
|
||
budget_year=int(year),
|
||
budget_month=int(month),
|
||
version=version,
|
||
status="active",
|
||
remark=data.get("remark", ""),
|
||
created_by=current_user.name if hasattr(current_user, "name") else "",
|
||
)
|
||
db.add(plan)
|
||
db.commit()
|
||
db.refresh(plan)
|
||
|
||
# 记录操作日志
|
||
log = OperationLog(
|
||
action="create",
|
||
target_type="budget",
|
||
target_id=plan.id,
|
||
detail=__import__("json").dumps({"kpi_id": kpi_id, "period": period, "value": budget_value}, ensure_ascii=False),
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
|
||
return {"message": "预算已创建", "id": plan.id}
|
||
|
||
|
||
@router.put("/plans/{plan_id}")
|
||
def update_budget_plan(
|
||
plan_id: int,
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""更新预算计划"""
|
||
plan = db.query(BudgetPlan).filter(BudgetPlan.id == plan_id).first()
|
||
if not plan:
|
||
raise HTTPException(404, "预算计划不存在")
|
||
|
||
if "budget_value" in data:
|
||
plan.budget_value = data["budget_value"]
|
||
if "remark" in data:
|
||
plan.remark = data["remark"]
|
||
if "version" in data:
|
||
plan.version = data["version"]
|
||
if "status" in data:
|
||
plan.status = data["status"]
|
||
|
||
db.commit()
|
||
return {"message": "预算已更新"}
|
||
|
||
|
||
@router.delete("/plans/{plan_id}")
|
||
def delete_budget_plan(
|
||
plan_id: int,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""删除预算计划"""
|
||
plan = db.query(BudgetPlan).filter(BudgetPlan.id == plan_id).first()
|
||
if not plan:
|
||
raise HTTPException(404, "预算计划不存在")
|
||
db.delete(plan)
|
||
db.commit()
|
||
return {"message": "预算已删除"}
|
||
|
||
|
||
@router.post("/auto-decompose")
|
||
def auto_decompose_budget(
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
current_user=Depends(require_auth),
|
||
):
|
||
"""自动分解年度预算到月度（均分或按历史权重）"""
|
||
kpi_id = data.get("kpi_id")
|
||
year = data.get("year", datetime.now().year)
|
||
annual_budget = data.get("annual_budget")
|
||
method = data.get("method", "equal") # equal / weighted
|
||
version = data.get("version", "v1.0")
|
||
|
||
if not kpi_id or annual_budget is None:
|
||
raise HTTPException(400, "缺少必要参数: kpi_id, annual_budget")
|
||
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
if not kpi:
|
||
raise HTTPException(404, "KPI不存在")
|
||
|
||
# 计算各月权重
|
||
if method == "weighted":
|
||
# 按去年各月实际值的比例分配
|
||
last_year = year - 1
|
||
values = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi_id,
|
||
KPIValue.period.like(f"{last_year}-%"),
|
||
KPIValue.actual_value.isnot(None),
|
||
).order_by(KPIValue.period.asc()).all()
|
||
|
||
total = sum(v.actual_value for v in values)
|
||
if total > 0:
|
||
weights = {v.period: v.actual_value / total for v in values}
|
||
else:
|
||
method = "equal"
|
||
created = []
|
||
for m in range(1, 13):
|
||
period = f"{year}-{m:02d}"
|
||
weight = weights.get(period, 1 / 12) if method == "weighted" else 1 / 12
|
||
monthly_value = round(annual_budget * weight, 2)
|
||
|
||
existing = db.query(BudgetPlan).filter(
|
||
BudgetPlan.kpi_id == kpi_id,
|
||
BudgetPlan.period == period,
|
||
BudgetPlan.version == version,
|
||
BudgetPlan.status == "active",
|
||
).first()
|
||
|
||
if existing:
|
||
existing.budget_value = monthly_value
|
||
else:
|
||
bp = BudgetPlan(
|
||
kpi_id=kpi_id, period=period,
|
||
budget_value=monthly_value, budget_year=year,
|
||
budget_month=m, version=version, status="active",
|
||
created_by=current_user.name if hasattr(current_user, "name") else "",
|
||
)
|
||
db.add(bp)
|
||
created.append({"period": period, "value": monthly_value})
|
||
else:
|
||
# 均分
|
||
monthly = round(annual_budget / 12, 2)
|
||
created = []
|
||
for m in range(1, 13):
|
||
period = f"{year}-{m:02d}"
|
||
existing = db.query(BudgetPlan).filter(
|
||
BudgetPlan.kpi_id == kpi_id,
|
||
BudgetPlan.period == period,
|
||
BudgetPlan.version == version,
|
||
BudgetPlan.status == "active",
|
||
).first()
|
||
if existing:
|
||
existing.budget_value = monthly
|
||
else:
|
||
bp = BudgetPlan(
|
||
kpi_id=kpi_id, period=period,
|
||
budget_value=monthly, budget_year=year,
|
||
budget_month=m, version=version, status="active",
|
||
created_by=current_user.name if hasattr(current_user, "name") else "",
|
||
)
|
||
db.add(bp)
|
||
created.append({"period": period, "value": monthly})
|
||
|
||
db.commit()
|
||
return {
|
||
"message": f"年度预算已分解为{len(created)}个月度预算",
|
||
"kpi_id": kpi_id,
|
||
"kpi_name": kpi.kpi_name,
|
||
"year": year,
|
||
"annual_budget": annual_budget,
|
||
"method": method,
|
||
"monthly_budgets": created,
|
||
}
|
||
|
||
|
||
@router.get("/deviation-report")
|
||
def get_deviation_report(
|
||
kpi_id: Optional[int] = Query(None),
|
||
period: Optional[str] = Query(None),
|
||
year: Optional[int] = Query(None),
|
||
month: Optional[int] = Query(None),
|
||
dimension: Optional[str] = Query(None),
|
||
alert_level: Optional[str] = Query(None),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""获取差异分析报告（汇总多个KPI的实际vs预算差异）"""
|
||
if period is None:
|
||
if year and month:
|
||
period = f"{year}-{month:02d}"
|
||
elif year:
|
||
period = f"{year}-{datetime.now().month:02d}"
|
||
else:
|
||
period = datetime.now().strftime("%Y-%m")
|
||
|
||
query = db.query(KPIDefinition).filter(KPIDefinition.status == "active")
|
||
if kpi_id:
|
||
query = query.filter(KPIDefinition.id == kpi_id)
|
||
if dimension:
|
||
query = query.filter(KPIDefinition.dimension == dimension)
|
||
|
||
kpis = query.all()
|
||
from app.utils.deviation_engine import calc_period_deviation, calc_period_diff
|
||
|
||
items = []
|
||
summary = {
|
||
"total_kpis": 0,
|
||
"has_budget": 0,
|
||
"over_budget": 0,
|
||
"under_budget": 0,
|
||
"avg_deviation_rate": 0,
|
||
}
|
||
|
||
rates = []
|
||
for kpi in kpis:
|
||
item = calc_period_deviation(db, kpi.id, period)
|
||
items.append(item)
|
||
summary["total_kpis"] += 1
|
||
|
||
if item.get("budget_value") is not None:
|
||
summary["has_budget"] += 1
|
||
if item.get("is_over_budget"):
|
||
summary["over_budget"] += 1
|
||
elif item.get("deviation_rate") is not None and item["deviation_rate"] < 0:
|
||
summary["under_budget"] += 1
|
||
if item.get("deviation_rate") is not None:
|
||
rates.append(abs(item["deviation_rate"]))
|
||
|
||
# 补充同比/环比
|
||
if item.get("actual_value") is not None:
|
||
item["yoy"] = calc_period_diff(db, kpi.id, period, "yoy")
|
||
item["mom"] = calc_period_diff(db, kpi.id, period, "mom")
|
||
|
||
summary["avg_deviation_rate"] = round(sum(rates) / len(rates), 2) if rates else 0
|
||
|
||
# 前端 alert_level 过滤
|
||
if alert_level:
|
||
def get_level(rate):
|
||
if rate is None:
|
||
return None
|
||
if rate > 20:
|
||
return "red"
|
||
if rate > 10:
|
||
return "yellow"
|
||
return "normal"
|
||
items = [i for i in items if get_level(i.get("deviation_rate")) == alert_level]
|
||
|
||
return {
|
||
"period": period,
|
||
"summary": summary,
|
||
"items": items,
|
||
}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 功能6: 预算方法三选一向导 (CMA P1 - 增量/零基/弹性)
|
||
# ──────────────────────────────────────────────
|
||
|
||
@router.post("/method-comparison")
|
||
def budget_method_comparison(data: dict):
|
||
"""
|
||
预算方法三选一对比计算
|
||
接收: { entity: "hanke", last_month_budget: 91, current_revenue: 122, ... }
|
||
返回三种方法的计算结果
|
||
"""
|
||
entity = data.get("entity", "hanke")
|
||
last_month_budget = data.get("last_month_budget", 91) # 上月预算（万）
|
||
current_revenue = data.get("current_revenue", 122) # 当前收入（万）
|
||
fixed_costs = data.get("fixed_costs", {
|
||
"rent": 15, # 房租（万）
|
||
"labor": 40, # 人工（万）
|
||
"entertainment": 16, # 招待费（万）
|
||
"misc": 12, # 杂项（万）
|
||
})
|
||
variable_cost_rate = data.get("variable_cost_rate", 0.4862) # 变动成本率
|
||
|
||
# 1. 增量预算: 基于上月统一调整
|
||
increment_rate = data.get("increment_rate", 0.05) # 5%增幅
|
||
incremental_result = round(last_month_budget * (1 + increment_rate), 1)
|
||
incremental_detail = f"上月{last_month_budget}万 × (1+{increment_rate*100:.0f}%) = {incremental_result}万"
|
||
|
||
# 2. 零基预算: 每项从零论证
|
||
zbb_entertainment = round(fixed_costs.get("entertainment", 16) / 2, 1) # 砍半
|
||
zbb_misc = round(fixed_costs.get("misc", 12) * 0.7, 1) # 压缩30%
|
||
zbb_total = round(
|
||
fixed_costs.get("rent", 15)
|
||
+ fixed_costs.get("labor", 40)
|
||
+ zbb_entertainment
|
||
+ zbb_misc,
|
||
1,
|
||
)
|
||
zbb_savings = round(last_month_budget - zbb_total, 1)
|
||
zbb_detail = (
|
||
f"房租{fixed_costs.get('rent', 15)}万(固定)+人工{fixed_costs.get('labor', 40)}万(砍不掉)"
|
||
f"+招待{zbb_entertainment}万(砍半)+杂项{zbb_misc}万(压缩)"
|
||
f"={zbb_total}万 ← 省{zbb_savings}万"
|
||
)
|
||
|
||
# 3. 弹性预算: 根据收入水平动态调整
|
||
flexible_fixed = round(fixed_costs.get("rent", 15) + fixed_costs.get("labor", 40) * 0.5, 1)
|
||
flexible_variable = round(current_revenue * variable_cost_rate * 0.4, 1)
|
||
flexible_total = round(flexible_fixed + flexible_variable, 1)
|
||
flexible_variance = round(last_month_budget - flexible_total, 1)
|
||
flex_detail = (
|
||
f"收入{current_revenue}万 → 对应费用预算 = {flexible_total}万"
|
||
f"（固定部分{flexible_fixed}万+变动部分{flexible_variable}万）"
|
||
f"，实际{last_month_budget}万 → 差异{flexible_variance}万 → {'效率问题' if flexible_variance > 0 else '节省'}"
|
||
)
|
||
|
||
# 推荐方法
|
||
recommended = "zero_based"
|
||
|
||
return {
|
||
"entity": entity,
|
||
"entity_name": "陕西酣客（白酒经销）" if entity == "hanke" else "陕西博海科技（IT服务）",
|
||
"methods": [
|
||
{
|
||
"id": "incremental",
|
||
"name": "增量预算",
|
||
"name_en": "Incremental Budgeting",
|
||
"result_value": incremental_result,
|
||
"detail": incremental_detail,
|
||
"pros": "简单快速",
|
||
"cons": "浪费持续",
|
||
"is_recommended": False,
|
||
},
|
||
{
|
||
"id": "zero_based",
|
||
"name": "零基预算",
|
||
"name_en": "Zero-Based Budgeting (ZBB)",
|
||
"result_value": zbb_total,
|
||
"savings": zbb_savings,
|
||
"detail": zbb_detail,
|
||
"pros": "最合理",
|
||
"cons": "耗时",
|
||
"is_recommended": True,
|
||
},
|
||
{
|
||
"id": "flexible",
|
||
"name": "弹性预算",
|
||
"name_en": "Flexible Budgeting",
|
||
"result_value": flexible_total,
|
||
"variance": flexible_variance,
|
||
"detail": flex_detail,
|
||
"pros": "动态响应",
|
||
"cons": "需要详细分类",
|
||
"is_recommended": False,
|
||
},
|
||
],
|
||
"recommended": recommended,
|
||
"recommended_name": "零基预算",
|
||
}
|