#!/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()