#!/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 "�" 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())