feat(aps): B1/B5 首批——搬入只读域模块并起一个能独立运行的 APS 服务(九个只读能力)

This commit is contained in:
zh 2026-09-30 14:04:23 +08:00
parent 8e10e86901
commit 6539924d56
16 changed files with 8805 additions and 1 deletions

View File

@ -6,7 +6,9 @@
APS 现在还在 `main` 分支里与平台混装(`server/aps_domain` 39,212 行、`server/engines` 7,551 行、`server/state`、`server/importers`,以及早期原型 `aps-frontend/`)。B0–B7 负责把它拆到这里并补齐独立运行所需的能力。
B0.1 已完成:`src/aps/` 子包骨架、`pyproject.toml`、`uv.lock` 与独立 CI job 就位;平台侧的 lint / typecheck / test 与 APS 的 ruff / pytest 互不阻塞。
B0 已完成:工具链与包骨架、边界清单与契约草案、端点分流清单。
B1 首批 / B5 首批已完成:9 个只读模块从 `main` 搬入并改写引用(`domain` 4,053 行)+ 演示数据快照 + FastAPI 服务,**现在能独立启动**(`uv run python -m aps.api`),九个只读能力按契约返回。剩余:APS 侧另外 134 个端点、APS 自己的库、界面对话之外的独立界面。
## 工具链(B0.1)
@ -27,6 +29,30 @@ uv run pytest -q
依赖取舍:依赖清单以 `main` 分支的 `requirements.txt` 为准,按 B 线边界做了两处增删——**去掉** LLM 与对话智能体相关依赖(D12:APS 不内置对话智能体),**保留**求解引擎(ortools / scipy)、文档与图纸解析(openpyxl / pdfplumber / python-docx / lxml / ezdxf)与持久层(sqlalchemy / alembic / pymysql)——它们都归 APS。
## 怎么跑(B1 首批 / B5 首批)
```bash
cd aps
uv sync
uv run python -m aps.api # http://127.0.0.1:8000,交互式文档在 /docs
```
```bash
curl http://127.0.0.1:8000/api/health
curl http://127.0.0.1:8000/api/manifest
curl -X POST http://127.0.0.1:8000/api/capabilities/aps.orders.query \
-H 'content-type: application/json' -d '{"params":{"limit":3}}'
```
已经能跑的九个只读能力:`aps.master.query` / `aps.orders.query` / `aps.capacity.query` / `aps.calendar.query` / `aps.routing.query` / `aps.schedule.query` / `aps.exceptions.query` / `aps.inventory.query` / `aps.forecast.query`。
## 数据从哪来
| 阶段 | 数据源 |
| --- | --- |
| B1 过渡(现在) | `data/demo_world.json`——由 `main` 的 `build_demo_world()` 导出(10 物料 / 7 销售订单 / 13 工位 / 180 条班次日历)。`APS_WORLD_PATH` 可指向别的快照 |
| B2 拆库后 | APS 自己的库(开发期 SQLite、生产 PostgreSQL)。`domain` 的读函数都吃 `World` 字典,换数据源不动它们 |
## 边界规则
| 规则 | 说明 |

4223
aps/data/demo_world.json Normal file

File diff suppressed because it is too large Load Diff

View File

