tegwick
35e86d7b85
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
"""Versioned implementation of the Custodian repository-classification contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
CONTRACT_VERSION = "1.0"
|
|
CATEGORIES = frozenset({"experimental", "research", "project", "tooling", "product", "business"})
|
|
DOMAINS = frozenset(
|
|
{
|
|
"infotech",
|
|
"financials",
|
|
"communication",
|
|
"consumer",
|
|
"health",
|
|
"industrials",
|
|
"energy",
|
|
"utilities",
|
|
"materials",
|
|
"realestate",
|
|
"crypto",
|
|
"agents",
|
|
"space",
|
|
"government",
|
|
}
|
|
)
|
|
BUSINESS_STAKE = frozenset(
|
|
{
|
|
"execution",
|
|
"intelligence",
|
|
"finance",
|
|
"legal",
|
|
"sales",
|
|
"experience",
|
|
"technology",
|
|
"operations",
|
|
"product",
|
|
"people",
|
|
"procurement",
|
|
"sustainability",
|
|
"automation",
|
|
}
|
|
)
|
|
BUSINESS_MECHANICS = frozenset({"intention", "control", "coordination", "operation", "adaptation"})
|
|
_TAG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClassificationIssue:
|
|
field: str
|
|
message: str
|
|
|
|
|
|
class ClassificationError(ValueError):
|
|
def __init__(self, issues: list[ClassificationIssue]):
|
|
self.issues = issues
|
|
super().__init__("; ".join(f"{issue.field}: {issue.message}" for issue in issues))
|
|
|
|
|
|
def _list(data: dict[str, Any], field: str, issues: list[ClassificationIssue]) -> list[str]:
|
|
value = data.get(field, [])
|
|
if value is None:
|
|
return []
|
|
if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
|
|
issues.append(ClassificationIssue(field, "must be a list of strings"))
|
|
return []
|
|
if len(value) != len(set(value)):
|
|
issues.append(ClassificationIssue(field, "must not contain duplicates"))
|
|
return value
|
|
|
|
|
|
def validate_classification(data: dict[str, Any]) -> list[ClassificationIssue]:
|
|
"""Validate required fields and controlled vocabularies from canon v1.0."""
|
|
issues: list[ClassificationIssue] = []
|
|
category = data.get("category")
|
|
domain = data.get("domain")
|
|
if category not in CATEGORIES:
|
|
issues.append(ClassificationIssue("category", f"must be one of {', '.join(sorted(CATEGORIES))}"))
|
|
if domain not in DOMAINS:
|
|
issues.append(ClassificationIssue("domain", f"must be one of {', '.join(sorted(DOMAINS))}"))
|
|
|
|
secondary = _list(data, "secondary_domains", issues)
|
|
unknown_domains = sorted(set(secondary) - DOMAINS)
|
|
if unknown_domains:
|
|
issues.append(ClassificationIssue("secondary_domains", f"unknown values: {', '.join(unknown_domains)}"))
|
|
if domain in secondary:
|
|
issues.append(ClassificationIssue("secondary_domains", "must not repeat the primary domain"))
|
|
|
|
tags = _list(data, "capability_tags", issues)
|
|
invalid_tags = sorted(tag for tag in tags if not _TAG_RE.fullmatch(tag))
|
|
if invalid_tags:
|
|
issues.append(ClassificationIssue("capability_tags", f"not lowercase kebab-case: {', '.join(invalid_tags)}"))
|
|
|
|
stake = _list(data, "business_stake", issues)
|
|
unknown_stake = sorted(set(stake) - BUSINESS_STAKE)
|
|
if unknown_stake:
|
|
issues.append(ClassificationIssue("business_stake", f"unknown values: {', '.join(unknown_stake)}"))
|
|
|
|
mechanics = _list(data, "business_mechanics", issues)
|
|
unknown_mechanics = sorted(set(mechanics) - BUSINESS_MECHANICS)
|
|
if unknown_mechanics:
|
|
issues.append(
|
|
ClassificationIssue("business_mechanics", f"unknown values: {', '.join(unknown_mechanics)}")
|
|
)
|
|
return issues
|
|
|
|
|
|
def require_valid_classification(data: dict[str, Any]) -> dict[str, Any]:
|
|
issues = validate_classification(data)
|
|
if issues:
|
|
raise ClassificationError(issues)
|
|
return data
|