Andrew Yim
1706a820fe
Unify the two schema-generation scenarios into a single slot-17 skill, both feeding one claims register -> build -> validate(16) pipeline: - Mode 1 (existing site): NEW scripts/extract_site_claims.py turns URLs / local HTML / a directory into a claims register. Existing JSON-LD -> CONFIRMED; title/OpenGraph -> PENDING (never auto-shipped). + site-extraction-methodology.md and bundled fixtures/site/ demo pages. - Mode 2 (not-yet-published site): land the source-to-schema engine (build_schema_drafts.py, type_templates.json, claims/source registers, 3 refs, sample_claims.csv) from the Desktop builder. - Rewrite SKILL.md (v2.0) around the two-mode framing; the claims register is the shared pivot. Only CONFIRMED, non-conflicting claims become schema; unfilled template slots are pruned, never emitted as placeholders. - Retire the old template-fill generator (code/ + desktop/); update root CLAUDE.md. Self-tested both chains end-to-end: Mode 2 sample -> build -> validate PASS (P0=0); Mode 1 fixtures -> extract -> build -> validate PASS (P0=0), JSON-LD round-trips with nested address intact. Fixed two adapter bugs (nested node promotion; relative-path URI). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
306 lines
13 KiB
Python
306 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
build_schema_drafts.py — Source-to-Schema draft generator (skill 17, pre-launch)
|
|
|
|
WHAT IT DOES
|
|
Turns a *claims register* (reconciled, provenance-tracked facts) into JSON-LD
|
|
drafts, then writes a dataset CSV that feeds straight into
|
|
16-seo-schema-validator/scripts/validate_schema.py.
|
|
|
|
WHY A CLAIMS REGISTER FIRST (the core idea)
|
|
Authoring schema for a site that does not exist yet is error-prone because the
|
|
facts live in many conflicting sources (DART, Wikipedia, brochures...). If you
|
|
pour raw, unreconciled facts straight into JSON-LD you reproduce every conflict
|
|
and gap as a schema defect. So we reconcile facts FIRST (one confirmed value per
|
|
property, with sources recorded), and only CONFIRMED, non-conflicting claims are
|
|
allowed to become schema. Everything else is reported, not shipped.
|
|
|
|
THE PRUNING RULE (placeholder leakage is the #1 pre-launch defect)
|
|
A template slot that has no confirmed value is DELETED — never emitted as
|
|
"{{...}}" or "TODO". An empty nested object (only @type left) is dropped too.
|
|
This guarantees drafts contain only backed facts.
|
|
|
|
INPUT (claims register): .csv or .xlsx with columns (case-insensitive, KR/EN aliases ok)
|
|
entity_id e.g. org:shilla, hotel:theshilla-seoul (groups rows into one node)
|
|
entity_type Organization | Hotel | Person | JobPosting | VideoObject | WebSite | FAQPage
|
|
property schema.org property, dotted for nesting: address.streetAddress
|
|
append nothing for scalars; arrays are handled automatically
|
|
value the confirmed value (pipe-separate multiple values: a|b|c)
|
|
lang optional (ko/en/ja/zh) -> produces one draft per language
|
|
url optional target URL (provisional ok) -> carried to validator
|
|
source_ids optional pipe-separated refs into the source register (provenance)
|
|
authority optional 1..n (1 = most authoritative) — for your audit trail
|
|
confidence optional high|med|low
|
|
conflict optional — any truthy value (Y/1/true/충돌) EXCLUDES the claim
|
|
status CONFIRMED (default if blank) | PENDING | REJECTED — only CONFIRMED ships
|
|
note optional
|
|
|
|
USAGE
|
|
python scripts/build_schema_drafts.py path/to/claims_register.csv \
|
|
--templates scripts/type_templates.json --out drafts_out
|
|
# then hand off to the validator:
|
|
python ../16-seo-schema-validator/scripts/validate_schema.py \
|
|
drafts_out/schema_drafts_dataset.csv --out qa_out
|
|
"""
|
|
|
|
import argparse, csv, json, os, sys, copy, re
|
|
from collections import defaultdict, OrderedDict
|
|
|
|
# ----------------------------------------------------------------------------- #
|
|
# Column aliasing — accept Korean/English header variants #
|
|
# ----------------------------------------------------------------------------- #
|
|
COL_ALIASES = {
|
|
"entity_id": ["entity_id", "entity", "엔티티", "엔티티id", "id"],
|
|
"entity_type": ["entity_type", "type", "타입", "유형", "스키마타입"],
|
|
"property": ["property", "prop", "속성", "프로퍼티", "path"],
|
|
"value": ["value", "값", "내용"],
|
|
"lang": ["lang", "language", "언어", "언어코드"],
|
|
"url": ["url", "메뉴 url", "메뉴url", "주소"],
|
|
"source_ids": ["source_ids", "source", "sources", "출처", "출처id"],
|
|
"authority": ["authority", "권위", "권위순위"],
|
|
"confidence": ["confidence", "신뢰도"],
|
|
"conflict": ["conflict", "충돌", "conflict_flag"],
|
|
"status": ["status", "상태"],
|
|
"note": ["note", "notes", "비고", "메모"],
|
|
}
|
|
TRUTHY = {"y", "yes", "1", "true", "t", "충돌", "conflict", "o"}
|
|
PLACEHOLDER_RE = re.compile(r"^\{\{(.+?)\}\}$") # matches an entire-string slot
|
|
UNFILLED = object() # sentinel: slot had no value
|
|
|
|
|
|
def _norm(s):
|
|
return (s or "").strip().lower().replace(" ", "")
|
|
|
|
|
|
def map_columns(headers):
|
|
"""Return {canonical_name: actual_header} using the alias table."""
|
|
lookup = {_norm(h): h for h in headers}
|
|
out = {}
|
|
for canon, aliases in COL_ALIASES.items():
|
|
for a in aliases:
|
|
if _norm(a) in lookup:
|
|
out[canon] = lookup[_norm(a)]
|
|
break
|
|
return out
|
|
|
|
|
|
# ----------------------------------------------------------------------------- #
|
|
# Loading the claims register #
|
|
# ----------------------------------------------------------------------------- #
|
|
def load_rows(path):
|
|
"""Yield dict rows from .csv or .xlsx. Keeps original header names."""
|
|
ext = os.path.splitext(path)[1].lower()
|
|
if ext in (".csv", ".tsv"):
|
|
delim = "\t" if ext == ".tsv" else ","
|
|
with open(path, encoding="utf-8-sig", newline="") as f:
|
|
for row in csv.DictReader(f, delimiter=delim):
|
|
yield row
|
|
elif ext in (".xlsx", ".xlsm"):
|
|
try:
|
|
from openpyxl import load_workbook
|
|
except ImportError:
|
|
sys.exit("openpyxl required for .xlsx — pip install openpyxl")
|
|
wb = load_workbook(path, read_only=True, data_only=True)
|
|
for ws in wb.worksheets:
|
|
rows = ws.iter_rows(values_only=True)
|
|
try:
|
|
headers = [str(h) if h is not None else "" for h in next(rows)]
|
|
except StopIteration:
|
|
continue
|
|
if not map_columns(headers).get("entity_id"):
|
|
continue # sheet without our schema -> skip
|
|
for r in rows:
|
|
yield {headers[i]: ("" if v is None else str(v))
|
|
for i, v in enumerate(r) if i < len(headers)}
|
|
else:
|
|
sys.exit(f"Unsupported claims register format: {ext}")
|
|
|
|
|
|
# ----------------------------------------------------------------------------- #
|
|
# Distil rows -> per-(entity, lang) confirmed claim maps + exclusion log #
|
|
# ----------------------------------------------------------------------------- #
|
|
def collect_claims(path):
|
|
rows = list(load_rows(path))
|
|
if not rows:
|
|
sys.exit("Claims register is empty.")
|
|
cmap = map_columns(rows[0].keys())
|
|
for req in ("entity_id", "entity_type", "property", "value"):
|
|
if req not in cmap:
|
|
sys.exit(f"Missing required column '{req}'. Found: {list(rows[0].keys())}")
|
|
|
|
def g(row, key):
|
|
col = cmap.get(key)
|
|
return (row.get(col, "") if col else "").strip()
|
|
|
|
# claims[(entity_id, lang)] -> {"type":..., "url":..., "props": {path: [values]}}
|
|
claims = OrderedDict()
|
|
excluded = [] # (entity, prop, reason, detail)
|
|
for row in rows:
|
|
eid = g(row, "entity_id")
|
|
if not eid:
|
|
continue
|
|
etype = g(row, "entity_type")
|
|
prop = g(row, "property")
|
|
val = g(row, "value")
|
|
lang = g(row, "lang") or ""
|
|
status = (g(row, "status") or "CONFIRMED").upper()
|
|
conflict = _norm(g(row, "conflict")) in TRUTHY
|
|
|
|
if conflict:
|
|
excluded.append((eid, prop, "CONFLICT", f"sources disagree -> resolve first"))
|
|
continue
|
|
if status == "REJECTED":
|
|
excluded.append((eid, prop, "REJECTED", g(row, "note")))
|
|
continue
|
|
if status == "PENDING":
|
|
excluded.append((eid, prop, "PENDING", "not yet confirmed by an authoritative source"))
|
|
continue
|
|
if not val:
|
|
excluded.append((eid, prop, "EMPTY", "confirmed row but value is blank"))
|
|
continue
|
|
|
|
key = (eid, lang)
|
|
node = claims.setdefault(key, {"type": etype, "url": g(row, "url"),
|
|
"props": defaultdict(list)})
|
|
if etype and not node["type"]:
|
|
node["type"] = etype
|
|
if g(row, "url") and not node["url"]:
|
|
node["url"] = g(row, "url")
|
|
# pipe-separated value -> multiple values (array support)
|
|
for v in (val.split("|") if "|" in val else [val]):
|
|
v = v.strip()
|
|
if v:
|
|
node["props"][prop].append(v)
|
|
return claims, excluded
|
|
|
|
|
|
# ----------------------------------------------------------------------------- #
|
|
# Fill a template, pruning every unfilled slot #
|
|
# ----------------------------------------------------------------------------- #
|
|
def fill(node_template, props):
|
|
"""Recursively fill {{slots}}; return UNFILLED when a branch has no real data."""
|
|
if isinstance(node_template, str):
|
|
m = PLACEHOLDER_RE.match(node_template.strip())
|
|
if not m:
|
|
return node_template # literal (e.g. "@type":"Hotel")
|
|
path = m.group(1)
|
|
is_array = path.endswith("[]")
|
|
if is_array:
|
|
path = path[:-2]
|
|
vals = props.get(path, [])
|
|
if not vals:
|
|
return UNFILLED
|
|
if is_array:
|
|
return list(vals)
|
|
if len(vals) > 1:
|
|
print(f" ! multiple values for scalar '{path}' — using first ({len(vals)} given)")
|
|
return vals[0]
|
|
|
|
if isinstance(node_template, dict):
|
|
out = OrderedDict()
|
|
for k, v in node_template.items():
|
|
filled = fill(v, props)
|
|
if filled is UNFILLED:
|
|
continue
|
|
out[k] = filled
|
|
# an object that only carries @type/@context (no real data, no @id ref) is empty
|
|
meaningful = [k for k in out if k not in ("@type", "@context")]
|
|
if not meaningful:
|
|
return UNFILLED
|
|
return out
|
|
|
|
if isinstance(node_template, list):
|
|
out = [x for x in (fill(i, props) for i in node_template) if x is not UNFILLED]
|
|
return out if out else UNFILLED
|
|
|
|
return node_template
|
|
|
|
|
|
# ----------------------------------------------------------------------------- #
|
|
# Main #
|
|
# ----------------------------------------------------------------------------- #
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Build JSON-LD drafts from a claims register.")
|
|
ap.add_argument("claims", help="claims register .csv/.xlsx")
|
|
ap.add_argument("--templates", default=os.path.join(os.path.dirname(__file__), "type_templates.json"))
|
|
ap.add_argument("--out", default="drafts_out")
|
|
args = ap.parse_args()
|
|
|
|
templates = json.load(open(args.templates, encoding="utf-8"))["templates"]
|
|
claims, excluded = collect_claims(args.claims)
|
|
|
|
os.makedirs(os.path.join(args.out, "drafts"), exist_ok=True)
|
|
dataset_rows = []
|
|
built, skipped_type = 0, []
|
|
|
|
for (eid, lang), node in claims.items():
|
|
etype = node["type"]
|
|
if etype not in templates:
|
|
skipped_type.append((eid, etype))
|
|
continue
|
|
filled = fill(copy.deepcopy(templates[etype]["tpl"]), node["props"])
|
|
if filled is UNFILLED or not filled:
|
|
skipped_type.append((eid, f"{etype} (no usable claims)"))
|
|
continue
|
|
jsonld = json.dumps(filled, ensure_ascii=False, indent=2)
|
|
safe = re.sub(r"[^A-Za-z0-9]+", "_", eid).strip("_")
|
|
fname = f"{safe}__{lang}.jsonld" if lang else f"{safe}.jsonld"
|
|
with open(os.path.join(args.out, "drafts", fname), "w", encoding="utf-8") as f:
|
|
f.write(jsonld)
|
|
dataset_rows.append(OrderedDict([
|
|
("entity_id", eid), ("entity_type", etype),
|
|
("url", node["url"]), ("lang", lang),
|
|
("jsonld", json.dumps(filled, ensure_ascii=False)),
|
|
]))
|
|
built += 1
|
|
|
|
# dataset CSV — directly consumable by validate_schema.py (auto-detects 'jsonld')
|
|
ds_path = os.path.join(args.out, "schema_drafts_dataset.csv")
|
|
with open(ds_path, "w", encoding="utf-8-sig", newline="") as f:
|
|
w = csv.DictWriter(f, fieldnames=["entity_id", "entity_type", "url", "lang", "jsonld"])
|
|
w.writeheader()
|
|
w.writerows(dataset_rows)
|
|
|
|
# build report
|
|
rep = [
|
|
"# Schema Draft Build Report",
|
|
"",
|
|
f"- Entities built: **{built}**",
|
|
f"- Claims excluded (not shipped): **{len(excluded)}**",
|
|
f"- Entities skipped (no template / no usable claims): **{len(skipped_type)}**",
|
|
"",
|
|
"## Excluded claims (resolve before they can become schema)",
|
|
]
|
|
if excluded:
|
|
rep.append("| entity | property | reason | detail |")
|
|
rep.append("|--------|----------|--------|--------|")
|
|
for eid, prop, reason, detail in excluded:
|
|
rep.append(f"| {eid} | {prop} | **{reason}** | {detail} |")
|
|
else:
|
|
rep.append("_None._")
|
|
if skipped_type:
|
|
rep += ["", "## Skipped entities", ""]
|
|
for eid, why in skipped_type:
|
|
rep.append(f"- {eid} — {why}")
|
|
rep += [
|
|
"", "## Next step",
|
|
"1. Resolve every excluded claim (confirm an authoritative value, clear conflicts).",
|
|
"2. Re-run this builder.",
|
|
"3. Validate the output (the QA gate):",
|
|
" ```bash",
|
|
f" python ../16-seo-schema-validator/scripts/validate_schema.py {ds_path} --out qa_out",
|
|
" ```",
|
|
"4. Fix all P0 from the validator, then proceed to client review.",
|
|
]
|
|
rep_path = os.path.join(args.out, "build_report.md")
|
|
open(rep_path, "w", encoding="utf-8").write("\n".join(rep))
|
|
|
|
print(f"Built {built} drafts | excluded {len(excluded)} claims | skipped {len(skipped_type)} entities")
|
|
print(f"Wrote: {ds_path}")
|
|
print(f" {rep_path}")
|
|
print(f" {os.path.join(args.out, 'drafts')}/*.jsonld")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|