1012 lines
43 KiB
Python
1012 lines
43 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
validate_schema.py — 5-layer offline JSON-LD schema validator.
|
||
|
||
WHY THIS EXISTS
|
||
---------------
|
||
When a client reviews hundreds of authored schema entries and says "there are too
|
||
many errors," the root cause is almost always that nobody ran a machine lint first.
|
||
Humans end up eyeballing raw JSON in a meeting. This tool moves every cheap,
|
||
machine-checkable error OUT of human review and INTO an automated gate that runs
|
||
first — so the client only ever sees clean, P0-free entries plus a defect report.
|
||
|
||
It is OFFLINE by design (the runtime cannot reach schema.org or Google). All rules
|
||
live in schema_rules.json; unknown types/properties degrade to warnings, never hard
|
||
errors, so the gate does not invent false positives.
|
||
|
||
THE 5 LAYERS
|
||
------------
|
||
L0 Coverage — URLs with no entry; entries whose URL isn't in the inventory.
|
||
L1 Syntax — invalid JSON, bad/missing @context, missing @type, encoding corruption.
|
||
L2 Vocabulary — unknown type, value-format errors (URL/date/lang/currency/number),
|
||
(strict only) unexpected properties on a known type.
|
||
L3 Rich-result — Google REQUIRED property missing (blocks rich result); recommended absent.
|
||
L4 Consistency — NAP mismatch, @id duplicates/dangling refs, swapped geo,
|
||
placeholder text, duplicate descriptions across entries.
|
||
|
||
GATE: PASS iff zero P0. Process exits 1 when the gate fails (so CI/`&&` chains stop).
|
||
|
||
Usage:
|
||
python validate_schema.py DATASET [--url-list URLLIST] [--out DIR]
|
||
[--strict] [--no-recommended]
|
||
[--live URL ...] [--rules schema_rules.json]
|
||
DATASET may be .xlsx / .csv (one row per entry, a JSON-LD column) / .jsonl / .json
|
||
/ a directory of .json|.jsonld files. With --live, validate live URLs instead.
|
||
"""
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
|
||
RULES_DEFAULT = Path(__file__).resolve().parent / "schema_rules.json"
|
||
|
||
SEVERITY_ORDER = {"P0": 0, "P1": 1, "P2": 2}
|
||
|
||
# Header aliases for tabular input. Keys are normalized (lowercased, spaces removed).
|
||
COLUMN_ALIASES = {
|
||
"jsonld": ["jsonld", "jsonld", "json-ld", "json_ld", "schema", "schemamarkup",
|
||
"structureddata", "structured_data", "markup", "스키마", "구조화데이터",
|
||
"구조화된데이터", "jsonldcode", "스키마코드"],
|
||
"url": ["url", "메뉴url", "pageurl", "주소", "링크", "loc", "uri", "캐노니컬", "canonical"],
|
||
"lang": ["lang", "language", "언어", "언어코드", "locale", "lng"],
|
||
"device": ["device", "pc/mobile", "pcmobile", "pc_mobile", "platform", "디바이스", "기기"],
|
||
"page_type": ["page_type", "pagetype", "type", "페이지유형", "페이지타입", "menulevel",
|
||
"menu_level", "메뉴레벨", "template", "템플릿", "유형"],
|
||
}
|
||
|
||
URL_RE = re.compile(r"^https?://[^\s]+$", re.IGNORECASE)
|
||
# ISO-8601 date or datetime (date, date+time, optional tz). Loose but rejects free text.
|
||
DATE_RE = re.compile(
|
||
r"^\d{4}-\d{2}-\d{2}"
|
||
r"(?:[T ]\d{2}:\d{2}(?::\d{2})?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)?$"
|
||
)
|
||
LANG_RE = re.compile(r"^[a-zA-Z]{2,3}(?:-[A-Za-z0-9]{2,4})?$")
|
||
JSONLD_SCRIPT_RE = re.compile(
|
||
r'<script[^>]+type=["\']application/ld\+json["\'][^>]*>(.*?)</script>',
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Defect collection
|
||
# --------------------------------------------------------------------------- #
|
||
class DefectLog:
|
||
"""Accumulates findings. One row per finding, ready for triage."""
|
||
|
||
def __init__(self):
|
||
self.rows = []
|
||
|
||
def add(self, severity, layer, code, message, entry_id="", url="", node_type=""):
|
||
self.rows.append({
|
||
"entry_id": str(entry_id),
|
||
"url": url or "",
|
||
"node_type": node_type or "",
|
||
"layer": layer,
|
||
"code": code,
|
||
"severity": severity,
|
||
"message": message,
|
||
"status": "open",
|
||
"owner": "",
|
||
"note": "",
|
||
})
|
||
|
||
def counts(self):
|
||
c = Counter(r["severity"] for r in self.rows)
|
||
return {"P0": c.get("P0", 0), "P1": c.get("P1", 0), "P2": c.get("P2", 0)}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Input adapters (Mode A: authored dataset / Mode B: live URLs)
|
||
# --------------------------------------------------------------------------- #
|
||
def _norm_header(h):
|
||
return re.sub(r"\s+", "", str(h or "").strip().lower())
|
||
|
||
|
||
def _detect_columns(headers):
|
||
"""Map normalized headers to canonical column roles. Returns {role: index}."""
|
||
found = {}
|
||
for idx, h in enumerate(headers):
|
||
nh = _norm_header(h)
|
||
for role, aliases in COLUMN_ALIASES.items():
|
||
if role in found:
|
||
continue
|
||
if nh in aliases:
|
||
found[role] = idx
|
||
return found
|
||
|
||
|
||
def _row_to_entry(row, cols, entry_id, source_ref):
|
||
def cell(role):
|
||
i = cols.get(role)
|
||
if i is None or i >= len(row):
|
||
return None
|
||
v = row[i]
|
||
return None if v is None else str(v).strip()
|
||
raw = cell("jsonld")
|
||
if not raw:
|
||
return None # blank JSON-LD cell → no entry to validate
|
||
return {
|
||
"entry_id": entry_id,
|
||
"url": cell("url") or "",
|
||
"lang": cell("lang") or "",
|
||
"device": cell("device") or "",
|
||
"page_type": cell("page_type") or "",
|
||
"raw": raw,
|
||
"source_ref": source_ref,
|
||
}
|
||
|
||
|
||
def _load_csv(path):
|
||
entries = []
|
||
with open(path, newline="", encoding="utf-8-sig") as f:
|
||
reader = csv.reader(f)
|
||
rows = list(reader)
|
||
if not rows:
|
||
return entries
|
||
cols = _detect_columns(rows[0])
|
||
if "jsonld" not in cols:
|
||
raise ValueError(
|
||
f"No JSON-LD column found in {path}. Looked for: "
|
||
f"{', '.join(COLUMN_ALIASES['jsonld'][:6])} … Headers were: {rows[0]}"
|
||
)
|
||
for n, row in enumerate(rows[1:], start=2):
|
||
e = _row_to_entry(row, cols, f"{Path(path).stem}#r{n}", f"{path}:row{n}")
|
||
if e:
|
||
entries.append(e)
|
||
return entries
|
||
|
||
|
||
def _load_xlsx(path):
|
||
try:
|
||
from openpyxl import load_workbook
|
||
except ImportError:
|
||
raise SystemExit(
|
||
"Reading .xlsx needs openpyxl: pip install openpyxl\n"
|
||
"(or export the sheet to .csv and pass that instead)."
|
||
)
|
||
entries = []
|
||
wb = load_workbook(path, read_only=True, data_only=True)
|
||
for sheet in wb.worksheets:
|
||
rows = list(sheet.iter_rows(values_only=True))
|
||
if not rows:
|
||
continue
|
||
cols = _detect_columns(rows[0])
|
||
if "jsonld" not in cols:
|
||
continue # a tab without a JSON-LD column (e.g. summary) — skip silently
|
||
for n, row in enumerate(rows[1:], start=2):
|
||
e = _row_to_entry(list(row), cols, f"{sheet.title}#r{n}",
|
||
f"{path}:{sheet.title}:row{n}")
|
||
if e:
|
||
entries.append(e)
|
||
if not entries:
|
||
raise ValueError(
|
||
f"No sheet in {path} had a recognizable JSON-LD column. "
|
||
f"Looked for: {', '.join(COLUMN_ALIASES['jsonld'][:6])} …"
|
||
)
|
||
return entries
|
||
|
||
|
||
def _looks_like_schema(obj):
|
||
"""True if a parsed object is itself JSON-LD (vs a wrapper row)."""
|
||
if isinstance(obj, list):
|
||
return True
|
||
if isinstance(obj, dict):
|
||
return any(k in obj for k in ("@context", "@type", "@graph"))
|
||
return False
|
||
|
||
|
||
def _wrapper_to_entry(obj, entry_id, source_ref):
|
||
"""A JSONL/JSON wrapper object that carries url/lang + a jsonld payload."""
|
||
cols = {k: k for k in obj.keys()}
|
||
norm = {_norm_header(k): k for k in obj.keys()}
|
||
def pick(role):
|
||
for alias in COLUMN_ALIASES[role]:
|
||
if alias in norm:
|
||
v = obj[norm[alias]]
|
||
return v
|
||
return None
|
||
payload = pick("jsonld")
|
||
raw = payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False)
|
||
url = pick("url")
|
||
return {
|
||
"entry_id": entry_id,
|
||
"url": str(url).strip() if url else "",
|
||
"lang": str(pick("lang") or "").strip(),
|
||
"device": str(pick("device") or "").strip(),
|
||
"page_type": str(pick("page_type") or "").strip(),
|
||
"raw": raw,
|
||
"source_ref": source_ref,
|
||
}
|
||
|
||
|
||
def _load_jsonl(path):
|
||
entries = []
|
||
with open(path, encoding="utf-8") as f:
|
||
for n, line in enumerate(f, start=1):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
sref = f"{path}:line{n}"
|
||
try:
|
||
obj = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
# Keep the bad line so L1 reports it as a syntax error.
|
||
entries.append({"entry_id": f"{Path(path).stem}#l{n}", "url": "",
|
||
"lang": "", "device": "", "page_type": "",
|
||
"raw": line, "source_ref": sref})
|
||
continue
|
||
eid = f"{Path(path).stem}#l{n}"
|
||
if _looks_like_schema(obj):
|
||
entries.append({"entry_id": eid, "url": "", "lang": "", "device": "",
|
||
"page_type": "", "raw": line, "source_ref": sref})
|
||
else:
|
||
entries.append(_wrapper_to_entry(obj, eid, sref))
|
||
return entries
|
||
|
||
|
||
def _load_json(path):
|
||
with open(path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
entries = []
|
||
if isinstance(data, dict) and not _looks_like_schema(data) and all(
|
||
isinstance(v, (dict, list, str)) for v in data.values()
|
||
) and not any(k.startswith("@") for k in data):
|
||
# url -> jsonld map
|
||
for url, payload in data.items():
|
||
raw = payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False)
|
||
entries.append({"entry_id": url, "url": url, "lang": "", "device": "",
|
||
"page_type": "", "raw": raw, "source_ref": f"{path}:{url}"})
|
||
elif isinstance(data, list):
|
||
for n, item in enumerate(data, start=1):
|
||
sref = f"{path}:[{n}]"
|
||
eid = f"{Path(path).stem}#{n}"
|
||
if _looks_like_schema(item) or not isinstance(item, dict):
|
||
raw = item if isinstance(item, str) else json.dumps(item, ensure_ascii=False)
|
||
entries.append({"entry_id": eid, "url": "", "lang": "", "device": "",
|
||
"page_type": "", "raw": raw, "source_ref": sref})
|
||
else:
|
||
entries.append(_wrapper_to_entry(item, eid, sref))
|
||
else:
|
||
entries.append({"entry_id": Path(path).stem, "url": "", "lang": "", "device": "",
|
||
"page_type": "", "raw": json.dumps(data, ensure_ascii=False),
|
||
"source_ref": path})
|
||
return entries
|
||
|
||
|
||
def _load_dir(path):
|
||
entries = []
|
||
for p in sorted(Path(path).rglob("*")):
|
||
if p.suffix.lower() in (".json", ".jsonld"):
|
||
entries.append({"entry_id": p.stem, "url": "", "lang": "", "device": "",
|
||
"page_type": "", "raw": p.read_text(encoding="utf-8"),
|
||
"source_ref": str(p)})
|
||
if not entries:
|
||
raise ValueError(f"No .json/.jsonld files found under {path}")
|
||
return entries
|
||
|
||
|
||
def _load_live(urls):
|
||
try:
|
||
import requests
|
||
except ImportError:
|
||
raise SystemExit("Live mode (--live) needs requests: pip install requests")
|
||
entries = []
|
||
headers = {"User-Agent": "Mozilla/5.0 (compatible; SchemaValidator/1.0)"}
|
||
for url in urls:
|
||
try:
|
||
resp = requests.get(url, headers=headers, timeout=20)
|
||
resp.raise_for_status()
|
||
except Exception as exc: # noqa: BLE001 — best-effort live fetch
|
||
entries.append({"entry_id": url, "url": url, "lang": "", "device": "",
|
||
"page_type": "", "raw": "", "source_ref": url,
|
||
"_fetch_error": str(exc)})
|
||
continue
|
||
scripts = JSONLD_SCRIPT_RE.findall(resp.text)
|
||
if not scripts:
|
||
entries.append({"entry_id": url, "url": url, "lang": "", "device": "",
|
||
"page_type": "", "raw": "", "source_ref": url,
|
||
"_no_schema": True})
|
||
continue
|
||
for i, block in enumerate(scripts, start=1):
|
||
entries.append({"entry_id": f"{url}#{i}", "url": url, "lang": "",
|
||
"device": "", "page_type": "", "raw": block.strip(),
|
||
"source_ref": f"{url} (script {i})"})
|
||
return entries
|
||
|
||
|
||
def load_entries(input_path, live_urls):
|
||
if live_urls:
|
||
return _load_live(live_urls)
|
||
p = Path(input_path)
|
||
if p.is_dir():
|
||
return _load_dir(p)
|
||
suffix = p.suffix.lower()
|
||
if suffix == ".csv":
|
||
return _load_csv(p)
|
||
if suffix in (".xlsx", ".xlsm"):
|
||
return _load_xlsx(p)
|
||
if suffix == ".jsonl":
|
||
return _load_jsonl(p)
|
||
if suffix in (".json", ".jsonld"):
|
||
return _load_json(p)
|
||
raise ValueError(f"Unsupported input: {input_path} (suffix {suffix!r})")
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Node helpers
|
||
# --------------------------------------------------------------------------- #
|
||
def type_of(node):
|
||
"""Return the primary @type as a string (first if it's a list)."""
|
||
t = node.get("@type")
|
||
if isinstance(t, list):
|
||
return t[0] if t else ""
|
||
return t or ""
|
||
|
||
|
||
def iter_typed_nodes(parsed):
|
||
"""Yield every dict that has an @type, top-level and nested (recursively)."""
|
||
seen = []
|
||
|
||
def walk(obj):
|
||
if isinstance(obj, dict):
|
||
if "@type" in obj:
|
||
seen.append(obj)
|
||
for v in obj.values():
|
||
walk(v)
|
||
elif isinstance(obj, list):
|
||
for v in obj:
|
||
walk(v)
|
||
|
||
# @graph documents: walk the graph; otherwise walk the object/array directly.
|
||
if isinstance(parsed, dict) and "@graph" in parsed:
|
||
walk(parsed["@graph"])
|
||
else:
|
||
walk(parsed)
|
||
return seen
|
||
|
||
|
||
def all_strings(obj):
|
||
"""Yield (key, value) for every string value anywhere in the structure."""
|
||
if isinstance(obj, dict):
|
||
for k, v in obj.items():
|
||
if isinstance(v, str):
|
||
yield k, v
|
||
else:
|
||
yield from all_strings(v)
|
||
elif isinstance(obj, list):
|
||
for v in obj:
|
||
yield from all_strings(v)
|
||
|
||
|
||
def normalize_name(s):
|
||
return re.sub(r"\s+", " ", str(s or "").strip().lower())
|
||
|
||
|
||
def first_text(value):
|
||
"""Coerce a property value to a comparable scalar (handles list/dict)."""
|
||
if isinstance(value, list):
|
||
return first_text(value[0]) if value else ""
|
||
if isinstance(value, dict):
|
||
return value.get("name") or value.get("@id") or value.get("streetAddress") or ""
|
||
return value
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Layer 0 — Coverage
|
||
# --------------------------------------------------------------------------- #
|
||
def load_url_inventory(url_list_path):
|
||
urls = set()
|
||
p = Path(url_list_path)
|
||
suffix = p.suffix.lower()
|
||
if suffix in (".xlsx", ".xlsm"):
|
||
from openpyxl import load_workbook
|
||
wb = load_workbook(p, read_only=True, data_only=True)
|
||
for sheet in wb.worksheets:
|
||
for row in sheet.iter_rows(values_only=True):
|
||
for cell in row:
|
||
if isinstance(cell, str) and URL_RE.match(cell.strip()):
|
||
urls.add(cell.strip())
|
||
elif suffix == ".csv":
|
||
with open(p, newline="", encoding="utf-8-sig") as f:
|
||
for row in csv.reader(f):
|
||
for cell in row:
|
||
if isinstance(cell, str) and URL_RE.match(cell.strip()):
|
||
urls.add(cell.strip())
|
||
else: # plain text, one URL per line
|
||
for line in p.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if URL_RE.match(line):
|
||
urls.add(line)
|
||
return urls
|
||
|
||
|
||
def layer0_coverage(entries, inventory, defects):
|
||
entry_urls = {e["url"] for e in entries if e.get("url")}
|
||
missing = inventory - entry_urls
|
||
for url in sorted(missing):
|
||
defects.add("P1", "L0", "COVERAGE_MISSING",
|
||
"Inventory URL has no authored schema entry.", url=url)
|
||
orphans = entry_urls - inventory
|
||
for url in sorted(orphans):
|
||
defects.add("P2", "L0", "COVERAGE_ORPHAN",
|
||
"Entry URL is not in the canonical URL inventory "
|
||
"(typo, stale path, or missing from list).", url=url)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Layer 1 — Syntax
|
||
# --------------------------------------------------------------------------- #
|
||
def layer1_syntax(entry, rules, defects):
|
||
"""Parse + structural checks. Returns parsed object or None (fatal)."""
|
||
eid, url = entry["entry_id"], entry["url"]
|
||
if entry.get("_fetch_error"):
|
||
defects.add("P1", "L1", "FETCH_ERROR",
|
||
f"Could not fetch live URL: {entry['_fetch_error']}", eid, url)
|
||
return None
|
||
if entry.get("_no_schema"):
|
||
defects.add("P0", "L1", "NO_SCHEMA_IN_HTML",
|
||
"Live page has no application/ld+json script block.", eid, url)
|
||
return None
|
||
raw = entry["raw"]
|
||
if "<EFBFBD>" in raw:
|
||
defects.add("P1", "L1", "ENCODING_CORRUPTION",
|
||
"Replacement character (\\ufffd) present — encoding corruption.",
|
||
eid, url)
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
defects.add("P0", "L1", "INVALID_JSON",
|
||
f"JSON does not parse: {exc.msg} at line {exc.lineno} col {exc.colno}.",
|
||
eid, url)
|
||
return None
|
||
|
||
nodes = iter_typed_nodes(parsed)
|
||
if not nodes:
|
||
defects.add("P1", "L1", "NO_TYPE",
|
||
"No @type found anywhere in the entry — not a usable schema object.",
|
||
eid, url)
|
||
|
||
# @context lives at the top of the document; nested nodes inherit it.
|
||
if isinstance(parsed, dict):
|
||
ctx = parsed.get("@context")
|
||
if ctx is None:
|
||
defects.add("P1", "L1", "MISSING_CONTEXT",
|
||
"Top-level @context is missing.", eid, url)
|
||
else:
|
||
ctx_urls = [ctx] if isinstance(ctx, str) else (
|
||
[c for c in ctx if isinstance(c, str)] if isinstance(ctx, list) else []
|
||
)
|
||
valid = rules["valid_contexts"]
|
||
if ctx_urls and not any(c.rstrip("/") in [v.rstrip("/") for v in valid]
|
||
for c in ctx_urls):
|
||
defects.add("P1", "L1", "WRONG_CONTEXT",
|
||
f"@context is not schema.org: {ctx_urls}.", eid, url)
|
||
return parsed
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Layer 2 — Vocabulary + value formats
|
||
# --------------------------------------------------------------------------- #
|
||
def _check_value_formats(node, rules, defects, eid, url, ntype, severity):
|
||
vf = rules["value_formats"]
|
||
|
||
def each(value):
|
||
if isinstance(value, list):
|
||
for v in value:
|
||
yield from each(v)
|
||
else:
|
||
yield value
|
||
|
||
for prop, value in node.items():
|
||
if prop.startswith("@"):
|
||
continue
|
||
if prop in vf["url_props"]:
|
||
for v in each(value):
|
||
if isinstance(v, str) and not URL_RE.match(v.strip()):
|
||
defects.add(severity, "L2", "BAD_URL",
|
||
f"'{prop}' is not an http(s) URL: {v!r}.", eid, url, ntype)
|
||
if prop in vf["date_props"]:
|
||
for v in each(value):
|
||
if isinstance(v, str) and not DATE_RE.match(v.strip()):
|
||
defects.add(severity, "L2", "BAD_DATE",
|
||
f"'{prop}' is not ISO-8601: {v!r}.", eid, url, ntype)
|
||
if prop in vf["lang_props"]:
|
||
for v in each(value):
|
||
if isinstance(v, str) and not LANG_RE.match(v.strip()):
|
||
defects.add(severity, "L2", "BAD_LANG",
|
||
f"'{prop}' is not a BCP-47 language code: {v!r}.",
|
||
eid, url, ntype)
|
||
if prop in vf["currency_props"]:
|
||
for v in each(value):
|
||
if isinstance(v, str) and not re.match(r"^[A-Z]{3}$", v.strip()):
|
||
defects.add(severity, "L2", "BAD_CURRENCY",
|
||
f"'{prop}' is not a 3-letter ISO-4217 code: {v!r}.",
|
||
eid, url, ntype)
|
||
if prop in vf["number_props"]:
|
||
for v in each(value):
|
||
if isinstance(v, str):
|
||
try:
|
||
float(v.replace(",", ""))
|
||
except ValueError:
|
||
defects.add(severity, "L2", "BAD_NUMBER",
|
||
f"'{prop}' is not numeric: {v!r}.", eid, url, ntype)
|
||
|
||
|
||
def layer2_vocabulary(node, rules, defects, eid, url, strict):
|
||
ntype = type_of(node)
|
||
known = rules["known_types"]
|
||
containers = set(rules["container_types"])
|
||
minor = "P1" if strict else "P2"
|
||
|
||
if ntype and ntype not in known and ntype not in containers:
|
||
defects.add(minor, "L2", "UNKNOWN_TYPE",
|
||
f"@type '{ntype}' is not in the curated rule set "
|
||
"(treated as a warning — add it to schema_rules.json if intended).",
|
||
eid, url, ntype)
|
||
|
||
_check_value_formats(node, rules, defects, eid, url, ntype, minor)
|
||
|
||
# Unexpected-property check is OPT-IN (--strict). Off by default to avoid the
|
||
# exact noise explosion that makes clients say "too many errors".
|
||
if strict and ntype in known:
|
||
spec = known[ntype]
|
||
allowed = set(spec["required"]) | set(spec["recommended"]) | set(spec["allowed"])
|
||
allowed |= set(rules["global_properties"])
|
||
for prop in node:
|
||
if prop.startswith("@"):
|
||
continue
|
||
if prop not in allowed:
|
||
defects.add("P1", "L2", "UNEXPECTED_PROPERTY",
|
||
f"'{prop}' is not a known property of {ntype} (strict mode).",
|
||
eid, url, ntype)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Layer 3 — Rich-result (required / recommended)
|
||
# --------------------------------------------------------------------------- #
|
||
def layer3_richresult(node, rules, defects, eid, url, no_recommended):
|
||
ntype = type_of(node)
|
||
known = rules["known_types"]
|
||
if ntype not in known:
|
||
return # containers + unknown types have no required-property contract
|
||
spec = known[ntype]
|
||
|
||
for prop in spec["required"]:
|
||
if not node.get(prop):
|
||
defects.add("P0", "L3", "MISSING_REQUIRED",
|
||
f"{ntype} is missing required property '{prop}' "
|
||
"(blocks the rich result).", eid, url, ntype)
|
||
|
||
if not no_recommended:
|
||
missing_rec = [p for p in spec["recommended"] if not node.get(p)]
|
||
if missing_rec:
|
||
# Aggregate to ONE line per node — never one defect per property.
|
||
defects.add("P2", "L3", "MISSING_RECOMMENDED",
|
||
f"{ntype} is missing recommended properties: "
|
||
f"{', '.join(missing_rec)}.", eid, url, ntype)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Layer 4 — Consistency (cross-node / cross-entry)
|
||
# --------------------------------------------------------------------------- #
|
||
NAP_TYPES = {"Organization", "Corporation", "LocalBusiness", "Hotel",
|
||
"LodgingBusiness", "Resort", "Restaurant", "FoodEstablishment", "BarOrPub"}
|
||
|
||
|
||
def _address_street(node):
|
||
addr = node.get("address")
|
||
if isinstance(addr, dict):
|
||
return normalize_name(addr.get("streetAddress"))
|
||
if isinstance(addr, list) and addr and isinstance(addr[0], dict):
|
||
return normalize_name(addr[0].get("streetAddress"))
|
||
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.
|
||
|
||
A *reference* is an object whose only key is @id (e.g. {"@id": "...#org"}).
|
||
A *definition* is any object carrying @id plus other content. References live
|
||
in untyped wrapper dicts, so this must walk the raw doc — not just typed nodes.
|
||
"""
|
||
if isinstance(obj, dict):
|
||
nid = obj.get("@id")
|
||
if nid:
|
||
if set(obj.keys()) == {"@id"}:
|
||
referenced.add(nid)
|
||
else:
|
||
defined.setdefault(nid, []).append(obj)
|
||
for v in obj.values():
|
||
_walk_ids(v, defined, referenced)
|
||
elif isinstance(obj, list):
|
||
for v in obj:
|
||
_walk_ids(v, defined, referenced)
|
||
|
||
|
||
def layer4_consistency(node_index, parsed_docs, rules, defects):
|
||
"""node_index: (entry, node) for every TYPED node.
|
||
parsed_docs: (entry, parsed) for every entry that parsed — used for @id scan."""
|
||
# ---- placeholder text (P0) ----
|
||
tokens = [t.lower() for t in rules["placeholder_tokens"]]
|
||
for entry, node in node_index:
|
||
ntype = type_of(node)
|
||
for key, val in all_strings(node):
|
||
low = val.lower()
|
||
hit = next((t for t in tokens if t in low), None)
|
||
if hit:
|
||
defects.add("P0", "L4", "PLACEHOLDER_TEXT",
|
||
f"Placeholder/boilerplate token {hit!r} in '{key}': {val[:60]!r}.",
|
||
entry["entry_id"], entry["url"], ntype)
|
||
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"):
|
||
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}'{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}'{loc} has conflicting streetAddress values across "
|
||
f"entries: {sorted(streets)}.", entry_id="(dataset)")
|
||
|
||
# ---- @id duplicates + dangling references (P1) ----
|
||
defined = {} # @id -> list of definition dicts (walked across all docs)
|
||
referenced = set() # @id values used purely as references
|
||
for _, parsed in parsed_docs:
|
||
_walk_ids(parsed, defined, referenced)
|
||
for nid, defs in defined.items():
|
||
if len(defs) > 1:
|
||
# duplicate only matters if the definitions actually differ
|
||
shapes = {json.dumps(n, sort_keys=True, ensure_ascii=False) for n in defs}
|
||
if len(shapes) > 1:
|
||
defects.add("P1", "L4", "DUPLICATE_ID",
|
||
f"@id {nid!r} is defined {len(defs)} times with differing content.",
|
||
entry_id="(dataset)")
|
||
for nid in sorted(referenced - set(defined)):
|
||
defects.add("P1", "L4", "DANGLING_ID",
|
||
f"@id reference {nid!r} points to a node that is never defined.",
|
||
entry_id="(dataset)")
|
||
|
||
# ---- swapped / out-of-range geo (P1) ----
|
||
g = rules["geo"]
|
||
for entry, node in node_index:
|
||
if type_of(node) != "GeoCoordinates":
|
||
continue
|
||
try:
|
||
lat = float(first_text(node.get("latitude")))
|
||
lon = float(first_text(node.get("longitude")))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
lat_ok = g["lat_min"] <= lat <= g["lat_max"]
|
||
lon_ok = g["lon_min"] <= lon <= g["lon_max"]
|
||
if lat_ok and lon_ok:
|
||
continue # both in valid global range
|
||
# Invalid — distinguish a clean transposition from plain garbage.
|
||
swap_ok = (g["lat_min"] <= lon <= g["lat_max"]) and (g["lon_min"] <= lat <= g["lon_max"])
|
||
if swap_ok:
|
||
defects.add("P1", "L4", "GEO_SWAPPED",
|
||
f"GeoCoordinates look transposed (latitude={lat}, longitude={lon}) "
|
||
"— swapping them yields valid coordinates.",
|
||
entry["entry_id"], entry["url"], "GeoCoordinates")
|
||
else:
|
||
defects.add("P1", "L4", "GEO_OUT_OF_RANGE",
|
||
f"GeoCoordinates out of range (latitude={lat}, longitude={lon}).",
|
||
entry["entry_id"], entry["url"], "GeoCoordinates")
|
||
|
||
# ---- duplicate descriptions across entries (P1) ----
|
||
desc_groups = defaultdict(set)
|
||
for entry, node in node_index:
|
||
d = first_text(node.get("description"))
|
||
if isinstance(d, str) and len(d.strip()) >= 30:
|
||
desc_groups[d.strip()].add(entry["entry_id"])
|
||
for desc, eids in desc_groups.items():
|
||
if len(eids) >= 3:
|
||
defects.add("P1", "L4", "DUPLICATE_DESCRIPTION",
|
||
f"Identical description reused across {len(eids)} entries "
|
||
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, verify_refs=False):
|
||
defects = DefectLog()
|
||
if inventory is not None:
|
||
layer0_coverage(entries, inventory, defects)
|
||
|
||
node_index = []
|
||
parsed_docs = []
|
||
valid_entries = 0
|
||
for entry in entries:
|
||
parsed = layer1_syntax(entry, rules, defects)
|
||
if parsed is None:
|
||
continue
|
||
valid_entries += 1
|
||
parsed_docs.append((entry, parsed))
|
||
for node in iter_typed_nodes(parsed):
|
||
layer2_vocabulary(node, rules, defects, entry["entry_id"], entry["url"], strict)
|
||
layer3_richresult(node, rules, defects, entry["entry_id"], entry["url"],
|
||
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)
|
||
|
||
|
||
def write_outputs(defects, outdir, meta):
|
||
outdir = Path(outdir)
|
||
outdir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# defect_log.csv — the client-facing triage artifact
|
||
fields = ["entry_id", "url", "node_type", "layer", "code", "severity",
|
||
"message", "status", "owner", "note"]
|
||
rows = sorted(defects.rows, key=lambda r: (SEVERITY_ORDER[r["severity"]],
|
||
r["layer"], r["code"]))
|
||
with open(outdir / "defect_log.csv", "w", newline="", encoding="utf-8-sig") as f:
|
||
w = csv.DictWriter(f, fieldnames=fields)
|
||
w.writeheader()
|
||
w.writerows(rows)
|
||
|
||
counts = defects.counts()
|
||
gate = "PASS" if counts["P0"] == 0 else "FAIL"
|
||
by_code = Counter((r["severity"], r["code"]) for r in defects.rows)
|
||
|
||
# results.json — machine-readable
|
||
results = {
|
||
"summary": {**meta, **counts, "total": len(rows), "gate": gate},
|
||
"by_code": [{"severity": s, "code": c, "count": n}
|
||
for (s, c), n in by_code.most_common()],
|
||
"defects": rows,
|
||
}
|
||
(outdir / "results.json").write_text(
|
||
json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
# report.md — human summary
|
||
lines = [
|
||
"# Schema Validation Report", "",
|
||
f"- Entries read: **{meta['entries']}** | parsed OK: **{meta['valid_entries']}** "
|
||
f"| nodes checked: **{meta['nodes']}**",
|
||
f"- Defects: **P0 {counts['P0']}** · **P1 {counts['P1']}** · **P2 {counts['P2']}** "
|
||
f"(total {len(rows)})",
|
||
"",
|
||
f"## Gate: **{gate}**",
|
||
("> ✅ Zero P0 — entries may advance to client review."
|
||
if gate == "PASS" else
|
||
"> ⛔ P0 present — these entries must NOT reach client review. Fix P0 first."),
|
||
"",
|
||
"## Defects by code", "",
|
||
"| Severity | Code | Count |", "|---|---|---|",
|
||
]
|
||
for (sev, code), n in by_code.most_common():
|
||
lines.append(f"| {sev} | {code} | {n} |")
|
||
|
||
p0 = [r for r in rows if r["severity"] == "P0"]
|
||
if p0:
|
||
lines += ["", "## P0 blockers (top 15)", "",
|
||
"| Entry | Type | Code | Message |", "|---|---|---|---|"]
|
||
for r in p0[:15]:
|
||
msg = r["message"].replace("|", "\\|")
|
||
lines.append(f"| {r['entry_id']} | {r['node_type']} | {r['code']} | {msg} |")
|
||
|
||
lines += ["", "## Next step",
|
||
("Triage P1 in `defect_log.csv`; client reviews the clean entries against this report."
|
||
if gate == "PASS" else
|
||
"Assign and fix every P0, re-run the validator, and only then open client review."),
|
||
""]
|
||
(outdir / "report.md").write_text("\n".join(lines), encoding="utf-8")
|
||
return gate, counts
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(description="5-layer offline JSON-LD schema validator.")
|
||
ap.add_argument("dataset", nargs="?", help="xlsx/csv/jsonl/json file or a directory")
|
||
ap.add_argument("--url-list", help="canonical URL inventory (xlsx/csv/txt) → enables Layer 0")
|
||
ap.add_argument("--out", default="schema_qa_out", help="output directory")
|
||
ap.add_argument("--strict", action="store_true",
|
||
help="unexpected props on known types → P1; unknown types → P1")
|
||
ap.add_argument("--no-recommended", action="store_true",
|
||
help="drop L3 recommended (P2) findings — highest-signal gate")
|
||
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:
|
||
ap.error("provide a DATASET path or --live URL ...")
|
||
|
||
rules = json.loads(Path(args.rules).read_text(encoding="utf-8"))
|
||
|
||
try:
|
||
entries = load_entries(args.dataset, args.live)
|
||
except (ValueError, FileNotFoundError) as exc:
|
||
print(f"ERROR loading input: {exc}", file=sys.stderr)
|
||
return 2
|
||
|
||
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,
|
||
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}
|
||
gate, counts = write_outputs(defects, args.out, meta)
|
||
|
||
print(f"[{gate}] entries={len(entries)} nodes={nodes} "
|
||
f"P0={counts['P0']} P1={counts['P1']} P2={counts['P2']} → {args.out}/")
|
||
# Exit 1 when the gate fails so CI and `&&` chains stop on P0.
|
||
return 0 if gate == "PASS" else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|