feat: advance repository records and provenance
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
This commit is contained in:
parent
329af60753
commit
35e86d7b85
24 changed files with 1618 additions and 51 deletions
114
src/repo_manager/classification.py
Normal file
114
src/repo_manager/classification.py
Normal file
|
|
@ -0,0 +1,114 @@
|
|||
"""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
|
||||
Loading…
Add table
Add a link
Reference in a new issue