our-claude-skills/custom-skills/95-ourdigital-presales-seo/scripts/kg_query.py

#!/usr/bin/env python3
"""Query Google Knowledge Graph (Korean) for a prospect's brand ecosystem + competitors.

Part of the ourdigital-presales-seo skill (Stage 3). Generalized from the
Sono Hotels & Resorts pre-sales script: entities are built from CLI args, not
hardcoded. Read-only GET to kgsearch.googleapis.com.

Key resolution: env GOOGLE_KG_API_KEY -> GOOGLE_API_KEY.

Example:
  python kg_query.py --brand "소노호텔앤리조트" \
    --aliases "소노호텔&리조트,SONO Hotels & Resorts" \
    --parent "소노인터내셔널" --legacy "대명소노그룹,대명리조트" \
    --subbrands "소노벨,소노캄,소노펠리체,쏠비치,소노문" \
    --properties "소노벨 비발디파크,소노캄 제주,쏠비치 양양" \
    --competitors "롯데호텔,신라호텔,조선호텔앤리조트,한화리조트,켄싱턴리조트" \
    --out-dir ./data
"""
import argparse
import json
import os
import sys
import time
import urllib.parse
import urllib.request

API_KEY = os.environ.get("GOOGLE_KG_API_KEY") or os.environ.get("GOOGLE_API_KEY")
ENDPOINT = "https://kgsearch.googleapis.com/v1/entities:search"
LODGING = {"Hotel", "LodgingBusiness", "Resort", "Place", "TouristAttraction"}


def query_kg(q, lang, limit):
    params = {"query": q, "key": API_KEY, "languages": lang, "limit": limit, "indent": "true"}
    url = ENDPOINT + "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={"User-Agent": "OurDigital-SEO-Audit/1.0"})
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read().decode("utf-8"))


def flatten(label, group, data):
    rows = []
    for item in data.get("itemListElement", []):
        r = item.get("result", {})
        detail = r.get("detailedDescription", {}) or {}
        rows.append({
            "group": group, "input_label": label, "kg_id": r.get("@id"),
            "name": r.get("name"), "types": r.get("@type"), "description": r.get("description"),
            "result_score": item.get("resultScore"), "has_detailed_desc": bool(detail.get("articleBody")),
            "detailed_source": detail.get("url"), "image": (r.get("image") or {}).get("contentUrl"),
            "url": r.get("url"),
        })
    if not rows:
        rows.append({"group": group, "input_label": label, "kg_id": None, "name": None,
                     "types": None, "description": "NO KG ENTITY FOUND", "result_score": 0,
                     "has_detailed_desc": False, "detailed_source": None, "image": None, "url": None})
    return rows


def build_entities(args):
    ents = []

    def add(group, raw):
        for it in (raw or "").split(","):
            it = it.strip()
            if it:
                ents.append((group, it, it))

    add("01_master", args.brand)
    add("01_master", args.aliases)
    add("02_corporate", args.parent)
    add("03_legacy", args.legacy)
    add("04_membership", args.membership)
    add("05_subbrand", args.subbrands)
    add("06_property", args.properties)
    add("09_competitor", args.competitors)
    return ents


def main():
    ap = argparse.ArgumentParser(description="KG entity examination for pre-sales SEO")
    ap.add_argument("--brand", required=True, help="Master brand (comma-separated allowed)")
    ap.add_argument("--aliases", default="", help="Brand name variants (& vs 앤, EN)")
    ap.add_argument("--parent", default="", help="Parent / corporate entity")
    ap.add_argument("--legacy", default="", help="Former / legacy names")
    ap.add_argument("--membership", default="", help="Membership / sales-rep units")
    ap.add_argument("--subbrands", default="", help="Sub-brands")
    ap.add_argument("--properties", default="", help="Flagship properties")
    ap.add_argument("--competitors", default="", help="Competitor benchmarks")
    ap.add_argument("--lang", default="ko")
    ap.add_argument("--limit", type=int, default=30)
    ap.add_argument("--out-dir", default=".")
    args = ap.parse_args()

    if not API_KEY:
        sys.exit("ERROR: set GOOGLE_KG_API_KEY or GOOGLE_API_KEY in the environment.")

    ents = build_entities(args)
    raw, flat = {}, []
    for group, label, q in ents:
        try:
            data = query_kg(q, args.lang, args.limit)
            raw[label] = data
            flat.extend(flatten(label, group, data))
        except Exception as e:  # network/quota — record, continue
            raw[label] = {"error": str(e)}
            flat.append({"group": group, "input_label": label, "kg_id": None, "name": None,
                         "types": None, "description": f"ERROR: {e}", "result_score": 0,
                         "has_detailed_desc": False, "detailed_source": None, "image": None, "url": None})
        time.sleep(0.3)

    os.makedirs(args.out_dir, exist_ok=True)
    with open(os.path.join(args.out_dir, "kg_korean_raw.json"), "w", encoding="utf-8") as f:
        json.dump(raw, f, ensure_ascii=False, indent=2)
    with open(os.path.join(args.out_dir, "kg_korean_flat.json"), "w", encoding="utf-8") as f:
        json.dump(flat, f, ensure_ascii=False, indent=2)

    # Console summary: top match per input label + lodging-type flag
    by_label = {}
    for r in flat:
        e = by_label.setdefault(r["input_label"], {"group": r["group"], "n": 0, "lodging": False, "top": r})
        e["group"] = r["group"]
        if r.get("kg_id"):
            e["n"] += 1
        if set(r.get("types") or []) & LODGING:
            e["lodging"] = True
        if (r.get("result_score") or 0) > (e["top"].get("result_score") or 0):
            e["top"] = r

    print(f"{'GROUP':<14}{'SCORE':>8} {'#':>3} {'LODG':<5} {'TOP TYPE':<22} {'DESC':<5} INPUT -> TOP KG NAME")
    print("-" * 120)
    for lbl, e in sorted(by_label.items(), key=lambda kv: kv[1]["group"]):
        top = e["top"]
        ttype = ",".join((top.get("types") or [])[:2]) or "-"
        print(f"{e['group']:<14}{top.get('result_score') or 0:>8.0f} {e['n']:>3} "
              f"{('YES' if e['lodging'] else '-'):<5} {ttype[:22]:<22} "
              f"{('yes' if top.get('has_detailed_desc') else 'no'):<5} {lbl[:40]} -> {top.get('name') or '(NONE)'}")
    print(f"\nWrote kg_korean_raw.json + kg_korean_flat.json to {args.out_dir}")


if __name__ == "__main__":
    main()