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>
347 lines
14 KiB
Python
347 lines
14 KiB
Python
#!/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())
|