353 lines
15 KiB
Python
353 lines
15 KiB
Python
# ============================================================
|
||
# 计划员约束规则库（moduleId: domain-constraint-rules, 可重生 ✅）
|
||
# 默认通用规则 = constraints.py 的 C1..C13 内建目录（只读基线，本模块不改它）
|
||
# 本模块只管理「计划员用自然语言提出、经 P2 确认后扩充的规则」：
|
||
# * 按类别归档（沿用 GROUP_LABELS 的 8 类 + 其他）
|
||
# * 按范围归档（仅本项目 / 全厂）
|
||
# * 没有接入排产引擎的规则如实标注，不假装生效
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import tempfile
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from server.aps_domain.constraints import (
|
||
DEFAULT_CONSTRAINT_DEFS,
|
||
GROUP_LABELS,
|
||
get_constraint,
|
||
get_constraint_profile,
|
||
)
|
||
|
||
World = dict[str, Any]
|
||
|
||
RULE_CATEGORIES: dict[str, str] = {**GROUP_LABELS, "other": "其他"}
|
||
RULE_SCOPES: dict[str, str] = {"project": "仅本项目", "tenant": "全厂"}
|
||
RULE_KINDS: dict[str, str] = {"hard": "必须满足", "soft": "尽量满足"}
|
||
RULE_OPS: tuple[str, ...] = ("add", "enable", "disable", "delete")
|
||
|
||
_ID_RE = re.compile(r"^R-(\d+)$")
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
|
||
|
||
def _safe_tenant(tenant_uuid: str | None) -> str:
|
||
raw = (tenant_uuid or "platform").strip() or "platform"
|
||
return re.sub(r"[^\w\-]+", "_", raw)[:64] or "platform"
|
||
|
||
|
||
def tenant_rules_path(tenant_uuid: str | None) -> Path:
|
||
"""全厂规则落盘位置：与 world.json 同一数据根（尊重 APS_HOME / APS_DATA_DIR）。"""
|
||
from server.aps_home import data_dir
|
||
|
||
tenant = _safe_tenant(tenant_uuid)
|
||
if tenant == "platform":
|
||
return data_dir() / "constraint-rules.json"
|
||
return data_dir() / "tenants" / tenant / "constraint-rules.json"
|
||
|
||
|
||
def load_tenant_rules(tenant_uuid: str | None) -> list[dict[str, Any]]:
|
||
"""读取全厂规则（文件不存在或损坏时返回空列表，不抛错）。"""
|
||
path = tenant_rules_path(tenant_uuid)
|
||
if not path.exists():
|
||
return []
|
||
try:
|
||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
return []
|
||
rules = raw.get("rules") if isinstance(raw, dict) else raw
|
||
if not isinstance(rules, list):
|
||
return []
|
||
return [r for r in rules if isinstance(r, dict)]
|
||
|
||
|
||
def save_tenant_rules(tenant_uuid: str | None, rules: list[dict[str, Any]]) -> str:
|
||
"""原子写入全厂规则文件（tmp + os.replace，与 WorldStore 同一写盘策略）。"""
|
||
path = tenant_rules_path(tenant_uuid)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
body = json.dumps(
|
||
{"version": 1, "scope": "tenant", "rules": rules},
|
||
ensure_ascii=False, indent=2,
|
||
)
|
||
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".constraint-rules-", suffix=".tmp")
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
||
fh.write(body + "\n")
|
||
os.replace(tmp, path)
|
||
except BaseException:
|
||
try:
|
||
os.unlink(tmp)
|
||
except OSError:
|
||
pass
|
||
raise
|
||
return str(path)
|
||
|
||
|
||
def project_rules(world: World) -> list[dict[str, Any]]:
|
||
raw = world.get("constraintRules")
|
||
if not isinstance(raw, list):
|
||
return []
|
||
return [r for r in raw if isinstance(r, dict)]
|
||
|
||
|
||
def all_rules(world: World, tenant_uuid: str | None = None) -> list[dict[str, Any]]:
|
||
"""全厂规则 + 本项目规则，统一补 scope 字段（缺省按所属容器判定）。"""
|
||
out: list[dict[str, Any]] = []
|
||
for r in load_tenant_rules(tenant_uuid):
|
||
out.append({**r, "scope": r.get("scope") or "tenant"})
|
||
for r in project_rules(world):
|
||
out.append({**r, "scope": r.get("scope") or "project"})
|
||
return out
|
||
|
||
|
||
def next_rule_id(rules: list[dict[str, Any]]) -> str:
|
||
top = 0
|
||
for r in rules:
|
||
m = _ID_RE.match(str(r.get("id") or ""))
|
||
if m:
|
||
top = max(top, int(m.group(1)))
|
||
return f"R-{top + 1:03d}"
|
||
|
||
|
||
def rule_effect(world: World, rule: dict[str, Any]) -> dict[str, Any]:
|
||
"""如实说明这条规则当前靠谁执行；没有接入引擎就直说。"""
|
||
bound = str(rule.get("boundConstraintId") or "").strip()
|
||
if not bound:
|
||
return {"bound": False, "enforced": False, "text": "已记录，暂未接入排产引擎"}
|
||
c = get_constraint(world, bound)
|
||
if not c:
|
||
return {"bound": True, "enforced": False,
|
||
"text": f"绑定的内置规则 {bound} 不在当前约束目录，暂未执行"}
|
||
state = "已启用" if c.get("enabled", True) else "已关闭"
|
||
kind_label = "硬约束" if c.get("kind") == "hard" else "软约束"
|
||
enforced = bool(c.get("enabled", True))
|
||
if not enforced:
|
||
return {"bound": True, "enforced": False,
|
||
"text": f"内置规则《{c['name']}》当前已关闭，本条暂不生效"}
|
||
return {"bound": True, "enforced": True,
|
||
"text": f"由内置规则《{c['name']}》执行（{state}／{kind_label}）"}
|
||
|
||
|
||
def _default_ids() -> set[str]:
|
||
return {str(d["id"]) for d in DEFAULT_CONSTRAINT_DEFS}
|
||
|
||
|
||
def normalize_rule_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
"""校验规则库载荷；只接受计划员语言字段，不接受 JSON 片段。"""
|
||
raw = payload if isinstance(payload, dict) else {}
|
||
op = str(raw.get("op") or "add").strip().lower() or "add"
|
||
if op not in RULE_OPS:
|
||
raise ValueError("规则操作只能是 新增/启用/停用/删除")
|
||
if op != "add":
|
||
rule_id = str(raw.get("ruleId") or raw.get("id") or "").strip()
|
||
if not rule_id:
|
||
raise ValueError("请指定要操作的规则")
|
||
return {"op": op, "ruleId": rule_id}
|
||
name = str(raw.get("name") or raw.get("text") or "").strip()
|
||
if not name:
|
||
raise ValueError("规则内容不能为空，请用一句话描述")
|
||
if len(name) > 40:
|
||
raise ValueError("规则内容请控制在 40 字以内，说清「什么时候、什么不能排」即可")
|
||
category = str(raw.get("category") or "other").strip().lower() or "other"
|
||
if category not in RULE_CATEGORIES:
|
||
raise ValueError(f"类别只能是：{'、'.join(RULE_CATEGORIES.values())}")
|
||
scope = str(raw.get("scope") or "project").strip().lower() or "project"
|
||
if scope not in RULE_SCOPES:
|
||
raise ValueError("适用范围只能是：仅本项目 / 全厂")
|
||
kind = str(raw.get("kind") or "soft").strip().lower() or "soft"
|
||
if kind not in RULE_KINDS:
|
||
raise ValueError("强度只能是：必须满足 / 尽量满足")
|
||
bound = str(raw.get("boundConstraintId") or "").strip()
|
||
if bound and bound not in _default_ids():
|
||
raise ValueError(f"绑定的内置规则 {bound} 不存在，请留空或改用现有内置规则")
|
||
out: dict[str, Any] = {
|
||
"op": "add", "name": name, "category": category, "scope": scope, "kind": kind,
|
||
}
|
||
if bound:
|
||
out["boundConstraintId"] = bound
|
||
source = str(raw.get("sourceText") or "").strip()
|
||
if source:
|
||
out["sourceText"] = source[:200]
|
||
return out
|
||
|
||
|
||
def _bound_apply(world: World, norm: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None]:
|
||
"""新增规则绑定内置可配项时，把强度真正落到排产引擎；返回 (补丁, 说明)。"""
|
||
bound = str(norm.get("boundConstraintId") or "")
|
||
if not bound:
|
||
return None, None
|
||
c = get_constraint(world, bound)
|
||
if not c or not c.get("kindConfigurable"):
|
||
return None, None
|
||
target = "hard" if norm.get("kind") == "hard" else "soft"
|
||
if c.get("kind") == target:
|
||
return None, None
|
||
label = "硬约束（违反阻止发布）" if target == "hard" else "软约束（只提示风险）"
|
||
return {"constraints": {bound: {"kind": target}}}, f"同时把内置规则《{c['name']}》调整为{label}"
|
||
|
||
|
||
def confirmation_for_rule_save(world: World, payload: dict[str, Any],
|
||
tenant_uuid: str | None = None) -> tuple[str, list[str]]:
|
||
"""P2 确认卡文案：计划员看得懂，写清范围、强度和是否真的接入引擎。"""
|
||
norm = normalize_rule_payload(payload)
|
||
if norm["op"] == "add":
|
||
title = f"登记约束规则：{norm['name']}"
|
||
lines = [
|
||
f"类别：{RULE_CATEGORIES[norm['category']]}；适用范围：{RULE_SCOPES[norm['scope']]}",
|
||
f"强度：{RULE_KINDS[norm['kind']]}",
|
||
]
|
||
patch, note = _bound_apply(world, norm)
|
||
if patch:
|
||
lines.append(f"接入排产：{note}")
|
||
elif norm.get("boundConstraintId"):
|
||
c = get_constraint(world, str(norm["boundConstraintId"]))
|
||
lines.append(f"接入排产：由内置规则《{(c or {}).get('name') or norm['boundConstraintId']}》执行")
|
||
else:
|
||
lines.append("接入排产：暂未接入排产引擎，先进入规则库备查")
|
||
lines.append("确认后写入规则库并记审计；不影响历史版本")
|
||
return title, lines
|
||
rules = {str(r.get("id")): r for r in all_rules(world, tenant_uuid)}
|
||
rule = rules.get(norm["ruleId"])
|
||
if not rule:
|
||
raise ValueError(f"规则库中没有 {norm['ruleId']} 这条规则")
|
||
name = str(rule.get("name") or norm["ruleId"])
|
||
effect = rule_effect(world, rule)["text"]
|
||
if norm["op"] == "delete":
|
||
return f"删除约束规则：{name}", [
|
||
f"原规则：{name}（{RULE_CATEGORIES.get(str(rule.get('category')), '其他')}）",
|
||
f"当前执行：{effect}",
|
||
"删除后不再出现在规则库；不影响历史版本与已发布计划",
|
||
]
|
||
label = "启用" if norm["op"] == "enable" else "停用"
|
||
lines = [
|
||
f"原规则：{name}（{RULE_CATEGORIES.get(str(rule.get('category')), '其他')}／"
|
||
f"{RULE_SCOPES.get(str(rule.get('scope')), '仅本项目')}）",
|
||
f"当前执行：{effect}",
|
||
"该操作只调整规则库里的状态，不改动内置约束的开关",
|
||
]
|
||
return f"{label}约束规则：{name}", lines
|
||
|
||
|
||
def _locate(world: World, tenant_uuid: str | None,
|
||
rule_id: str) -> tuple[dict[str, Any] | None, str | None]:
|
||
for r in load_tenant_rules(tenant_uuid):
|
||
if str(r.get("id")) == rule_id:
|
||
return r, "tenant"
|
||
for r in project_rules(world):
|
||
if str(r.get("id")) == rule_id:
|
||
return r, "project"
|
||
return None, None
|
||
|
||
|
||
def _write_container(world: World, tenant_uuid: str | None, where: str,
|
||
rules: list[dict[str, Any]]) -> None:
|
||
if where == "tenant":
|
||
save_tenant_rules(tenant_uuid, rules)
|
||
else:
|
||
world["constraintRules"] = rules
|
||
|
||
|
||
def apply_rule_save(world: World, payload: dict[str, Any],
|
||
tenant_uuid: str | None = None) -> dict[str, Any]:
|
||
"""落库：新增/启用/停用/删除计划员扩充规则（调用方负责审计与 save）。"""
|
||
norm = normalize_rule_payload(payload)
|
||
if norm["op"] == "add":
|
||
existing = all_rules(world, tenant_uuid)
|
||
rule_id = next_rule_id(existing)
|
||
rec: dict[str, Any] = {
|
||
"id": rule_id,
|
||
"name": norm["name"],
|
||
"category": norm["category"],
|
||
"scope": norm["scope"],
|
||
"kind": norm["kind"],
|
||
"enabled": True,
|
||
"createdAt": _now(),
|
||
"updatedAt": _now(),
|
||
}
|
||
if norm.get("boundConstraintId"):
|
||
rec["boundConstraintId"] = norm["boundConstraintId"]
|
||
if norm.get("sourceText"):
|
||
rec["source"] = {"type": "PLANNER_CHAT", "text": norm["sourceText"]}
|
||
patch, note = _bound_apply(world, norm)
|
||
if patch:
|
||
from server.aps_domain.constraints import apply_profile_save
|
||
|
||
apply_profile_save(world, patch)
|
||
if norm["scope"] == "tenant":
|
||
save_tenant_rules(tenant_uuid, load_tenant_rules(tenant_uuid) + [rec])
|
||
else:
|
||
world["constraintRules"] = project_rules(world) + [rec]
|
||
effect = rule_effect(world, rec)["text"]
|
||
message = (f"约束规则已登记 ✅ {rule_id} {rec['name']}"
|
||
f"（{RULE_CATEGORIES[norm['category']]}／{RULE_SCOPES[norm['scope']]}）。"
|
||
f"\n执行情况：{effect}。" + (f"\n{note}。" if note else ""))
|
||
return {"ruleId": rule_id, "op": "add", "scope": norm["scope"],
|
||
"rule": rec, "message": message}
|
||
rule_id = norm["ruleId"]
|
||
rule, where = _locate(world, tenant_uuid, rule_id)
|
||
if not rule or not where:
|
||
raise ValueError(f"规则库中没有 {rule_id} 这条规则")
|
||
container = load_tenant_rules(tenant_uuid) if where == "tenant" else project_rules(world)
|
||
name = str(rule.get("name") or rule_id)
|
||
if norm["op"] == "delete":
|
||
kept = [r for r in container if str(r.get("id")) != rule_id]
|
||
_write_container(world, tenant_uuid, where, kept)
|
||
return {"ruleId": rule_id, "op": "delete", "scope": where,
|
||
"message": f"约束规则已删除 ✅ {rule_id} {name}。"}
|
||
enabled = norm["op"] == "enable"
|
||
updated = []
|
||
for r in container:
|
||
if str(r.get("id")) == rule_id:
|
||
updated.append({**r, "enabled": enabled, "updatedAt": _now()})
|
||
else:
|
||
updated.append(r)
|
||
_write_container(world, tenant_uuid, where, updated)
|
||
label = "启用" if enabled else "停用"
|
||
return {"ruleId": rule_id, "op": norm["op"], "scope": where,
|
||
"message": f"约束规则已{label} ✅ {rule_id} {name}。"}
|
||
|
||
|
||
def constraint_library(world: World, tenant_uuid: str | None = None) -> dict[str, Any]:
|
||
"""设置页数据：默认通用规则概览 + 计划员扩充规则（按类别分组）。"""
|
||
builtin = get_constraint_profile(world)["constraints"]
|
||
custom: list[dict[str, Any]] = []
|
||
for r in all_rules(world, tenant_uuid):
|
||
effect = rule_effect(world, r)
|
||
category = str(r.get("category") or "other")
|
||
custom.append({
|
||
**r,
|
||
"category": category,
|
||
"categoryLabel": RULE_CATEGORIES.get(category, "其他"),
|
||
"scopeLabel": RULE_SCOPES.get(str(r.get("scope")), "仅本项目"),
|
||
"kindLabel": RULE_KINDS.get(str(r.get("kind")), "尽量满足"),
|
||
"effect": effect,
|
||
})
|
||
groups = []
|
||
for gid, label in RULE_CATEGORIES.items():
|
||
items = [r for r in custom if r["category"] == gid]
|
||
if items:
|
||
groups.append({"id": gid, "label": label, "items": items})
|
||
return {
|
||
"builtin": {
|
||
"total": len(builtin),
|
||
"enabled": sum(1 for c in builtin if c.get("enabled", True)),
|
||
"configurable": sum(1 for c in builtin if c.get("configurable")),
|
||
},
|
||
"custom": custom,
|
||
"groups": groups,
|
||
"categories": [{"id": k, "label": v} for k, v in RULE_CATEGORIES.items()],
|
||
"scopes": [{"id": k, "label": v} for k, v in RULE_SCOPES.items()],
|
||
"summary": {
|
||
"builtinTotal": len(builtin),
|
||
"customTotal": len(custom),
|
||
"engineBound": sum(1 for r in custom if r["effect"]["enforced"]),
|
||
"recordOnly": sum(1 for r in custom if not r["effect"]["enforced"]),
|
||
},
|
||
}
|