, <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_

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