332 lines
15 KiB
Python
332 lines
15 KiB
Python
# ============================================================
|
||
# 主动引导（moduleId: domain-guidance, 可重生 ✅）
|
||
# AG-07：按世界状态给出下一步建议（空态 / 冲突高 / 知识未命中）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
World = dict[str, Any]
|
||
|
||
|
||
def build_guidance(world: World, *, context: str | None = None) -> dict[str, Any]:
|
||
"""根据世界状态产出信号与可点击下一步建议。
|
||
|
||
context: 可选强化场景（knowledge_miss / high_conflict / empty_schedule）
|
||
"""
|
||
signals: list[dict[str, Any]] = []
|
||
suggestions: list[dict[str, Any]] = []
|
||
|
||
versions = world.get("scheduleVersions") or []
|
||
flex_versions = world.get("flexScheduleVersions") or []
|
||
conflicts = [c for c in (world.get("conflicts") or []) if c.get("status") != "RESOLVED"]
|
||
flex_conflicts = [c for c in (world.get("flexConflicts") or []) if c.get("status") != "RESOLVED"]
|
||
open_conflicts = len(conflicts) + len(flex_conflicts)
|
||
latest = versions[-1] if versions else None
|
||
latest_flex = flex_versions[-1] if flex_versions else None
|
||
current = latest_flex or latest
|
||
draft = current and current.get("status") == "DRAFT"
|
||
flex_orders = world.get("flexOrders") or []
|
||
|
||
# 没有现成方案时，排产齐备度优先于“甘特为空”。零订单或零可排订单
|
||
# 只能引导补数，不能给用户任何试排入口。
|
||
if not current:
|
||
from server.aps_domain.readiness import check_readiness
|
||
|
||
readiness = check_readiness(world)
|
||
readiness_summary = readiness.get("summary") or {}
|
||
if not readiness_summary.get("total") or not readiness_summary.get("ready"):
|
||
guide = scheduling_data_guide(world)
|
||
guide["context"] = context or guide.get("context")
|
||
return guide
|
||
|
||
from server.importers.workbook_profiles import has_adoption_flow
|
||
|
||
if has_adoption_flow((world.get("planningContext") or {}).get("sourceProfile")):
|
||
from server.aps_domain.readiness import check_readiness
|
||
|
||
readiness = check_readiness(world)
|
||
pending = readiness.get("summary") or {}
|
||
if pending.get("blocked", 0) or readiness.get("globalIssues"):
|
||
return {
|
||
"signals": [{"id": "master_review", "severity": "high",
|
||
"title": "排产资料仍有待核对信息",
|
||
"reason": "资料已采用，请先处理人员、在制或供料问题，再生成可靠方案。"}],
|
||
"suggestions": [{"id": "review_missing", "label": "查看需要补充的信息",
|
||
"command": "排产还缺什么", "priority": 20,
|
||
"reason": "按订单列出实际阻断项"},
|
||
{"id": "wizard", "label": "带我补齐并排产",
|
||
"command": "带我排一版", "priority": 15,
|
||
"reason": "按阻断项逐步补齐，不跳过缺失数据"}],
|
||
"hint": "资料已采用，请先核对影响排产的问题；系统会保留你在主数据中的修改。",
|
||
"context": context,
|
||
"stats": {"scheduleVersions": len(versions), "flexScheduleVersions": len(flex_versions),
|
||
"blockedOrders": int(pending.get("blocked") or 0), "openConflicts": open_conflicts},
|
||
}
|
||
|
||
# ---- 信号检测 ----
|
||
if not versions and not flex_versions:
|
||
signals.append({
|
||
"id": "no_version", "severity": "high",
|
||
"title": "还没有排产方案",
|
||
"reason": "甘特为空，需先试排生成草稿。",
|
||
})
|
||
suggestions.append({
|
||
"id": "run_delivery", "label": "生成交期优先试排方案",
|
||
"command": "生成一版交期优先试排方案", "priority": 10,
|
||
"reason": "无版本时空态首选",
|
||
})
|
||
suggestions.append({
|
||
"id": "scenario_compare", "label": "多策略方案对比",
|
||
"command": "多策略方案对比", "priority": 8,
|
||
"reason": "不确定策略时先沙盒对比",
|
||
})
|
||
|
||
conflict_n = open_conflicts
|
||
if latest:
|
||
conflict_n = max(conflict_n, int(latest.get("conflictCount") or 0))
|
||
if latest_flex:
|
||
conflict_n = max(conflict_n, int(latest_flex.get("conflictCount") or 0))
|
||
|
||
if conflict_n >= 3 or context == "high_conflict":
|
||
signals.append({
|
||
"id": "high_conflict", "severity": "high" if conflict_n >= 5 else "medium",
|
||
"title": f"冲突偏高（约 {conflict_n} 项）",
|
||
"reason": "建议先看冲突中心，或换策略对比后再发布。",
|
||
})
|
||
suggestions.append({
|
||
"id": "view_conflicts", "label": "查看冲突",
|
||
"command": "查看冲突", "priority": 20,
|
||
"reason": "冲突高时优先定位",
|
||
})
|
||
suggestions.append({
|
||
"id": "compare_strategies", "label": "多策略方案对比",
|
||
"command": "多策略方案对比", "priority": 18,
|
||
"reason": "用沙盒评估缓解冲突的策略",
|
||
})
|
||
|
||
if current:
|
||
suggestions.extend([
|
||
{"id": "view_schedule", "label": "查看排产结果",
|
||
"command": "查看排产结果",
|
||
"priority": 15, "reason": "先检查已经生成的方案"},
|
||
{"id": "export_schedule", "label": "导出排产表",
|
||
"command": "生成排产方案报告", "priority": 7,
|
||
"reason": "导出当前方案供核对"},
|
||
])
|
||
if draft and current:
|
||
trial = current.get("trialOnly") is True or current.get("productionReady") is False
|
||
signals.append({
|
||
"id": "draft_ready", "severity": "low",
|
||
"title": f"已生成方案 {current.get('versionNo') or ''}",
|
||
"reason": ("这是试排方案，尚未下发到车间。请先核对订单、设备和交期。"
|
||
if trial else "方案已生成。请先检查交期和安排，再决定是否发布。"),
|
||
})
|
||
if not trial and conflict_n == 0:
|
||
suggestions.append({
|
||
"id": "publish", "label": "确认发布方案",
|
||
"command": "发布当前排产版本", "priority": 6,
|
||
"reason": "草稿已生成，发布仍需确认",
|
||
})
|
||
elif current:
|
||
signals.append({
|
||
"id": "schedule_exists", "severity": "low",
|
||
"title": f"已有排产方案 {current.get('versionNo') or ''}",
|
||
"reason": "先查看现有安排；如有插单、缺料或设备变化，再调整方案。",
|
||
})
|
||
|
||
if flex_orders and not flex_versions and not versions:
|
||
signals.append({
|
||
"id": "no_flex", "severity": "medium",
|
||
"title": "订单已准备好，等待生成方案",
|
||
"reason": "先核对订单和设备，再生成试排方案。",
|
||
})
|
||
suggestions.append({
|
||
"id": "flex_run", "label": "生成排产方案",
|
||
"command": "生成一版柔性排产方案", "priority": 12,
|
||
"reason": "柔性轨空态",
|
||
})
|
||
|
||
if context == "knowledge_miss":
|
||
signals.append({
|
||
"id": "knowledge_miss", "severity": "medium",
|
||
"title": "知识未命中",
|
||
"reason": "可换说法，或浏览已有知识资产。",
|
||
})
|
||
suggestions.append({
|
||
"id": "kb_list", "label": "查看知识库清单",
|
||
"command": "知识库", "priority": 25,
|
||
"reason": "未命中时的下一步",
|
||
})
|
||
suggestions.append({
|
||
"id": "kb_machining", "label": "机加工工艺怎么生成",
|
||
"command": "机加工工艺怎么生成", "priority": 24,
|
||
"reason": "机械加工工艺路线总则",
|
||
})
|
||
suggestions.append({
|
||
"id": "kb_sop", "label": "问主数据录入顺序",
|
||
"command": "主数据录入顺序有什么规定", "priority": 22,
|
||
"reason": "常见 SOP 样例",
|
||
})
|
||
|
||
# 去重（按 command 保留最高 priority）
|
||
by_cmd: dict[str, dict[str, Any]] = {}
|
||
for s in suggestions:
|
||
prev = by_cmd.get(s["command"])
|
||
if prev is None or s["priority"] > prev["priority"]:
|
||
by_cmd[s["command"]] = s
|
||
ranked = sorted(by_cmd.values(), key=lambda x: -x["priority"])[:5]
|
||
|
||
if not ranked:
|
||
ranked = [{
|
||
"id": "help", "label": "查看可用操作",
|
||
"command": "帮助", "priority": 1,
|
||
"reason": "默认兜底",
|
||
}]
|
||
|
||
hint = "根据当前状态，建议下一步："
|
||
if signals:
|
||
hint = signals[0]["reason"]
|
||
|
||
return {
|
||
"signals": signals,
|
||
"suggestions": ranked,
|
||
"hint": hint,
|
||
"context": context,
|
||
"stats": {
|
||
"scheduleVersions": len(versions),
|
||
"flexScheduleVersions": len(flex_versions),
|
||
"openConflicts": open_conflicts,
|
||
"conflictScore": conflict_n,
|
||
},
|
||
}
|
||
|
||
|
||
def guidance_ui_block(world: World, *, context: str | None = None) -> dict[str, Any]:
|
||
"""构造 guidance UI 块 props（由调用方包进 UIBlock）。"""
|
||
return build_guidance(world, context=context)
|
||
|
||
|
||
def _guide_step(step_id: str, title: str, detail: str, how: str,
|
||
actions: list[dict[str, str]]) -> dict[str, Any]:
|
||
return {
|
||
"id": step_id,
|
||
"title": title,
|
||
"detail": detail,
|
||
"how": how,
|
||
"actions": actions,
|
||
}
|
||
|
||
|
||
def scheduling_data_guide(world: World) -> dict[str, Any]:
|
||
"""排产缺数据的傻瓜式引导：按顺序列出缺什么、为什么、怎么补。
|
||
|
||
输出与 guidance 块同构（steps + suggestions + stats），前端可按步骤
|
||
一键回填口令，非专业人员跟着做即可。
|
||
"""
|
||
from server.aps_domain.readiness import check_readiness
|
||
|
||
report = check_readiness(world)
|
||
s = report["summary"]
|
||
orders = [o for o in (world.get("flexOrders") or [])
|
||
if (o.get("status") or "") not in ("DONE", "CANCELLED")]
|
||
materials = world.get("flexMaterials") or []
|
||
routings = world.get("flexRoutings") or []
|
||
equipment = [e for e in (world.get("flexEquipment") or []) if e.get("status") == "RUNNING"]
|
||
calendar = world.get("flexCalendar") or []
|
||
|
||
issue_types = {i["type"] for r in report["orders"] for i in r["issues"]}
|
||
steps: list[dict[str, Any]] = []
|
||
|
||
if not orders:
|
||
steps.append(_guide_step(
|
||
"orders", "先告诉我排哪些单",
|
||
"现在还没有待排订单。",
|
||
"直接发一张订单表（Excel/CSV），或在对话里说「新建订单 产品A 100件 8月20日交」；也可以让我带你一步一步录。",
|
||
[{"label": "带我录订单并排产", "command": "带我排一版"}],
|
||
))
|
||
if not materials or "NO_PRODUCT" in issue_types:
|
||
missing_n = sum(1 for r in report["orders"]
|
||
if any(i["type"] == "NO_PRODUCT" for i in r["issues"]))
|
||
steps.append(_guide_step(
|
||
"products", "把要做的产品告诉系统",
|
||
"还没有产品/材料档案" if not materials else f"有 {missing_n} 张订单的产品查不到档案。",
|
||
"发一份产品/材料名录，或说「帮我分析项目文件数据」，我自动从表格里补。",
|
||
[{"label": "带我排一版", "command": "带我排一版"}],
|
||
))
|
||
if not routings or "NO_ROUTING" in issue_types:
|
||
missing_n = sum(1 for r in report["orders"]
|
||
if any(i["type"] == "NO_ROUTING" for i in r["issues"]))
|
||
steps.append(_guide_step(
|
||
"routing", "产品怎么做还没定",
|
||
"还没有工艺路线" if not routings else f"有 {missing_n} 张订单的产品没有工艺路线。",
|
||
"说「带我排一版」，我会先推荐常用工艺模板，你挑一个或改一下就行。",
|
||
[{"label": "让我选工艺模板", "command": "带我排一版"}],
|
||
))
|
||
if not equipment or "NO_CAPABLE_EQUIPMENT" in issue_types:
|
||
missing_n = sum(1 for r in report["orders"]
|
||
if any(i["type"] == "NO_CAPABLE_EQUIPMENT" for i in r["issues"]))
|
||
steps.append(_guide_step(
|
||
"equipment", "还缺能干活的设备/工序能力",
|
||
"没有运行中的设备" if not equipment else f"有 {missing_n} 张订单的工序找不到能做它的设备。",
|
||
"发设备台账，或在主数据页给设备勾上会做的工序；拿不准就说「带我排一版」，我告诉你缺在哪。",
|
||
[{"label": "带我排一版", "command": "带我排一版"}],
|
||
))
|
||
if not calendar:
|
||
steps.append(_guide_step(
|
||
"calendar", "还没告诉系统几点上班",
|
||
"没有班次日历，系统不知道每天几点开工、一周开几天。",
|
||
"说「默认日历」用常见班制（周一至周五 8:00-17:00），或告诉我班制，比如「两班 8点到23点 周一到周六」。",
|
||
[{"label": "带我排一版", "command": "带我排一版"}],
|
||
))
|
||
if s.get("timePending"):
|
||
steps.append(_guide_step(
|
||
"time", "有几道工序工时还没填",
|
||
f"还有 {s['timePending']} 道工序没有单件工时，硬排会失真。",
|
||
"说「带我排一版」，我会一条一条问你「这步每件多少分钟」，你回数字就行。",
|
||
[{"label": "逐条补工时", "command": "带我排一版"}],
|
||
))
|
||
if "MATERIAL_GAP" in issue_types:
|
||
gap_n = sum(1 for r in report["orders"]
|
||
if any(i["type"] == "MATERIAL_GAP" for i in r["issues"]))
|
||
steps.append(_guide_step(
|
||
"material", "部分物料可能不够",
|
||
f"有 {gap_n} 张订单的 BOM 物料库存/在途不足（提醒，不影响先试排）。",
|
||
"补库存表或采购到货信息；不补也能先试排，结果里会标出来。",
|
||
[{"label": "先试排看看", "command": "直接排一版"}],
|
||
))
|
||
|
||
suggestions = [{
|
||
"id": "wizard", "label": "带我补齐并排产", "command": "带我排一版",
|
||
"priority": 30, "reason": "一步步跟着补最稳",
|
||
}]
|
||
if not steps:
|
||
suggestions = [{
|
||
"id": "run", "label": "直接排一版", "command": "直接排一版",
|
||
"priority": 30, "reason": "数据基本齐了",
|
||
}, {
|
||
"id": "recheck", "label": "再看还缺什么", "command": "排产还缺什么",
|
||
"priority": 25, "reason": "需要时可复查",
|
||
}]
|
||
|
||
return {
|
||
"mode": "ready" if not steps else "data-missing",
|
||
"hint": ("数据基本齐了，可以直接试排。" if not steps
|
||
else f"要开排还差 {len(steps)} 样东西，我按顺序列好了，补一样少一样。"),
|
||
"signals": [{
|
||
"id": "data-missing", "severity": "high",
|
||
"title": f"还差 {len(steps)} 样数据",
|
||
"reason": "补齐后才能排出可信方案。",
|
||
}] if steps else [],
|
||
"steps": steps,
|
||
"suggestions": suggestions,
|
||
"context": "data-missing" if steps else "ready",
|
||
"stats": {
|
||
"orders": len(orders), "materials": len(materials),
|
||
"routings": len(routings), "equipment": len(equipment),
|
||
"calendar": len(calendar),
|
||
"ready": s.get("ready"), "blocked": s.get("blocked"),
|
||
"timePending": s.get("timePending"),
|
||
},
|
||
}
|