feat(seo-schema-generator): merge site-extraction + source-to-schema into one skill
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>
This commit is contained in:
@@ -0,0 +1,346 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
extract_site_claims.py — Scenario 1 adapter: an EXISTING website → claims register.
|
||||
|
||||
THE MERGE, IN ONE SENTENCE
|
||||
Schema generation has two scenarios that differ only in WHERE facts come from:
|
||||
1) from a given website — the live pages ARE the source of truth (this script)
|
||||
2) from collected sources — a not-yet-published site, facts scattered & conflicting
|
||||
Both emit the SAME claims_register.csv, which build_schema_drafts.py then turns into
|
||||
drafts and 16-seo-schema-validator gates. The claims register is the shared pivot
|
||||
that lets one skill cover both scenarios.
|
||||
|
||||
WHAT THIS DOES
|
||||
Reads pages of an existing site and seeds a claims register from them:
|
||||
- existing JSON-LD (<script type="application/ld+json">) -> CONFIRMED (authority 1):
|
||||
the site already published these facts about itself.
|
||||
- <title> / meta description / OpenGraph / <html lang> / canonical -> PENDING:
|
||||
inferred, not authoritative. The builder EXCLUDES PENDING claims until a human
|
||||
confirms them — so inference never silently ships (the skill's core principle).
|
||||
|
||||
You then review/edit claims_register.csv, run build_schema_drafts.py, and validate.
|
||||
|
||||
INPUT (any mix): URLs (needs `requests`), local .html files, or a directory of .html
|
||||
OUTPUT (in --out): claims_register.csv + extraction_report.md
|
||||
|
||||
USAGE
|
||||
python extract_site_claims.py https://example.com/ https://example.com/about --out site_claims
|
||||
python extract_site_claims.py ./snapshot/ --out site_claims # offline, local HTML
|
||||
# then:
|
||||
python build_schema_drafts.py site_claims/claims_register.csv --out drafts_out
|
||||
python ../16-seo-schema-validator/scripts/validate_schema.py drafts_out/schema_drafts_dataset.csv --out qa_out
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from collections import OrderedDict
|
||||
from html.parser import HTMLParser
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
JSONLD_RE = re.compile(
|
||||
r'<script[^>]+type=["\']application/ld\+json["\'][^>]*>(.*?)</script>',
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
|
||||
# entity_id prefix per schema type — keeps ids readable and groupable.
|
||||
TYPE_PREFIX = {
|
||||
"Organization": "org", "Corporation": "org", "LocalBusiness": "biz",
|
||||
"WebSite": "site", "WebPage": "page",
|
||||
"Hotel": "hotel", "LodgingBusiness": "hotel", "Resort": "hotel",
|
||||
"Restaurant": "dining", "FoodEstablishment": "dining", "BarOrPub": "dining",
|
||||
"Person": "person", "Product": "product", "Article": "article",
|
||||
"NewsArticle": "article", "BlogPosting": "article", "Event": "event",
|
||||
"FAQPage": "faq", "BreadcrumbList": "crumb",
|
||||
}
|
||||
# OpenGraph og:type -> our seed @type for the meta-only fallback.
|
||||
OG_TYPE_MAP = {"website": "WebSite", "article": "Article", "product": "Product",
|
||||
"business.business": "LocalBusiness", "profile": "Person"}
|
||||
|
||||
CLAIM_FIELDS = ["entity_id", "entity_type", "property", "value", "lang", "url",
|
||||
"source_ids", "authority", "confidence", "conflict", "status", "note"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Lightweight HTML meta extraction (stdlib only)
|
||||
# --------------------------------------------------------------------------- #
|
||||
class MetaParser(HTMLParser):
|
||||
"""Pull <title>, <html lang>, <link rel=canonical>, and <meta> name/property."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.title_parts = []
|
||||
self._in_title = False
|
||||
self.lang = ""
|
||||
self.canonical = ""
|
||||
self.meta = {} # name/property (lowercased) -> content
|
||||
|
||||
def handle_starttag(self, tag, attrs):
|
||||
a = {k.lower(): (v or "") for k, v in attrs}
|
||||
if tag == "html" and a.get("lang"):
|
||||
self.lang = a["lang"].strip()
|
||||
elif tag == "title":
|
||||
self._in_title = True
|
||||
elif tag == "link" and a.get("rel", "").lower() == "canonical" and a.get("href"):
|
||||
self.canonical = a["href"].strip()
|
||||
elif tag == "meta":
|
||||
key = (a.get("property") or a.get("name") or "").lower().strip()
|
||||
if key and "content" in a:
|
||||
self.meta.setdefault(key, a["content"].strip())
|
||||
|
||||
def handle_endtag(self, tag):
|
||||
if tag == "title":
|
||||
self._in_title = False
|
||||
|
||||
def handle_data(self, data):
|
||||
if self._in_title:
|
||||
self.title_parts.append(data)
|
||||
|
||||
@property
|
||||
def title(self):
|
||||
return re.sub(r"\s+", " ", "".join(self.title_parts)).strip()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Page acquisition
|
||||
# --------------------------------------------------------------------------- #
|
||||
def gather_pages(inputs):
|
||||
"""Yield (url, html). Accepts http(s) URLs, .html files, and directories."""
|
||||
for item in inputs:
|
||||
if re.match(r"^https?://", item, re.IGNORECASE):
|
||||
try:
|
||||
import requests
|
||||
except ImportError:
|
||||
sys.exit("Fetching URLs needs requests: pip install requests "
|
||||
"(or pass local .html files / a directory instead).")
|
||||
try:
|
||||
r = requests.get(item, timeout=20,
|
||||
headers={"User-Agent": "Mozilla/5.0 (SchemaGen/1.0)"})
|
||||
r.raise_for_status()
|
||||
yield item, r.text
|
||||
except Exception as exc: # noqa: BLE001 — best-effort fetch
|
||||
print(f" ! could not fetch {item}: {exc}", file=sys.stderr)
|
||||
else:
|
||||
p = Path(item)
|
||||
if p.is_dir():
|
||||
for hp in sorted(p.rglob("*.html")):
|
||||
hp = hp.resolve()
|
||||
yield hp.as_uri(), hp.read_text(encoding="utf-8", errors="replace")
|
||||
elif p.is_file():
|
||||
p = p.resolve()
|
||||
yield p.as_uri(), p.read_text(encoding="utf-8", errors="replace")
|
||||
else:
|
||||
print(f" ! not a URL/file/dir, skipped: {item}", file=sys.stderr)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# JSON-LD node -> flat (dotted-path, value) claims
|
||||
# --------------------------------------------------------------------------- #
|
||||
def primary_type(node):
|
||||
t = node.get("@type")
|
||||
if isinstance(t, list):
|
||||
return t[0] if t else ""
|
||||
return t or ""
|
||||
|
||||
|
||||
def top_level_nodes(parsed):
|
||||
"""Return the ENTITY nodes only — @graph members, array items, or a single object.
|
||||
|
||||
Deliberately NOT recursive: nested objects (PostalAddress, GeoCoordinates, …) belong
|
||||
to their parent and are captured by flatten() as dotted paths. Recursing here would
|
||||
wrongly promote a nested address into its own entity.
|
||||
"""
|
||||
if isinstance(parsed, dict) and "@graph" in parsed:
|
||||
graph = parsed["@graph"]
|
||||
return [n for n in graph if isinstance(n, dict) and "@type" in n]
|
||||
if isinstance(parsed, list):
|
||||
return [n for n in parsed if isinstance(n, dict) and "@type" in n]
|
||||
if isinstance(parsed, dict) and "@type" in parsed:
|
||||
return [parsed]
|
||||
return []
|
||||
|
||||
|
||||
def flatten(node, prefix=""):
|
||||
"""Yield (property_path, value) pairs matching the template {{dotted.path}} slots.
|
||||
|
||||
- scalars -> ("name", "X")
|
||||
- scalar arrays -> ("sameAs", "a|b|c") (pipe-joined; builder splits on '|')
|
||||
- nested objects -> recurse with dotted prefix ("address.streetAddress", ...)
|
||||
- @type inside a nested object is structural (templates hard-code it) -> skipped
|
||||
- a bare {"@id": "..."} reference -> ("parentOrganization.@id", "...")
|
||||
"""
|
||||
for key, val in node.items():
|
||||
if key == "@type":
|
||||
continue
|
||||
path = f"{prefix}{key}"
|
||||
if isinstance(val, (str, int, float, bool)):
|
||||
yield path, str(val)
|
||||
elif isinstance(val, list):
|
||||
scalars = [str(v) for v in val if isinstance(v, (str, int, float, bool))]
|
||||
if scalars:
|
||||
yield path, "|".join(scalars)
|
||||
for v in val: # objects inside arrays -> recurse (best-effort)
|
||||
if isinstance(v, dict):
|
||||
yield from flatten(v, prefix=f"{path}.")
|
||||
elif isinstance(val, dict):
|
||||
yield from flatten(val, prefix=f"{path}.")
|
||||
|
||||
|
||||
def slugify(text, maxlen=40):
|
||||
s = re.sub(r"^https?://", "", text or "")
|
||||
s = re.sub(r"[^A-Za-z0-9]+", "-", s).strip("-").lower()
|
||||
return s[:maxlen] or "node"
|
||||
|
||||
|
||||
def entity_id_for(node, url, idx):
|
||||
"""Readable, groupable id like `hotel:westin-hotel` from an @id or URL."""
|
||||
etype = primary_type(node)
|
||||
prefix = TYPE_PREFIX.get(etype, "node")
|
||||
nid = node.get("@id")
|
||||
if nid:
|
||||
pr = urlparse(str(nid))
|
||||
tail = [s for s in pr.path.split("/") if s]
|
||||
parts = (tail[-1:] if tail else []) + ([pr.fragment] if pr.fragment else [])
|
||||
base = "-".join(parts) or pr.netloc or str(nid)
|
||||
else:
|
||||
base = url or f"n{idx}"
|
||||
return f"{prefix}:{slugify(base)}"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Build claims rows from one page
|
||||
# --------------------------------------------------------------------------- #
|
||||
def claims_from_page(url, html, default_lang, rows, seen_props):
|
||||
"""Append claim rows for one page. seen_props tracks (entity_id, property)
|
||||
already taken from authoritative JSON-LD, so meta only fills genuine gaps."""
|
||||
page_lang = ""
|
||||
found_jsonld = False
|
||||
|
||||
# --- 1) existing JSON-LD -> CONFIRMED (authority 1) ---
|
||||
for block in JSONLD_RE.findall(html):
|
||||
try:
|
||||
parsed = json.loads(block.strip())
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
for i, node in enumerate(top_level_nodes(parsed)):
|
||||
etype = primary_type(node)
|
||||
if not etype:
|
||||
continue
|
||||
found_jsonld = True
|
||||
eid = entity_id_for(node, url, i)
|
||||
node_lang = node.get("inLanguage") if isinstance(node.get("inLanguage"), str) else ""
|
||||
lang = node_lang or default_lang
|
||||
for prop, value in flatten(node):
|
||||
rows.append(_row(eid, etype, prop, value, lang, url,
|
||||
"S-SITE", 1, "high", "CONFIRMED",
|
||||
"extracted from existing JSON-LD"))
|
||||
seen_props.add((eid, prop))
|
||||
|
||||
# --- 2) meta / OpenGraph -> PENDING (inferred, needs confirmation) ---
|
||||
mp = MetaParser()
|
||||
try:
|
||||
mp.feed(html)
|
||||
except Exception: # noqa: BLE001 — tolerate malformed HTML
|
||||
pass
|
||||
page_lang = mp.lang or default_lang
|
||||
og_type = mp.meta.get("og:type", "").lower()
|
||||
etype = OG_TYPE_MAP.get(og_type, "WebPage")
|
||||
eid = f"{TYPE_PREFIX.get(etype, 'page')}:{slugify(mp.canonical or url)}"
|
||||
|
||||
inferred = {
|
||||
"name": mp.meta.get("og:title") or mp.title,
|
||||
"url": mp.meta.get("og:url") or mp.canonical or (url if url.startswith("http") else ""),
|
||||
"description": mp.meta.get("og:description") or mp.meta.get("description"),
|
||||
"image": mp.meta.get("og:image"),
|
||||
"inLanguage": page_lang,
|
||||
}
|
||||
# Only emit meta claims when this page contributed NO JSON-LD (else JSON-LD wins).
|
||||
if not found_jsonld:
|
||||
for prop, value in inferred.items():
|
||||
if value and (eid, prop) not in seen_props:
|
||||
rows.append(_row(eid, etype, prop, value, page_lang, url,
|
||||
"S-SITE-META", 2, "med", "PENDING",
|
||||
"inferred from <title>/OpenGraph — confirm before shipping"))
|
||||
|
||||
|
||||
def _row(eid, etype, prop, value, lang, url, src, authority, conf, status, note):
|
||||
return OrderedDict([
|
||||
("entity_id", eid), ("entity_type", etype), ("property", prop),
|
||||
("value", value), ("lang", lang or ""), ("url", url if url.startswith("http") else ""),
|
||||
("source_ids", src), ("authority", authority), ("confidence", conf),
|
||||
("conflict", ""), ("status", status), ("note", note),
|
||||
])
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Main
|
||||
# --------------------------------------------------------------------------- #
|
||||
def main(argv=None):
|
||||
ap = argparse.ArgumentParser(
|
||||
description="Scenario-1 adapter: existing website -> claims register.")
|
||||
ap.add_argument("inputs", nargs="+", help="URLs, .html files, or a directory")
|
||||
ap.add_argument("--out", default="site_claims", help="output directory")
|
||||
ap.add_argument("--default-lang", default="", help="fallback language code (e.g. ko)")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
rows, seen = [], set()
|
||||
pages = 0
|
||||
for url, html in gather_pages(args.inputs):
|
||||
pages += 1
|
||||
before = len(rows)
|
||||
claims_from_page(url, html, args.default_lang, rows, seen)
|
||||
print(f" · {url} → {len(rows) - before} claims")
|
||||
|
||||
if not rows:
|
||||
print("No claims extracted (no JSON-LD or usable meta on the given pages).",
|
||||
file=sys.stderr)
|
||||
return 1
|
||||
|
||||
outdir = Path(args.out)
|
||||
outdir.mkdir(parents=True, exist_ok=True)
|
||||
reg = outdir / "claims_register.csv"
|
||||
with open(reg, "w", encoding="utf-8-sig", newline="") as f:
|
||||
w = csv.DictWriter(f, fieldnames=CLAIM_FIELDS)
|
||||
w.writeheader()
|
||||
w.writerows(rows)
|
||||
|
||||
confirmed = sum(1 for r in rows if r["status"] == "CONFIRMED")
|
||||
pending = sum(1 for r in rows if r["status"] == "PENDING")
|
||||
entities = sorted({r["entity_id"] for r in rows})
|
||||
rep = [
|
||||
"# Site Extraction Report", "",
|
||||
f"- Pages read: **{pages}**",
|
||||
f"- Claims extracted: **{len(rows)}** "
|
||||
f"(CONFIRMED from JSON-LD: {confirmed} · PENDING from meta: {pending})",
|
||||
f"- Entities seeded: **{len(entities)}**", "",
|
||||
"## Entities", "",
|
||||
]
|
||||
rep += [f"- `{e}`" for e in entities]
|
||||
rep += [
|
||||
"", "## Review before building",
|
||||
"1. **CONFIRMED** rows came from the site's own JSON-LD — spot-check accuracy.",
|
||||
"2. **PENDING** rows were inferred from `<title>`/OpenGraph and will NOT ship until "
|
||||
"you set `status=CONFIRMED` (and clear any `conflict`).",
|
||||
"3. Add anything the pages didn't expose (telephone, address, geo, sameAs).",
|
||||
"", "## Next step",
|
||||
"```bash",
|
||||
f"python build_schema_drafts.py {reg} --out drafts_out",
|
||||
"python ../16-seo-schema-validator/scripts/validate_schema.py "
|
||||
"drafts_out/schema_drafts_dataset.csv --out qa_out",
|
||||
"```",
|
||||
]
|
||||
(outdir / "extraction_report.md").write_text("\n".join(rep), encoding="utf-8")
|
||||
|
||||
print(f"\nWrote {len(rows)} claims ({confirmed} CONFIRMED, {pending} PENDING) "
|
||||
f"for {len(entities)} entities → {reg}")
|
||||
print(f" {outdir / 'extraction_report.md'}")
|
||||
print("Review the register (confirm PENDING rows), then run build_schema_drafts.py.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user