Hermes CI Fix
ac645a56ee
- ai_analysis: generate_rule_suggestions pending预警按entity过滤; kpi_analysis归属校验404; dashboard-analysis-stream/ask/review-plans 补 get_entity_id + KPI/ActionPlan/KPIAlert 全按账套过滤; stream cache key 含 entity_id (原跨企业串缓存) - dashboard: summary alert_count 补 KPIAlert.entity_id 过滤 (kpi_total 已有) - bot_kpis: update_bot_kpi_value 归属校验 404 (原任意登录可改他企 bot KPI) - cost: ActualCost/StandardCost 确认多企业共用 → cost_overview/variance/breakdown/dashboard 补 entity 过滤(engine 函数加 entity_id 透传); dashboard total_cost 既有 KeyError 修复 - kpis: create_kpi 字段白名单 (仿 UPDATE_KPI_WHITELIST+kpi_code), 拦截 status/important_flag 注入 - tests: 新增 test_multitenant_ai_20260907.py 11条越权用例 (跨entity 404/空返回断言) - 全量 pytest: 745 passed / 10 failed(与基线 HEAD 一致, 存量问题非本次引入) / 1 xfailed
298 lines
12 KiB
Python
298 lines
12 KiB
Python
"""成本分析引擎 — 管理会计OS
|
||
标准成本vs实际成本差异分析（量差/价差/效率差异）
|
||
ABC作业成本法分配
|
||
"""
|
||
import logging, os
|
||
from datetime import datetime
|
||
from typing import Optional, List, Dict
|
||
from app.database import get_session_local
|
||
from app.models import StandardCost, ActualCost, AbcActivity, AbcAllocation, KPIDefinition, KPIValue
|
||
|
||
logger = logging.getLogger("cma.cost")
|
||
|
||
ERP_API = "http://127.0.0.1:8300"
|
||
ERP_KEY = os.environ.get("ERP_API_KEY", "")
|
||
|
||
|
||
# ============================================================
|
||
# 差异计算
|
||
# ============================================================
|
||
|
||
def calc_variance(standard_qty: float, actual_qty: float,
|
||
standard_price: float, actual_price: float) -> dict:
|
||
"""计算量差和价差
|
||
|
||
量差 = (实际用量 - 标准用量) × 标准价格
|
||
价差 = (实际价格 - 标准价格) × 实际用量
|
||
总差异 = 量差 + 价差
|
||
"""
|
||
qty_variance = round((actual_qty - standard_qty) * standard_price, 2)
|
||
price_variance = round((actual_price - standard_price) * actual_qty, 2)
|
||
total_variance = round(qty_variance + price_variance, 2)
|
||
return {
|
||
"qty_variance": qty_variance, # 量差
|
||
"price_variance": price_variance, # 价差
|
||
"total_variance": total_variance, # 总差异
|
||
"standard_cost": round(standard_qty * standard_price, 2),
|
||
"actual_cost": round(actual_qty * actual_price, 2),
|
||
}
|
||
|
||
|
||
def calc_efficiency_variance(standard_hours: float, actual_hours: float,
|
||
standard_rate: float) -> dict:
|
||
"""计算效率差异（人工/制造费用）
|
||
|
||
效率差异 = (实际工时 - 标准工时) × 标准分配率
|
||
分配率差异 = (实际分配率 - 标准分配率) × 实际工时
|
||
"""
|
||
eff = round((actual_hours - standard_hours) * standard_rate, 2)
|
||
# 假设实际分配率从外面传入
|
||
return {
|
||
"efficiency_variance": eff,
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# 产品级差异分析
|
||
# ============================================================
|
||
|
||
def calc_product_variance(product_code: str, period: str, entity_id: Optional[int] = None) -> dict:
|
||
"""计算指定产品在指定期间的成本差异（entity_id=None 时不过滤，API 调用必传——多租户隔离 2026-09-07）"""
|
||
db = get_session_local()()
|
||
try:
|
||
q_std = db.query(StandardCost).filter(
|
||
StandardCost.product_code == product_code,
|
||
StandardCost.status == "active",
|
||
)
|
||
q_act = db.query(ActualCost).filter(
|
||
ActualCost.product_code == product_code,
|
||
ActualCost.period == period,
|
||
)
|
||
if entity_id is not None:
|
||
q_std = q_std.filter(StandardCost.entity_id == entity_id)
|
||
q_act = q_act.filter(ActualCost.entity_id == entity_id)
|
||
standards = q_std.all()
|
||
actuals = q_act.all()
|
||
|
||
if not standards or not actuals:
|
||
return {"product_code": product_code, "error": "标准成本或实际成本数据不足", "items": [], "summary": {}}
|
||
|
||
# 按 cost_type 分组
|
||
cost_types = set()
|
||
for s in standards: cost_types.add(s.cost_type)
|
||
for a in actuals: cost_types.add(a.cost_type)
|
||
|
||
items = []
|
||
total_std = 0
|
||
total_act = 0
|
||
total_qty_var = 0
|
||
total_price_var = 0
|
||
|
||
for ct in sorted(cost_types):
|
||
std_items = [s for s in standards if s.cost_type == ct]
|
||
act_items = [a for a in actuals if a.cost_type == ct]
|
||
|
||
if std_items and act_items:
|
||
s = std_items[0]
|
||
a = act_items[0]
|
||
var = calc_variance(s.standard_quantity, a.actual_quantity,
|
||
s.standard_price, a.actual_price)
|
||
items.append({
|
||
"cost_type": ct,
|
||
"item_name": s.item_name,
|
||
"standard_quantity": s.standard_quantity,
|
||
"actual_quantity": a.actual_quantity,
|
||
"standard_price": s.standard_price,
|
||
"actual_price": a.actual_price,
|
||
"standard_cost": var["standard_cost"],
|
||
"actual_cost": var["actual_cost"],
|
||
"qty_variance": var["qty_variance"],
|
||
"price_variance": var["price_variance"],
|
||
"total_variance": var["total_variance"],
|
||
})
|
||
total_std += var["standard_cost"]
|
||
total_act += var["actual_cost"]
|
||
total_qty_var += var["qty_variance"]
|
||
total_price_var += var["price_variance"]
|
||
|
||
return {
|
||
"product_code": product_code,
|
||
"period": period,
|
||
"items": items,
|
||
"summary": {
|
||
"total_standard_cost": round(total_std, 2),
|
||
"total_actual_cost": round(total_act, 2),
|
||
"total_variance": round(total_act - total_std, 2),
|
||
"total_qty_variance": round(total_qty_var, 2),
|
||
"total_price_variance": round(total_price_var, 2),
|
||
"variance_rate": round((total_act - total_std) / total_std * 100, 2) if total_std else 0,
|
||
}
|
||
}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# ABC 作业成本分配
|
||
# ============================================================
|
||
|
||
def calc_driver_rate(activity_id: int) -> dict:
|
||
"""计算作业动因分配率 = 总成本 / 动因总量"""
|
||
db = get_session_local()()
|
||
try:
|
||
act = db.query(AbcActivity).filter(AbcActivity.id == activity_id).first()
|
||
if not act or not act.driver_volume:
|
||
return {"error": "作业中心不存在或动因总量为0"}
|
||
rate = round(act.total_cost / act.driver_volume, 4) if act.driver_volume > 0 else 0
|
||
act.driver_rate = rate
|
||
db.commit()
|
||
return {
|
||
"activity_code": act.activity_code,
|
||
"activity_name": act.activity_name,
|
||
"total_cost": act.total_cost,
|
||
"driver_volume": act.driver_volume,
|
||
"driver_rate": rate,
|
||
}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def allocate_cost(activity_id: int, period: str, product_code: str,
|
||
product_name: str, driver_consumed: float) -> dict:
|
||
"""按动因分配成本到产品"""
|
||
db = get_session_local()()
|
||
try:
|
||
act = db.query(AbcActivity).filter(AbcActivity.id == activity_id).first()
|
||
if not act or act.driver_rate == 0:
|
||
# 自动计算分配率
|
||
if act and act.driver_volume > 0:
|
||
act.driver_rate = round(act.total_cost / act.driver_volume, 4)
|
||
db.commit()
|
||
if not act or act.driver_rate == 0:
|
||
return {"error": "分配率未设置"}
|
||
allocated = round(driver_consumed * act.driver_rate, 2)
|
||
alloc = AbcAllocation(
|
||
period=period,
|
||
activity_id=activity_id,
|
||
product_code=product_code,
|
||
product_name=product_name,
|
||
driver_consumed=driver_consumed,
|
||
allocated_cost=allocated,
|
||
)
|
||
db.add(alloc)
|
||
db.commit()
|
||
return {
|
||
"activity_code": act.activity_code,
|
||
"product_code": product_code,
|
||
"driver_consumed": driver_consumed,
|
||
"driver_rate": act.driver_rate,
|
||
"allocated_cost": allocated,
|
||
}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# 成本总览数据
|
||
# ============================================================
|
||
|
||
def get_cost_overview(period: str, entity_id: Optional[int] = None) -> dict:
|
||
"""获取成本总览数据（总成本、结构占比、趋势）
|
||
entity_id=None 时不过滤（兼容存量调用）；API 调用必须传 entity_id —— 多租户隔离 2026-09-07"""
|
||
db = get_session_local()()
|
||
try:
|
||
# 从实际成本表汇总（按账套过滤）
|
||
q_costs = db.query(ActualCost).filter(ActualCost.period == period)
|
||
if entity_id is not None:
|
||
q_costs = q_costs.filter(ActualCost.entity_id == entity_id)
|
||
actual_costs = q_costs.all()
|
||
total_cost = sum(a.actual_cost for a in actual_costs)
|
||
|
||
# 按成本类型分组
|
||
by_type: Dict[str, float] = {}
|
||
for a in actual_costs:
|
||
by_type[a.cost_type] = by_type.get(a.cost_type, 0) + a.actual_cost
|
||
|
||
structure = [
|
||
{"cost_type": k, "amount": round(v, 2), "ratio": round(v / total_cost * 100, 1) if total_cost else 0}
|
||
for k, v in sorted(by_type.items())
|
||
]
|
||
|
||
# 从ERP获取成本数据做补充
|
||
import httpx
|
||
try:
|
||
resp = httpx.get(f"{ERP_API}/api/v1/query", params={"table": "BalanceInfo", "limit": 100},
|
||
headers={"X-API-Key": ERP_KEY}, timeout=10)
|
||
balance_data = resp.json().get("data", [])
|
||
erp_cost = 0
|
||
for b in balance_data:
|
||
if b.get("Act_ID") == 5: # 营业成本
|
||
erp_cost += float(b.get("Act_Hap", 0) or 0) + float(b.get("Act_Tot", 0) or 0)
|
||
except Exception:
|
||
erp_cost = 0
|
||
|
||
# 历史趋势（近6个月）
|
||
from sqlalchemy import text
|
||
year = period[:4]
|
||
months_texts = []
|
||
try:
|
||
m = int(period.split("-")[1])
|
||
for i in range(6):
|
||
pm = m - i
|
||
py = int(year)
|
||
while pm <= 0:
|
||
pm += 12
|
||
py -= 1
|
||
months_texts.append(f"{py}-{pm:02d}")
|
||
|
||
trend = []
|
||
for p in reversed(months_texts):
|
||
q_trend = db.query(ActualCost).filter(ActualCost.period == p)
|
||
if entity_id is not None:
|
||
q_trend = q_trend.filter(ActualCost.entity_id == entity_id)
|
||
costs = q_trend.all()
|
||
total = round(sum(c.actual_cost for c in costs), 2)
|
||
trend.append({"period": p, "total_cost": total})
|
||
except Exception:
|
||
trend = []
|
||
|
||
return {
|
||
"period": period,
|
||
"total_cost": round(total_cost + erp_cost, 2),
|
||
"erp_cost": round(erp_cost, 2),
|
||
"manual_cost": round(total_cost, 2),
|
||
"structure": structure,
|
||
"trend": trend,
|
||
}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def get_cost_breakdown(product_code: str, period: str, entity_id: Optional[int] = None) -> dict:
|
||
"""获取成本构成（料/工/费占比）（entity_id=None 不过滤——多租户隔离 2026-09-07，API 必传）"""
|
||
db = get_session_local()()
|
||
try:
|
||
q_actuals = db.query(ActualCost).filter(
|
||
ActualCost.product_code == product_code,
|
||
ActualCost.period == period,
|
||
)
|
||
if entity_id is not None:
|
||
q_actuals = q_actuals.filter(ActualCost.entity_id == entity_id)
|
||
actuals = q_actuals.all()
|
||
|
||
material = sum(a.actual_cost for a in actuals if a.cost_type == "material")
|
||
labor = sum(a.actual_cost for a in actuals if a.cost_type == "labor")
|
||
overhead = sum(a.actual_cost for a in actuals if a.cost_type == "overhead")
|
||
total = material + labor + overhead
|
||
|
||
return {
|
||
"product_code": product_code,
|
||
"period": period,
|
||
"material": {"amount": round(material, 2), "ratio": round(material / total * 100, 1) if total else 0},
|
||
"labor": {"amount": round(labor, 2), "ratio": round(labor / total * 100, 1) if total else 0},
|
||
"overhead": {"amount": round(overhead, 2), "ratio": round(overhead / total * 100, 1) if total else 0},
|
||
"total": round(total, 2),
|
||
}
|
||
finally:
|
||
db.close()
|