"""成本分析API — 管理会计OS""" import logging from datetime import datetime from typing import Optional from fastapi import APIRouter, Depends, Query, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import StandardCost, ActualCost, AbcActivity, AbcAllocation from app.utils.cost_engine import ( calc_product_variance, get_cost_overview, get_cost_breakdown, calc_driver_rate, allocate_cost ) logger = logging.getLogger("cma.cost") router = APIRouter(prefix="/api/cma/cost", tags=["成本分析"]) # ============================================================ # 标准成本卡片 CRUD # ============================================================ @router.get("/standard-costs") def list_standard_costs(product_code: Optional[str] = Query(None), cost_type: Optional[str] = Query(None), db: Session = Depends(get_db)): """查询标准成本卡片""" query = db.query(StandardCost).filter(StandardCost.status == "active") if product_code: query = query.filter(StandardCost.product_code == product_code) if cost_type: query = query.filter(StandardCost.cost_type == cost_type) items = query.order_by(StandardCost.product_code, StandardCost.cost_type).all() return {"data": items} @router.post("/standard-costs") def create_standard_cost(data: dict, db: Session = Depends(get_db)): """创建标准成本卡片""" sc = StandardCost( product_code=data["product_code"], product_name=data.get("product_name", ""), cost_type=data["cost_type"], item_name=data["item_name"], standard_quantity=data["standard_quantity"], unit=data.get("unit", ""), standard_price=data["standard_price"], standard_cost=round(data["standard_quantity"] * data["standard_price"], 2), version=data.get("version", "v1.0"), remark=data.get("remark"), ) db.add(sc) db.commit() return {"message": "标准成本已创建", "id": sc.id} @router.put("/standard-costs/{cost_id}") def update_standard_cost(cost_id: int, data: dict, db: Session = Depends(get_db)): """修改标准成本卡片""" sc = db.query(StandardCost).filter(StandardCost.id == cost_id).first() if not sc: raise HTTPException(404, "标准成本记录不存在") for k in ("product_code", "product_name", "cost_type", "item_name", "standard_quantity", "unit", "standard_price", "version", "remark"): if k in data: setattr(sc, k, data[k]) sc.standard_cost = round(sc.standard_quantity * sc.standard_price, 2) db.commit() return {"message": "已更新"} @router.delete("/standard-costs/{cost_id}") def delete_standard_cost(cost_id: int, db: Session = Depends(get_db)): """删除标准成本卡片""" sc = db.query(StandardCost).filter(StandardCost.id == cost_id).first() if not sc: raise HTTPException(404, "标准成本记录不存在") sc.status = "archived" db.commit() return {"message": "已归档"} # ============================================================ # 实际成本 CRUD # ============================================================ @router.get("/actual-costs") def list_actual_costs(period: Optional[str] = Query(None), product_code: Optional[str] = Query(None), db: Session = Depends(get_db)): """查询实际成本""" query = db.query(ActualCost) if period: query = query.filter(ActualCost.period == period) if product_code: query = query.filter(ActualCost.product_code == product_code) items = query.order_by(ActualCost.period.desc(), ActualCost.product_code).all() return {"data": items} @router.post("/actual-costs") def create_actual_cost(data: dict, db: Session = Depends(get_db)): """录入实际成本""" ac = ActualCost( period=data["period"], product_code=data["product_code"], product_name=data.get("product_name", ""), cost_type=data["cost_type"], item_name=data.get("item_name", ""), actual_quantity=data["actual_quantity"], actual_price=data["actual_price"], actual_cost=round(data["actual_quantity"] * data["actual_price"], 2), source=data.get("source", "manual"), ) db.add(ac) db.commit() return {"message": "实际成本已录入", "id": ac.id} # ============================================================ # ABC 作业成本 # ============================================================ @router.get("/abc/activities") def list_abc_activities(db: Session = Depends(get_db)): """查询ABC作业中心列表""" items = db.query(AbcActivity).order_by(AbcActivity.activity_code).all() return {"data": items} @router.post("/abc/activities") def create_abc_activity(data: dict, db: Session = Depends(get_db)): """创建ABC作业中心""" act = AbcActivity( activity_code=data["activity_code"], activity_name=data["activity_name"], activity_desc=data.get("activity_desc"), cost_driver=data["cost_driver"], driver_unit=data.get("driver_unit"), total_cost=data.get("total_cost", 0), driver_volume=data.get("driver_volume", 0), ) act.driver_rate = round(act.total_cost / act.driver_volume, 4) if act.driver_volume > 0 else 0 db.add(act) db.commit() return {"message": "作业中心已创建", "id": act.id} @router.post("/abc/allocate") def do_allocate(data: dict, db: Session = Depends(get_db)): """执行ABC成本分配""" result = allocate_cost( activity_id=data["activity_id"], period=data.get("period", datetime.now().strftime("%Y-%m")), product_code=data["product_code"], product_name=data.get("product_name", ""), driver_consumed=data["driver_consumed"], ) return result @router.get("/abc/allocations") def list_allocations(period: Optional[str] = Query(None), product_code: Optional[str] = Query(None), db: Session = Depends(get_db)): """查询ABC分配记录""" query = db.query(AbcAllocation) if period: query = query.filter(AbcAllocation.period == period) if product_code: query = query.filter(AbcAllocation.product_code == product_code) items = query.order_by(AbcAllocation.period.desc()).all() return {"data": items} # ============================================================ # 分析看板 # ============================================================ @router.get("/overview") def cost_overview(period: Optional[str] = Query(None)): """成本总览（总成本、结构占比、趋势）""" if period is None: period = datetime.now().strftime("%Y-%m") return get_cost_overview(period) @router.get("/variance") def cost_variance(product_code: str = Query(...), period: Optional[str] = Query(None)): """量差价差分析""" if period is None: period = datetime.now().strftime("%Y-%m") return calc_product_variance(product_code, period) @router.get("/breakdown") def cost_breakdown(product_code: str = Query(...), period: Optional[str] = Query(None)): """成本构成（料/工/费占比）""" if period is None: period = datetime.now().strftime("%Y-%m") return get_cost_breakdown(product_code, period) @router.get("/dashboard") def cost_dashboard(period: Optional[str] = Query(None)): """成本分析首页—汇总数据""" if period is None: period = datetime.now().strftime("%Y-%m") overview = get_cost_overview(period) # 获取所有产品列表 db = get_db().__next__() try: products = db.query(ActualCost.product_code, ActualCost.product_name).filter( ActualCost.period == period ).distinct().all() product_list = [{"code": p[0], "name": p[1]} for p in products] # 各产品成本 product_costs = [] for code, name in products: costs = db.query(ActualCost).filter( ActualCost.product_code == code, ActualCost.period == period, ).all() total = round(sum(c.actual_cost for c in costs), 2) product_costs.append({"product_code": code, "product_name": name, "total_cost": total}) finally: db.close() return { "period": period, "overview": overview, "products": product_list, "total_cost": round(sum(p["total_cost"] for p in product_list) + overview.get("erp_cost", 0), 2) if product_list else 0, }