320 lines
12 KiB
Python
320 lines
12 KiB
Python
# ============================================================
|
||
# SOP → 约束规则包（moduleId: domain-sop-rules, IND-02 / C13，可重生 ✅）
|
||
# 知识库 SOP 编译为可应用的约束/换型策略补丁；预览 P0，应用 P2
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from server.timeutil import fmt_dt
|
||
|
||
World = dict[str, Any]
|
||
|
||
|
||
def _extract_minutes(text: str, *patterns: str, default: float | None = None) -> float | None:
|
||
for pat in patterns:
|
||
m = re.search(pat, text)
|
||
if m:
|
||
try:
|
||
return float(m.group(1))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
return default
|
||
|
||
|
||
def list_compilable_sops(kb_assets: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""列出可编译的 SOP（kind=sop 且已审批）。"""
|
||
out = []
|
||
for a in kb_assets:
|
||
if a.get("kind") != "sop" or not a.get("approved", True):
|
||
continue
|
||
pack = compile_sop_asset(a)
|
||
out.append({
|
||
"assetId": a.get("assetId"),
|
||
"title": a.get("title"),
|
||
"tags": a.get("tags") or [],
|
||
"effectCount": len(pack.get("effects") or []),
|
||
"compilable": bool(pack.get("effects")),
|
||
"summary": pack.get("summary") or "",
|
||
})
|
||
return out
|
||
|
||
|
||
def compile_sop_asset(asset: dict[str, Any]) -> dict[str, Any]:
|
||
"""
|
||
将单条 SOP 编译为规则包预览（确定性模板+关键词，非 LLM）。
|
||
返回 {packId, title, source, effects[], summary, warnings[]}
|
||
"""
|
||
title = str(asset.get("title") or "")
|
||
content = str(asset.get("content") or "")
|
||
tags = [str(t) for t in (asset.get("tags") or [])]
|
||
blob = f"{title}\n{content}\n{' '.join(tags)}"
|
||
effects: list[dict[str, Any]] = []
|
||
warnings: list[str] = []
|
||
|
||
# —— 换线标准 SOP ——
|
||
if re.search(r"换线|换型", blob):
|
||
cross = _extract_minutes(
|
||
content,
|
||
r"跨产品族换线\s*(\d+)\s*分钟",
|
||
r"跨族[^\d]{0,6}(\d+)\s*分",
|
||
default=60.0,
|
||
)
|
||
std = _extract_minutes(
|
||
content,
|
||
r"换线作业标准\s*(\d+)\s*分钟",
|
||
r"标准\s*(\d+)\s*分钟",
|
||
default=30.0,
|
||
)
|
||
effects.append({
|
||
"op": "constraint.set",
|
||
"id": "C10_changeover",
|
||
"enabled": True,
|
||
"kind": "soft",
|
||
"rationale": "SOP 要求顺序相关换型受控",
|
||
})
|
||
effects.append({
|
||
"op": "changeover.policy",
|
||
"defaultCrossFamilyMin": cross,
|
||
"standardSetupMin": std,
|
||
"noChangeoverAfter": "16:00" if re.search(r"16:00|十六点|夜班", content) else None,
|
||
"rationale": f"跨族默认 {cross:.0f} 分；标准 {std:.0f} 分"
|
||
+ ("；16:00 后不安排换线" if re.search(r"16:00", content) else ""),
|
||
})
|
||
effects.append({
|
||
"op": "strategy.suggest",
|
||
"strategy": "CHANGEOVER_MIN",
|
||
"rationale": "同产品族连续生产优先",
|
||
})
|
||
|
||
# —— 齐套 / 库存 SOP ——
|
||
if re.search(r"齐套|缺料|库存", blob) and not re.search(r"换线", title):
|
||
effects.append({
|
||
"op": "constraint.set",
|
||
"id": "C6_material_kit",
|
||
"enabled": True,
|
||
"kind": "soft",
|
||
"rationale": "SOP：缺料须可见，不默认阻断草案",
|
||
})
|
||
|
||
# —— 插单审批 SOP ——
|
||
if re.search(r"插单|急单", blob):
|
||
effects.append({
|
||
"op": "process.note",
|
||
"text": "VIP 插单须影响评估 + 主管审批（本系统 rush.apply 已是 P2）",
|
||
"rationale": "与门禁对齐，无需改引擎开关",
|
||
})
|
||
|
||
if not effects:
|
||
warnings.append("未识别可落地的约束模板；可扩展编译器或改用约束配置中心手工调整")
|
||
|
||
# C13 标记：有可应用效果时启用
|
||
if any(e.get("op") in ("constraint.set", "changeover.policy") for e in effects):
|
||
effects.append({
|
||
"op": "constraint.set",
|
||
"id": "C13_sop",
|
||
"enabled": True,
|
||
"kind": "soft",
|
||
"params": {
|
||
"sourceAssetId": asset.get("assetId"),
|
||
"sourceTitle": title,
|
||
},
|
||
"rationale": "标记本剖面含 SOP 编译规则",
|
||
})
|
||
|
||
summary_bits = []
|
||
for e in effects:
|
||
if e["op"] == "constraint.set":
|
||
summary_bits.append(f"{e['id']}→{'开' if e.get('enabled') else '关'}/{e.get('kind', '')}")
|
||
elif e["op"] == "changeover.policy":
|
||
summary_bits.append(f"跨族换型默认{e.get('defaultCrossFamilyMin')}分")
|
||
elif e["op"] == "strategy.suggest":
|
||
summary_bits.append(f"建议策略{e.get('strategy')}")
|
||
elif e["op"] == "process.note":
|
||
summary_bits.append("流程提示")
|
||
|
||
return {
|
||
"packId": f"pack-{asset.get('assetId') or uuid.uuid4().hex[:8]}",
|
||
"title": title,
|
||
"source": {
|
||
"assetId": asset.get("assetId"),
|
||
"kind": asset.get("kind"),
|
||
"version": asset.get("version"),
|
||
"title": title,
|
||
},
|
||
"effects": effects,
|
||
"summary": "；".join(summary_bits) if summary_bits else "无可应用效果",
|
||
"warnings": warnings,
|
||
}
|
||
|
||
|
||
def compile_sop_by_asset(kb_assets: list[dict[str, Any]], asset_id: str | None = None) -> dict[str, Any]:
|
||
"""按 Pi 明确给出的 assetId 精确选择并编译 SOP。"""
|
||
requested = str(asset_id or "").strip()
|
||
sops = [a for a in kb_assets if a.get("kind") == "sop" and a.get("approved", True)]
|
||
available = [
|
||
{"assetId": a.get("assetId"), "title": a.get("title")}
|
||
for a in sops[:12]
|
||
]
|
||
if not requested:
|
||
return {
|
||
"ok": False,
|
||
"error": "请提供明确的 assetId，不能根据原始问句自动选择 SOP。",
|
||
"available": available,
|
||
}
|
||
hit = next(
|
||
(a for a in sops if str(a.get("assetId") or "") == requested),
|
||
None,
|
||
)
|
||
if hit is None:
|
||
return {
|
||
"ok": False,
|
||
"error": f"未找到可编译的 SOP：assetId={requested}。",
|
||
"available": available,
|
||
}
|
||
return {"ok": True, "pack": compile_sop_asset(hit)}
|
||
|
||
|
||
def compile_sop_by_query(kb_assets: list[dict[str, Any]], query: str | None = None) -> dict[str, Any]:
|
||
"""兼容旧调用；禁止用原始问句选择 SOP，调用方必须改用 assetId。"""
|
||
del query
|
||
sops = [a for a in kb_assets if a.get("kind") == "sop" and a.get("approved", True)]
|
||
return {
|
||
"ok": False,
|
||
"error": "请提供明确的 assetId，不能根据原始问句自动选择 SOP。",
|
||
"available": [
|
||
{"assetId": a.get("assetId"), "title": a.get("title")}
|
||
for a in sops[:12]
|
||
],
|
||
}
|
||
|
||
|
||
def confirmation_for_sop_apply(pack: dict[str, Any]) -> tuple[str, list[str]]:
|
||
title = f"应用 SOP 规则包：{pack.get('title') or pack.get('packId')}"
|
||
lines = [pack.get("summary") or "应用编译效果"]
|
||
src = (pack.get("source") or {}).get("title")
|
||
if src:
|
||
lines.append(f"出处：{src}")
|
||
for e in pack.get("effects") or []:
|
||
if e.get("rationale"):
|
||
lines.append(f"· {e['op']}: {e['rationale']}")
|
||
lines.append("P2 写主干；执行前自动建档")
|
||
for w in pack.get("warnings") or []:
|
||
lines.append(f"⚠ {w}")
|
||
return title, lines
|
||
|
||
|
||
def apply_sop_pack(world: World, pack: dict[str, Any]) -> dict[str, Any]:
|
||
"""把规则包写入世界（约束剖面 / 换型策略 / 规则包登记）。"""
|
||
from server.aps_domain.constraints import apply_profile_save
|
||
|
||
effects = pack.get("effects") or []
|
||
constraint_payload: dict[str, Any] = {"constraints": {}}
|
||
changeover_policy: dict[str, Any] = dict(world.get("changeoverPolicy") or {})
|
||
notes: list[str] = []
|
||
applied: list[str] = []
|
||
|
||
for e in effects:
|
||
op = e.get("op")
|
||
if op == "constraint.set":
|
||
cid = str(e.get("id") or "")
|
||
if not cid:
|
||
continue
|
||
patch: dict[str, Any] = {}
|
||
if "enabled" in e:
|
||
patch["enabled"] = bool(e["enabled"])
|
||
if e.get("kind") in ("hard", "soft"):
|
||
patch["kind"] = e["kind"]
|
||
if isinstance(e.get("params"), dict):
|
||
patch["params"] = e["params"]
|
||
constraint_payload["constraints"][cid] = patch
|
||
applied.append(f"constraint:{cid}")
|
||
elif op == "changeover.policy":
|
||
if e.get("defaultCrossFamilyMin") is not None:
|
||
changeover_policy["defaultCrossFamilyMin"] = float(e["defaultCrossFamilyMin"])
|
||
if e.get("standardSetupMin") is not None:
|
||
changeover_policy["standardSetupMin"] = float(e["standardSetupMin"])
|
||
if e.get("noChangeoverAfter"):
|
||
changeover_policy["noChangeoverAfter"] = str(e["noChangeoverAfter"])
|
||
applied.append("changeover.policy")
|
||
elif op == "strategy.suggest":
|
||
notes.append(f"建议后续试排使用策略 {e.get('strategy')}")
|
||
applied.append(f"suggest:{e.get('strategy')}")
|
||
elif op == "process.note":
|
||
notes.append(str(e.get("text") or ""))
|
||
applied.append("process.note")
|
||
|
||
if constraint_payload["constraints"]:
|
||
apply_profile_save(world, constraint_payload)
|
||
|
||
if any(e.get("op") == "changeover.policy" for e in effects):
|
||
world["changeoverPolicy"] = changeover_policy
|
||
|
||
packs = world.setdefault("rulePacks", [])
|
||
rec = {
|
||
"packId": pack.get("packId") or f"pack-{uuid.uuid4().hex[:8]}",
|
||
"title": pack.get("title"),
|
||
"source": pack.get("source"),
|
||
"appliedAt": fmt_dt(datetime.now()),
|
||
"effects": effects,
|
||
"notes": notes,
|
||
}
|
||
# 同 asset 再应用则替换
|
||
src_id = (pack.get("source") or {}).get("assetId")
|
||
packs[:] = [p for p in packs if (p.get("source") or {}).get("assetId") != src_id]
|
||
packs.append(rec)
|
||
|
||
return {
|
||
"packId": rec["packId"],
|
||
"applied": applied,
|
||
"notes": notes,
|
||
"changeoverPolicy": changeover_policy,
|
||
"rulePackCount": len(packs),
|
||
}
|
||
|
||
|
||
def get_changeover_policy(world: World) -> dict[str, Any]:
|
||
pol = world.get("changeoverPolicy") or {}
|
||
return {
|
||
"defaultCrossFamilyMin": float(pol.get("defaultCrossFamilyMin") or 30),
|
||
"standardSetupMin": float(pol.get("standardSetupMin") or 30),
|
||
"noChangeoverAfter": pol.get("noChangeoverAfter"),
|
||
}
|
||
|
||
|
||
def sop_as_block(pack: dict[str, Any]) -> Any:
|
||
from server.contracts import UIBlock
|
||
return UIBlock(
|
||
blockId=f"sop-pack-{uuid.uuid4().hex[:8]}",
|
||
type="report",
|
||
props={
|
||
"title": f"SOP 编译：{pack.get('title')}",
|
||
"reportType": "sop-rule-pack",
|
||
"markdown": _pack_markdown(pack),
|
||
"pack": pack,
|
||
},
|
||
)
|
||
|
||
|
||
def _pack_markdown(pack: dict[str, Any]) -> str:
|
||
lines = [
|
||
f"# {pack.get('title')}",
|
||
"",
|
||
f"**摘要**：{pack.get('summary')}",
|
||
"",
|
||
"## 效果",
|
||
]
|
||
for e in pack.get("effects") or []:
|
||
lines.append(f"- `{e.get('op')}` — {e.get('rationale') or e}")
|
||
if pack.get("warnings"):
|
||
lines.append("")
|
||
lines.append("## 警告")
|
||
for w in pack["warnings"]:
|
||
lines.append(f"- {w}")
|
||
src = pack.get("source") or {}
|
||
lines += ["", f"出处：{src.get('title')}（{src.get('assetId')}）"]
|
||
return "\n".join(lines)
|