@ -48,6 +48,12 @@ src = ["src", "tests"]
[tool.ruff.lint]
select = ["E", "F", "W", "I", "UP", "B", "SIM"]
[tool.ruff.lint.per-file-ignores]
# B1 从 main 分支原样搬入的旧代码:先保证行为与旧系统一致,格式与风格清理放到 B3(那时删掉这一段)
"src/aps/domain/*.py" = ["E501", "I001", "B007", "SIM103", "SIM108"]
"src/aps/engines/*.py" = ["E501", "I001"]
"src/aps/timeutil.py" = ["I001"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

View File

@ -0,0 +1,20 @@
"""本地启动:``uv run python -m aps.api``(默认 127.0.0.1:8000,APS_HOST / APS_PORT 可覆盖)。"""
from __future__ import annotations
import os
import uvicorn
def main() -> None:
uvicorn.run(
"aps.api.app:app",
host=os.environ.get("APS_HOST", "127.0.0.1"),
port=int(os.environ.get("APS_PORT", "8000")),
log_level=os.environ.get("APS_LOG_LEVEL", "info"),
)
if __name__ == "__main__":
main()

346
aps/src/aps/api/app.py Normal file
View File

@ -0,0 +1,346 @@
"""APS 服务:对平台的契约化接口层(B5 首批)。
只读优先:七个只读能力对应 `docs/contract-draft.md` 的清单,返回体与平台
`contracts/schemas/receipt.schema.json` 的 `ExecutionReceipt` 同形。
平台怎么用它:
GET /api/manifest → ProviderManifest(平台侧 Provider 适配器读这个)
POST /api/capabilities/{capabilityId} → 执行一次能力调用,返回 ExecutionReceipt
GET /api/health → 健康探测
权限不在这里判:按 D9,权限归业务系统。平台放行只是"允许提交",APS 仍要按传入的
`identity.accountId` 校验自己的权限(本批只读,账号校验在 B2 拆库后接上)。
"""
from __future__ import annotations
import logging
import uuid
from collections.abc import Callable
from typing import Any
from fastapi import FastAPI
from pydantic import BaseModel, Field
from aps import __version__
from aps.domain import changeover, inventory, master_query, masterdata, orders
from aps.world import World, load_world
logger = logging.getLogger("aps.api")
Handler = Callable[[World, dict[str, Any]], dict[str, Any]]
ERROR_KINDS = ("invalid_argument", "forbidden", "conflict", "rate_limited", "unavailable", "unknown_result")
class InvokeRequest(BaseModel):
"""平台侧 Core 传来的调用请求(形状见 docs/contract-draft.md 第 1 节)。"""
params: dict[str, Any] = Field(default_factory=dict)
executionId: str | None = None
idempotencyKey: str | None = None
deadlineAt: int | None = None
identity: dict[str, Any] | None = None
class CapabilitySpec(BaseModel):
"""能力声明,字段与平台 provider-sdk 的 Capability 一致。"""
id: str
title: str
description: str
effect: str = "read"
mode: str = "sync"
whenNotToUse: str | None = None
synonyms: list[str] = Field(default_factory=list)
inputSchema: dict[str, Any] = Field(default_factory=dict)
outputSchema: dict[str, Any] = Field(default_factory=dict)
idempotency: str = "none"
errors: list[str] = Field(default_factory=lambda: ["invalid_argument", "unavailable", "unknown_result"])
def _limit(params: dict[str, Any], default: int, cap: int = 200) -> int:
try:
value = int(params.get("limit", default))
except (TypeError, ValueError):
raise ValueError("limit 必须是整数") from None
return max(1, min(value, cap))
def _contains(item: Any, keyword: str) -> bool:
import json
return keyword.lower() in json.dumps(item, ensure_ascii=False).lower()
def _master_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
return master_query.run_master_query(
world,
entity=params.get("entity"),
code=params.get("code"),
text=str(params.get("text") or ""),
limit=_limit(params, 40),
)
def _orders_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
items = orders.list_orders(world)
code = params.get("code")
if code:
items = [item for item in items if _contains(item, str(code))]
limit = _limit(params, 50)
return {"items": items[:limit], "total": len(items)}
def _capacity_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
workstations = masterdata.active_workstations(world)
limit = _limit(params, 50)
return {"items": workstations[:limit], "total": len(workstations), "pool": orders.pool_summary(world)}
def _calendar_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
entries = list(world.get("shiftCalendar") or [])
limit = _limit(params, 60)
view = changeover.build_changeover_view(world)
return {"items": entries[:limit], "total": len(entries), "changeover": view}
def _routing_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
return master_query.query_routing(world, params.get("code"), limit=_limit(params, 40))
def _schedule_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
versions = list(world.get("scheduleVersions") or []) or list(world.get("flexScheduleVersions") or [])
limit = _limit(params, 20)
return {"items": versions[:limit], "total": len(versions)}
def _exceptions_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
items = list(world.get("conflicts") or []) + list(world.get("flexConflicts") or [])
limit = _limit(params, 50)
return {"items": items[:limit], "total": len(items)}
def _inventory_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
report = inventory.build_inventory_projection(
world,
horizon_days=_limit(params, 30),
bucket=str(params.get("bucket") or "DAY"),
include_forecast=bool(params.get("includeForecast", True)),
)
items = report.get("items", []) if isinstance(report, dict) else []
return {"items": items, "total": len(items), "projection": report}
def _forecast_query(world: World, params: dict[str, Any]) -> dict[str, Any]:
items = list(world.get("forecastOrders") or [])
limit = _limit(params, 50)
return {"items": items[:limit], "total": len(items)}
_FILTER_SCHEMA: dict[str, Any] = {
"type": "object",
"properties": {
"project": {"type": "string", "description": "项目 id,缺省用当前项目"},
"code": {"type": "string", "description": "物料号 / 订单号等业务编码"},
"limit": {"type": "integer", "description": "返回条数上限,默认 20–50"},
},
}
CAPABILITIES: dict[str, tuple[CapabilitySpec, Handler]] = {}
def _register(spec: CapabilitySpec, handler: Handler) -> None:
CAPABILITIES[spec.id] = (spec, handler)
_register(
CapabilitySpec(
id="aps.master.query",
title="主数据查询",
description="查物料、工序、BOM、工艺路线等主数据",
whenNotToUse="要查订单用 aps.orders.query;要查排产结果用 aps.schedule.query",
synonyms=["主数据", "基础资料", "物料表"],
inputSchema={
"type": "object",
"properties": {
**_FILTER_SCHEMA["properties"],
"entity": {
"type": "string",
"enum": ["overview", "material", "operation", "bom", "routing", "flex", "mrp", "order"],
},
"text": {"type": "string", "description": "自然语言问法,用于兜底识别实体"},
},
},
outputSchema={
"type": "object",
"properties": {"entity": {"type": "string"}, "items": {"type": "array"}, "counts": {"type": "object"}},
},
),
_master_query,
)
_register(
CapabilitySpec(
id="aps.orders.query",
title="订单查询",
description="查销售订单、生产订单与交期状态",
whenNotToUse="要看排产结果用 aps.schedule.query;要改订单用写能力",
synonyms=["订单", "交期", "承诺日期", "工单"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={"type": "object", "properties": {"items": {"type": "array"}, "total": {"type": "integer"}}},
),
_orders_query,
)
_register(
CapabilitySpec(
id="aps.capacity.query",
title="产能查询",
description="查工位、设备与产能池负荷",
whenNotToUse="要查某版方案的负荷用 aps.schedule.query",
synonyms=["产能", "负荷", "能力池", "工位"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={
"type": "object",
"properties": {"items": {"type": "array"}, "total": {"type": "integer"}, "pool": {"type": "object"}},
},
),
_capacity_query,
)
_register(
CapabilitySpec(
id="aps.calendar.query",
title="日历查询",
description="查班次日历与换型矩阵",
whenNotToUse="不用于查排产结果",
synonyms=["日历", "班次", "换型", "setup", "换模"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={"type": "object", "properties": {"items": {"type": "array"}, "total": {"type": "integer"}}},
),
_calendar_query,
)
_register(
CapabilitySpec(
id="aps.routing.query",
title="工艺路线查询",
description="查工艺路线模板与工序顺序",
whenNotToUse="图纸解析不在本能力内",
synonyms=["工艺路线", "工序", "路线模板"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={"type": "object", "properties": {"items": {"type": "array"}}},
),
_routing_query,
)
_register(
CapabilitySpec(
id="aps.schedule.query",
title="排产结果查询",
description="查现有排产方案版本与结果",
whenNotToUse="要新算一版试排用 aps.schedule.preview(写能力,尚未实现)",
synonyms=["排产结果", "计划", "方案版本"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={"type": "object", "properties": {"items": {"type": "array"}, "total": {"type": "integer"}}},
),
_schedule_query,
)
_register(
CapabilitySpec(
id="aps.exceptions.query",
title="异常查询",
description="查缺料、冲突与超期等异常",
whenNotToUse="要看某个订单细节用 aps.orders.query",
synonyms=["异常", "冲突", "缺料", "超期"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={"type": "object", "properties": {"items": {"type": "array"}, "total": {"type": "integer"}}},
),
_exceptions_query,
)
_register(
CapabilitySpec(
id="aps.inventory.query",
title="库存查询",
description="查可用库存与缺料缺口",
whenNotToUse="要看排产结果里的投料用 aps.schedule.query",
synonyms=["库存", "可用量", "缺料"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={"type": "object", "properties": {"items": {"type": "array"}}},
),
_inventory_query,
)
_register(
CapabilitySpec(
id="aps.forecast.query",
title="预测查询",
description="查需求预测订单与时间分布",
whenNotToUse="实单在 aps.orders.query",
synonyms=["预测", "预测量", "需求预测"],
inputSchema={"type": "object", "properties": _FILTER_SCHEMA["properties"]},
outputSchema={"type": "object", "properties": {"items": {"type": "array"}, "total": {"type": "integer"}}},
),
_forecast_query,
)
def manifest() -> dict[str, Any]:
"""ProviderManifest:平台侧 Provider 适配器按这个挂能力。"""
return {
"id": "aps",
"displayName": "APS 排产系统",
"version": __version__,
"platformVersion": 1,
# ADR-0012:APS 有账号体系,登录态由用户绑定;首版走"账号密码一次性换 token"
"requiresIdentity": True,
"auth": {"kind": "password-exchange"},
"capabilities": [spec.model_dump() for spec, _ in CAPABILITIES.values()],
}
app = FastAPI(title="APS", version=__version__, description="APS 独立业务系统 · 契约化接口层(B5 首批,只读)")
@app.get("/api/health")
async def health() -> dict[str, Any]:
import time
world = load_world()
return {
"status": "ok",
"detail": f"{len(world)} 个世界字段",
"checkedAt": int(time.time() * 1000),
}
@app.get("/api/manifest")
async def get_manifest() -> dict[str, Any]:
return manifest()
@app.post("/api/capabilities/{capability_id}")
async def invoke(capability_id: str, body: InvokeRequest) -> dict[str, Any]:
execution_id = body.executionId or f"exec-aps-{uuid.uuid4()}"
entry = CAPABILITIES.get(capability_id)
if entry is None:
return {
"executionId": execution_id,
"status": "failed",
"errorKind": "invalid_argument",
"result": {"message": f"APS 没有能力 {capability_id}"},
}
spec, handler = entry
try:
result = handler(load_world(), body.params)
except (ValueError, TypeError, KeyError) as error:
return {
"executionId": execution_id,
"status": "failed",
"errorKind": "invalid_argument",
"result": {"message": str(error) or "参数不合法"},
}
except Exception as error: # noqa: BLE001 - 兜底归一为 unknown_result,与平台六分类对齐
logger.exception("能力 %s 执行失败", capability_id)
return {
"executionId": execution_id,
"status": "failed",
"errorKind": "unknown_result",
"result": {"message": f"{type(error).__name__}: {error}"},
}
return {"executionId": execution_id, "status": "succeeded", "result": result, "capabilityId": spec.id}

View File

@ -0,0 +1,258 @@
# ============================================================
# 换型矩阵主数据(moduleId: domain-changeover, MD-06 / C10,可重生 ✅)
# 产品族 × 产品族 setup 分钟;相邻工单换型耗时进 RuleEngine 占槽
# ============================================================
from __future__ import annotations
from typing import Any
World = dict[str, Any]
# 缺矩阵条目时的跨族默认换型(分钟);同族为 0
DEFAULT_CROSS_FAMILY_MIN = 30.0
def ensure_changeover_table(world: World) -> None:
"""旧 world 缺表时自愈。"""
if "changeoverMatrix" not in world or not isinstance(world.get("changeoverMatrix"), list):
world["changeoverMatrix"] = []
def product_family(mat: dict[str, Any] | None) -> str:
"""取产品族编码:显式 productFamily,否则用物料编码前缀(CTRL-A → CTRL)。"""
if not mat:
return "?"
fam = str(mat.get("productFamily") or "").strip()
if fam:
return fam
code = str(mat.get("code") or "").strip()
if "-" in code:
return code.split("-", 1)[0]
return code or f"P{mat.get('id', '?')}"
def family_of_product(world: World, product_id: int | None) -> str | None:
if product_id is None:
return None
mat = next((m for m in world.get("materials", []) if m["id"] == product_id), None)
return product_family(mat) if mat else None
def lookup_setup_minutes(
world: World,
from_family: str | None,
to_family: str | None,
*,
default_cross: float | None = None,
) -> float:
"""
查换型分钟。同族 / 无前序 → 0;有矩阵条目用条目;否则跨族用 default_cross。
default_cross 缺省读 world.changeoverPolicy(IND-02),再回落模块常量。
"""
if not to_family:
return 0.0
if not from_family or from_family == to_family:
return 0.0
ensure_changeover_table(world)
for row in world.get("changeoverMatrix") or []:
if row.get("fromFamily") == from_family and row.get("toFamily") == to_family:
return float(row.get("setupMinutes") or 0)
# 对称回退:若只录了反向
for row in world.get("changeoverMatrix") or []:
if row.get("fromFamily") == to_family and row.get("toFamily") == from_family:
return float(row.get("setupMinutes") or 0)
if default_cross is None:
pol = world.get("changeoverPolicy") or {}
default_cross = float(pol.get("defaultCrossFamilyMin") or DEFAULT_CROSS_FAMILY_MIN)
return float(default_cross)
def extra_setup_minutes(world: World, from_family: str | None, to_family: str | None) -> float:
"""RuleEngine 用:相对「无换型」的额外准备分钟。"""
return lookup_setup_minutes(world, from_family, to_family)
def list_families(world: World) -> list[dict[str, Any]]:
"""成品/半成品族清单(主数据页下拉)。"""
bags: dict[str, list[dict[str, Any]]] = {}
for m in world.get("materials", []):
if m.get("type") not in ("FINISHED_PRODUCT", "SEMI_FINISHED"):
continue
if m.get("status", "ACTIVE") != "ACTIVE":
continue
fam = product_family(m)
bags.setdefault(fam, []).append({"id": m["id"], "code": m["code"], "name": m["name"]})
return [
{"code": fam, "productCount": len(prods), "products": prods}
for fam, prods in sorted(bags.items(), key=lambda x: x[0])
]
def build_changeover_view(world: World) -> dict[str, Any]:
"""P0 投影:矩阵行 + 族清单 + 缺省跨族分钟。"""
ensure_changeover_table(world)
families = list_families(world)
fam_codes = [f["code"] for f in families]
rows = []
for r in world.get("changeoverMatrix") or []:
rows.append({
"id": r.get("id"),
"fromFamily": r.get("fromFamily"),
"toFamily": r.get("toFamily"),
"setupMinutes": float(r.get("setupMinutes") or 0),
"note": r.get("note") or "",
})
rows.sort(key=lambda x: (x["fromFamily"] or "", x["toFamily"] or ""))
# 完整方阵:有录入用录入,同族 0,缺省跨族 default
grid = []
for a in fam_codes:
for b in fam_codes:
mins = lookup_setup_minutes(world, a, b)
explicit = next(
(r for r in rows if r["fromFamily"] == a and r["toFamily"] == b), None
)
grid.append({
"fromFamily": a, "toFamily": b, "setupMinutes": mins,
"explicit": explicit is not None or a == b,
"id": explicit["id"] if explicit else None,
})
return {
"defaultCrossFamilyMin": float((world.get("changeoverPolicy") or {}).get("defaultCrossFamilyMin")
or DEFAULT_CROSS_FAMILY_MIN),
"families": families,
"rows": rows,
"grid": grid,
"hint": "同族换型 0 分;跨族查矩阵,缺省见 changeoverPolicy(SOP 可编译覆盖)。排产首道工序叠加额外换型(C10)。",
"policy": world.get("changeoverPolicy") or {},
}
def normalize_changeover_payload(world: World, payload: dict[str, Any]) -> dict[str, Any]:
"""upsert / delete 归一化。"""
ensure_changeover_table(world)
if payload.get("delete") or str(payload.get("op") or "").lower() == "delete":
rid = int(payload.get("id") or 0)
row = next((r for r in world["changeoverMatrix"] if r.get("id") == rid), None)
if row is None:
raise ValueError(f"换型矩阵行不存在:{rid}")
return {
"op": "delete", "id": rid,
"fromFamily": row.get("fromFamily"), "toFamily": row.get("toFamily"),
}
frm = str(payload.get("fromFamily") or "").strip()
to = str(payload.get("toFamily") or "").strip()
if not frm or not to:
raise ValueError("必须提供 fromFamily 与 toFamily")
if frm == to:
raise ValueError("同族换型固定为 0,无需录入对角线")
mins = float(payload.get("setupMinutes") if payload.get("setupMinutes") is not None else 30)
if mins < 0:
raise ValueError("换型分钟不能为负")
note = str(payload.get("note") or "").strip()
existing = next(
(r for r in world["changeoverMatrix"]
if r.get("fromFamily") == frm and r.get("toFamily") == to),
None,
)
out: dict[str, Any] = {
"op": "upsert",
"fromFamily": frm, "toFamily": to,
"setupMinutes": mins, "note": note,
}
if payload.get("id"):
out["id"] = int(payload["id"])
elif existing:
out["id"] = existing["id"]
return out
def apply_changeover_action(world: World, next_id, payload: dict[str, Any]) -> dict[str, Any]:
"""写入换型矩阵(P2)。"""
p = normalize_changeover_payload(world, payload)
ensure_changeover_table(world)
if p["op"] == "delete":
world["changeoverMatrix"] = [r for r in world["changeoverMatrix"] if r.get("id") != p["id"]]
return {"kind": "CHANGEOVER", "op": "delete", "id": p["id"],
"fromFamily": p["fromFamily"], "toFamily": p["toFamily"]}
if p.get("id"):
row = next((r for r in world["changeoverMatrix"] if r.get("id") == p["id"]), None)
if row is None:
# id 指定但找不到 → 按族键更新或新建
row = next(
(r for r in world["changeoverMatrix"]
if r.get("fromFamily") == p["fromFamily"] and r.get("toFamily") == p["toFamily"]),
None,
)
if row:
row.update({
"fromFamily": p["fromFamily"], "toFamily": p["toFamily"],
"setupMinutes": p["setupMinutes"], "note": p["note"],
})
return {"kind": "CHANGEOVER", "op": "update", "id": row["id"],
"fromFamily": p["fromFamily"], "toFamily": p["toFamily"],
"setupMinutes": p["setupMinutes"]}
# 新建或按族键覆盖
existing = next(
(r for r in world["changeoverMatrix"]
if r.get("fromFamily") == p["fromFamily"] and r.get("toFamily") == p["toFamily"]),
None,
)
if existing:
existing.update({"setupMinutes": p["setupMinutes"], "note": p["note"]})
return {"kind": "CHANGEOVER", "op": "update", "id": existing["id"],
"fromFamily": p["fromFamily"], "toFamily": p["toFamily"],
"setupMinutes": p["setupMinutes"]}
rid = next_id("changeoverMatrix")
world["changeoverMatrix"].append({
"id": rid, "fromFamily": p["fromFamily"], "toFamily": p["toFamily"],
"setupMinutes": p["setupMinutes"], "note": p["note"],
})
return {"kind": "CHANGEOVER", "op": "create", "id": rid,
"fromFamily": p["fromFamily"], "toFamily": p["toFamily"],
"setupMinutes": p["setupMinutes"]}
def sort_entries_changeover_min(
world: World,
entries: list[dict[str, Any]],
*,
level_weight_fn,
) -> list[dict[str, Any]]:
"""
SC-07:换型最小化排序(贪心最近邻)。
种子取交期最早(预测垫后、插单优先);其后每次选相对上一产品族换型分钟最小的项,
平手再比交期/优先级。非全局最优,毫秒级可解释。
"""
if len(entries) <= 1:
return list(entries)
def _fam(entry: dict[str, Any]) -> str:
pid = entry["item"]["productId"]
mat = next((m for m in world.get("materials", []) if m["id"] == pid), None)
return product_family(mat)
def _seed_key(entry: dict[str, Any]):
so = entry["so"]
forecast = 1 if so.get("isForecast") else 0
rush = 0 if so.get("isRush") else 1
lw = -level_weight_fn(world, so.get("customerLevel"))
return (forecast, rush, so["deliveryDate"], lw, so["priority"], so.get("orderNo") or "")
remaining = sorted(entries, key=_seed_key)
ordered: list[dict[str, Any]] = [remaining.pop(0)]
last_fam = _fam(ordered[0])
while remaining:
best_i = 0
best_score: tuple | None = None
for i, entry in enumerate(remaining):
so = entry["so"]
cost = extra_setup_minutes(world, last_fam, _fam(entry))
score = (cost, so["deliveryDate"], so["priority"], so.get("orderNo") or "")
if best_score is None or score < best_score:
best_score = score
best_i = i
nxt = remaining.pop(best_i)
ordered.append(nxt)
last_fam = _fam(nxt)
return ordered

View File

@ -0,0 +1,260 @@
# ============================================================
# 预测 / 长周期订单(moduleId: domain-forecast, OR-05 首切片,可重生 ✅)
# 规则:预测需求独立于销售订单;默认不进正式排产;可试排纳入 / 转正
# ============================================================
from __future__ import annotations
from datetime import datetime
from typing import Any
from aps.domain.orders import apply_order_action, find_product_by_hint
from aps.timeutil import add_minutes, fmt_date, fmt_dt, parse_dt, today0
World = dict[str, Any]
FORECAST_STATUSES = {"DRAFT", "ACTIVE", "CONSUMED", "CANCELLED"}
FORECAST_BUCKETS = {"DAY", "WEEK", "MONTH"}
SCHEDULABLE_FORECAST = {"ACTIVE"}
def _now() -> str:
return fmt_dt(datetime.now())
def ensure_forecast_table(world: World) -> None:
if "forecastOrders" not in world or not isinstance(world.get("forecastOrders"), list):
world["forecastOrders"] = []
def list_forecasts(world: World) -> list[dict[str, Any]]:
ensure_forecast_table(world)
rows = list(world["forecastOrders"])
rows.sort(key=lambda f: (f.get("status") != "ACTIVE", f.get("dueDate") or "", f.get("id") or 0))
return rows
def forecast_summary(world: World) -> dict[str, int]:
ensure_forecast_table(world)
counts = {"all": 0, "draft": 0, "active": 0, "consumed": 0, "cancelled": 0}
for f in world["forecastOrders"]:
counts["all"] += 1
st = str(f.get("status") or "").lower()
if st in counts:
counts[st] += 1
return counts
def find_forecast(world: World, *, forecast_id: int | None = None,
forecast_no: str | None = None) -> dict[str, Any] | None:
ensure_forecast_table(world)
if forecast_id:
return next((f for f in world["forecastOrders"] if f["id"] == forecast_id), None)
if forecast_no:
no = forecast_no.strip().upper()
return next((f for f in world["forecastOrders"] if f.get("forecastNo") == no), None)
return None
def _product(world: World, product_id: int) -> dict[str, Any]:
product = next((m for m in world["materials"]
if m["id"] == product_id and m["type"] == "FINISHED_PRODUCT"), None)
if product is None:
raise ValueError(f"产品不存在或不是成品:{product_id}")
return product
def normalize_forecast_payload(world: World, payload: dict[str, Any]) -> dict[str, Any]:
out = dict(payload or {})
hint = str(out.get("productCode") or out.get("productName") or "").strip()
if not out.get("productId") and hint:
product = find_product_by_hint(world, hint)
if product is None:
raise ValueError(f"找不到成品:{hint}")
out["productId"] = product["id"]
if not out.get("productId"):
raise ValueError("预测订单必须指定 productId / productCode")
product = _product(world, int(out["productId"]))
qty = int(out.get("quantity") or 0)
if qty <= 0:
raise ValueError("预测数量必须为正整数")
bucket = str(out.get("bucket") or "WEEK").upper()
if bucket not in FORECAST_BUCKETS:
raise ValueError("bucket 必须是 DAY/WEEK/MONTH")
period_start = str(out.get("periodStart") or "").strip()
if not period_start:
period_start = fmt_date(add_minutes(today0(), 7 * 24 * 60))
# 默认展望:日+0 / 周+6 / 月+29
span = 0 if bucket == "DAY" else 6 if bucket == "WEEK" else 29
period_end = str(out.get("periodEnd") or "").strip()
if not period_end:
period_end = fmt_date(add_minutes(parse_dt(period_start + " 00:00"), span * 24 * 60))
due = str(out.get("dueDate") or period_end).strip()
status = str(out.get("status") or "ACTIVE").upper()
if status not in FORECAST_STATUSES:
raise ValueError("状态必须是 DRAFT/ACTIVE/CONSUMED/CANCELLED")
conf = float(out.get("confidence") if out.get("confidence") is not None else 0.7)
conf = max(0.0, min(1.0, conf))
return {
"id": int(out["id"]) if out.get("id") else None,
"productId": product["id"],
"productCode": product["code"],
"productName": product["name"],
"quantity": qty,
"unit": product.get("unit") or "件",
"bucket": bucket,
"periodStart": period_start[:10],
"periodEnd": period_end[:10],
"dueDate": due[:10],
"confidence": round(conf, 2),
"status": status,
"customerName": str(out.get("customerName") or "预测需求").strip() or "预测需求",
"customerLevel": str(out.get("customerLevel") or "C").upper(),
"note": str(out.get("note") or "").strip(),
"source": str(out.get("source") or "MANUAL").upper(),
}
def confirmation_for_forecast_action(world: World, action: str, payload: dict[str, Any]) -> tuple[str, list[str]]:
if action == "forecast.upsert":
p = normalize_forecast_payload(world, payload)
verb = "编辑" if p.get("id") else "新建"
return f"{verb}预测订单", [
f"产品:{p['productName']} × {p['quantity']} {p['unit']}",
f"时段:{p['periodStart']} ~ {p['periodEnd']}({p['bucket']})· 交期 {p['dueDate']}",
f"置信度 {int(p['confidence'] * 100)}% · 状态 {p['status']}",
"预测默认不进正式排产;可「预测纳入试排」或「预测转正」(P2)",
]
if action == "forecast.delete":
f = find_forecast(world, forecast_id=int(payload.get("id") or 0),
forecast_no=payload.get("forecastNo"))
if f is None:
raise ValueError("目标预测订单不存在")
return f"删除预测 {f['forecastNo']}", [
f"产品:{f.get('productName')} × {f.get('quantity')}",
"仅删除预测台账,不影响已转正销售订单与世界排产版本",
]
if action == "forecast.convert":
f = find_forecast(world, forecast_id=int(payload.get("id") or 0),
forecast_no=payload.get("forecastNo"))
if f is None:
raise ValueError("目标预测订单不存在")
if f.get("status") == "CONSUMED":
raise ValueError(f"{f['forecastNo']} 已转正,勿重复操作")
if f.get("status") == "CANCELLED":
raise ValueError(f"{f['forecastNo']} 已取消,不能转正")
return f"预测转正 {f['forecastNo']}", [
f"将生成 APPROVED 销售订单:{f.get('productName')} × {f.get('quantity')}",
f"交期 {f.get('dueDate')};预测状态 → CONSUMED",
"转正后参与正式排产;执行前自动建档(P2)",
]
raise ValueError(f"不支持的预测动作:{action}")
def apply_forecast_action(world: World, next_id, action: str, payload: dict[str, Any]) -> dict[str, Any]:
ensure_forecast_table(world)
if action == "forecast.upsert":
p = normalize_forecast_payload(world, payload)
if p.get("id"):
row = find_forecast(world, forecast_id=p["id"])
if row is None:
raise ValueError("目标预测订单不存在")
before = row.get("status")
row.update({
"productId": p["productId"], "productCode": p["productCode"],
"productName": p["productName"], "quantity": p["quantity"], "unit": p["unit"],
"bucket": p["bucket"], "periodStart": p["periodStart"], "periodEnd": p["periodEnd"],
"dueDate": p["dueDate"], "confidence": p["confidence"], "status": p["status"],
"customerName": p["customerName"], "customerLevel": p["customerLevel"],
"note": p["note"], "updatedAt": _now(),
})
return {"forecast": row, "created": False, "beforeStatus": before}
fid = next_id("forecastOrder")
row = {
"id": fid,
"forecastNo": "FC" + today0().strftime("%Y%m%d") + f"{fid:03d}",
"productId": p["productId"], "productCode": p["productCode"],
"productName": p["productName"], "quantity": p["quantity"], "unit": p["unit"],
"bucket": p["bucket"], "periodStart": p["periodStart"], "periodEnd": p["periodEnd"],
"dueDate": p["dueDate"], "confidence": p["confidence"], "status": p["status"],
"customerName": p["customerName"], "customerLevel": p["customerLevel"],
"note": p["note"], "source": p["source"],
"createdAt": _now(), "updatedAt": _now(), "convertedOrderNo": None,
}
world["forecastOrders"].append(row)
return {"forecast": row, "created": True, "beforeStatus": None}
if action == "forecast.delete":
row = find_forecast(world, forecast_id=int(payload.get("id") or 0),
forecast_no=payload.get("forecastNo"))
if row is None:
raise ValueError("目标预测订单不存在")
before = row.get("status")
world["forecastOrders"] = [f for f in world["forecastOrders"] if f["id"] != row["id"]]
return {"forecast": {"id": row["id"], "forecastNo": row["forecastNo"], "status": "DELETED"},
"created": False, "beforeStatus": before}
if action == "forecast.convert":
row = find_forecast(world, forecast_id=int(payload.get("id") or 0),
forecast_no=payload.get("forecastNo"))
if row is None:
raise ValueError("目标预测订单不存在")
if row.get("status") in ("CONSUMED", "CANCELLED"):
raise ValueError(f"{row['forecastNo']} 状态为 {row['status']},不能转正")
applied = apply_order_action(world, next_id, "order.upsert", {
"customerName": row.get("customerName") or "预测转正客户",
"customerLevel": row.get("customerLevel") or "B",
"deliveryDate": row.get("dueDate"),
"priority": 5,
"status": "APPROVED",
"productId": row["productId"],
"quantity": row["quantity"],
"isRush": False,
"specialRequirements": f"由预测 {row['forecastNo']} 转正",
})
order = applied["order"]
before = row.get("status")
row["status"] = "CONSUMED"
row["convertedOrderNo"] = order["orderNo"]
row["updatedAt"] = _now()
return {
"forecast": row, "order": order, "created": False, "beforeStatus": before,
"convertedOrderNo": order["orderNo"],
}
raise ValueError(f"不支持的预测动作:{action}")
def forecasts_as_schedule_entries(world: World) -> list[dict[str, Any]]:
"""把 ACTIVE 预测投影为引擎可消费的伪销售订单项(不改主干表)。"""
ensure_forecast_table(world)
entries: list[dict[str, Any]] = []
for fc in world["forecastOrders"]:
if fc.get("status") not in SCHEDULABLE_FORECAST:
continue
so = {
"id": 900000 + int(fc["id"]),
"orderNo": fc["forecastNo"],
"customerName": fc.get("customerName") or "预测需求",
"customerLevel": fc.get("customerLevel") or "C",
"orderDate": fc.get("periodStart"),
"deliveryDate": fc.get("dueDate"),
"priority": 9,
"status": "APPROVED",
"isRush": False,
"rushStrategy": None,
"isForecast": True,
"forecastId": fc["id"],
"items": [{
"id": 900000 + int(fc["id"]),
"orderId": 900000 + int(fc["id"]),
"lineNo": 1,
"productId": fc["productId"],
"productName": fc.get("productName"),
"productCode": fc.get("productCode"),
"quantity": fc["quantity"],
"unit": fc.get("unit") or "件",
"status": "PENDING",
}],
}
entries.append({"so": so, "item": so["items"][0]})
return entries

View File

@ -0,0 +1,302 @@
# ============================================================
# 库存投影(moduleId: domain-inventory, PL-04 首切片,可重生 ✅)
# 规则:期初库存+在途 → 按日/周扣减需求、加上计划到货 → 投影可用量;
# 低于安全库存 / 断料告警。只读,不写主干。
# ============================================================
from __future__ import annotations
from typing import Any
from aps.domain.orders import SCHEDULABLE_STATUSES
from aps.engines.queries import find_bom_items
from aps.timeutil import add_minutes, fmt_date, parse_dt, today0
World = dict[str, Any]
BUCKETS = {"DAY", "WEEK"}
PO_RECEIPT_STATUSES = {"DRAFT", "RELEASED", "CONFIRMED", "ORDERED", "OPEN"}
def _as_date(s: str):
return parse_dt(s[:10] + " 00:00")
def _add_days(d, n: int):
return add_minutes(d, n * 24 * 60)
def _day_list(start, days: int) -> list[str]:
return [fmt_date(_add_days(start, i)) for i in range(max(1, days))]
def _week_keys(start, days: int) -> list[tuple[str, str, str]]:
"""返回 (key, start, end) 按周桶,覆盖 horizon。"""
from datetime import timedelta
end = _add_days(start, days - 1)
out: list[tuple[str, str, str]] = []
cur = start - timedelta(days=start.weekday())
while cur <= end:
w_end = cur + timedelta(days=6)
b_s = max(cur, start)
b_e = min(w_end, end)
if b_s <= b_e:
iso = b_s.isocalendar()
out.append((f"W{iso.week:02d}", fmt_date(b_s), fmt_date(b_e)))
cur = cur + timedelta(days=7)
return out
def _collect_fg_demands(world: World, *, include_forecast: bool) -> list[dict[str, Any]]:
"""成品需求事件:交期当日消耗成品库存(MTS/发货口径)。"""
events: list[dict[str, Any]] = []
for so in world.get("salesOrders", []):
if so.get("status") not in SCHEDULABLE_STATUSES:
continue
due = str(so.get("deliveryDate") or "")[:10]
if not due:
continue
for item in so.get("items") or []:
if item.get("status") in ("CANCELLED", "COMPLETED"):
continue
qty = float(item.get("quantity") or 0)
if qty <= 0:
continue
events.append({
"date": due, "productId": int(item["productId"]),
"qty": qty, "ref": so.get("orderNo"), "source": "FIRM",
})
if include_forecast:
from aps.domain.forecast import ensure_forecast_table
ensure_forecast_table(world)
for fc in world.get("forecastOrders", []):
if fc.get("status") != "ACTIVE":
continue
due = str(fc.get("dueDate") or fc.get("periodEnd") or "")[:10]
if not due:
continue
qty = float(fc.get("quantity") or 0)
conf = float(fc.get("confidence") if fc.get("confidence") is not None else 1.0)
conf = max(0.0, min(1.0, conf))
wqty = qty * conf
if wqty <= 0:
continue
events.append({
"date": due, "productId": int(fc["productId"]),
"qty": wqty, "ref": fc.get("forecastNo"), "source": "FORECAST",
})
return events
def _explode_component_demand(world: World, fg_events: list[dict[str, Any]]) -> dict[int, dict[str, float]]:
"""成品需求 → 组件毛需求(按日聚合)。"""
by_mat: dict[int, dict[str, float]] = {}
for ev in fg_events:
for bi in find_bom_items(world, ev["productId"]):
mid = int(bi["materialId"])
need = float(bi.get("quantity") or 0) * float(ev["qty"])
if need <= 0:
continue
slot = by_mat.setdefault(mid, {})
slot[ev["date"]] = round(slot.get(ev["date"], 0.0) + need, 3)
return by_mat
def _fg_demand_by_product(fg_events: list[dict[str, Any]]) -> dict[int, dict[str, float]]:
by_p: dict[int, dict[str, float]] = {}
for ev in fg_events:
slot = by_p.setdefault(int(ev["productId"]), {})
slot[ev["date"]] = round(slot.get(ev["date"], 0.0) + float(ev["qty"]), 3)
return by_p
def _po_receipts(world: World) -> dict[int, dict[str, float]]:
"""采购建议到货:到货日 = suggestedOrderDate + leadTimeDays(缺省用 requiredDate)。"""
by_mat: dict[int, dict[str, float]] = {}
for po in world.get("purchaseOrders", []):
if str(po.get("status") or "").upper() not in PO_RECEIPT_STATUSES:
continue
mid = po.get("materialId")
if mid is None:
continue
qty = float(po.get("quantity") or 0)
if qty <= 0:
continue
lead = int(po.get("leadTimeDays") or 0)
order_d = str(po.get("suggestedOrderDate") or po.get("requiredDate") or "")[:10]
if not order_d:
continue
if po.get("suggestedOrderDate"):
arrive = fmt_date(_add_days(_as_date(order_d), lead))
else:
arrive = str(po.get("requiredDate") or order_d)[:10]
slot = by_mat.setdefault(int(mid), {})
slot[arrive] = round(slot.get(arrive, 0.0) + qty, 3)
return by_mat
def _project_series(
*,
opening: float,
safety: float,
days: list[str],
demand_by_day: dict[str, float],
receipt_by_day: dict[str, float],
) -> list[dict[str, Any]]:
bal = float(opening)
series: list[dict[str, Any]] = []
for d in days:
rec = float(receipt_by_day.get(d, 0.0))
dem = float(demand_by_day.get(d, 0.0))
bal = round(bal + rec - dem, 3)
if bal < 0:
status = "STOCKOUT"
elif bal < safety:
status = "BELOW_SAFETY"
else:
status = "OK"
series.append({
"date": d, "receipt": round(rec, 3), "demand": round(dem, 3),
"projected": bal, "safetyStock": safety, "status": status,
})
return series
def _aggregate_week(series: list[dict[str, Any]], weeks: list[tuple[str, str, str]]) -> list[dict[str, Any]]:
out = []
for key, ws, we in weeks:
cells = [c for c in series if ws <= c["date"] <= we]
if not cells:
continue
receipt = round(sum(c["receipt"] for c in cells), 3)
demand = round(sum(c["demand"] for c in cells), 3)
projected = cells[-1]["projected"]
safety = cells[-1]["safetyStock"]
if projected < 0:
status = "STOCKOUT"
elif projected < safety:
status = "BELOW_SAFETY"
else:
status = "OK"
out.append({
"date": key, "periodStart": ws, "periodEnd": we,
"receipt": receipt, "demand": demand,
"projected": projected, "safetyStock": safety, "status": status,
})
return out
def build_inventory_projection(
world: World,
*,
material_code: str | None = None,
material_type: str | None = None,
horizon_days: int = 30,
bucket: str = "DAY",
include_forecast: bool = True,
start_date: str | None = None,
) -> dict[str, Any]:
"""
PL-04:库存时间投影。
- 成品:交期扣减成品库存
- 原料:BOM 展开毛需求按交期扣减;采购建议按预计到货增加
- 期初 = stock;在途计入起点日到货
"""
bucket = (bucket or "DAY").upper()
if bucket not in BUCKETS:
raise ValueError("bucket 必须是 DAY/WEEK")
horizon_days = max(1, min(int(horizon_days or 30), 180))
start = _as_date(start_date) if start_date else today0()
start_s = fmt_date(start)
days = _day_list(start, horizon_days)
mats = list(world.get("materials") or [])
if material_code:
code = material_code.strip().upper()
mats = [m for m in mats if str(m.get("code") or "").upper() == code]
if not mats:
raise ValueError(f"找不到物料:{material_code}")
if material_type:
mt = material_type.strip().upper()
mats = [m for m in mats if str(m.get("type") or "").upper() == mt]
fg_events = _collect_fg_demands(world, include_forecast=include_forecast)
fg_demand = _fg_demand_by_product(fg_events)
comp_demand = _explode_component_demand(world, fg_events)
receipts = _po_receipts(world)
# 在途:计入起点日到货
for m in mats:
mid = m["id"]
transit = float(m.get("inTransit") or 0)
if transit > 0:
slot = receipts.setdefault(mid, {})
slot[start_s] = round(slot.get(start_s, 0.0) + transit, 3)
weeks = _week_keys(start, horizon_days) if bucket == "WEEK" else []
materials_out: list[dict[str, Any]] = []
alert_stockout = 0
alert_safety = 0
for m in sorted(mats, key=lambda x: (x.get("type") != "FINISHED_PRODUCT", x.get("code") or "")):
mid = int(m["id"])
mtype = str(m.get("type") or "")
opening = float(m.get("stock") or 0)
safety = float(m.get("safetyStock") or 0)
if mtype == "FINISHED_PRODUCT":
dem = fg_demand.get(mid, {})
else:
dem = comp_demand.get(mid, {})
rec = receipts.get(mid, {})
series = _project_series(
opening=opening, safety=safety, days=days,
demand_by_day=dem, receipt_by_day=rec,
)
view = _aggregate_week(series, weeks) if bucket == "WEEK" else series
has_stockout = any(c["status"] == "STOCKOUT" for c in view)
has_safety = any(c["status"] == "BELOW_SAFETY" for c in view)
if has_stockout:
alert_stockout += 1
alert = "STOCKOUT"
elif has_safety:
alert_safety += 1
alert = "BELOW_SAFETY"
else:
alert = "OK"
first_bad = next((c["date"] for c in view if c["status"] != "OK"), None)
end_proj = view[-1]["projected"] if view else opening
materials_out.append({
"materialId": mid,
"code": m.get("code"),
"name": m.get("name"),
"type": mtype,
"unit": m.get("unit") or "件",
"openingStock": opening,
"inTransit": float(m.get("inTransit") or 0),
"safetyStock": safety,
"endingProjected": end_proj,
"alert": alert,
"firstAlertDate": first_bad,
"totalDemand": round(sum(c["demand"] for c in view), 3),
"totalReceipt": round(sum(c["receipt"] for c in view), 3),
"series": view,
})
return {
"startDate": start_s,
"horizonDays": horizon_days,
"bucket": bucket,
"includeForecast": include_forecast,
"materialFilter": material_code,
"typeFilter": material_type,
"summary": {
"materialCount": len(materials_out),
"stockoutCount": alert_stockout,
"belowSafetyCount": alert_safety,
"okCount": len(materials_out) - alert_stockout - alert_safety,
},
"materials": materials_out,
"hint": (
"库存投影:期初=现有库存,在途计入首日到货;"
"成品按交期扣减,原料按 BOM 毛需求扣减;采购建议按预计到货增加。"
),
}

View File

@ -0,0 +1,500 @@
# ============================================================
# 主数据/订单自然语言只读查询(moduleId: domain-master-query, 可重生 ✅)
# 让对话能检索 world 中的销售订单、物料、工序、BOM、工艺路线、柔性资源。
# ============================================================
from __future__ import annotations
import re
from typing import Any
from aps.engines.queries import find_bom_items, find_routing_steps
World = dict[str, Any]
ENTITIES = ("order", "material", "operation", "bom", "routing", "overview", "flex", "mrp")
def detect_entity(text: str) -> str:
t = text or ""
if re.search(r"概览|主数据总览|资源树|工厂有什么", t):
return "overview"
if re.search(r"委外|采购建议|分解建议|外协|MRP建议", t, re.I):
return "mrp"
if re.search(r"柔性|能力池|虚拟产线|flex", t, re.I):
return "flex"
if re.search(r"BOM|物料清单|用料|子件", t, re.I):
return "bom"
if re.search(r"工艺路线|路线步骤|工序流|怎么排工序", t):
return "routing"
if re.search(r"工序库|有哪些工序|查工序|工序编码", t):
return "operation"
if re.search(r"物料|成品|原料|库存|料号", t):
return "material"
if re.search(r"订单|交期|客户|WBS", t):
return "order"
return "overview"
def extract_code(text: str) -> str | None:
"""抽订单号 / 物料码 / 工序码。"""
t = text or ""
m = re.search(r"(SO\d{6,}|\d{6,}|FO-?\d{3,}|(?:CTRL|PCB|HV|PDU)-[A-Z0-9\-]+|"
r"[0-9A-Za-z]{6,}Z1M|[LM]\d{6,}|\d{10,})", t, re.I)
if m:
return m.group(1)
m2 = re.search(r"(?:订单|物料|料号|工序|产品)\s*[::]?\s*([0-9A-Za-z\-]{4,})", t)
return m2.group(1) if m2 else None
def _find_product(world: World, code_or_name: str | None) -> dict | None:
mats = world.get("materials") or []
if not code_or_name:
return next((m for m in mats if m.get("type") == "FINISHED_PRODUCT"), None)
key = code_or_name.strip().lower()
for m in mats:
if str(m.get("code", "")).lower() == key:
return m
if key in str(m.get("name", "")).lower():
return m
# flex 成品
for m in world.get("flexMaterials") or []:
if m.get("type") == "FINISHED_PRODUCT" and (
str(m.get("code", "")).lower() == key or key in str(m.get("name", "")).lower()
):
return {"id": None, "code": m["code"], "name": m["name"], "type": "FINISHED_PRODUCT",
"_flex": True}
return None
def query_orders(world: World, code: str | None = None, limit: int = 20) -> dict[str, Any]:
rows = []
for so in world.get("salesOrders") or []:
if code and code not in str(so.get("orderNo", "")) and code not in str(so.get("customerName", "")):
continue
item = (so.get("items") or [{}])[0]
rows.append({
"orderNo": so.get("orderNo"),
"status": so.get("status"),
"customer": so.get("customerName"),
"level": so.get("customerLevel"),
"productCode": item.get("productCode"),
"productName": item.get("productName"),
"quantity": item.get("quantity"),
"deliveryDate": so.get("deliveryDate") or item.get("deliveryDate"),
"wbs": so.get("wbs") or "",
"special": so.get("specialRequirements") or "",
})
# 柔性订单补充(若固定轨无命中)
if not rows or (code and not any(r["orderNo"] == code for r in rows)):
for fo in world.get("flexOrders") or []:
if code and code not in str(fo.get("orderNo", "")):
continue
rows.append({
"orderNo": fo.get("orderNo"),
"status": fo.get("status"),
"customer": fo.get("project") or "柔性订单",
"level": "",
"productCode": fo.get("productCode"),
"productName": fo.get("productCode"),
"quantity": fo.get("quantity"),
"deliveryDate": fo.get("dueDate"),
"wbs": fo.get("wbs") or "",
"special": "flex",
})
return {"entity": "order", "count": len(rows), "rows": rows[:limit]}
def query_materials(world: World, code: str | None = None, limit: int = 40) -> dict[str, Any]:
rows = []
for m in world.get("materials") or []:
if code:
c = code.lower()
if c not in str(m.get("code", "")).lower() and c not in str(m.get("name", "")).lower():
continue
rows.append({
"code": m.get("code"), "name": m.get("name"), "type": m.get("type"),
"unit": m.get("unit"), "stock": m.get("stock"), "inTransit": m.get("inTransit"),
"spec": m.get("spec") or "",
})
if not rows:
for m in world.get("flexMaterials") or []:
if code:
c = code.lower()
if c not in str(m.get("code", "")).lower() and c not in str(m.get("name", "")).lower():
continue
rows.append({
"code": m.get("code"), "name": m.get("name"), "type": m.get("type"),
"unit": m.get("unit"), "stock": m.get("stock"), "inTransit": m.get("inTransit"),
"spec": "",
})
return {"entity": "material", "count": len(rows), "rows": rows[:limit]}
def query_operations(world: World, code: str | None = None, limit: int = 40) -> dict[str, Any]:
rows = []
ops = world.get("operations") or []
if not ops:
ops = [
{"code": o.get("code"), "name": o.get("name"),
"type": "INTERNAL", "standardTime": o.get("changeoverMin")}
for o in (world.get("flexOperations") or [])
]
for o in ops:
if code:
c = code.lower()
if c not in str(o.get("code", "")).lower() and c not in str(o.get("name", "")).lower():
continue
rows.append({
"code": o.get("code"), "name": o.get("name"),
"type": o.get("type"), "standardTime": o.get("standardTime"),
})
return {"entity": "operation", "count": len(rows), "rows": rows[:limit]}
def query_bom(world: World, code: str | None = None, limit: int = 50) -> dict[str, Any]:
product = _find_product(world, code)
if not product:
return {"entity": "bom", "count": 0, "rows": [], "hint": "未找到成品,请带料号如 28200003654300"}
if product.get("_flex") or product.get("id") is None:
pc = product["code"]
rows = []
mats = {m["code"]: m for m in (world.get("flexMaterials") or [])}
for b in world.get("flexBom") or []:
if b.get("productCode") != pc:
continue
mat = mats.get(b.get("materialCode") or "", {})
rows.append({
"materialCode": b.get("materialCode"),
"materialName": mat.get("name") or b.get("materialCode"),
"quantity": b.get("quantity"),
"consumeOp": b.get("consumeOp"),
"isKey": b.get("isKey"),
})
return {"entity": "bom", "productCode": pc, "productName": product.get("name"),
"count": len(rows), "rows": rows[:limit]}
pid = product["id"]
mats = {m["id"]: m for m in world.get("materials") or []}
ops = {o["id"]: o for o in world.get("operations") or []}
rows = []
for it in find_bom_items(world, pid):
mat = mats.get(it["materialId"], {})
op = ops.get(it.get("operationId"), {})
rows.append({
"materialCode": mat.get("code"),
"materialName": mat.get("name"),
"quantity": it.get("quantity"),
"consumeOp": op.get("code") or op.get("name"),
"isKey": it.get("isKeyMaterial"),
})
return {"entity": "bom", "productCode": product.get("code"), "productName": product.get("name"),
"count": len(rows), "rows": rows[:limit]}
def query_routing(world: World, code: str | None = None, limit: int = 40) -> dict[str, Any]:
product = _find_product(world, code)
if not product:
return {"entity": "routing", "count": 0, "rows": [], "hint": "未找到成品"}
if product.get("_flex") or product.get("id") is None:
pc = product["code"]
ops = {o["code"]: o for o in (world.get("flexOperations") or [])}
rows = []
for r in sorted((world.get("flexRoutings") or []), key=lambda x: x.get("seq", 0)):
if r.get("productCode") != pc:
continue
op = ops.get(r.get("operationCode") or "", {})
rows.append({
"seq": r.get("seq"),
"operationCode": r.get("operationCode"),
"operationName": op.get("name") or r.get("operationCode"),
"stdTimePerUnit": r.get("stdTimePerUnit"),
"requireMold": r.get("requireMold"),
})
return {"entity": "routing", "productCode": pc, "productName": product.get("name"),
"count": len(rows), "rows": rows[:limit]}
pid = product["id"]
ops = {o["id"]: o for o in world.get("operations") or []}
rows = []
for s in find_routing_steps(world, pid):
op = ops.get(s["operationId"], {})
rows.append({
"seq": s.get("sequenceNo"),
"operationCode": op.get("code"),
"operationName": op.get("name"),
"setupTime": s.get("setupTime"),
"runTimePerUnit": s.get("runTimePerUnit"),
"isExternal": s.get("isExternal"),
})
return {"entity": "routing", "productCode": product.get("code"), "productName": product.get("name"),
"count": len(rows), "rows": rows[:limit]}
def query_overview(world: World) -> dict[str, Any]:
factories = [f.get("name") for f in (world.get("factories") or [])]
return {
"entity": "overview",
"factory": factories[0] if factories else "(无)",
"counts": {
"factories": len(world.get("factories") or []),
"lines": len(world.get("lines") or []),
"workstations": len(world.get("workstations") or []),
"materials": len(world.get("materials") or []),
"operations": len(world.get("operations") or []),
"boms": len(world.get("boms") or []),
"routings": len(world.get("routings") or []),
"salesOrders": len(world.get("salesOrders") or []),
"flexOrders": len(world.get("flexOrders") or []),
"flexEquipment": len(world.get("flexEquipment") or []),
},
}
def query_flex(world: World, limit: int = 20) -> dict[str, Any]:
return {
"entity": "flex",
"orders": [
{"orderNo": o.get("orderNo"), "productCode": o.get("productCode"),
"qty": o.get("quantity"), "due": o.get("dueDate")}
for o in (world.get("flexOrders") or [])[:limit]
],
"equipment": [
{"code": e.get("code"), "name": e.get("name"), "zone": e.get("zone"),
"caps": len(e.get("capabilities") or [])}
for e in (world.get("flexEquipment") or [])[:limit]
],
"operations": [
{"code": o.get("code"), "name": o.get("name"), "bottleneck": o.get("isBottleneck")}
for o in (world.get("flexOperations") or [])[:limit]
],
}
def _mrp_aspect(text: str) -> str:
t = text or ""
if re.search(r"委外|外协", t) and not re.search(r"采购", t):
return "outsource"
if re.search(r"采购", t) and not re.search(r"委外|外协", t):
return "purchase"
return "all"
def query_mrp(world: World, order_no: str | None = None, *, aspect: str = "all",
text: str = "") -> dict[str, Any]:
"""查某单(或全部)的采购/委外建议,并核对工艺是否有外协步骤。"""
aspect = aspect if aspect in ("outsource", "purchase", "all") else _mrp_aspect(text)
order_no = (order_no or "").strip() or None
def _match_so(row: dict) -> bool:
if not order_no:
return True
return str(row.get("salesOrderNo") or "") == order_no
purchase = [r for r in (world.get("purchaseOrders") or []) if _match_so(r)]
outsource = [r for r in (world.get("outsourceOrders") or []) if _match_so(r)]
# 工艺外协核对(即使尚未分解也能答「有没有委外」)
external_steps: list[dict] = []
product_code = None
if order_no:
so = next((s for s in (world.get("salesOrders") or [])
if str(s.get("orderNo")) == order_no), None)
fo = next((o for o in (world.get("flexOrders") or [])
if str(o.get("orderNo")) == order_no), None)
product_id = None
if so:
item = (so.get("items") or [{}])[0]
product_code = item.get("productCode")
product_id = item.get("productId")
if product_id is None and product_code:
mat = next((m for m in (world.get("materials") or [])
if m.get("code") == product_code), None)
product_id = mat["id"] if mat else None
elif fo:
product_code = fo.get("productCode")
mat = next((m for m in (world.get("materials") or [])
if m.get("code") == product_code), None)
product_id = mat["id"] if mat else None
for s in (world.get("flexRoutings") or []):
if s.get("productCode") == product_code and s.get("isExternal"):
external_steps.append({
"seq": s.get("seq"), "operationCode": s.get("operationCode"),
"operationName": s.get("operationName") or s.get("operationCode"),
})
if product_id is not None and not external_steps:
for step in find_routing_steps(world, product_id):
if step.get("isExternal"):
op = next((o for o in (world.get("operations") or [])
if o.get("id") == step.get("operationId")), {})
external_steps.append({
"seq": step.get("sequenceNo"),
"operationCode": op.get("code"),
"operationName": op.get("name") or op.get("code"),
})
return {
"entity": "mrp",
"aspect": aspect,
"orderNo": order_no,
"productCode": product_code,
"purchase": [
{"orderNo": r.get("orderNo"), "materialCode": r.get("materialCode"),
"materialName": r.get("materialName"), "quantity": r.get("quantity"),
"status": r.get("status"), "suggestedOrderDate": r.get("suggestedOrderDate")}
for r in purchase
],
"outsource": [
{"orderNo": r.get("orderNo"), "operationName": r.get("operationName"),
"quantity": r.get("quantity"), "status": r.get("status"),
"requiredDate": r.get("requiredDate")}
for r in outsource
],
"externalSteps": external_steps,
"decomposed": bool(purchase or outsource),
}
def run_master_query(world: World, *, entity: str | None = None, code: str | None = None,
text: str = "", limit: int = 40, aspect: str | None = None) -> dict[str, Any]:
ent = entity or detect_entity(text)
if ent not in ENTITIES:
ent = "overview"
code = code or extract_code(text)
if ent == "order":
return query_orders(world, code, limit=limit)
if ent == "material":
return query_materials(world, code, limit=limit)
if ent == "operation":
return query_operations(world, code, limit=limit)
if ent == "bom":
return query_bom(world, code, limit=limit)
if ent == "routing":
return query_routing(world, code, limit=limit)
if ent == "flex":
return query_flex(world, limit=limit)
if ent == "mrp":
return query_mrp(world, code, aspect=aspect or _mrp_aspect(text), text=text)
return query_overview(world)
def format_master_query(result: dict[str, Any]) -> str:
"""把查询结果格式化为对话可读文本。"""
ent = result.get("entity")
if ent == "overview":
c = result.get("counts") or {}
return (
f"主数据概览 · 工厂【{result.get('factory')}】\n"
f"· 产线 {c.get('lines', 0)} / 工位 {c.get('workstations', 0)}\n"
f"· 物料 {c.get('materials', 0)} / 工序 {c.get('operations', 0)} / "
f"BOM {c.get('boms', 0)} / 路线 {c.get('routings', 0)}\n"
f"· 销售订单 {c.get('salesOrders', 0)} / 柔性订单 {c.get('flexOrders', 0)} / "
f"柔性设备 {c.get('flexEquipment', 0)}\n"
"可继续说:「查订单」「查物料」「查 BOM 28200003654300」「查工艺路线」。"
)
if ent == "order":
rows = result.get("rows") or []
if not rows:
return "未找到匹配订单。可说「有哪些订单」或带订单号如「查订单 102285668」。"
lines = [f"订单共 {result.get('count')} 条:"]
for r in rows:
lines.append(
f"· {r['orderNo']} [{r.get('status')}] {r.get('productName') or r.get('productCode')}"
f" ×{r.get('quantity')} 交期 {r.get('deliveryDate')} "
f"客户 {r.get('customer')}"
+ (f" WBS={r['wbs']}" if r.get("wbs") else "")
)
return "\n".join(lines)
if ent == "material":
rows = result.get("rows") or []
if not rows:
return "未找到匹配物料。"
lines = [f"物料共 {result.get('count')} 条(展示前 {len(rows)}):"]
for r in rows:
lines.append(
f"· {r['code']} {r.get('name')} [{r.get('type')}] "
f"库存 {r.get('stock')}{r.get('unit') or ''}"
)
return "\n".join(lines)
if ent == "operation":
rows = result.get("rows") or []
if not rows:
return "工序库为空。"
lines = [f"工序共 {result.get('count')} 条:"]
for r in rows:
st = r.get("standardTime")
lines.append(f"· {r['code']} {r.get('name')} 标准工时={st}")
return "\n".join(lines)
if ent == "bom":
if result.get("hint") and not result.get("rows"):
return result["hint"]
rows = result.get("rows") or []
head = f"产品 {result.get('productCode')} {result.get('productName') or ''} BOM · {result.get('count')} 项:"
lines = [head]
for r in rows:
key = "★" if r.get("isKey") else " "
lines.append(
f"·{key} {r.get('materialCode')} {r.get('materialName')} ×{r.get('quantity')}"
+ (f" @{r.get('consumeOp')}" if r.get("consumeOp") else "")
)
return "\n".join(lines) if rows else head + "(空)"
if ent == "routing":
if result.get("hint") and not result.get("rows"):
return result["hint"]
rows = result.get("rows") or []
head = f"产品 {result.get('productCode')} 工艺路线 · {result.get('count')} 步:"
lines = [head]
for r in rows:
t = r.get("runTimePerUnit") or r.get("stdTimePerUnit")
lines.append(
f"· seq={r.get('seq')} {r.get('operationCode')} {r.get('operationName')} "
f"单件 {t} 分"
)
return "\n".join(lines) if rows else head + "(空)"
if ent == "flex":
lines = ["柔性资源:"]
for o in result.get("orders") or []:
lines.append(f"· 订单 {o['orderNo']} {o['productCode']} ×{o['qty']} 交期 {o['due']}")
for e in result.get("equipment") or []:
lines.append(f"· 设备 {e['code']} {e['name']} 区={e.get('zone')} 能力数={e.get('caps')}")
for o in result.get("operations") or []:
flag = "★瓶颈" if o.get("bottleneck") else ""
lines.append(f"· 工序 {o['code']} {o['name']} {flag}")
return "\n".join(lines) if len(lines) > 1 else "柔性资源为空。"
if ent == "mrp":
order_no = result.get("orderNo") or "全部订单"
aspect = result.get("aspect") or "all"
purchase = result.get("purchase") or []
outsource = result.get("outsource") or []
external = result.get("externalSteps") or []
lines = [f"订单 {order_no} 的采购/委外情况:"]
if aspect in ("outsource", "all"):
if outsource:
lines.append(f"· 委外建议:{len(outsource)} 条")
for r in outsource[:8]:
lines.append(
f" - {r.get('orderNo')} {r.get('operationName')} ×{r.get('quantity')} "
f"[{r.get('status')}] 需求日 {r.get('requiredDate')}"
)
elif external:
lines.append(
f"· 委外建议:尚未分解,但工艺有 {len(external)} 道外协步骤"
f"({ '、'.join(str(s.get('operationName') or s.get('operationCode')) for s in external[:5]) })。"
"可说「分解订单」生成委外建议单。"
)
else:
lines.append("· 委外建议:无。工艺路线未标记外协步骤,当前没有委外订单。")
if aspect in ("purchase", "all"):
if purchase:
lines.append(f"· 采购建议:{len(purchase)} 条")
for r in purchase[:8]:
lines.append(
f" - {r.get('orderNo')} {r.get('materialName') or r.get('materialCode')} "
f"×{r.get('quantity')} [{r.get('status')}] 建议下单 {r.get('suggestedOrderDate')}"
)
else:
tip = "(可先说「分解订单」生成建议)" if not result.get("decomposed") else ""
lines.append(f"· 采购建议:无{tip}")
if aspect == "outsource" and not outsource and not external:
lines.append("结论:这个订单没有委外。")
elif aspect == "outsource" and (outsource or external):
lines.append("结论:这个订单涉及委外。")
return "\n".join(lines)
return str(result)

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,794 @@
# ============================================================
# 订单管理领域服务(moduleId: domain-orders, 可重生 ✅)
# 职责:销售订单只读投影 + P2 订单写入动作的纯应用逻辑
# 规则:新增/编辑/取消/完成/删除订单均写主干世界,必须由调用方走 Harness 确认卡
# ============================================================
from __future__ import annotations # 前向类型引用
from datetime import datetime # 时间戳
from typing import Any # 类型标注
from aps.timeutil import fmt_dt, today0 # 时间格式化
World = dict[str, Any] # 世界状态类型别名
# 当前数据模型的订单状态。后续订单池审批会扩展为更完整状态机。
ORDER_STATUSES = {
"DRAFT", "SUBMITTED", "APPROVED", "REJECTED", "CHANGED",
"CANCELLED", "COMPLETED",
"CONFIRMED", # 遗留别名 ≡ APPROVED(排产资格相同)
}
# 正式排产默认纳入的状态(OR-03)
SCHEDULABLE_STATUSES = {"APPROVED", "CONFIRMED"}
# 可提交审核
SUBMITTABLE_STATUSES = {"DRAFT", "REJECTED", "CHANGED"}
# 可批准/驳回
REVIEWABLE_STATUSES = {"SUBMITTED", "CHANGED"}
CUSTOMER_LEVELS = {"VIP", "A", "B", "C"}
ORDER_ACTIONS = (
"order.upsert", "order.cancel", "order.complete", "order.delete", "order.clear",
"order.submit", "order.approve", "order.reject",
)
POOL_LANES = {
"pending": ("SUBMITTED",),
"approved": ("APPROVED", "CONFIRMED"),
"changed": ("CHANGED",),
"draft": ("DRAFT", "REJECTED"),
"closed": ("CANCELLED", "COMPLETED"),
}
def normalize_status(status: str | None, *, default: str = "DRAFT") -> str:
s = str(status or default).upper()
if s == "CONFIRMED":
return "APPROVED" # 新写入统一为 APPROVED
if s not in ORDER_STATUSES:
raise ValueError(
"订单状态必须是 DRAFT/SUBMITTED/APPROVED/REJECTED/CHANGED/CANCELLED/COMPLETED")
return s
def is_schedulable(status: str | None, *, include_unapproved: bool = False) -> bool:
s = str(status or "")
if s in SCHEDULABLE_STATUSES:
return True
if include_unapproved and s in ("DRAFT", "SUBMITTED", "CHANGED", "REJECTED"):
return True
return False
def _now() -> str:
"""当前时间字符串(与世界状态 createdAt/updatedAt 口径一致)。"""
return fmt_dt(datetime.now())
def _product(world: World, product_id: int) -> dict[str, Any]:
"""取成品物料;不存在则抛出 ValueError,阻止脏订单进入世界状态。"""
product = next((m for m in world["materials"]
if m["id"] == product_id and m["type"] == "FINISHED_PRODUCT"), None)
if product is None:
raise ValueError(f"产品不存在或不是成品:{product_id}")
return product
def _next_item_id(world: World) -> int:
"""销售订单明细发号:扫描现有 items,避免改 WorldStore 计数器契约。"""
max_id = 0
for so in world.get("salesOrders", []):
for item in so.get("items", []):
if isinstance(item.get("id"), int):
max_id = max(max_id, item["id"])
return max_id + 1
def list_products(world: World) -> list[dict[str, Any]]:
"""成品下拉选项(P0 只读)。"""
return [{
"id": m["id"],
"code": m["code"],
"name": m["name"],
"unit": m["unit"],
} for m in world["materials"] if m["type"] == "FINISHED_PRODUCT"]
def _flex_status_to_sales(status: str | None) -> str:
"""现场/SQL 柔性订单状态 → 销售订单状态机。"""
s = str(status or "").upper()
if s in ("RELEASED", "RELEASE", "OPEN", "IN_PROGRESS", "PRODUCING", "PLANNED"):
return "APPROVED"
if s in ("DONE", "COMPLETED", "FINISHED", "CLOSED"):
return "COMPLETED"
if s in ("CANCELLED", "CANCELED"):
return "CANCELLED"
if s in ("DRAFT", "CREATED"):
return "DRAFT"
if s in ORDER_STATUSES or s == "CONFIRMED":
try:
return normalize_status(s)
except ValueError:
return "APPROVED"
return "APPROVED"
def sync_flex_orders_to_sales(world: World) -> int:
"""Idempotently project flexible shop-floor orders into sales orders."""
demo_mat = {
"HV-HARNESS", "PDU-UNIT", "CHARGE-GUN", "HV-CONN",
"CTRL-A", "CTRL-B", "CTRL-C",
}
flex: list[dict[str, Any]] = []
for order in world.get("flexOrders") or []:
if str(order.get("source") or "").upper() in {"CLOSED_LOOP", "MRP"}:
continue
if str(order.get("status") or "").upper() in {"DONE", "CANCELLED", "COMPLETED"}:
continue
order_no = str(order.get("orderNo") or "")
product_code = str(order.get("productCode") or "")
if order_no.startswith(("FO-26", "SO-DEMO")) or product_code in demo_mat:
continue
flex.append(order)
existing_rows = list(world.get("salesOrders") or [])
projected_sources = {"SQL", "FLEX", "SITE"}
manual_rows = [
row for row in existing_rows
if str(row.get("source") or "MANUAL").upper() not in projected_sources
]
existing_projected = {
str(row.get("orderNo") or ""): row
for row in existing_rows
if str(row.get("source") or "MANUAL").upper() in projected_sources
and str(row.get("orderNo") or "")
}
mats = {material.get("code"): material for material in (world.get("materials") or []) if material.get("code")}
flex_mats = {material.get("code"): material for material in (world.get("flexMaterials") or []) if material.get("code")}
next_mat = max(
(material.get("id", 0) for material in (world.get("materials") or []) if isinstance(material.get("id"), int)),
default=0,
) + 1
next_so = max(
(row.get("id", 0) for row in existing_rows if isinstance(row.get("id"), int)),
default=0,
) + 1
next_item = _next_item_id(world)
business_day = str(world.get("businessDate") or fmt_dt(today0())[:10])[:10]
projected_rows: list[dict[str, Any]] = []
changed = 0
seen_order_nos: set[str] = set()
for flex_order in flex:
order_no = str(flex_order.get("orderNo") or "").strip()
product_code = str(flex_order.get("productCode") or "").strip()
if not order_no or not product_code or order_no in seen_order_nos:
continue
seen_order_nos.add(order_no)
material = mats.get(product_code)
if material is None:
flex_material = flex_mats.get(product_code) or {
"code": product_code,
"name": flex_order.get("productName") or product_code,
"type": "FINISHED_PRODUCT",
"unit": "件",
}
material = {
"id": next_mat,
"code": product_code,
"name": flex_material.get("name") or product_code,
"spec": flex_material.get("spec") or "",
"type": flex_material.get("type") or "FINISHED_PRODUCT",
"unit": flex_material.get("unit") or "件",
"productFamily": "",
"safetyStock": float(flex_material.get("safetyStock") or 0),
"procurementLeadTime": float(flex_material.get("procurementLeadTime") or 0),
"stock": float(flex_material.get("stock") or 0),
"inTransit": float(flex_material.get("inTransit") or 0),
"status": "ACTIVE",
}
world.setdefault("materials", []).append(material)
mats[product_code] = material
next_mat += 1
product_id = material.get("id")
if not isinstance(product_id, int) or product_id <= 0:
continue
status = _flex_status_to_sales(flex_order.get("status"))
quantity = float(flex_order.get("quantity") or 0) or 1
desired_order = {
"orderNo": order_no,
"customerName": flex_order.get("customerName") or "现场客户",
"customerLevel": flex_order.get("customerLevel") or "B",
"orderDate": str(flex_order.get("orderDate") or business_day)[:10],
"deliveryDate": str(flex_order.get("dueDate") or business_day)[:10],
"priority": int(flex_order.get("priority") or 5),
"status": status,
"source": "SQL" if flex_order.get("craftlCode") else "FLEX",
"isRush": bool(flex_order.get("isRush")),
"specialRequirements": flex_order.get("drawingNo") or "",
"totalAmount": 0,
}
desired_item = {
"productId": product_id,
"productCode": product_code,
"productName": flex_order.get("productName") or material.get("name") or product_code,
"quantity": quantity,
"unit": material.get("unit") or "件",
"status": "PENDING" if status == "APPROVED" else status,
}
existing = existing_projected.get(order_no)
existing_item = ((existing or {}).get("items") or [{}])[0]
order_matches = bool(existing) and all(existing.get(key) == value for key, value in desired_order.items())
item_matches = bool(existing) and all(existing_item.get(key) == value for key, value in desired_item.items())
if order_matches and item_matches and len((existing or {}).get("items") or []) == 1:
projected_rows.append(existing)
continue
now = _now()
if existing:
row = {**existing, **desired_order}
item_id = existing_item.get("id") if isinstance(existing_item.get("id"), int) else next_item
if item_id == next_item:
next_item += 1
row["items"] = [{**existing_item, **desired_item, "id": item_id}]
row["createdAt"] = existing.get("createdAt") or now
row["updatedAt"] = now
else:
row = {
"id": next_so,
**desired_order,
"createdAt": now,
"updatedAt": now,
"items": [{"id": next_item, **desired_item}],
}
next_so += 1
next_item += 1
projected_rows.append(row)
changed += 1
removed = set(existing_projected) - seen_order_nos
changed += len(removed)
next_rows = manual_rows + projected_rows
if changed or next_rows != existing_rows:
if not changed:
changed = 1
world["salesOrders"] = next_rows
return changed
def list_orders(world: World) -> list[dict[str, Any]]:
"""销售订单列表投影(含已投影的 SQL/柔性单;只读兜底不写盘)。"""
rows: list[dict[str, Any]] = []
for so in world.get("salesOrders") or []:
items = so.get("items", [])
first = items[0] if items else {}
rows.append({
"id": so["id"],
"orderNo": so["orderNo"],
"customerName": so["customerName"],
"customerLevel": so.get("customerLevel") or "B",
"orderDate": so.get("orderDate") or "",
"deliveryDate": so.get("deliveryDate") or so.get("dueDate") or "",
"priority": so.get("priority") or 5,
"status": _flex_status_to_sales(so.get("status")),
"source": so.get("source", "MANUAL"),
"isRush": so.get("isRush", False),
"rushStrategy": so.get("rushStrategy"),
"specialRequirements": so.get("specialRequirements", ""),
"totalAmount": so.get("totalAmount", 0),
"itemCount": len(items),
"productId": first.get("productId"),
"productCode": first.get("productCode", ""),
"productName": first.get("productName", ""),
"quantity": first.get("quantity", 0),
"unit": first.get("unit", "件"),
"updatedAt": so.get("updatedAt") or so.get("createdAt"),
"reviewNote": so.get("reviewNote") or "",
"schedulable": is_schedulable(_flex_status_to_sales(so.get("status"))),
})
# 兜底:销售表空时只读展柔性订单(不写 salesOrders,避免污染演示世界)
if not rows:
mats = {m.get("code"): m for m in (world.get("materials") or [])}
for i, fo in enumerate(world.get("flexOrders") or []):
if (fo.get("status") or "") in ("DONE", "CANCELLED", "COMPLETED"):
continue
ono = str(fo.get("orderNo") or "")
if ono.startswith(("FO-26", "SO-DEMO")):
continue
pc = fo.get("productCode") or ""
st = _flex_status_to_sales(fo.get("status"))
rows.append({
"id": 10_000_000 + int(fo.get("id") or i),
"orderNo": fo.get("orderNo"),
"customerName": fo.get("customerName") or "现场客户",
"customerLevel": "B",
"orderDate": "",
"deliveryDate": fo.get("dueDate") or "",
"priority": fo.get("priority") or 5,
"status": st,
"source": "FLEX",
"isRush": False,
"itemCount": 1,
"productId": (mats.get(pc) or {}).get("id"),
"productCode": pc,
"productName": fo.get("productName") or (mats.get(pc) or {}).get("name") or pc,
"quantity": fo.get("quantity") or 0,
"unit": "件",
"schedulable": is_schedulable(st),
})
rows.sort(key=lambda x: (
x["status"] in ("CANCELLED", "COMPLETED"),
x["status"] not in ("SUBMITTED", "CHANGED", "APPROVED"),
x["deliveryDate"], x["priority"],
))
return rows
def pool_summary(world: World) -> dict[str, Any]:
"""订单池看板计数(只读,不写盘)。"""
counts = {k: 0 for k in POOL_LANES}
counts["all"] = 0
counts["schedulable"] = 0
for row in list_orders(world):
counts["all"] += 1
st = row.get("status")
if is_schedulable(st):
counts["schedulable"] += 1
for lane, statuses in POOL_LANES.items():
if st in statuses:
counts[lane] += 1
break
return counts
def list_production_orders(world: World) -> list[dict[str, Any]]:
"""生产订单/虚拟产线投影:优先最新柔性排产版本,其次固定轨 productionOrders。"""
rows: list[dict[str, Any]] = []
flex_vers = world.get("flexScheduleVersions") or []
if flex_vers:
ver = flex_vers[-1]
vid = ver.get("id")
orders_by_no = {o.get("orderNo"): o for o in (world.get("flexOrders") or [])}
for vl in world.get("flexVirtualLines") or []:
if vl.get("versionId") != vid:
continue
ono = vl.get("orderNo")
o = orders_by_no.get(ono) or {}
wos = [w for w in (world.get("flexWorkOrders") or [])
if w.get("versionId") == vid and w.get("vlId") == vl.get("id")]
rows.append({
"id": vl.get("id"),
"orderNo": ono,
"vlNo": vl.get("vlNo"),
"productCode": vl.get("productCode") or o.get("productCode"),
"productName": o.get("productName") or vl.get("productCode"),
"quantity": vl.get("quantity") or o.get("quantity"),
"dueDate": o.get("dueDate"),
"plannedStart": vl.get("plannedStart"),
"plannedEnd": vl.get("plannedEnd"),
"stepCount": len(wos) or vl.get("stepCount") or 0,
"status": vl.get("status") or "PLANNED",
"versionNo": ver.get("versionNo"),
"track": "flex",
})
return rows
for po in world.get("productionOrders") or []:
rows.append({
"id": po.get("id"),
"orderNo": po.get("orderNo") or po.get("salesOrderNo"),
"vlNo": po.get("poNo") or po.get("code"),
"productCode": po.get("productCode"),
"productName": po.get("productName"),
"quantity": po.get("quantity"),
"dueDate": po.get("dueDate"),
"plannedStart": po.get("plannedStart"),
"plannedEnd": po.get("plannedEnd"),
"stepCount": po.get("woCount") or 0,
"status": po.get("status") or "PLANNED",
"versionNo": po.get("schedulingVersionId"),
"track": "fixed",
})
return rows
def find_product_by_hint(world: World, hint: str) -> dict[str, Any] | None:
"""按编码/名称模糊解析成品(对话建单用):先精确编码,再名称包含匹配。"""
h = (hint or "").strip()
if not h:
return None
products = [m for m in world.get("materials") or [] if m.get("type") == "FINISHED_PRODUCT"]
if not products:
products = [m for m in world.get("flexMaterials") or [] if m.get("type") == "FINISHED_PRODUCT"]
exact = next((m for m in products if str(m.get("code") or "").lower() == h.lower()), None)
if exact:
return exact
return next(
(m for m in products
if h in str(m.get("name") or "") or str(m.get("name") or "") in h
or h.lower() in str(m.get("code") or "").lower()),
None,
)
def find_order(world: World, *, order_id: int | None = None, order_no: str | None = None) -> dict[str, Any] | None:
"""按 ID 或订单号查订单(P0 只读;含柔性表只读回退)。"""
if order_id is not None:
hit = next((so for so in world.get("salesOrders") or [] if so["id"] == order_id), None)
if hit:
return hit
if order_no:
hit = next((so for so in world.get("salesOrders") or []
if str(so.get("orderNo") or "").lower() == order_no.lower()), None)
if hit:
return hit
fo = next((o for o in world.get("flexOrders") or []
if str(o.get("orderNo") or "").lower() == order_no.lower()), None)
if fo:
return {
"id": fo.get("id"), "orderNo": fo.get("orderNo"),
"customerName": fo.get("customerName") or "现场客户",
"customerLevel": "B", "deliveryDate": fo.get("dueDate"),
"priority": fo.get("priority") or 5,
"status": _flex_status_to_sales(fo.get("status")),
"source": "FLEX", "items": [{
"productCode": fo.get("productCode"),
"productName": fo.get("productName"),
"quantity": fo.get("quantity"), "unit": "件",
}],
}
return None
def normalize_order_payload(world: World, payload: dict[str, Any]) -> dict[str, Any]:
"""订单写入载荷归一化与校验(P0 纯校验)。"""
out = dict(payload)
if out.get("id") in ("", None):
out.pop("id", None)
elif not isinstance(out["id"], int):
out["id"] = int(out["id"])
out["customerName"] = str(out.get("customerName") or "").strip()
if not out["customerName"]:
raise ValueError("客户名称不能为空")
out["customerLevel"] = str(out.get("customerLevel") or "A").upper()
if out["customerLevel"] not in CUSTOMER_LEVELS:
raise ValueError("客户等级必须是 VIP/A/B/C")
out["deliveryDate"] = str(out.get("deliveryDate") or "").strip()
if not out["deliveryDate"]:
raise ValueError("交期不能为空")
out["priority"] = max(1, min(9, int(out.get("priority") or 5)))
out["productId"] = int(out.get("productId") or 0)
product = _product(world, out["productId"])
out["productName"] = product["name"]
out["productCode"] = product["code"]
out["unit"] = product["unit"]
out["quantity"] = max(1, int(float(out.get("quantity") or 1)))
# 新建默认 DRAFT;遗留 CONFIRMED 写入时归一为 APPROVED
default_status = "DRAFT" if not out.get("id") else "APPROVED"
out["status"] = normalize_status(out.get("status"), default=default_status)
out["isRush"] = bool(out.get("isRush", False))
out["rushStrategy"] = "STRATEGY_SHIFT" if out["isRush"] else None
out["specialRequirements"] = str(out.get("specialRequirements") or "").strip()
out["reviewNote"] = str(out.get("reviewNote") or "").strip()
return out
def confirmation_for_order_action(world: World, action: str, payload: dict[str, Any]) -> tuple[str, list[str]]:
"""生成订单 P2 确认卡标题与影响摘要。"""
if action == "order.upsert":
p = normalize_order_payload(world, payload)
verb = "编辑" if p.get("id") else "新增"
title = f"{verb}销售订单"
lines = [
f"客户:{p['customerName']}({p['customerLevel']})",
f"产品:{p['productName']} × {p['quantity']} {p['unit']},交期 {p['deliveryDate']}",
"批准后写入主干订单池,并影响后续新排产版本(P2)",
]
return title, lines
if action == "order.clear":
n = len(world.get("salesOrders", []))
po = len(world.get("purchaseOrders", []))
os = len(world.get("outsourceOrders", []))
if n == 0 and po == 0 and os == 0:
raise ValueError("订单池已空,无需清理")
return "一键清理全部订单", [
f"将清空销售订单 {n} 条",
f"同步清空采购建议 {po} / 委外建议 {os} 条",
"历史排产版本与工单保留不回写;后续试排将无订单可排",
"执行前自动建档,可回滚(P2)",
]
if action == "order.approve":
ids = _resolve_order_ids(world, payload)
if not ids:
raise ValueError("没有可批准的订单(需 SUBMITTED/CHANGED)")
nos = [find_order(world, order_id=i)["orderNo"] for i in ids] # type: ignore[index]
return f"批准订单 {len(ids)} 条", [
"、".join(nos[:5]) + ("…" if len(nos) > 5 else ""),
"批准后状态 → APPROVED,可进入正式排产",
"执行前自动建档,可回滚(P2)",
]
if action == "order.reject":
ids = _resolve_order_ids(world, payload, for_reject=True)
if not ids:
raise ValueError("没有可驳回的订单(需 SUBMITTED/CHANGED)")
note = str(payload.get("reviewNote") or "").strip() or "(未填写原因)"
nos = [find_order(world, order_id=i)["orderNo"] for i in ids] # type: ignore[index]
return f"驳回订单 {len(ids)} 条", [
"、".join(nos[:5]) + ("…" if len(nos) > 5 else ""),
f"驳回原因:{note}",
"驳回后状态 → REJECTED,不参与排产;可修改后重新提交",
"执行前自动建档,可回滚(P2)",
]
order = find_order(world, order_id=int(payload.get("id") or 0), order_no=payload.get("orderNo"))
if order is None:
raise ValueError("目标订单不存在")
if action == "order.cancel":
return f"取消订单 {order['orderNo']}", [
f"客户:{order['customerName']},当前状态 {order['status']}",
"批准后该订单不再参与后续排产,历史版本不回写",
"执行前自动建档,可回滚",
]
if action == "order.complete":
return f"完成订单 {order['orderNo']}", [
f"客户:{order['customerName']},当前状态 {order['status']}",
"批准后订单及明细置为 COMPLETED,不再参与后续排产",
"执行前自动建档,可回滚",
]
if action == "order.delete":
draft_po = sum(1 for p in world.get("purchaseOrders", [])
if p.get("salesOrderId") == order["id"] and p.get("status") == "DRAFT")
draft_os = sum(1 for o in world.get("outsourceOrders", [])
if o.get("salesOrderId") == order["id"] and o.get("status") == "DRAFT")
hist = sum(1 for po in world.get("productionOrders", []) if po.get("salesOrderId") == order["id"])
lines = [
f"客户:{order['customerName']} · {order['orderNo']}({order['status']})",
"批准后从订单池物理删除,后续排产不再纳入",
]
if draft_po or draft_os:
lines.append(f"同步清除草稿采购 {draft_po} / 委外 {draft_os} 条")
if hist:
lines.append(f"历史生产订单 {hist} 条保留不回写(仅断联销售单)")
lines.append("执行前自动建档,可回滚")
return f"删除订单 {order['orderNo']}", lines
raise ValueError(f"不支持的订单动作:{action}")
def _resolve_order_ids(world: World, payload: dict[str, Any], *, for_reject: bool = False) -> list[int]:
"""解析单笔/批量订单 ID,并过滤到可审核状态。"""
allowed = REVIEWABLE_STATUSES
raw_ids = payload.get("orderIds") or payload.get("ids") or []
ids: list[int] = []
if raw_ids == "pending" or raw_ids == ["pending"]:
return [so["id"] for so in world.get("salesOrders", []) if so.get("status") in allowed]
if isinstance(raw_ids, list) and raw_ids:
ids = [int(x) for x in raw_ids]
elif payload.get("id"):
ids = [int(payload["id"])]
elif payload.get("orderNo"):
o = find_order(world, order_no=str(payload["orderNo"]))
if o:
ids = [o["id"]]
out = []
for oid in ids:
o = find_order(world, order_id=oid)
if o and o["status"] in allowed:
out.append(oid)
return out
def apply_order_action(world: World, next_id, action: str, payload: dict[str, Any]) -> dict[str, Any]:
"""应用订单写入动作到内存世界(P2 动作的业务侧实现;调用方负责门禁/快照/审计/落盘)。"""
if action == "order.clear":
n = len(world.get("salesOrders", []))
po_n = len(world.get("purchaseOrders", []))
os_n = len(world.get("outsourceOrders", []))
world["salesOrders"] = []
world["purchaseOrders"] = []
world["outsourceOrders"] = []
return {
"order": {"id": 0, "orderNo": "ALL", "status": "CLEARED"},
"created": False, "beforeStatus": None,
"clearedOrders": n, "clearedPurchase": po_n, "clearedOutsource": os_n,
}
if action == "order.upsert":
p = normalize_order_payload(world, payload)
if p.get("id"):
order = find_order(world, order_id=p["id"])
if order is None:
raise ValueError("目标订单不存在")
before_status = order["status"]
# 已批准订单被改内容 → CHANGED,需重新审核(除非显式仍写 APPROVED 且字段未变)
next_status = p["status"]
content_changed = any([
order.get("customerName") != p["customerName"],
order.get("customerLevel") != p["customerLevel"],
order.get("deliveryDate") != p["deliveryDate"],
order.get("priority") != p["priority"],
(order.get("items") or [{}])[0].get("productId") != p["productId"],
(order.get("items") or [{}])[0].get("quantity") != p["quantity"],
bool(order.get("isRush")) != bool(p["isRush"]),
])
if before_status in SCHEDULABLE_STATUSES and content_changed and next_status in SCHEDULABLE_STATUSES:
next_status = "CHANGED"
order.update({
"customerName": p["customerName"],
"customerLevel": p["customerLevel"],
"deliveryDate": p["deliveryDate"],
"priority": p["priority"],
"status": next_status,
"isRush": p["isRush"],
"rushStrategy": p["rushStrategy"],
"specialRequirements": p["specialRequirements"],
"updatedAt": _now(),
})
if next_status != before_status:
order.setdefault("changes", []).append({
"at": _now(), "action": "order.upsert",
"from": before_status, "to": next_status,
})
if not order.get("items"):
order["items"] = []
item = order["items"][0] if order["items"] else None
if item is None:
item = {"id": _next_item_id(world), "orderId": order["id"], "lineNo": 1}
order["items"].append(item)
item.update({
"productId": p["productId"],
"productName": p["productName"],
"productCode": p["productCode"],
"quantity": p["quantity"],
"unit": p["unit"],
"bomVersion": item.get("bomVersion", "V1.0"),
"routingVersion": item.get("routingVersion", "V1.0"),
"status": "COMPLETED" if next_status == "COMPLETED"
else "CANCELLED" if next_status == "CANCELLED"
else "PENDING",
"note": item.get("note", ""),
})
return {"order": order, "created": False, "beforeStatus": before_status}
order_id = next_id("salesOrder")
order_date = today0().strftime("%Y-%m-%d")
product = _product(world, p["productId"])
order = {
"id": order_id,
"orderNo": "SO" + today0().strftime("%Y%m%d") + f"{order_id:03d}",
"customerId": f"CUST{order_id:03d}",
"customerName": p["customerName"],
"customerLevel": p["customerLevel"],
"orderDate": order_date,
"deliveryDate": p["deliveryDate"],
"priority": p["priority"],
"manualPriority": None,
"status": p["status"],
"source": "MANUAL",
"specialRequirements": p["specialRequirements"],
"totalAmount": p["quantity"] * 100,
"isRush": p["isRush"],
"rushStrategy": p["rushStrategy"],
"changes": [],
"createdBy": "planner",
"createdAt": _now(),
"updatedAt": _now(),
"items": [{
"id": _next_item_id(world),
"orderId": order_id,
"lineNo": 1,
"productId": product["id"],
"productName": product["name"],
"productCode": product["code"],
"quantity": p["quantity"],
"unit": product["unit"],
"bomVersion": "V1.0",
"routingVersion": "V1.0",
"status": "PENDING",
"note": "",
}],
}
world["salesOrders"].append(order)
return {"order": order, "created": True, "beforeStatus": None}
# 批准/驳回支持 orderIds / pending,须在单笔 id 查找之前
if action == "order.approve":
ids = _resolve_order_ids(world, payload)
if not ids:
raise ValueError("没有可批准的订单")
updated = []
for oid in ids:
o = find_order(world, order_id=oid)
if not o:
continue
before = o["status"]
o["status"] = "APPROVED"
o["reviewNote"] = str(payload.get("reviewNote") or "").strip()
o["updatedAt"] = _now()
o.setdefault("changes", []).append({
"at": _now(), "action": "order.approve", "from": before, "to": "APPROVED",
})
updated.append(o)
first = updated[0]
return {
"order": first, "created": False,
"beforeStatus": first.get("changes", [{}])[-1].get("from"),
"approvedCount": len(updated),
"orderNos": [o["orderNo"] for o in updated],
}
if action == "order.reject":
ids = _resolve_order_ids(world, payload, for_reject=True)
if not ids:
raise ValueError("没有可驳回的订单")
note = str(payload.get("reviewNote") or payload.get("reason") or "").strip()
updated = []
for oid in ids:
o = find_order(world, order_id=oid)
if not o:
continue
before = o["status"]
o["status"] = "REJECTED"
o["reviewNote"] = note or "已驳回"
o["updatedAt"] = _now()
o.setdefault("changes", []).append({
"at": _now(), "action": "order.reject", "from": before, "to": "REJECTED",
"note": note,
})
updated.append(o)
first = updated[0]
return {
"order": first, "created": False, "beforeStatus": "SUBMITTED",
"rejectedCount": len(updated),
"orderNos": [o["orderNo"] for o in updated],
}
order = find_order(world, order_id=int(payload.get("id") or 0), order_no=payload.get("orderNo"))
if order is None:
raise ValueError("目标订单不存在")
before_status = order["status"]
if action == "order.delete":
oid = order["id"]
order_no = order["orderNo"]
# 清除该单关联的草稿 MRP;已下达建议保留但断开关联语义由业务自行处理
world["purchaseOrders"] = [
p for p in world.get("purchaseOrders", [])
if not (p.get("salesOrderId") == oid and p.get("status") == "DRAFT")
]
world["outsourceOrders"] = [
o for o in world.get("outsourceOrders", [])
if not (o.get("salesOrderId") == oid and o.get("status") == "DRAFT")
]
world["salesOrders"] = [so for so in world["salesOrders"] if so["id"] != oid]
return {"order": {"id": oid, "orderNo": order_no, "status": "DELETED"},
"created": False, "beforeStatus": before_status}
if action == "order.cancel":
order["status"] = "CANCELLED"
for item in order.get("items", []):
item["status"] = "CANCELLED"
elif action == "order.complete":
order["status"] = "COMPLETED"
for item in order.get("items", []):
item["status"] = "COMPLETED"
else:
raise ValueError(f"不支持的订单动作:{action}")
order["updatedAt"] = _now()
order.setdefault("changes", []).append({
"at": _now(),
"action": action,
"from": before_status,
"to": order["status"],
})
return {"order": order, "created": False, "beforeStatus": before_status}
def apply_order_submit(world: World, payload: dict[str, Any]) -> dict[str, Any]:
"""提交审核(P1):DRAFT/REJECTED/CHANGED → SUBMITTED。"""
order = find_order(world, order_id=int(payload.get("id") or 0), order_no=payload.get("orderNo"))
if order is None:
raise ValueError("目标订单不存在")
before = order["status"]
if before not in SUBMITTABLE_STATUSES:
raise ValueError(f"订单 {order['orderNo']} 状态为 {before},不能提交审核")
order["status"] = "SUBMITTED"
order["reviewNote"] = ""
order["updatedAt"] = _now()
order.setdefault("changes", []).append({
"at": _now(), "action": "order.submit", "from": before, "to": "SUBMITTED",
})
return {"order": order, "created": False, "beforeStatus": before}

View File

@ -0,0 +1,179 @@
# ============================================================
# 排产参数(moduleId: domain-params, 可重生 ✅)
# OR-02:客户等级权重 / 目标权重;变更走 P2,只影响后续新版本
# ============================================================
from __future__ import annotations
from copy import deepcopy
from typing import Any
World = dict[str, Any]
DEFAULT_LEVEL_WEIGHTS = {"VIP": 3.0, "A": 2.0, "B": 1.0, "C": 1.0}
DEFAULT_OBJECTIVE_WEIGHTS = {"tardiness": 0.4, "cost": 0.3, "utilization": 0.2, "balance": 0.1}
def default_schedule_params() -> dict[str, Any]:
return {
"defaultEngine": "RULE",
"planningHorizonDays": 14,
"timeGranularityMinutes": 15,
"deliveryBufferRatio": 0.95,
"weights": dict(DEFAULT_OBJECTIVE_WEIGHTS),
"customerLevelWeights": dict(DEFAULT_LEVEL_WEIGHTS),
"freezeWindowHours": 24,
"cpTimeLimitSeconds": 8, # SC-03 CP-SAT 时限
}
def get_schedule_params(world: World) -> dict[str, Any]:
"""合并默认值后的排产参数投影(P0)。"""
base = default_schedule_params()
cur = world.get("scheduleParams") or {}
out = {**base, **cur}
out["weights"] = {**DEFAULT_OBJECTIVE_WEIGHTS, **(cur.get("weights") or {})}
out["customerLevelWeights"] = {
**DEFAULT_LEVEL_WEIGHTS,
**(cur.get("customerLevelWeights") or {}),
}
return out
def level_weight(world: World, level: str | None) -> float:
"""取客户等级排序权重(越高越优先排)。"""
weights = get_schedule_params(world)["customerLevelWeights"]
return float(weights.get(str(level or "C").upper(), 1.0))
def normalize_params_payload(payload: dict[str, Any]) -> dict[str, Any]:
"""校验并规范化 params.update 载荷。"""
out: dict[str, Any] = {}
if payload.get("resetDefaults"):
out["resetDefaults"] = True
return out
if "customerLevelWeights" in payload and payload["customerLevelWeights"] is not None:
raw = payload["customerLevelWeights"]
if not isinstance(raw, dict):
raise ValueError("customerLevelWeights 必须是对象")
levels: dict[str, float] = {}
for k, v in raw.items():
key = str(k).upper()
if key not in DEFAULT_LEVEL_WEIGHTS:
raise ValueError(f"不支持的客户等级:{k}")
try:
val = float(v)
except (TypeError, ValueError) as exc:
raise ValueError(f"{key} 权重必须是数字") from exc
if val < 0 or val > 100:
raise ValueError(f"{key} 权重须在 0~100")
levels[key] = val
out["customerLevelWeights"] = levels
if "weights" in payload and payload["weights"] is not None:
raw = payload["weights"]
if not isinstance(raw, dict):
raise ValueError("weights 必须是对象")
obj: dict[str, float] = {}
for k, v in raw.items():
key = str(k)
if key not in DEFAULT_OBJECTIVE_WEIGHTS:
raise ValueError(f"不支持的目标权重键:{k}")
try:
val = float(v)
except (TypeError, ValueError) as exc:
raise ValueError(f"{key} 权重必须是数字") from exc
if val < 0 or val > 1:
raise ValueError(f"{key} 目标权重须在 0~1")
obj[key] = val
out["weights"] = obj
if "planningHorizonDays" in payload and payload["planningHorizonDays"] is not None:
days = int(payload["planningHorizonDays"])
if days < 1 or days > 90:
raise ValueError("planningHorizonDays 须在 1~90")
out["planningHorizonDays"] = days
if "defaultEngine" in payload and payload["defaultEngine"] is not None:
eng = str(payload["defaultEngine"]).upper()
if eng not in ("RULE", "CP", "GA", "HYBRID"):
raise ValueError("defaultEngine 须为 RULE/CP/GA/HYBRID")
out["defaultEngine"] = eng
if "cpTimeLimitSeconds" in payload and payload["cpTimeLimitSeconds"] is not None:
sec = float(payload["cpTimeLimitSeconds"])
if sec < 0.5 or sec > 120:
raise ValueError("cpTimeLimitSeconds 须在 0.5~120")
out["cpTimeLimitSeconds"] = sec
if not out:
raise ValueError("未提供可更新的排产参数字段")
return out
def confirmation_for_params_update(world: World, payload: dict[str, Any]) -> tuple[str, list[str]]:
"""生成 params.update 确认卡文案。"""
norm = normalize_params_payload(payload)
cur = get_schedule_params(world)
if norm.get("resetDefaults"):
return "恢复排产参数默认值", [
"客户等级权重恢复 VIP=3 / A=2 / B=1 / C=1",
"目标权重恢复交期/成本/利用率/均衡默认比",
"只影响后续新排产版本,不回写历史版本",
]
lines: list[str] = []
if "customerLevelWeights" in norm:
before = cur["customerLevelWeights"]
after = {**before, **norm["customerLevelWeights"]}
parts = [f"{k}:{before.get(k)}→{after[k]}" for k in ("VIP", "A", "B", "C")]
lines.append("客户等级权重:" + " / ".join(parts))
if "weights" in norm:
before = cur["weights"]
after = {**before, **norm["weights"]}
parts = [f"{k}:{before.get(k)}→{after[k]}" for k in after]
lines.append("目标权重:" + " / ".join(parts))
if "planningHorizonDays" in norm:
lines.append(f"展望期:{cur['planningHorizonDays']}→{norm['planningHorizonDays']} 天")
if "defaultEngine" in norm:
lines.append(f"默认引擎:{cur.get('defaultEngine')}→{norm['defaultEngine']}")
if "cpTimeLimitSeconds" in norm:
lines.append(f"CP 时限:{cur.get('cpTimeLimitSeconds')}→{norm['cpTimeLimitSeconds']} 秒")
lines.append("只影响后续新排产版本,不回写历史版本")
return "更新排产参数", lines
def apply_params_update(world: World, payload: dict[str, Any]) -> dict[str, Any]:
"""写入 scheduleParams(P2 批准后调用)。"""
norm = normalize_params_payload(payload)
before = deepcopy(get_schedule_params(world))
if norm.get("resetDefaults"):
world["scheduleParams"] = default_schedule_params()
else:
sp = world.setdefault("scheduleParams", default_schedule_params())
if "customerLevelWeights" in norm:
sp["customerLevelWeights"] = {
**DEFAULT_LEVEL_WEIGHTS,
**(sp.get("customerLevelWeights") or {}),
**norm["customerLevelWeights"],
}
if "weights" in norm:
sp["weights"] = {
**DEFAULT_OBJECTIVE_WEIGHTS,
**(sp.get("weights") or {}),
**norm["weights"],
}
if "planningHorizonDays" in norm:
sp["planningHorizonDays"] = norm["planningHorizonDays"]
if "defaultEngine" in norm:
sp["defaultEngine"] = norm["defaultEngine"]
if "cpTimeLimitSeconds" in norm:
sp["cpTimeLimitSeconds"] = norm["cpTimeLimitSeconds"]
after = get_schedule_params(world)
return {
"kind": "SCHEDULE_PARAMS",
"id": "scheduleParams",
"name": "排产参数",
"before": before,
"after": after,
"reset": bool(norm.get("resetDefaults")),
}

View File

@ -0,0 +1,74 @@
# ============================================================
# 主数据查询辅助(moduleId: engines-queries, 可重生 ✅)
# 从 legacy common.js 的全局查找器按行移植(P0 只读):
# findRoutingSteps / findBomItems / findProductLines /
# findWorkstationForOperation / getShiftMinutes / getLineShifts / getAvailableMinutes
# ============================================================
from __future__ import annotations # 前向类型引用
from typing import Any # 类型标注
# 世界状态的类型别名(camelCase 键的字典)
World = dict[str, Any]
def find_routing_steps(world: World, product_id: int) -> list[dict]:
"""取产品默认工艺路线的步骤(按顺序号排序)。"""
routing = next((r for r in world["routings"] if r["productId"] == product_id and r["isDefault"]), None) # 默认路线
if not routing: # 无路线 → 空列表(上游生成 NO_LINE 类冲突)
return []
steps = [s for s in world["routingSteps"] if s["routingId"] == routing["id"]] # 该路线全部步骤
return sorted(steps, key=lambda s: s["sequenceNo"]) # 按工序顺序返回
def find_bom_items(world: World, product_id: int) -> list[dict]:
"""取产品默认 BOM 的明细行。"""
bom = next((b for b in world["boms"] if b["productId"] == product_id and b["isDefault"]), None) # 默认 BOM
if not bom: # 无 BOM → 空(视为无料耗)
return []
return [i for i in world["bomItems"] if i["bomId"] == bom["id"]] # 该 BOM 全部明细
def find_product_lines(world: World, product_id: int) -> list[dict]:
"""取产品可用产线配置(priority 越小越优先;仅 ACTIVE 产线,停用产线不参与新排产 MD-01)。"""
active_line_ids = {ln["id"] for ln in world["lines"] if ln.get("status", "ACTIVE") == "ACTIVE"} # 在用产线集
rows = [lp for lp in world["lineProducts"]
if lp["productId"] == product_id and lp["lineId"] in active_line_ids] # 该产品的在用产线配置
return sorted(rows, key=lambda lp: lp["priority"]) # 按优先级排序
def find_workstation_for_operation(world: World, line_id: int, operation_id: int) -> dict | None:
"""在指定产线上找能执行某工序的工位(工位工序配置 ∩ 产线工位;仅 ACTIVE 工位 MD-01)。"""
ws_ids = {wo["workstationId"] for wo in world["workstationOperations"] if wo["operationId"] == operation_id} # 能做该工序的工位集
return next((ws for ws in world["workstations"]
if ws["lineId"] == line_id and ws["id"] in ws_ids
and ws.get("status", "ACTIVE") == "ACTIVE"), None) # 限定产线取第一个在用工位
def _hm_to_minutes(hm: str) -> int:
"""'HH:MM' → 当日分钟数(班次时间换算的基元)。"""
h, m = hm.split(":") # 拆时与分
return int(h) * 60 + int(m) # 换算为分钟
def get_shift_minutes(world: World, shift_id: int) -> int:
"""某班次的有效工作分钟(扣除休息段,与 legacy getShiftMinutes 对齐)。"""
shift = next((s for s in world["shifts"] if s["id"] == shift_id), None) # 定位班次
if not shift: # 班次缺失防御
return 0
total = _hm_to_minutes(shift["endTime"]) - _hm_to_minutes(shift["startTime"]) # 毛时长
for bp in shift.get("breakPeriods", []): # 逐个休息段扣减
total -= _hm_to_minutes(bp["end"]) - _hm_to_minutes(bp["start"])
return max(total, 0) # 不为负
def get_line_shifts(world: World, line_id: int, date_str: str) -> list[dict]:
"""某产线某日的工作班次列表(班次日历 isWorking 过滤)。"""
shift_ids = [sc["shiftId"] for sc in world["shiftCalendar"] # 扫描班次日历
if sc["lineId"] == line_id and sc["date"] == date_str and sc["isWorking"]]
return [s for s in world["shifts"] if s["id"] in shift_ids] # 映射回班次对象
def get_available_minutes(world: World, line_id: int, date_str: str) -> int:
"""某产线某日的总可用分钟(各工作班次有效分钟之和)。"""
return sum(get_shift_minutes(world, s["id"]) for s in get_line_shifts(world, line_id, date_str)) # 累加

36
aps/src/aps/timeutil.py Normal file
View File

@ -0,0 +1,36 @@
# ============================================================
# 日期时间工具(moduleId: server-timeutil, 可重生 ✅)
# 与 legacy aps-frontend/js/data.js 的 $utils 语义一致:
# 字符串格式统一 "YYYY-MM-DD" 与 "YYYY-MM-DD HH:MM"
# ============================================================
from __future__ import annotations # 前向类型引用
from datetime import datetime, timedelta # 标准日期时间类型
def fmt_date(d: datetime) -> str:
"""格式化为日期字符串 YYYY-MM-DD(与 data.js fmtDate 对齐)。"""
return d.strftime("%Y-%m-%d") # 固定零填充格式
def fmt_dt(d: datetime) -> str:
"""格式化为日期时间字符串 YYYY-MM-DD HH:MM(与 data.js fmtDateTime 对齐)。"""
return d.strftime("%Y-%m-%d %H:%M") # 分钟精度(排产粒度 15 分钟足够)
def parse_dt(s: str) -> datetime:
"""解析日期/日期时间字符串(兼容两种格式,与 data.js parseDate 对齐)。"""
if len(s) <= 10: # 仅日期(10 个字符)
return datetime.strptime(s, "%Y-%m-%d") # 解析为当日零点
return datetime.strptime(s, "%Y-%m-%d %H:%M") # 含时分的完整解析
def add_minutes(d: datetime, minutes: float) -> datetime:
"""时间加分钟(与 data.js addMinutes 对齐)。"""
return d + timedelta(minutes=minutes) # timedelta 直接支持浮点分钟
def today0() -> datetime:
"""今天零点(与 data.js today 对齐),作为种子/排产的相对基准。"""
now = datetime.now() # 当前本地时间
return now.replace(hour=0, minute=0, second=0, microsecond=0) # 归零到当日 00:00

49
aps/src/aps/world.py Normal file
View File

@ -0,0 +1,49 @@
"""世界状态:APS 业务数据的读路径入口。
B1 首批按旧系统的形态读 JSON 快照(`data/demo_world.json` 由 `main` 的 `build_demo_world()` 导出,
含 10 物料 / 7 销售订单 / 13 工位 / 180 条班次日历)。B2 拆库后换成 APS 自己的库
(开发期 SQLite、生产 PostgreSQL)——`domain` 里的读函数都吃 `World` 字典,换数据源不动它们。
环境变量 `APS_WORLD_PATH` 可指向别的快照,便于测试与多项目。
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any
# 旧代码里的 World 就是一个 dict:masterdata / orders / master_query 的读函数都按它取数
World = dict[str, Any]
DEFAULT_WORLD_PATH = Path(__file__).resolve().parents[2] / "data" / "demo_world.json"
_cached: World | None = None
_cached_path: str | None = None
def world_path() -> Path:
"""当前使用的世界快照路径。"""
return Path(os.environ.get("APS_WORLD_PATH") or DEFAULT_WORLD_PATH)
def load_world(path: str | Path | None = None) -> World:
"""读入世界快照;同路径只读一次(进程内缓存,测试可用 set_world 覆盖)。"""
global _cached, _cached_path
target = str(path or world_path())
if _cached is not None and _cached_path == target:
return _cached
with open(target, encoding="utf-8") as handle:
data = json.load(handle)
if not isinstance(data, dict):
raise ValueError(f"世界快照必须是 JSON 对象:{target}")
_cached, _cached_path = data, target
return data
def set_world(world: World | None) -> None:
"""测试与本地试验用:直接塞一份内存世界,传 None 清掉缓存。"""
global _cached, _cached_path
_cached = world
_cached_path = None if world is None else "<memory>"

90
aps/tests/test_api.py Normal file
View File

@ -0,0 +1,90 @@
"""B5 首批冒烟:APS 服务能起来、能按契约返回、能力清单与错误分类对得上。
只读能力全部基于 main 的 build_demo_world() 导出的快照(aps/data/demo_world.json),
所以断言里的数量是确定的:7 张销售订单、13 个可用工位。
"""
from __future__ import annotations
from fastapi.testclient import TestClient
from aps.api.app import CAPABILITIES, app, manifest
from aps.world import load_world
client = TestClient(app)
def test_health() -> None:
response = client.get("/api/health")
assert response.status_code == 200
assert response.json()["status"] == "ok"
def test_manifest_shape_matches_platform_contract() -> None:
body = manifest()
assert body["id"] == "aps"
assert body["platformVersion"] == 1, "平台只接受 platformVersion = 1"
assert body["requiresIdentity"] is True, "APS 有账号体系(ADR-0012)"
assert body["auth"]["kind"] in {"oauth", "password-exchange", "apikey"}
assert len(body["capabilities"]) >= 7, "只读能力至少 7 个"
def test_every_capability_is_described_for_the_model() -> None:
"""能力描述规范七要素:id 命名、用途 ≤40 字、写清副作用、声明参数 schema。"""
for spec in manifest()["capabilities"]:
assert spec["id"].startswith("aps."), spec["id"]
assert 0 < len(spec["description"]) <= 40, f"{spec['id']} 用途超 40 字"
assert spec["effect"] in {"read", "draft", "write"}
assert spec["mode"] in {"sync", "async"}
assert spec["idempotency"] in {"none", "key", "natural"}
assert spec["inputSchema"], f"{spec['id']} 缺 inputSchema"
assert spec["errors"], f"{spec['id']} 缺错误分类"
def test_read_capabilities_return_data() -> None:
orders = client.post("/api/capabilities/aps.orders.query", json={"params": {"limit": 5}}).json()
assert orders["status"] == "succeeded"
assert orders["result"]["total"] == 7
capacity = client.post("/api/capabilities/aps.capacity.query", json={"params": {}}).json()
assert capacity["status"] == "succeeded"
assert capacity["result"]["total"] == 13
def test_every_registered_capability_invokes() -> None:
for capability_id in CAPABILITIES:
body = client.post(f"/api/capabilities/{capability_id}", json={"params": {"limit": 2}}).json()
assert body["status"] == "succeeded", f"{capability_id} 调用失败:{body}"
assert body["executionId"].startswith("exec-aps-"), "回执要带 executionId"
def test_unknown_capability_is_invalid_argument() -> None:
body = client.post("/api/capabilities/aps.not-exist", json={"params": {}}).json()
assert body["status"] == "failed"
assert body["errorKind"] == "invalid_argument"
def test_bad_params_map_to_invalid_argument() -> None:
body = client.post("/api/capabilities/aps.orders.query", json={"params": {"limit": "不是数字"}}).json()
assert body["status"] == "failed"
assert body["errorKind"] == "invalid_argument"
def test_execution_id_is_echoed_back() -> None:
body = client.post(
"/api/capabilities/aps.orders.query",
json={"params": {}, "executionId": "exec-from-platform-1"},
).json()
assert body["executionId"] == "exec-from-platform-1", "平台给的 executionId 必须原样带回"
def test_demo_world_is_loaded_from_snapshot() -> None:
world = load_world()
assert len(world["salesOrders"]) == 7
assert len(world["workstations"]) == 13