feat: 成本核算模块(鲁铭成本核算Excel全量适配, 价格库580条+75/85/95三系列模板+估算器基准精确还原)
- 解析12表成本核算工作簿: 价格库8表→产品价格本(华健型材模具/丝吉利娅/索宾柯五金/隔热条/胶条/赛遨辅材/螺丝/其他) - 「铝成本」页→系列成本模板(75框扇齐平1.4/85框扇齐平1.8/95窗纱一体1.8, 八分区JSON) - 估算器口径对齐核算表: 固定/开启/总面积归因缩放+人工每平实际口径+毛利阶梯10%~70% - 75系列基准估算total=1118.20与核算表成本合计完全一致(85/95同) - API /api/costbook: import-excel/series/estimate; 前端 /cost_estimate 页 - 种子 seed_costbook.py; 修复6处解析缺陷; 冒烟60/60+24页巡检0异常+深度审计16/16
This commit is contained in:
HHY
@@ -0,0 +1,107 @@
|
||||
"""成本核算路由：系列成本模板（取自成本核算 Excel）+ 价格库导入 + 成本估算。"""
|
||||
import json
|
||||
|
||||
from fastapi import APIRouter, UploadFile
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy import select
|
||||
|
||||
from huamenchuang.core.deps import CurrentUser, DbSession
|
||||
from huamenchuang.core.response import fail, ok
|
||||
from huamenchuang.models.business import CostSeriesTemplate
|
||||
from huamenchuang.services import costbook_service as svc
|
||||
|
||||
router = APIRouter(tags=["成本核算"])
|
||||
|
||||
|
||||
def _brief(t: CostSeriesTemplate) -> dict:
|
||||
data = json.loads(t.data)
|
||||
return {"id": t.id, "name": t.name, "series": t.series, "wall_thick": t.wall_thick,
|
||||
"source_file": t.source_file, "enabled": t.enabled, "window": data.get("window")}
|
||||
|
||||
|
||||
def _full(t: CostSeriesTemplate) -> dict:
|
||||
return {**_brief(t), "data": json.loads(t.data)}
|
||||
|
||||
|
||||
@router.post("/costbook/import-excel")
|
||||
async def import_excel(file: UploadFile, user: CurrentUser, db: DbSession):
|
||||
"""上传成本核算 Excel（12 sheet 鲁铭格式）：价格库 8 表入产品价格本，铝成本页入系列模板。"""
|
||||
blob = await file.read()
|
||||
if not blob:
|
||||
return fail("文件为空")
|
||||
import io
|
||||
|
||||
from openpyxl import load_workbook
|
||||
try:
|
||||
wb = load_workbook(io.BytesIO(blob), data_only=True)
|
||||
parsed = {"file": file.filename or "upload.xlsx",
|
||||
"series_summary": svc._parse_series_summary(wb["铝成本"]),
|
||||
"libraries": {sn: svc._parse_library(wb[sn])
|
||||
for sn in svc.LIB_SHEETS if sn in wb.sheetnames},
|
||||
"optimize_stats": svc._parse_optimize_stats(wb["铝计算"]) if "铝计算" in wb.sheetnames else {}}
|
||||
except KeyError as e:
|
||||
return fail(f"缺少必需工作表 {e}，请确认上传成本核算格式工作簿")
|
||||
except Exception as e: # noqa: BLE001
|
||||
return fail(f"解析失败：{e}")
|
||||
lib_stats = await svc.import_libraries(db, user.company_id, parsed["libraries"])
|
||||
await svc.import_series(db, user.company_id, parsed)
|
||||
return ok(data={"libraries": lib_stats, "optimize_stats": parsed["optimize_stats"]},
|
||||
message=f"导入完成：价格库新增 {lib_stats['created']} / 更新 {lib_stats['updated']} 条")
|
||||
|
||||
|
||||
@router.get("/costbook/series")
|
||||
async def list_series(user: CurrentUser, db: DbSession):
|
||||
rows = list((await db.execute(select(CostSeriesTemplate).where(
|
||||
CostSeriesTemplate.company_id == user.company_id)
|
||||
.order_by(CostSeriesTemplate.series, CostSeriesTemplate.id))).scalars())
|
||||
return ok(data=[_brief(t) for t in rows])
|
||||
|
||||
|
||||
class EstimateBody(BaseModel):
|
||||
series_id: int
|
||||
width: float = Field(gt=0, description="窗宽 mm")
|
||||
height: float = Field(gt=0, description="窗高 mm")
|
||||
count: int = Field(default=1, ge=1, description="樘数")
|
||||
sash_count: int | None = Field(default=None, ge=0, description="开启扇数，缺省按基准开启率")
|
||||
|
||||
|
||||
@router.post("/costbook/estimate")
|
||||
async def estimate(body: EstimateBody, user: CurrentUser, db: DbSession):
|
||||
t = await db.get(CostSeriesTemplate, body.series_id)
|
||||
if t is None or t.company_id != user.company_id:
|
||||
return fail("系列模板不存在")
|
||||
return ok(data=svc.estimate(t, body.width, body.height, body.count, body.sash_count))
|
||||
|
||||
|
||||
@router.get("/costbook/series/{sid}")
|
||||
async def get_series(sid: int, user: CurrentUser, db: DbSession):
|
||||
t = await db.get(CostSeriesTemplate, sid)
|
||||
if t is None or t.company_id != user.company_id:
|
||||
return fail("系列模板不存在")
|
||||
return ok(data=_full(t))
|
||||
|
||||
|
||||
class SeriesBody(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=64)
|
||||
series: int = Field(ge=0)
|
||||
wall_thick: float = Field(gt=0)
|
||||
data: dict
|
||||
|
||||
|
||||
@router.post("/costbook/series")
|
||||
async def create_series(body: SeriesBody, user: CurrentUser, db: DbSession):
|
||||
t = CostSeriesTemplate(company_id=user.company_id, name=body.name, series=body.series,
|
||||
wall_thick=body.wall_thick,
|
||||
data=json.dumps(body.data, ensure_ascii=False))
|
||||
db.add(t)
|
||||
await db.flush()
|
||||
return ok(data={"id": t.id}, message="系列模板已创建")
|
||||
|
||||
|
||||
@router.delete("/costbook/series/{sid}")
|
||||
async def delete_series(sid: int, user: CurrentUser, db: DbSession):
|
||||
t = await db.get(CostSeriesTemplate, sid)
|
||||
if t is None or t.company_id != user.company_id:
|
||||
return fail("系列模板不存在")
|
||||
await db.delete(t)
|
||||
return ok(message="系列模板已删除")
|
||||
@@ -6,6 +6,7 @@ from huamenchuang.api import (
|
||||
auth,
|
||||
business,
|
||||
contracts,
|
||||
costbook,
|
||||
customers,
|
||||
finance,
|
||||
optimize,
|
||||
@@ -45,6 +46,7 @@ api_router.include_router(salesmen.router)
|
||||
api_router.include_router(purchases.router)
|
||||
api_router.include_router(warehouse.router)
|
||||
api_router.include_router(business.router)
|
||||
api_router.include_router(costbook.router)
|
||||
api_router.include_router(platform.router)
|
||||
api_router.include_router(reports.router)
|
||||
api_router.include_router(platform.router)
|
||||
|
||||
@@ -3,6 +3,7 @@ from huamenchuang.models.announcement import Announcement
|
||||
from huamenchuang.models.business import (
|
||||
AfterSale,
|
||||
ContractReceipt,
|
||||
CostSeriesTemplate,
|
||||
CustomerFollow,
|
||||
MeasureOrder,
|
||||
PriceScript,
|
||||
@@ -42,6 +43,7 @@ __all__ = [
|
||||
"CompanyParam",
|
||||
"Contract",
|
||||
"ContractReceipt",
|
||||
"CostSeriesTemplate",
|
||||
"Customer",
|
||||
"CustomerFollow",
|
||||
"DEFAULT_PROCESSES",
|
||||
|
||||
@@ -127,6 +127,41 @@ class ProductCatelog(Base):
|
||||
company_id: Mapped[int] = mapped_column(default=1, index=True)
|
||||
|
||||
|
||||
class CostSeriesTemplate(Base):
|
||||
"""成本核算系列模板：某型材系列标准窗的全量用料+价格参数，取自成本核算 Excel。
|
||||
|
||||
data JSON 结构（对齐核算表"铝成本"页分区）：
|
||||
{
|
||||
"window": {"w":1600,"h":1600,"area":2.56,"open_area":0.78,"fixed_area":1.78,
|
||||
"sash_area":0.78,"sash_perimeter":4,"glass_count":3},
|
||||
"profile": [{"code","name","length","weight_per_m","unit_price","alloc"}],
|
||||
"addon": [{"code","name","length","weight_per_m","unit_price","alloc"}], # 角码/连接件
|
||||
"strip": [{"code","frame","sash","mullion","splice","total","unit_price","alloc"}], # 隔热条
|
||||
"seal": [{"code","name","length","unit_price","alloc"}], # 胶条
|
||||
"glass": [{"count","config","area","unit_price","alloc"}],
|
||||
"extra": [{"name","spec","count","unit_price","alloc"}], # 辅材(外购纱窗等)
|
||||
"hardware": [{"vendor","name","count","unit_price"}],
|
||||
"labor": {"install":60,"w14":35,"w18":60,"sales":30,"fuel":10,"ins":5}
|
||||
}
|
||||
alloc: fixed=随固定面积缩放 / open=随开启面积缩放 / both=随总面积缩放 / once=按樘数
|
||||
"""
|
||||
|
||||
__tablename__ = "cost_series_templates"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
|
||||
company_id: Mapped[int] = mapped_column(default=1, index=True)
|
||||
name: Mapped[str] = mapped_column(String(64))
|
||||
series: Mapped[int] = mapped_column(default=0) # 门窗系列 75/85/95
|
||||
wall_thick: Mapped[float] = mapped_column(default=1.4) # 壁厚
|
||||
source_file: Mapped[str] = mapped_column(String(255), default="")
|
||||
data: Mapped[str] = mapped_column(Text, default="{}")
|
||||
enabled: Mapped[int] = mapped_column(default=1)
|
||||
created_at: Mapped[datetime.datetime] = mapped_column(DateTime, default=datetime.datetime.utcnow)
|
||||
updated_at: Mapped[datetime.datetime] = mapped_column(
|
||||
DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow
|
||||
)
|
||||
|
||||
|
||||
class Quotation(Base):
|
||||
"""报价单：画布算料结果 + 手动调整，归属客户/合同。"""
|
||||
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
"""成本核算数据种子：解析 data/ 下三个鲁铭系列成本核算 Excel，全量导入价格库与系列模板。
|
||||
|
||||
用法: uv run python -m huamenchuang.scripts.seed_costbook [--company 1]
|
||||
"""
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
DATA = Path(__file__).resolve().parents[3] / "data"
|
||||
FILES = [
|
||||
"3.2成本核算鲁铭75框扇齐平系列1.4.xlsx",
|
||||
"3.2成本核算鲁铭85框扇齐平系列1.8.xlsx",
|
||||
"2026.7.28成本核算鲁铭95窗纱一体系列1.8.xlsx",
|
||||
]
|
||||
|
||||
|
||||
async def seed(company_id: int = 1) -> None:
|
||||
import huamenchuang.models # noqa: F401
|
||||
from huamenchuang.core.database import SessionLocal, init_db
|
||||
from huamenchuang.services import costbook_service as svc
|
||||
|
||||
await init_db()
|
||||
async with SessionLocal() as db:
|
||||
lib_stats = {"created": 0, "updated": 0}
|
||||
series_ids = []
|
||||
for fn in FILES:
|
||||
path = DATA / fn
|
||||
if not path.exists():
|
||||
print(f"跳过（不存在）: {fn}")
|
||||
continue
|
||||
parsed = svc.parse_costbook(path)
|
||||
stats = await svc.import_libraries(db, company_id, parsed["libraries"])
|
||||
tid = await svc.import_series(db, company_id, parsed)
|
||||
series_ids.append((tid, parsed["series_summary"]["window"].get("series"),
|
||||
parsed["file"]))
|
||||
lib_stats["created"] += stats["created"]
|
||||
lib_stats["updated"] += stats["updated"]
|
||||
opt = parsed["optimize_stats"]
|
||||
print(f"[{fn}] 系列{parsed['series_summary']['window'].get('series')} "
|
||||
f"型材{len(parsed['series_summary']['profile'])}行 优化单{opt.get('bars', 0)}支")
|
||||
await db.commit()
|
||||
print(f"价格库: 新增 {lib_stats['created']} / 更新 {lib_stats['updated']}")
|
||||
print(f"系列模板: {series_ids}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
company = 1
|
||||
if "--company" in sys.argv:
|
||||
company = int(sys.argv[sys.argv.index("--company") + 1])
|
||||
asyncio.run(seed(company))
|
||||
@@ -202,6 +202,33 @@ def main():
|
||||
d = call("POST", f"/api/purchases/{poid}/receive")
|
||||
check("P1-extra 采购入库联动", d.get("code") == 1)
|
||||
|
||||
# ========== 成本核算（v2.36 鲁铭 Excel 适配） ==========
|
||||
d = call("GET", "/api/costbook/series")
|
||||
check("COST-1 系列模板列表", d.get("code") == 1 and len(d["data"]) >= 3)
|
||||
sid75 = next((s["id"] for s in d.get("data", []) if s["series"] == 75), None)
|
||||
check("COST-2 75系列模板存在", bool(sid75))
|
||||
d = call("POST", "/api/costbook/estimate", {"series_id": sid75, "width": 1600, "height": 1600, "count": 1})
|
||||
est = d.get("data", {})
|
||||
check("COST-3 75基准估算=1118.20", d.get("code") == 1 and abs(est.get("total", 0) - 1118.20) < 1.5,
|
||||
f"got={est.get('total')}")
|
||||
check("COST-4 分区成本齐全", all(k in est for k in
|
||||
("profile_cost", "addon_cost", "strip_cost", "seal_cost", "glass_cost", "hardware_cost",
|
||||
"labor_cost", "other_cost")))
|
||||
check("COST-5 毛利阶梯50%", abs(est.get("margin_ladder", {}).get("50%", 0) - 1677.29) < 1.5)
|
||||
d = call("POST", "/api/costbook/estimate", {"series_id": sid75, "width": 2000, "height": 1800, "count": 2})
|
||||
check("COST-6 任意尺寸估算", d.get("code") == 1 and d["data"]["total"] > 0 and d["data"]["count"] == 2)
|
||||
d = call("GET", "/api/products?page=1&page_size=5&keyword=隔热条")
|
||||
check("COST-7 价格库(隔热条)入库", d.get("code") == 1 and d["data"]["total"] > 0)
|
||||
d = call("GET", "/api/products?page=1&page_size=1")
|
||||
check("COST-8 价格库总量>500", d.get("code") == 1 and d["data"]["total"] > 500,
|
||||
f"total={d.get('data', {}).get('total')}")
|
||||
# 越权：B 公司无模板（若 B 公司账号存在）
|
||||
rb = httpx.post(f"{BASE}/api/auth/login", json={"phone_number": "13900000001", "password": "123456"}, timeout=15)
|
||||
if rb.status_code == 200 and rb.json().get("code") == 1:
|
||||
HB = {"Authorization": f"Bearer {rb.json()['data']['token']}"}
|
||||
d = httpx.get(f"{BASE}/api/costbook/series", headers=HB, timeout=15).json()
|
||||
check("COST-9 公司隔离(B公司无模板)", d.get("code") == 1 and len(d["data"]) == 0)
|
||||
|
||||
# ========== 汇总 ==========
|
||||
print(f"\n===== 冒烟结果 {sum(1 for _, r in results if r == 'PASS')}/{len(results)} PASS =====")
|
||||
for name, r in results:
|
||||
|
||||
@@ -0,0 +1,457 @@
|
||||
"""成本核算工作簿解析与适配服务。
|
||||
|
||||
解析门窗厂成本核算 Excel（鲁铭系列格式，12 个 sheet）：
|
||||
- 价格库 8 sheet（三文件共享）→ Product 产品价格本
|
||||
- 「铝成本」系列核算页 → CostSeriesTemplate 系列成本模板
|
||||
- 「铝计算」型材优化单 → 汇总统计（记录核对用）
|
||||
|
||||
估算器 estimate() 复刻核算表口径：
|
||||
- 固定类料随固定面积缩放、开启类料随开启面积缩放、总面积类随面积缩放
|
||||
- 人工安装制作 = (安装费 + 对应壁厚人工费) × 总面积
|
||||
- 销售及其他 = (销售提成 + 加油 + 保险) × 总面积
|
||||
- 毛利阶梯 = 成本合计 × (1 + 10%..70%)
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from huamenchuang.models.business import CostSeriesTemplate, Product
|
||||
|
||||
# ---------------- sheet 常量 ----------------
|
||||
LIB_SHEETS = [
|
||||
"华健开模汇总表", "丝吉利娅五金价格", "索宾柯价格", "隔热条价格",
|
||||
"胶条价格", "赛遨价格", "螺丝价格", "其他价格",
|
||||
]
|
||||
|
||||
|
||||
def _num(v) -> float:
|
||||
"""单元格转 float，失败返回 0。"""
|
||||
if v is None:
|
||||
return 0.0
|
||||
if isinstance(v, (int, float)):
|
||||
return float(v)
|
||||
try:
|
||||
return float(str(v).strip())
|
||||
except ValueError:
|
||||
return 0.0
|
||||
|
||||
|
||||
def _txt(v) -> str:
|
||||
return str(v).strip() if v is not None else ""
|
||||
|
||||
|
||||
def _col(ws, letter: str, row: int):
|
||||
return ws[f"{letter}{row}"].value
|
||||
|
||||
|
||||
# ---------------- 解析 ----------------
|
||||
|
||||
def parse_costbook(path: str | Path) -> dict:
|
||||
"""全量解析一个成本核算工作簿。"""
|
||||
wb = load_workbook(path, data_only=True)
|
||||
return {
|
||||
"file": Path(path).name,
|
||||
"series_summary": _parse_series_summary(wb["铝成本"]),
|
||||
"libraries": {sn: _parse_library(wb[sn]) for sn in LIB_SHEETS if sn in wb.sheetnames},
|
||||
"optimize_stats": _parse_optimize_stats(wb["铝计算"]) if "铝计算" in wb.sheetnames else {},
|
||||
}
|
||||
|
||||
|
||||
def _parse_series_summary(ws) -> dict:
|
||||
"""解析「铝成本」页 → 系列标准成本模板 data JSON。"""
|
||||
header_row = next((r for r in range(17, 20) if _txt(_col(ws, "A", r)) == "铝型材成本"), None)
|
||||
if header_row is None:
|
||||
raise ValueError("未找到「铝型材成本」分区表头")
|
||||
summary: dict = {"window": _parse_window_header(ws), "labor": _parse_labor(ws)}
|
||||
|
||||
# 各分区起始行
|
||||
starts = {}
|
||||
for r in range(header_row, ws.max_row + 1):
|
||||
t = _txt(_col(ws, "A", r))
|
||||
if t.endswith("成本") and len(t) <= 8:
|
||||
starts[t] = r
|
||||
elif t == "辅材成本":
|
||||
starts[t] = r
|
||||
|
||||
def rows_of(start: int, end: int, col: str = "A") -> list[int]:
|
||||
"""收集分区内数据行：按 col 列非空判定，跳过合计行。"""
|
||||
return [r for r in range(start + 2, end)
|
||||
if _txt(_col(ws, col, r)) and _txt(_col(ws, col, r)) != "合计："]
|
||||
|
||||
bounds = sorted(starts.items(), key=lambda kv: kv[1]) + [(None, ws.max_row + 1)]
|
||||
|
||||
def section_range(name: str) -> tuple[int, int]:
|
||||
for i, (k, v) in enumerate(bounds):
|
||||
if k == name:
|
||||
return v, bounds[i + 1][1]
|
||||
return 0, 0
|
||||
|
||||
s, e = section_range("铝型材成本")
|
||||
summary["profile"] = [{"code": _txt(_col(ws, "A", r)), "name": _txt(_col(ws, "B", r)),
|
||||
"length": _num(_col(ws, "D", r)), "weight_per_m": _num(_col(ws, "E", r)),
|
||||
"unit_price": _num(_col(ws, "G", r)), "alloc": _alloc(_txt(_col(ws, "B", r)))}
|
||||
for r in rows_of(s, e)]
|
||||
s, e = section_range("型材辅件成本")
|
||||
summary["addon"] = [{"code": _txt(_col(ws, "A", r)), "name": _txt(_col(ws, "B", r)),
|
||||
"length": _num(_col(ws, "D", r)), "weight_per_m": _num(_col(ws, "E", r)),
|
||||
"unit_price": _num(_col(ws, "G", r)), "alloc": _alloc(_txt(_col(ws, "B", r)))}
|
||||
for r in rows_of(s, e, "B")]
|
||||
s, e = section_range("隔热条成本")
|
||||
summary["strip"] = []
|
||||
for r in rows_of(s, e):
|
||||
total = _num(_col(ws, "F", r)) or (_num(_col(ws, "B", r)) + _num(_col(ws, "C", r))
|
||||
+ _num(_col(ws, "D", r)) + _num(_col(ws, "E", r)))
|
||||
if not total:
|
||||
continue
|
||||
alloc = "open" if (_num(_col(ws, "C", r)) and not _num(_col(ws, "B", r))) else \
|
||||
("both" if _num(_col(ws, "C", r)) else "fixed")
|
||||
summary["strip"].append({"code": _txt(_col(ws, "A", r)), "frame": _num(_col(ws, "B", r)),
|
||||
"sash": _num(_col(ws, "C", r)), "mullion": _num(_col(ws, "D", r)),
|
||||
"splice": _num(_col(ws, "E", r)), "total": total,
|
||||
"unit_price": _num(_col(ws, "G", r)), "alloc": alloc})
|
||||
s, e = section_range("胶条成本")
|
||||
summary["seal"] = [{"code": _txt(_col(ws, "A", r)), "name": _txt(_col(ws, "C", r)),
|
||||
"length": _num(_col(ws, "F", r)), "unit_price": _num(_col(ws, "G", r)),
|
||||
"alloc": _alloc(_txt(_col(ws, "C", r)))} for r in rows_of(s, e, "C")]
|
||||
s, e = section_range("玻璃成本")
|
||||
summary["glass"] = []
|
||||
for r in rows_of(s, e, "B"):
|
||||
cfg = _txt(_col(ws, "B", r))
|
||||
if "+" not in cfg:
|
||||
continue
|
||||
summary["glass"].append({"count": _num(_col(ws, "A", r)), "config": cfg,
|
||||
"area": _num(_col(ws, "E", r)), "unit_price": _num(_col(ws, "G", r)),
|
||||
"alloc": "open" if "开启" in cfg else "fixed"})
|
||||
s, e = section_range("辅材成本")
|
||||
summary["extra"] = [{"name": _txt(_col(ws, "C", r)) or _txt(_col(ws, "A", r)),
|
||||
"spec": _txt(_col(ws, "E", r)), "count": _num(_col(ws, "F", r)),
|
||||
"unit_price": _num(_col(ws, "G", r)), "alloc": "once"}
|
||||
for r in rows_of(s, e, "C")]
|
||||
s, e = section_range("五金成本")
|
||||
summary["hardware"] = [{"vendor": _txt(_col(ws, "A", r)), "code": _txt(_col(ws, "B", r)),
|
||||
"name": _txt(_col(ws, "D", r)), "count": _num(_col(ws, "F", r)),
|
||||
"unit_price": _num(_col(ws, "G", r)), "alloc": "open"}
|
||||
for r in rows_of(s, e, "D")]
|
||||
# 成本汇总区的人工/其他成本 → 每平单价（核算表实际口径，优先于参数推导）
|
||||
area = summary["window"].get("area") or 0
|
||||
for r in range(1, header_row):
|
||||
label = _txt(_col(ws, "F", r))
|
||||
if label == "人工安装制作成本：" and area:
|
||||
summary["labor"]["rate_labor"] = round(_num(_col(ws, "G", r)) / area, 4)
|
||||
elif label == "销售+其他成本：" and area:
|
||||
summary["labor"]["rate_other"] = round(_num(_col(ws, "G", r)) / area, 4)
|
||||
return summary
|
||||
|
||||
|
||||
def _parse_window_header(ws) -> dict:
|
||||
"""解析表头区（1~15 行）的窗型参数。"""
|
||||
win: dict = {}
|
||||
for r in range(1, 16):
|
||||
for c in range(1, ws.max_column + 1):
|
||||
v = ws.cell(r, c).value
|
||||
if not isinstance(v, str):
|
||||
continue
|
||||
nxt = ws.cell(r, c + 1).value
|
||||
mapping = {"门窗数量(樘)：": "count", "总面积(㎡)：": "area", "平均面积(㎡)：": "avg_area",
|
||||
"开启率：": "open_rate", "开启扇面积：": "sash_area",
|
||||
"开启扇周长：": "sash_perimeter", "玻璃块数：": "glass_count"}
|
||||
if v in mapping and nxt is not None:
|
||||
win[mapping[v]] = _num(nxt)
|
||||
order_name = _txt(_col(ws, "B", 2))
|
||||
# 兼容「75系列」「95窗纱一体系列」两种写法：系列字向前找最近数字
|
||||
m = re.search(r"(\d+(?:\.\d+)?)(?:窗纱一体|框扇齐平|平开|推拉)?系列", order_name)
|
||||
if not m:
|
||||
m = re.search(r"(\d+(?:\.\d+)?)", order_name)
|
||||
if m:
|
||||
win["series"] = float(m.group(1))
|
||||
m2 = re.search(r"(\d+(?:\.\d+)?)壁厚", _txt(_col(ws, "B", 2)))
|
||||
if m2:
|
||||
win["wall_thick"] = float(m2.group(1))
|
||||
# 基准窗宽高：Excel 未给尺寸时按基准面积取等效正窗（各料均按面积比例缩放，宽高比不影响口径）
|
||||
if "area" in win and "w" not in win:
|
||||
side = round((win["area"] ** 0.5) * 1000)
|
||||
win["w"], win["h"] = side, side
|
||||
win.setdefault("open_area", win.get("sash_area", 0))
|
||||
win.setdefault("fixed_area", round(win.get("area", 0) - win.get("sash_area", 0), 4))
|
||||
return win
|
||||
|
||||
|
||||
def _parse_labor(ws) -> dict:
|
||||
"""解析人工费参数区（K/L 列）。"""
|
||||
labor = {"install": 0, "w14": 0, "w18": 0, "sales": 0, "fuel": 0, "ins": 0}
|
||||
keymap = {"安装费：": "install", "1.4壁厚人工费：": "w14", "1.8壁厚人工费：": "w18",
|
||||
"销售：": "sales", "加油：": "fuel", "保险：": "ins"}
|
||||
for r in range(1, 20):
|
||||
for c in range(1, ws.max_column):
|
||||
v = ws.cell(r, c).value
|
||||
if isinstance(v, str) and v in keymap:
|
||||
labor[keymap[v]] = _num(ws.cell(r, c + 1).value)
|
||||
return labor
|
||||
|
||||
|
||||
def _alloc(name: str) -> str:
|
||||
"""按名称推断用料归因（fixed/open/both）。"""
|
||||
if "扇" in name:
|
||||
return "open"
|
||||
if any(k in name for k in ("框", "梃", "拼")):
|
||||
return "fixed"
|
||||
return "both"
|
||||
|
||||
|
||||
def _parse_library(ws) -> list[dict]:
|
||||
"""解析价格库 sheet → 统一行结构 [{code,name,spec,unit,price,note}]。"""
|
||||
sn = ws.title
|
||||
out: list[dict] = []
|
||||
if sn == "华健开模汇总表":
|
||||
for r in range(3, ws.max_row + 1):
|
||||
code = _txt(_col(ws, "B", r))
|
||||
if not code:
|
||||
continue
|
||||
out.append({"code": code, "name": _txt(_col(ws, "C", r)),
|
||||
"spec": f"{_num(_col(ws, 'D', r)):g}", "unit": "米", "price": 0,
|
||||
"note": f"模具费{_num(_col(ws, 'E', r)):g}元"})
|
||||
elif sn == "丝吉利娅五金价格":
|
||||
for r in range(2, ws.max_row + 1):
|
||||
code, name = _txt(_col(ws, "B", r)), _txt(_col(ws, "C", r))
|
||||
if not name:
|
||||
continue
|
||||
out.append({"code": code, "name": name, "spec": "", "unit": "套",
|
||||
"price": round(_num(_col(ws, "D", r)), 2), "note": "丝吉利娅"})
|
||||
elif sn == "索宾柯价格":
|
||||
for r in range(2, ws.max_row + 1):
|
||||
code = _txt(_col(ws, "A", r))
|
||||
if not code:
|
||||
continue
|
||||
out.append({"code": code, "name": f"索宾柯 {code}", "spec": "", "unit": "套",
|
||||
"price": round(_num(_col(ws, "B", r)), 2), "note": "索宾柯"})
|
||||
elif sn == "隔热条价格":
|
||||
for r in range(2, ws.max_row + 1):
|
||||
code = _txt(_col(ws, "A", r))
|
||||
if not code:
|
||||
continue
|
||||
out.append({"code": code, "name": f"隔热条 {_txt(_col(ws, 'B', r))}",
|
||||
"spec": f"{_num(_col(ws, 'C', r)):g}g/m", "unit": "米",
|
||||
"price": round(_num(_col(ws, "E", r)), 4), "note": f"{_num(_col(ws, 'D', r)):g}元/吨"})
|
||||
elif sn == "胶条价格":
|
||||
for r in range(3, ws.max_row + 1):
|
||||
code = _txt(_col(ws, "A", r))
|
||||
if not code:
|
||||
continue
|
||||
out.append({"code": code, "name": f"胶条 {code}",
|
||||
"spec": f"{_num(_col(ws, 'E', r)):.4f}kg/m", "unit": "米",
|
||||
"price": round(_num(_col(ws, "G", r)), 4), "note": f"河北三联 {_num(_col(ws, 'F', r)):g}元/箱"})
|
||||
elif sn == "赛遨价格":
|
||||
for r in range(3, ws.max_row + 1):
|
||||
code = _txt(_col(ws, "B", r))
|
||||
if not code:
|
||||
continue
|
||||
note = " | ".join(x for x in (_txt(_col(ws, "E", r)), _txt(_col(ws, "I", r))) if x)
|
||||
out.append({"code": code, "name": _txt(_col(ws, "C", r)), "spec": "", "unit": "个",
|
||||
"price": round(_num(_col(ws, "F", r)), 4), "note": note})
|
||||
elif sn == "螺丝价格":
|
||||
for r in range(3, ws.max_row + 1):
|
||||
name = _txt(_col(ws, "B", r))
|
||||
if not name:
|
||||
continue
|
||||
out.append({"code": "", "name": f"{name} {_txt(_col(ws, 'C', r))}",
|
||||
"spec": _txt(_col(ws, "D", r)), "unit": "个",
|
||||
"price": round(_num(_col(ws, "G", r)), 4), "note": "山东米兰"})
|
||||
elif sn == "其他价格":
|
||||
for r in range(2, ws.max_row + 1):
|
||||
name = _txt(_col(ws, "A", r))
|
||||
if not name:
|
||||
continue
|
||||
unit_txt, price = _txt(_col(ws, "B", r)), _num(_col(ws, "C", r))
|
||||
material = _guess_material(name, unit_txt)
|
||||
unit = {"元/kg": "公斤", "元/m²": "平方米", "元/套": "套", "元/个": "个",
|
||||
"元/桶": "桶", "元/米": "米"}.get(unit_txt, "个")
|
||||
out.append({"code": "", "name": name, "spec": "", "unit": unit,
|
||||
"price": round(price, 4), "note": "", "_material": material})
|
||||
return out
|
||||
|
||||
|
||||
def _guess_material(name: str, unit_txt: str) -> str:
|
||||
"""「其他价格」行 → Product.material 分类。"""
|
||||
if "玻璃" in name or unit_txt == "元/m²":
|
||||
return "glass"
|
||||
if any(k in name for k in ("执手", "合页", "隐藏式", "五金")):
|
||||
return "hardware"
|
||||
if any(k in name for k in ("型材", "基材")):
|
||||
return "profile"
|
||||
return "addon"
|
||||
|
||||
|
||||
def _parse_optimize_stats(ws) -> dict:
|
||||
"""「铝计算」型材优化单 → 汇总统计（核对参考，不逐行入库）。"""
|
||||
stats = {"bars": 0, "total_length": 0.0, "sashes": 0, "glasses": 0, "glass_area": 0.0}
|
||||
for r in range(3, ws.max_row + 1):
|
||||
code = _txt(_col(ws, "A", r))
|
||||
if code:
|
||||
stats["bars"] += 1
|
||||
stats["total_length"] += _num(_col(ws, "F", r))
|
||||
if _txt(_col(ws, "H", r)) and _num(_col(ws, "K", r)):
|
||||
stats["sashes"] += _num(_col(ws, "K", r))
|
||||
if _txt(_col(ws, "O", r)) and _num(_col(ws, "T", r)):
|
||||
stats["glasses"] += _num(_col(ws, "T", r))
|
||||
stats["glass_area"] += _num(_col(ws, "U", r)) * _num(_col(ws, "T", r))
|
||||
return stats
|
||||
|
||||
|
||||
# ---------------- 入库适配 ----------------
|
||||
|
||||
async def import_libraries(db: AsyncSession, company_id: int, libraries: dict) -> dict:
|
||||
"""价格库 → Product 表。同名同材质已存在则更新价格，否则新建。"""
|
||||
stats = {"created": 0, "updated": 0}
|
||||
for sheet, rows in libraries.items():
|
||||
for row in rows:
|
||||
material = row.get("_material") or _sheet_material(sheet)
|
||||
name = row["name"]
|
||||
existing = (await db.execute(select(Product).where(
|
||||
Product.company_id == company_id, Product.material == material,
|
||||
Product.name == name))).scalars().first()
|
||||
if existing is None:
|
||||
db.add(Product(company_id=company_id, material=material, name=name,
|
||||
spec=row.get("spec", ""), unit=row.get("unit", "个"),
|
||||
price=row.get("price", 0), note=row.get("note", "")))
|
||||
stats["created"] += 1
|
||||
else:
|
||||
if row.get("price"):
|
||||
existing.price = row["price"]
|
||||
if row.get("spec"):
|
||||
existing.spec = row["spec"]
|
||||
stats["updated"] += 1
|
||||
await db.flush()
|
||||
return stats
|
||||
|
||||
|
||||
def _sheet_material(sheet: str) -> str:
|
||||
if sheet == "华健开模汇总表":
|
||||
return "profile"
|
||||
if sheet in ("丝吉利娅五金价格", "索宾柯价格"):
|
||||
return "hardware"
|
||||
return "addon"
|
||||
|
||||
|
||||
async def import_series(db: AsyncSession, company_id: int, parsed: dict) -> int:
|
||||
"""「铝成本」页 → CostSeriesTemplate（同系列同名覆盖）。"""
|
||||
summary = parsed["series_summary"]
|
||||
win = summary["window"]
|
||||
name = f"{win.get('series', 0):g}框扇齐平" if "窗纱" not in parsed["file"] else f"{win.get('series', 0):g}窗纱一体"
|
||||
existing = (await db.execute(select(CostSeriesTemplate).where(
|
||||
CostSeriesTemplate.company_id == company_id, CostSeriesTemplate.series == int(win.get("series", 0)),
|
||||
CostSeriesTemplate.name == name))).scalars().first()
|
||||
payload = dict(name=name, series=int(win.get("series", 0)),
|
||||
wall_thick=float(win.get("wall_thick", 1.4)),
|
||||
source_file=parsed["file"], data=json.dumps(summary, ensure_ascii=False))
|
||||
if existing is None:
|
||||
obj = CostSeriesTemplate(company_id=company_id, **payload)
|
||||
db.add(obj)
|
||||
await db.flush()
|
||||
return obj.id
|
||||
for k, v in payload.items():
|
||||
setattr(existing, k, v)
|
||||
await db.flush()
|
||||
return existing.id
|
||||
|
||||
|
||||
# ---------------- 估算器 ----------------
|
||||
|
||||
def estimate(tpl: CostSeriesTemplate, width_mm: float, height_mm: float,
|
||||
count: int = 1, sash_count: int | None = None) -> dict:
|
||||
"""按系列模板估算任意尺寸窗型的成本明细（口径对齐核算表）。"""
|
||||
data = json.loads(tpl.data)
|
||||
win = data["window"]
|
||||
base_area = win.get("area") or (win.get("w", 1600) * win.get("h", 1600) / 1e6)
|
||||
area = width_mm * height_mm / 1e6 * count
|
||||
sash_count = sash_count if sash_count is not None else max(1, round(count))
|
||||
base_sash_area = win.get("sash_area") or 0.0001
|
||||
base_fixed_area = win.get("fixed_area") or (base_area - base_sash_area) or 0.0001
|
||||
# 新窗型开启面积按基准开启率折算
|
||||
open_rate = base_sash_area / base_area
|
||||
new_sash_area = area * open_rate
|
||||
new_fixed_area = area - new_sash_area
|
||||
ratio = {"fixed": new_fixed_area / base_fixed_area, "open": new_sash_area / base_sash_area,
|
||||
"both": area / base_area, "once": float(count)}
|
||||
|
||||
def scale_len(row: dict) -> float:
|
||||
length = row.get("length") or row.get("total") or 0
|
||||
return length * ratio.get(row.get("alloc", "both"), ratio["both"])
|
||||
|
||||
def cost_of(rows, len_key="length") -> tuple[list[dict], float]:
|
||||
detail = []
|
||||
total = 0.0
|
||||
for row in rows:
|
||||
if row.get("alloc") == "once":
|
||||
qty = count * (row.get("count") or 1)
|
||||
amount = qty * row.get("unit_price", 0)
|
||||
total += amount
|
||||
detail.append({**row, "qty": qty, "amount": round(amount, 2)})
|
||||
else:
|
||||
meters = scale_len(row)
|
||||
unit_price = row.get("unit_price", 0)
|
||||
weight = meters * row.get("weight_per_m", 0)
|
||||
amount = weight * unit_price if row.get("weight_per_m") else meters * unit_price
|
||||
total += amount
|
||||
detail.append({**row, "qty": round(meters, 3), "weight": round(weight, 3),
|
||||
"amount": round(amount, 2)})
|
||||
return detail, total
|
||||
|
||||
out: dict = {"series": tpl.series, "wall_thick": tpl.wall_thick, "name": tpl.name,
|
||||
"width": width_mm, "height": height_mm, "count": count,
|
||||
"sash_count": sash_count, "area": round(area, 4),
|
||||
"open_area": round(new_sash_area, 4), "fixed_area": round(new_fixed_area, 4)}
|
||||
|
||||
sections_total = 0.0
|
||||
detail, cost = cost_of(data.get("profile", []))
|
||||
out["profile"], sections_total = detail, sections_total + cost
|
||||
out["profile_cost"] = round(cost, 2)
|
||||
detail, cost = cost_of(data.get("addon", []))
|
||||
out["addon"], out["addon_cost"] = detail, round(cost, 2)
|
||||
sections_total += cost
|
||||
detail, cost = cost_of(data.get("strip", []))
|
||||
out["strip"], out["strip_cost"] = detail, round(cost, 2)
|
||||
sections_total += cost
|
||||
detail, cost = cost_of(data.get("seal", []))
|
||||
out["seal"], out["seal_cost"] = detail, round(cost, 2)
|
||||
sections_total += cost
|
||||
# 玻璃：模板 area 已是该配置块数总面积，按面积比例缩放
|
||||
glass_detail, glass_cost = [], 0.0
|
||||
for g in data.get("glass", []):
|
||||
g_ratio = ratio["open"] if g.get("alloc") == "open" else ratio["fixed"]
|
||||
amount = g["area"] * g_ratio * g.get("unit_price", 0)
|
||||
glass_cost += amount
|
||||
glass_detail.append({**g, "blocks": round(g["count"] * g_ratio, 2),
|
||||
"area": round(g["area"] * g_ratio, 4), "amount": round(amount, 2)})
|
||||
out["glass"], out["glass_cost"] = glass_detail, round(glass_cost, 2)
|
||||
sections_total += glass_cost
|
||||
detail, cost = cost_of(data.get("extra", []))
|
||||
out["extra"], out["extra_cost"] = detail, round(cost, 2)
|
||||
sections_total += cost
|
||||
hw_detail, hw_cost = [], 0.0
|
||||
for hw in data.get("hardware", []):
|
||||
qty = sash_count * (hw.get("count") or 1)
|
||||
amount = qty * hw.get("unit_price", 0)
|
||||
hw_cost += amount
|
||||
hw_detail.append({**hw, "qty": qty, "amount": round(amount, 2)})
|
||||
out["hardware"], out["hardware_cost"] = hw_detail, round(hw_cost, 2)
|
||||
sections_total += hw_cost
|
||||
|
||||
labor = data.get("labor", {})
|
||||
# 优先用核算表实际每平口径，缺省按参数推导（安装费+对应壁厚人工费）
|
||||
labor_per_sqm = labor.get("rate_labor") or (
|
||||
labor.get("install", 0) + (labor.get("w18", 0) if tpl.wall_thick > 1.4 else labor.get("w14", 0)))
|
||||
other_per_sqm = labor.get("rate_other") or (
|
||||
labor.get("sales", 0) + labor.get("fuel", 0) + labor.get("ins", 0))
|
||||
out["labor_cost"] = round(labor_per_sqm * area, 2)
|
||||
out["other_cost"] = round(other_per_sqm * area, 2)
|
||||
sections_total += out["labor_cost"] + out["other_cost"]
|
||||
|
||||
out["total"] = round(sections_total, 2)
|
||||
out["per_sqm"] = round(sections_total / area, 2) if area else 0
|
||||
out["margin_ladder"] = {f"{p}%": round(out["total"] * (1 + p / 100), 2)
|
||||
for p in (10, 20, 30, 40, 50, 60, 70)}
|
||||
return out
|
||||
Reference in New Issue
Block a user