ourdigital/our-claude-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SEO schema validator skill update

Browse Source
This commit is contained in:
Andrew Yim
2026-06-08 13:21:09 +09:00
parent 8ffb6bec6b
commit 7daa4cda68
5 changed files with 3076 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
custom-skills/16-seo-schema-validator/scripts/validate_schema.py
View File

@@ -607,6 +607,16 @@ def _address_street(node):
return ""
def _address_locality(node):
"""City/region key, used to keep distinct same-name locations of a chain apart."""
addr = node.get("address")
if isinstance(addr, list) and addr and isinstance(addr[0], dict):
addr = addr[0]
if isinstance(addr, dict):
return normalize_name(addr.get("addressLocality") or addr.get("addressRegion"))
return ""
def _walk_ids(obj, defined, referenced):
"""Collect @id definitions vs pure references by walking the whole document.
@@ -645,21 +655,26 @@ def layer4_consistency(node_index, parsed_docs, rules, defects):
break # one placeholder defect per node is enough signal
# ---- NAP consistency (P0) ----
# Group by (name, locality): a multi-location chain legitimately shares a name
# across cities (e.g. "더 파크뷰" in Seoul AND Jeju). A real NAP conflict is a
# SINGLE location with contradictory phone/street, so scope the check per city.
by_name = defaultdict(list)
for entry, node in node_index:
if type_of(node) in NAP_TYPES and node.get("name"):
by_name[normalize_name(first_text(node.get("name")))].append((entry, node))
for name, group in by_name.items():
key = (normalize_name(first_text(node.get("name"))), _address_locality(node))
by_name[key].append((entry, node))
for (name, locality), group in by_name.items():
loc = f" ({locality})" if locality else ""
phones = {str(first_text(n.get("telephone"))).strip()
for _, n in group if n.get("telephone")}
streets = {_address_street(n) for _, n in group if _address_street(n)}
if len(phones) > 1:
defects.add("P0", "L4", "NAP_PHONE_MISMATCH",
f"Business '{name}' has conflicting telephone values across "
f"Business '{name}'{loc} has conflicting telephone values across "
f"entries: {sorted(phones)}.", entry_id="(dataset)")
if len(streets) > 1:
defects.add("P0", "L4", "NAP_ADDRESS_MISMATCH",
f"Business '{name}' has conflicting streetAddress values across "
f"Business '{name}'{loc} has conflicting streetAddress values across "
f"entries: {sorted(streets)}.", entry_id="(dataset)")
# ---- @id duplicates + dangling references (P1) ----
@@ -719,10 +734,147 @@ def layer4_consistency(node_index, parsed_docs, rules, defects):
f"(e.g. {sorted(eids)[:3]}): {desc[:50]!r}…", entry_id="(dataset)")
# --------------------------------------------------------------------------- #
# Layer R — Reference-URL integrity (sameAs / external identity links)
# Hardened after the Shilla incident (2026-05-29): LLM-fabricated Wikidata IDs
# (Q-numbers pointing to unrelated entities) and google.com/search reference
# URLs shipped undetected. Offline: forbid search-result URLs (P0) and flag any
# external identity ref as REFERENCE_UNVERIFIED (P1). Online (--verify-refs):
# resolve every ref; for Wikidata, fetch the label and compare to the entity
# name — a mismatch is a FALSE_REFERENCE (P0).
# --------------------------------------------------------------------------- #
def _http_get(url, timeout=12, accept=None):
import urllib.request, urllib.parse
# percent-encode non-ASCII path/query (e.g. ko.wikipedia.org/wiki/호텔신라)
p = urllib.parse.urlsplit(url)
url = urllib.parse.urlunsplit((p.scheme, p.netloc,
urllib.parse.quote(p.path),
urllib.parse.quote(p.query, safe="=&?"),
p.fragment))
req = urllib.request.Request(url, headers={
"User-Agent": "schema-ref-validator/1.0 (+offline-qa)",
**({"Accept": accept} if accept else {})})
return urllib.request.urlopen(req, timeout=timeout)
def _wikidata_labels(qid, timeout=12):
import urllib.request, json as _json
url = f"https://www.wikidata.org/w/api.php?action=wbgetentities&ids={qid}&props=labels&format=json"
with _http_get(url, timeout=timeout, accept="application/json") as r:
data = _json.loads(r.read().decode("utf-8"))
ent = data.get("entities", {}).get(qid, {})
if "missing" in ent:
return None
return {lang: v.get("value", "") for lang, v in ent.get("labels", {}).items()}
def layer_references(node_index, rules, defects, verify_refs=False):
policy = rules.get("reference_policy")
if not policy:
return
forbidden = policy.get("forbidden_url_substrings", [])
ref_props = set(policy.get("identity_ref_props", ["sameAs"]))
import re as _re
qid_re = _re.compile(r"wikidata\.org/(?:wiki|entity)/(Q\d+)")
def every_string(obj):
# Unlike all_strings(), this also yields strings nested inside lists
# (e.g. each URL in a sameAs array) — the exact case missed before.
if isinstance(obj, dict):
for v in obj.values():
yield from every_string(v)
elif isinstance(obj, list):
for v in obj:
yield from every_string(v)
elif isinstance(obj, str):
yield obj
for entry, node in node_index:
eid, url, ntype = entry["entry_id"], entry["url"], type_of(node)
# (1) forbidden search-result URLs anywhere in the node (incl. list items) -> P0
for val in every_string(node):
low = val.lower()
hit = next((s for s in forbidden if s in low), None)
if hit:
defects.add("P0", "LR", "FORBIDDEN_REFERENCE",
f"Search-result URL used as reference (contains {hit!r}); "
f"not a valid entity reference: {val[:80]!r}.", eid, url, ntype)
# (2) external identity references (sameAs)
refs = []
for prop in ref_props:
v = node.get(prop)
if isinstance(v, str):
refs.append(v)
elif isinstance(v, list):
refs += [x for x in v if isinstance(x, str)]
if not refs:
continue
# (2a) discouraged reference sources (policy: prefer Wikipedia over Wikidata)
for dd in policy.get("discouraged_ref_domains", []):
for ref in refs:
if dd in ref:
defects.add("P1", "LR", "DISCOURAGED_REFERENCE",
f"{ref} uses a discouraged source ({dd}). Policy: prefer "
"Wikipedia; if none verified, omit — never fabricate.",
eid, url, ntype)
if not verify_refs:
defects.add("P1", "LR", "REFERENCE_UNVERIFIED",
f"{len(refs)} external reference(s) on '{ntype}' not machine-verified "
f"(run with --verify-refs / confirm online): {refs}.", eid, url, ntype)
continue
# online verification
name = normalize_name(first_text(node.get("name")))
alts = node.get("alternateName") or []
if isinstance(alts, str):
alts = [alts]
names = {name} | {normalize_name(a) for a in alts if isinstance(a, str)}
for ref in refs:
m = qid_re.search(ref)
if m:
try:
labels = _wikidata_labels(m.group(1))
except Exception as e:
defects.add("P1", "LR", "REFERENCE_UNREACHABLE",
f"Could not fetch Wikidata {m.group(1)} ({e}).", eid, url, ntype)
continue
if labels is None:
defects.add("P0", "LR", "FALSE_REFERENCE",
f"sameAs {ref} → Wikidata item is missing/deleted.", eid, url, ntype)
continue
lab = {normalize_name(v) for v in labels.values()}
# match if any entity name appears in any label or vice-versa
ok = any(n and (n in l or l in n) for n in names for l in lab)
if not ok:
defects.add("P0", "LR", "FALSE_REFERENCE",
f"sameAs {ref} label {sorted(lab)[:3]} does NOT match entity "
f"name {sorted(n for n in names if n)[:3]} — fabricated/incorrect ID.",
eid, url, ntype)
else:
is_social = any(d in ref for d in policy.get("social_profile_domains", []))
try:
code = _http_get(ref).status
except Exception as e:
code = f"error: {e}"
if is_social:
# HTTP 200 does NOT prove official ownership, and platforms
# often bot-block live pages (e.g. Facebook 400). Always hand
# social/profile refs to a human. (Shilla: a 200 YouTube
# channel was not the official one; FB page was closed.)
defects.add("P1", "LR", "SOCIAL_UNVERIFIED",
f"sameAs {ref} is a social/profile URL (HTTP {code}). "
"Confirm official ownership AND active status manually — "
"a 200 is not proof of ownership.", eid, url, ntype)
elif isinstance(code, int) and code >= 400:
defects.add("P0", "LR", "BROKEN_REFERENCE",
f"sameAs {ref} returned HTTP {code}.", eid, url, ntype)
elif not isinstance(code, int):
defects.add("P1", "LR", "REFERENCE_UNREACHABLE",
f"sameAs {ref} not reachable ({code}).", eid, url, ntype)
# --------------------------------------------------------------------------- #
# Orchestration + output
# --------------------------------------------------------------------------- #
def run(entries, rules, inventory, strict, no_recommended):
def run(entries, rules, inventory, strict, no_recommended, verify_refs=False):
defects = DefectLog()
if inventory is not None:
layer0_coverage(entries, inventory, defects)
@@ -743,6 +895,7 @@ def run(entries, rules, inventory, strict, no_recommended):
node_index.append((entry, node))
layer4_consistency(node_index, parsed_docs, rules, defects)
layer_references(node_index, rules, defects, verify_refs=verify_refs)
return defects, valid_entries, len(node_index)
@@ -822,6 +975,9 @@ def main(argv=None):
ap.add_argument("--live", nargs="+", metavar="URL",
help="Mode B: validate live URLs (extract embedded JSON-LD)")
ap.add_argument("--rules", default=str(RULES_DEFAULT), help="path to schema_rules.json")
ap.add_argument("--verify-refs", action="store_true",
help="online: resolve every sameAs and verify Wikidata labels match the "
"entity name (catches fabricated/incorrect reference IDs). Needs network.")
args = ap.parse_args(argv)
if not args.dataset and not args.live:
@@ -838,7 +994,8 @@ def main(argv=None):
inventory = load_url_inventory(args.url_list) if args.url_list else None
defects, valid_entries, nodes = run(entries, rules, inventory,
args.strict, args.no_recommended)
args.strict, args.no_recommended,
verify_refs=args.verify_refs)
meta = {"entries": len(entries), "valid_entries": valid_entries, "nodes": nodes,
"mode": "B-live" if args.live else "A-dataset", "strict": args.strict,
"coverage": inventory is not None}