o2o-site-AEO/geo/scripts/postflight.py

"""**발행한 뒤에** — 알리고, 제대로 나갔는지 본다.

    solution/backend/.venv/bin/python geo/scripts/postflight.py <slug>
    solution/backend/.venv/bin/python geo/scripts/postflight.py <slug> --place "스테이,머뭄"
    solution/backend/.venv/bin/python geo/scripts/postflight.py --all      (사이트맵의 전부)
    solution/backend/.venv/bin/python geo/scripts/postflight.py <slug> --dry-run

★★ **왜 "발행 전" 이 아니라 "발행 후" 인가.** 아직 없는 주소를 통보하면 검색엔진이 404 를
  받는다. 알리지 않은 것보다 나쁘다 — 헛주소를 보내는 호스트로 기록된다.
  그래서 통보는 **구워진 것을 확인한 뒤**에만 보낸다(`_live` 가 그 확인이다).

★ 순서가 뜻을 갖는다:
    1) 살아 있나   200 이 아니면 통보하지 않는다
    2) 알린다      루트 사이트맵에서 이 사이트 주소를 골라 IndexNow 로
    3) 기록한다    **성공한 것만.** 실패를 성공으로 기억하면 영영 다시 안 보낸다
    4) 본다        소유확인·색인·역방향 링크 (선택)
"""
import argparse
import asyncio
import json
import os
import sys
from datetime import datetime, timezone

_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
for _p in (_ROOT, os.path.join(_ROOT, "solution", "backend")):
    if _p not in sys.path:
        sys.path.insert(0, _p)
os.environ.setdefault("APP_ENV", "local")

import httpx  # noqa: E402

from geo import state  # noqa: E402
from geo.naver import notify  # noqa: E402
from geo.naver._http import TIMEOUT_SEC, get  # noqa: E402
from services.site_payload import publish_origin  # noqa: E402

STATE_NAME = "indexnow"


async def _live(client: httpx.AsyncClient, url: str) -> bool:
    """통보 전에 실제로 열리는지 본다. **404 통보를 막는 유일한 방어다.**"""
    res = await get(client, url)
    return res is not None and res.status_code == 200


async def run(origin: str, slugs: list[str] | None, *, dry_run: bool = False) -> list[dict]:
    origin = origin.rstrip("/")
    results: list[dict] = []
    seen = state.load(STATE_NAME)

    async with httpx.AsyncClient(timeout=TIMEOUT_SEC, follow_redirects=True) as client:
        locs = await notify.root_sitemap_locs(client, origin)
        if not locs:
            return [{"slug": None, "ok": False, "error": "루트 사이트맵을 읽지 못했다"}]

        if not slugs:
            # 사이트맵에서 slug 를 뽑는다 — 주소 규칙을 여기서 다시 만들지 않는다.
            slugs = sorted({loc.split("/s/", 1)[1].split("/", 1)[0].split("?")[0]
                            for loc in locs if "/s/" in loc})

        for slug in slugs:
            urls = notify.site_urls(locs, slug)
            if not urls:
                results.append({"slug": slug, "ok": False, "error": "사이트맵에 이 사이트 주소가 없다"})
                continue
            if not await _live(client, urls[0]):
                results.append({"slug": slug, "ok": False, "error": f"{urls[0]} 이 아직 200 이 아니다 — 통보하지 않는다"})
                continue
            if dry_run:
                results.append({"slug": slug, "ok": True, "urls": urls, "dry_run": True})
                continue

            res = await notify.notify_site(client, origin, slug, locs=locs)
            row = res.as_dict()
            results.append(row)
            if res.ok:
                # ★ 성공한 것만 기록한다(머리주석 3번).
                seen[slug] = {"at": datetime.now(timezone.utc).isoformat(), "urls": len(res.urls)}

    if not dry_run:
        state.save(STATE_NAME, seen)
    return results


async def main() -> int:
    parser = argparse.ArgumentParser(description="발행 후 — 알리고 확인한다")
    parser.add_argument("slug", nargs="*", help="알릴 사이트(여럿 가능). 비우면 --all 이 필요하다")
    parser.add_argument("--all", action="store_true", help="사이트맵에 있는 전부")
    parser.add_argument("--origin", default=None)
    parser.add_argument("--dry-run", action="store_true", help="보내지 않고 무엇을 보낼지만 본다")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    if not args.slug and not args.all:
        parser.error("slug 를 주거나 --all 을 써라")

    origin = args.origin or publish_origin()
    results = await run(origin, args.slug or None, dry_run=args.dry_run)

    if args.json:
        print(json.dumps(results, ensure_ascii=False, indent=2))
    else:
        head = "[발행 후] " + origin + (" (dry-run)" if args.dry_run else "")
        print(head + "\n")
        for r in results:
            mark = "  ✓" if r.get("ok") else "  ✗"
            n = len(r.get("urls") or [])
            detail = r.get("error") or f"URL {n}개" + (f" · HTTP {r['status']}" if r.get("status") else "")
            print(f"{mark} {r.get('slug')} — {detail}")
        bad = [r for r in results if not r.get("ok")]
        print(f"\n[결과] 통보 {len(results) - len(bad)} · 실패 {len(bad)}")
    return 1 if any(not r.get("ok") for r in results) else 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))