From 4f48ba3c59db5e061294c4902f4e696dc204b45b Mon Sep 17 00:00:00 2001 From: Andrew Yim Date: Wed, 27 May 2026 23:48:51 +0900 Subject: [PATCH] feat(seo-schema-validator): back the upgraded SKILL.md with a working 5-layer pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The "Upgrade Schema Validator" commit added SKILL.md referencing files that did not exist. Implement them so the skill actually runs: - scripts/validate_schema.py — 5-layer offline validator (L0 coverage, L1 syntax, L2 vocabulary/value-format, L3 rich-result, L4 consistency) with xlsx/csv/jsonl/ json/dir/live-URL adapters. Gate = zero P0; exits 1 on failure. - scripts/schema_rules.json — curated hotel-focused, offline rule set (edit-only extension point). - scripts/make_sample.py + fixtures/sample_schema.csv — deliberately flawed fixture seeding ≥1 defect per layer; used to self-test. - references/ — validation-methodology, defect-taxonomy (25 codes), hotel-type-map. - templates/ — client-qa-report, decision-log. - code/CLAUDE.md — redirect legacy single-URL tool to the new pipeline. Noise control: MISSING_RECOMMENDED aggregated one-line-per-node; unexpected-property checks opt-in via --strict. Generalized client-specific shilla-type-map → hotel-type-map. Self-tested: default P0=5/P1=4/P2=14 FAIL, --strict --no-recommended P2=0, adapters verified. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../16-seo-schema-validator/SKILL.md | 4 +- .../16-seo-schema-validator/code/CLAUDE.md | 165 +--- .../fixtures/sample_schema.csv | 16 + .../references/defect-taxonomy.md | 74 ++ .../references/hotel-type-map.md | 68 ++ .../references/validation-methodology.md | 123 +++ .../scripts/make_sample.py | 160 ++++ .../scripts/requirements.txt | 6 + .../scripts/schema_rules.json | 200 ++++ .../scripts/validate_schema.py | 854 ++++++++++++++++++ .../templates/client-qa-report-template.md | 57 ++ .../templates/decision-log.md | 39 + 12 files changed, 1636 insertions(+), 130 deletions(-) create mode 100644 custom-skills/16-seo-schema-validator/fixtures/sample_schema.csv create mode 100644 custom-skills/16-seo-schema-validator/references/defect-taxonomy.md create mode 100644 custom-skills/16-seo-schema-validator/references/hotel-type-map.md create mode 100644 custom-skills/16-seo-schema-validator/references/validation-methodology.md create mode 100644 custom-skills/16-seo-schema-validator/scripts/make_sample.py create mode 100644 custom-skills/16-seo-schema-validator/scripts/requirements.txt create mode 100644 custom-skills/16-seo-schema-validator/scripts/schema_rules.json create mode 100644 custom-skills/16-seo-schema-validator/scripts/validate_schema.py create mode 100644 custom-skills/16-seo-schema-validator/templates/client-qa-report-template.md create mode 100644 custom-skills/16-seo-schema-validator/templates/decision-log.md diff --git a/custom-skills/16-seo-schema-validator/SKILL.md b/custom-skills/16-seo-schema-validator/SKILL.md index 7d350a1..5567884 100644 --- a/custom-skills/16-seo-schema-validator/SKILL.md +++ b/custom-skills/16-seo-schema-validator/SKILL.md @@ -42,12 +42,12 @@ This skill's primary job is **Mode A**: catch errors before the client sees them Full rationale and the type→requirement matrix: `references/validation-methodology.md`. Severity + category codes: `references/defect-taxonomy.md`. -Shilla page-type → schema-type map: `references/shilla-type-map.md`. +Hotel page-type → schema-type map: `references/hotel-type-map.md`. Client-facing report + P1 decision log: `templates/client-qa-report-template.md`, `templates/decision-log.md`. ## Stage gates (aligned to the project lifecycle 설계→개발→테스트→안정화→런칭 후) -- **G1 설계** — Lock the schema spec and the page-type→type map (`shilla-type-map.md`). Approve the entry template. *DoD:* every page template has an assigned schema type and a required-property list. +- **G1 설계** — Lock the schema spec and the page-type→type map (`hotel-type-map.md`). Approve the entry template. *DoD:* every page template has an assigned schema type and a required-property list. - **G2 개발** — Authors produce entries. Run the validator with `--strict`. *DoD (gate):* **zero P0**, JSON parses for 100% of entries. Entries failing this NEVER reach client review. - **G3 테스트** — Re-run; triage P1 in `defect_log.csv` (assign owner/status). Client reviews ONLY the clean, P0-free entries, against a defect report — not raw JSON. *DoD:* P1 triaged, decisions logged in `templates/decision-log.md`. - **G4 안정화** — Fix → re-run → confirm no regressions. Spot-check a sample in Google Rich Results Test (online, outside this runtime). *DoD:* P0=0, P1 accepted/closed, online validator green on sample. diff --git a/custom-skills/16-seo-schema-validator/code/CLAUDE.md b/custom-skills/16-seo-schema-validator/code/CLAUDE.md index 9f44c68..7bd903d 100644 --- a/custom-skills/16-seo-schema-validator/code/CLAUDE.md +++ b/custom-skills/16-seo-schema-validator/code/CLAUDE.md @@ -1,148 +1,57 @@ -# CLAUDE.md +# CLAUDE.md — seo-schema-validator (Claude Code) -## Overview +## Canonical entry point -Structured data validator: extract, parse, and validate JSON-LD, Microdata, and RDFa markup against schema.org vocabulary. +This skill was upgraded to a **5-layer dataset-QA pipeline**. The authoritative +directive and run instructions live in the skill root: -## Quick Start +- **`../SKILL.md`** — modes, the 5 layers, stage gates, how to run. +- **`../scripts/validate_schema.py`** — the validator (run this, not the legacy script below). +- **`../scripts/schema_rules.json`** — the offline rule set (edit this to add a type/rule). +- **`../references/`** — `validation-methodology.md`, `defect-taxonomy.md`, `hotel-type-map.md`. +- **`../templates/`** — `client-qa-report-template.md`, `decision-log.md`. ```bash -pip install -r scripts/requirements.txt -python scripts/schema_validator.py --url https://example.com +# Primary use — QA an AUTHORED dataset before the client sees it (Mode A) +python ../scripts/validate_schema.py DATASET.xlsx --url-list URLLIST.xlsx --out schema_qa_out + +# Highest-signal pre-review gate +python ../scripts/validate_schema.py DATASET.csv --strict --no-recommended --out qa_strict + +# Try the bundled flawed fixture first +python ../scripts/make_sample.py +python ../scripts/validate_schema.py ../fixtures/sample_schema.csv --out demo_out + +# Post-deploy live audit (Mode B) — feeds seo-comprehensive-audit stage 4 +python ../scripts/validate_schema.py --live https://example.com --out live_out ``` -## Scripts +Gate rule: **PASS = zero P0.** The process exits 1 when the gate fails, so it stops +`&&` chains and CI. Only P0-free entries advance to client review. -| Script | Purpose | -|--------|---------| -| `schema_validator.py` | Extract and validate structured data | -| `base_client.py` | Shared utilities | +## Legacy single-URL tool (kept for quick one-offs) -## Usage +`scripts/schema_validator.py --url ` extracts and validates structured data from +one live page (JSON-LD / Microdata / RDFa via extruct). It predates the pipeline and is +**not** the gate. For any dataset or client-facing QA, use `validate_schema.py` above. ```bash -# Validate page schema -python scripts/schema_validator.py --url https://example.com - -# JSON output +pip install -r scripts/requirements.txt # extruct, jsonschema, rdflib, lxml, requests python scripts/schema_validator.py --url https://example.com --json - -# Validate local file -python scripts/schema_validator.py --file schema.json - -# Check Rich Results eligibility -python scripts/schema_validator.py --url https://example.com --rich-results ``` -## Supported Formats +## Notion output (OurDigital SEO Audit Log) -| Format | Detection | -|--------|-----------| -| JSON-LD | `', + 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 _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) ---- + 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(): + 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"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"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)") + + +# --------------------------------------------------------------------------- # +# Orchestration + output +# --------------------------------------------------------------------------- # +def run(entries, rules, inventory, strict, no_recommended): + 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) + 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") + 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) + 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()) diff --git a/custom-skills/16-seo-schema-validator/templates/client-qa-report-template.md b/custom-skills/16-seo-schema-validator/templates/client-qa-report-template.md new file mode 100644 index 0000000..c396e24 --- /dev/null +++ b/custom-skills/16-seo-schema-validator/templates/client-qa-report-template.md @@ -0,0 +1,57 @@ +# 구조화 데이터 QA 리포트 — {{프로젝트명}} + +> 클라이언트 검토용. **원본 JSON이 아니라 결함 리포트를 검토합니다.** +> 이 리포트에 오른 엔트리는 모두 기계 검증(Layer 0–4)을 통과한 **P0 0건** 상태입니다. + +| 항목 | 값 | +|---|---| +| 데이터셋 | `{{dataset_파일명}}` | +| 검증 일시 | {{YYYY-MM-DD HH:MM}} | +| 검증 모드 | A — Dataset QA (배포 전) / B — Live audit (배포 후) | +| 엔트리 수 | {{entries}} (파싱 성공 {{valid_entries}}, 노드 {{nodes}}) | +| **게이트** | **{{PASS / FAIL}}** (PASS = P0 0건) | +| 결함 | P0 {{n}} · P1 {{n}} · P2 {{n}} | +| Audit ID | SCHEMA-{{YYYYMMDD}}-{{NNN}} | + +## 1. 한눈에 보기 + +- ✅ **검토 가능 엔트리**: P0 0건을 통과한 {{n}}개 — 아래 판단 항목만 확인해 주세요. +- ⛔ **보류 엔트리**(있다면): P0 {{n}}건으로 검토 대상에서 제외. 수정 후 재검증합니다. +- 이번 검토에서 **사람의 판단이 필요한 것**은 기계가 잡지 못하는 두 가지뿐입니다: + 1. 페이지에 맞는 스키마 **타입**이 선택되었는가 + 2. 표시되는 **문구(설명·이름)**가 사실과 정확히 일치하는가 + +## 2. 결함 요약 (코드별) + +| 심각도 | 코드 | 건수 | 의미 | +|---|---|---|---| +| P0 | {{CODE}} | {{n}} | {{한 줄 설명}} | +| P1 | {{CODE}} | {{n}} | {{한 줄 설명}} | +| P2 | {{CODE}} | {{n}} | {{한 줄 설명}} | + +> 코드 정의: `references/defect-taxonomy.md`. 전체 목록: 첨부 `defect_log.csv`. + +## 3. P0 블로커 (있을 경우 — 검토 전 수정 필수) + +| 엔트리 | 타입 | 코드 | 내용 | 담당 | 상태 | +|---|---|---|---|---|---| +| {{entry_id}} | {{type}} | {{CODE}} | {{message}} | {{owner}} | open | + +## 4. 클라이언트 확인 요청 (판단 항목) + +기계가 통과시킨 엔트리 중, 사람의 확인이 필요한 항목입니다. + +| # | URL / 페이지 | 확인 요청 | 비고 | +|---|---|---|---| +| 1 | {{url}} | 이 페이지에 `{{@type}}` 타입이 맞습니까? | | +| 2 | {{url}} | 설명/이름 문구가 정확합니까? | | + +## 5. 다음 단계 + +- **PASS인 경우**: 위 4번 판단 항목 확정 → 배포 단계(G4 안정화)로 이동, 샘플을 Google + Rich Results Test로 최종 확인. +- **FAIL인 경우**: P0 담당 배정 → 수정 → 재검증(`validate_schema.py`) → 본 리포트 갱신. +- P1 처리 방침(수정/수용)은 `decision-log.md`에 기록합니다. + +--- +*생성: 16-seo-schema-validator · 첨부: `report.md`, `defect_log.csv`, `results.json`* diff --git a/custom-skills/16-seo-schema-validator/templates/decision-log.md b/custom-skills/16-seo-schema-validator/templates/decision-log.md new file mode 100644 index 0000000..5c88ad1 --- /dev/null +++ b/custom-skills/16-seo-schema-validator/templates/decision-log.md @@ -0,0 +1,39 @@ +# P1 Decision Log — {{프로젝트명}} + +P0 is non-negotiable: every P0 is fixed before launch (the gate enforces it). **P1 is +where judgement lives** — some P1s get fixed, some get consciously accepted. This log +records *which, by whom, and why*, so an accepted P1 is a decision on the record, not a +silently ignored defect. It is the G3 (테스트) deliverable. + +## How to use + +1. Open `defect_log.csv`, filter to `severity = P1`. +2. For each P1 (or each group of identical P1s), add a row below. +3. Decision is one of: **Fix** (will correct before launch) / **Accept** (ship as-is, + with rationale) / **Defer** (post-launch backlog). +4. An `Accept`/`Defer` needs a named approver. `Fix` needs an owner + target date. +5. Re-run the validator after the fixes; confirm the fixed P1s are gone. + +## Log + +| # | Code | Entry / scope | Summary | Decision | Owner / Approver | Target / Date | Rationale | +|---|---|---|---|---|---|---|---| +| 1 | {{CODE}} | {{entry_id or (dataset)}} | {{one line}} | Fix / Accept / Defer | {{name}} | {{YYYY-MM-DD}} | {{why}} | +| 2 | | | | | | | | +| 3 | | | | | | | | + +## Standing decisions (apply to all entries unless overridden) + +Record cross-cutting calls once here instead of per row — e.g. "MISSING_RECOMMENDED for +`starRating` is accepted group-wide: not contractually rated." Reduces log noise. + +| Code | Standing decision | Approver | Date | +|---|---|---|---| +| {{CODE}} | Accept group-wide — {{reason}} | {{name}} | {{YYYY-MM-DD}} | + +## Sign-off + +| Stage gate | Condition | Confirmed by | Date | +|---|---|---|---| +| G3 테스트 | All P1 triaged (Fix/Accept/Defer), decisions logged above | {{name}} | {{date}} | +| G4 안정화 | P0 = 0, all "Fix" P1 closed, online validator green on sample | {{name}} | {{date}} |