146 lines
5.3 KiB
Python
146 lines
5.3 KiB
Python
"""Legacy ConceptFormula reader and one-way migration into Curated Evidence.
|
|
|
|
The ``formulas/*.sql.md`` store is retained only for the migration window. Runtime
|
|
lookup uses typed, published ``kind=formula`` Evidence instead. A session reviewer
|
|
may still approve a formula locally; that is a proposal, not publication.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
import yaml
|
|
from pydantic import BaseModel
|
|
|
|
from tht.evidence.canonical import CuratedEvidence
|
|
|
|
FORMULAS_SUBDIR = "formulas"
|
|
_SUFFIX_RE = re.compile(r"^(.*?)-(\d+)\.sql\.md$")
|
|
|
|
|
|
class ConceptFormula(BaseModel):
|
|
concept: str
|
|
columns: list[str] = []
|
|
sql: str
|
|
# auto = sintetizzata dal modello (non ancora rivista); draft = bozza umana;
|
|
# reviewed = approvata da un revisore. (spec §4.7.2: status auto/draft/reviewed)
|
|
status: Literal["auto", "draft", "reviewed"] = "draft"
|
|
sources: list[str] = []
|
|
|
|
@property
|
|
def _slug(self) -> str:
|
|
"""ASCII slug for the filename (matches textutil.slugify shape)."""
|
|
import unicodedata
|
|
|
|
text = unicodedata.normalize("NFKD", self.concept).encode("ascii", "ignore").decode()
|
|
return re.sub(r"[^a-z0-9_]+", "-", text.lower()).strip("-") or "formula"
|
|
|
|
def dump(self) -> str:
|
|
meta = self.model_dump(exclude={"sql"}, mode="json")
|
|
fm = yaml.safe_dump(meta, sort_keys=False, allow_unicode=True)
|
|
return f"---\n{fm}---\n{self.sql}\n"
|
|
|
|
@classmethod
|
|
def parse(cls, text: str) -> ConceptFormula:
|
|
if not text.startswith("---\n"):
|
|
raise ValueError("frontmatter mancante (atteso '---\\n' iniziale)")
|
|
try:
|
|
_, fm, body = text.split("---\n", 2)
|
|
except ValueError as e:
|
|
raise ValueError("frontmatter malformato") from e
|
|
meta = yaml.safe_load(fm)
|
|
if not isinstance(meta, dict):
|
|
raise TypeError("frontmatter non valido")
|
|
return cls.model_validate({**meta, "sql": body.strip("\n")})
|
|
|
|
|
|
def _next_path(root: Path, slug: str) -> Path:
|
|
"""First free <slug>-<n>.sql.md path under root (n starts at 1)."""
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
existing = sorted(root.glob(f"{slug}-*.sql.md"))
|
|
n = 0
|
|
for p in existing:
|
|
m = _SUFFIX_RE.match(p.name)
|
|
if m:
|
|
n = max(n, int(m.group(2)))
|
|
return root / f"{slug}-{n + 1}.sql.md"
|
|
|
|
|
|
def save_formula(root: Path | str, formula: ConceptFormula) -> Path:
|
|
"""Persist a single concept->formula unit under <root>/formulas/. Returns the
|
|
written path. Append-only: each save writes a new file (so competing drafts and
|
|
reviewed versions coexist until a curator prunes)."""
|
|
root = Path(root)
|
|
formulas_dir = root / FORMULAS_SUBDIR
|
|
path = _next_path(formulas_dir, formula._slug)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(formula.dump())
|
|
return path
|
|
|
|
|
|
def _load_all(root: Path) -> list[ConceptFormula]:
|
|
formulas_dir = root / FORMULAS_SUBDIR
|
|
if not formulas_dir.is_dir():
|
|
return []
|
|
out: list[ConceptFormula] = []
|
|
for f in sorted(formulas_dir.glob("*.sql.md")):
|
|
try:
|
|
out.append(ConceptFormula.parse(f.read_text()))
|
|
except ValueError:
|
|
continue # malformed file: skip, don't crash retrieval
|
|
return out
|
|
|
|
|
|
def retrieve_formula(root: Path | str, concept: str) -> list[ConceptFormula]:
|
|
"""All formulas matching `concept` exactly under <root>/formulas/. Empty list if
|
|
none (or if the dir is absent). Multiple results mean competing drafts/versions for
|
|
the same concept -- the caller (gate) lets the reviewer pick."""
|
|
return [f for f in _load_all(Path(root)) if f.concept == concept]
|
|
|
|
|
|
def search_formulas(root: Path | str, query: str) -> list[ConceptFormula]:
|
|
"""Read legacy formulas for migration tooling only (case-insensitive concept match)."""
|
|
q = query.strip().lower()
|
|
return [f for f in _load_all(Path(root)) if q in f.concept.lower()]
|
|
|
|
|
|
def legacy_formula_to_curated(
|
|
formula: ConceptFormula,
|
|
*,
|
|
legacy_path: str,
|
|
) -> CuratedEvidence | None:
|
|
"""Convert one reviewed legacy formula into its deterministic curated counterpart.
|
|
|
|
Drafts and model-generated formulas have no global publication status. Their
|
|
caller must project them as session-local Formula proposals instead.
|
|
"""
|
|
if formula.status != "reviewed":
|
|
return None
|
|
source_notes = tuple(formula.sources) or (
|
|
"Legacy formula migrated without a recorded provenance note.",
|
|
)
|
|
source_sha256 = hashlib.sha256(formula.dump().encode("utf-8")).hexdigest()
|
|
source_file = legacy_path if legacy_path.startswith("source/") else f"source/{legacy_path}"
|
|
return CuratedEvidence.model_validate({
|
|
"schema_version": 1,
|
|
"id": f"evidence:{formula._slug}",
|
|
"title": formula.concept[:1].upper() + formula.concept[1:],
|
|
"kind": "formula",
|
|
"purposes": ["schema_linking", "sql_generation"],
|
|
"applies_to": {"concepts": [formula.concept], "columns": formula.columns},
|
|
"language": "it",
|
|
"provenance": {
|
|
"source_file": source_file,
|
|
"source_sha256": f"sha256:{source_sha256}",
|
|
"supporting_excerpts": source_notes,
|
|
},
|
|
"review_items": [],
|
|
"payload": {
|
|
"concept": formula.concept,
|
|
"columns": formula.columns,
|
|
"sql": formula.sql,
|
|
},
|
|
})
|