Selaa lähdekoodia

feat(输入兼容): 建立 accepted input 单一规则源

新增纯领域函数统一判断 direct-child 与 explicit 引用。集中定义任务类型白名单、祖先后代作用域兼容关系,以及 ElementSet 显式引用 Paragraph 的跨作用域 placement 例外。
SamLee 1 päivä sitten
vanhempi
commit
bf77ad94de

+ 93 - 0
script_build_host/src/script_build_host/domain/input_compatibility.py

@@ -0,0 +1,93 @@
+"""Single source of truth for accepted Script Task inputs."""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+from script_build_host.domain.task_contracts import ScriptTaskKind
+
+
+class AcceptedInputSource(StrEnum):
+    DIRECT_CHILD = "direct_child"
+    EXPLICIT = "explicit"
+
+
+class AcceptedInputCompatibilityError(ValueError):
+    pass
+
+
+_RETRIEVAL = frozenset(
+    {
+        ScriptTaskKind.PATTERN_RETRIEVAL,
+        ScriptTaskKind.DECODE_RETRIEVAL,
+        ScriptTaskKind.EXTERNAL_RETRIEVAL,
+        ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+    }
+)
+_CREATIVE = frozenset(
+    {
+        ScriptTaskKind.STRUCTURE,
+        ScriptTaskKind.PARAGRAPH,
+        ScriptTaskKind.ELEMENT_SET,
+        ScriptTaskKind.COMPARE,
+    }
+)
+_DIRECT_ALLOWED = {
+    ScriptTaskKind.DIRECTION: _RETRIEVAL,
+    ScriptTaskKind.CANDIDATE_PORTFOLIO: frozenset({ScriptTaskKind.COMPOSE}),
+    ScriptTaskKind.COMPOSE: _CREATIVE | _RETRIEVAL,
+    ScriptTaskKind.STRUCTURE: _CREATIVE | _RETRIEVAL,
+    ScriptTaskKind.PARAGRAPH: _CREATIVE | _RETRIEVAL,
+    ScriptTaskKind.ELEMENT_SET: _CREATIVE | _RETRIEVAL,
+    ScriptTaskKind.COMPARE: _CREATIVE | _RETRIEVAL,
+}
+_EXPLICIT_ALLOWED = {
+    **_DIRECT_ALLOWED,
+    ScriptTaskKind.CANDIDATE_PORTFOLIO: frozenset({ScriptTaskKind.COMPOSE, ScriptTaskKind.COMPARE}),
+    ScriptTaskKind.COMPOSE: _CREATIVE | _RETRIEVAL | {ScriptTaskKind.DIRECTION},
+    ScriptTaskKind.STRUCTURE: _CREATIVE | _RETRIEVAL | {ScriptTaskKind.DIRECTION},
+    ScriptTaskKind.PARAGRAPH: _CREATIVE | _RETRIEVAL | {ScriptTaskKind.DIRECTION},
+    ScriptTaskKind.ELEMENT_SET: _CREATIVE | _RETRIEVAL | {ScriptTaskKind.DIRECTION},
+    ScriptTaskKind.COMPARE: _CREATIVE | _RETRIEVAL | {ScriptTaskKind.DIRECTION},
+    ScriptTaskKind.ROOT_DELIVERY: frozenset(
+        {ScriptTaskKind.DIRECTION, ScriptTaskKind.CANDIDATE_PORTFOLIO}
+    ),
+}
+
+
+def scopes_compatible(left: str, right: str) -> bool:
+    return left == right or left.startswith(f"{right}/") or right.startswith(f"{left}/")
+
+
+def validate_accepted_input(
+    *,
+    consumer_kind: ScriptTaskKind,
+    consumer_scope: str,
+    producer_kind: ScriptTaskKind,
+    producer_scope: str,
+    source: AcceptedInputSource,
+) -> None:
+    placement = (
+        source is AcceptedInputSource.EXPLICIT
+        and consumer_kind is ScriptTaskKind.ELEMENT_SET
+        and producer_kind is ScriptTaskKind.PARAGRAPH
+    )
+    if not scopes_compatible(consumer_scope, producer_scope) and not placement:
+        raise AcceptedInputCompatibilityError("accepted input is outside the consumer scope")
+    allowed = (
+        _DIRECT_ALLOWED.get(consumer_kind, frozenset())
+        if source is AcceptedInputSource.DIRECT_CHILD
+        else _EXPLICIT_ALLOWED.get(consumer_kind, frozenset())
+    )
+    if producer_kind not in allowed:
+        raise AcceptedInputCompatibilityError(
+            "accepted Task kind is not valid for the consumer contract and input source"
+        )
+
+
+__all__ = [
+    "AcceptedInputCompatibilityError",
+    "AcceptedInputSource",
+    "scopes_compatible",
+    "validate_accepted_input",
+]