Hermes CI Fix
3dddd36866
包含前后端完整代码： - 前端：Vue3+Vite+ElementPlus - 后端：FastAPI+SQLAlchemy - 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动 - 当前版本：v1.0.0
238 lines
8.5 KiB
Python
238 lines
8.5 KiB
Python
"""成本分析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,
|
|
}
|