o2o-site-AEO/geo/naver/notify.py

"""발행본 주소를 검색엔진에 **알린다** — 크롤러가 지나가길 기다리지 않는다.

★ 어디에 닿고 어디에 안 닿나 — 이게 이 파일의 존재 이유다.
  닿는다   네이버(2023-07~) · Bing · Yandex · Seznam.
           ★ 네이버는 **여기 말고 자동 통로가 없다.** 서치어드바이저는 사람이 눌러야 한다.
  안 닿는다 구글. IndexNow 를 채택하지 않았다 — 사이트맵 제출이 유일한 자동화다.

★★ **왜 `geo` 가 이 일을 하나.**
  원래 담당은 `solution/backend/services/indexnow.py` 다. 그런데 그 코드가 읽는 파일
  (`<out>/s/<slug>/sitemap.xml`)을 프리렌더가 **더는 굽지 않는다** — 사이트가 한 장이 되면서
  루트 사이트맵 한 장으로 합쳤기 때문이다. 파일이 없으면 빈 목록을 돌려주고 경고 한 줄만
  남긴 채 **발행 잡은 성공한다.** 즉 통보가 조용히 0건이다.
  백엔드를 고치지 않기로 해서(2026-09-11), `geo` 가 **루트 사이트맵을 읽어** 대신 보낸다.

  ⚠️ **담당이 두 곳이 되면 안 된다.** 지금은 백엔드가 0건이라 중복이 없지만, 누가 백엔드를
  고치면 **같은 URL 이 두 번 나간다**(429 과다 요청 대상). 그때는 둘 중 하나를 꺼야 한다 —
  이 모듈은 `GEO_NOTIFY_ENABLED=0` 으로 끈다.

★ 보낼 URL 을 여기서 조립하지 않는다. **사이트맵에 있는 것만 보낸다** — 거기 있는 것이
  실제로 구워진 페이지다. 라우트 규칙을 또 두면 사이트맵에 없는 URL 을 통보하게 되고,
  그건 404 통보라 신뢰만 깎인다.

★ 실패해도 발행을 되돌리지 않는다. 정적 파일은 이미 올라가 있어서 되돌릴 것이 없다.
  대신 **조용히 지나가지 않게** 결과를 돌려준다 — 그게 지금 고장의 재발 방지다.
"""
import os
import re
from dataclasses import asdict, dataclass

import httpx

from geo.naver._http import TIMEOUT_SEC, get

ENDPOINT = "https://api.indexnow.org/indexnow"
# 규격 상한은 한 번에 10,000개다. 사이트 하나는 한 장이라 넉넉하다.
MAX_URLS = 10_000

KEY_ENV = "INDEXNOW_KEY"
ENABLED_ENV = "GEO_NOTIFY_ENABLED"


@dataclass
class NotifyResult:
    slug: str
    urls: list[str]
    sent: bool
    status: int | None = None
    error: str | None = None

    @property
    def ok(self) -> bool:
        # 200 OK · 202 Accepted 가 정상이다. 그 밖은 규격상 원인이 정해져 있다:
        #   400 형식 · 403 키 불일치 · 422 호스트 불일치 · 429 과다 요청
        return self.sent and self.status in (200, 202)

    def as_dict(self) -> dict:
        return {**asdict(self), "ok": self.ok}


def key() -> str:
    return os.environ.get(KEY_ENV, "").strip()


def enabled() -> bool:
    """키가 있고, 꺼져 있지 않을 때만 보낸다.

    ★ 스위치를 둔 이유는 머리주석의 "담당이 두 곳" 경고 때문이다. 백엔드 쪽이 고쳐지는
      날 여기를 꺼야 하는데, 코드를 지우는 것보다 환경변수 하나가 되돌리기 쉽다."""
    return bool(key()) and os.environ.get(ENABLED_ENV, "1").strip() != "0"


async def root_sitemap_locs(client: httpx.AsyncClient, origin: str) -> list[str]:
    """루트 사이트맵의 `<loc>` 전부. **이 호스트가 실제로 발행한 주소 목록**이다.

    ★ XML 파서를 쓰지 않고 정규식으로 뽑는다. 우리가 굽는 파일이라 형태가 고정이고,
      파서를 쓰면 한 글자 깨졌을 때 전부를 잃는다 — 통보는 부분 성공이 낫다."""
    res = await get(client, origin.rstrip("/") + "/sitemap.xml")
    if res is None or res.status_code != 200:
        return []
    return [u.strip() for u in re.findall(r"<loc>\s*([^<\s]+)\s*</loc>", res.text) if u.strip()]


def site_urls(locs: list[str], slug: str) -> list[str]:
    """이 사이트에 속한 주소만 고른다.

    ★ `/s/<slug>` 로 시작하는 것만 본다. `startswith` 가 아니라 경계까지 보는 이유:
      `/s/joy` 로 거르면 `/s/joy-cafe` 까지 끌려온다."""
    marker = f"/s/{slug}"
    picked = []
    for loc in locs:
        idx = loc.find(marker)
        if idx < 0:
            continue
        rest = loc[idx + len(marker):]
        if rest in ("", "/") or rest.startswith(("/", "?", "#")):
            picked.append(loc)
    return picked[:MAX_URLS]


async def submit(client: httpx.AsyncClient, urls: list[str]) -> NotifyResult | None:
    """한 호스트분을 보낸다. 호출측이 slug 를 채워 돌려받는다."""
    host = urls[0].split("//", 1)[-1].split("/", 1)[0]
    body = {
        "host": host,
        "key": key(),
        # 키 파일은 오리진 루트에 있다(프리렌더가 굽는다). 검색엔진이 이걸 열어
        # 같은 키가 있는지 보고 "이 호스트를 제어하는 쪽이 보냈다" 를 확인한다.
        # 비밀이 아니다 — 공개되어야 작동하는 값이다.
        "keyLocation": f"https://{host}/{key()}.txt",
        # 한 요청의 URL 은 전부 같은 호스트여야 한다(규격). 섞이면 422 다.
        "urlList": [u for u in urls if u.split("//", 1)[-1].split("/", 1)[0] == host],
    }
    try:
        res = await client.post(ENDPOINT, json=body, timeout=TIMEOUT_SEC)
    except httpx.HTTPError as ex:
        return NotifyResult("", body["urlList"], sent=True, error=f"{type(ex).__name__}: {ex}")
    return NotifyResult("", body["urlList"], sent=True, status=res.status_code)


async def notify_site(client: httpx.AsyncClient, origin: str, slug: str, locs: list[str] | None = None) -> NotifyResult:
    """사이트 하나를 알린다. `locs` 를 주면 사이트맵을 다시 읽지 않는다(여러 건 처리용)."""
    if not enabled():
        return NotifyResult(slug, [], sent=False, error=f"{KEY_ENV} 가 없거나 {ENABLED_ENV}=0")

    if locs is None:
        locs = await root_sitemap_locs(client, origin)
    urls = site_urls(locs, slug)
    if not urls:
        return NotifyResult(slug, [], sent=False, error="루트 사이트맵에 이 사이트 주소가 없다")

    result = await submit(client, urls)
    result.slug = slug
    return result


async def check_key_file(client: httpx.AsyncClient, origin: str) -> tuple[bool, str]:
    """키 파일이 열리는가 — **이게 없으면 통보가 403 으로 전부 거절된다.**

    발행 전에 봐야 하는 항목이다. 통보를 보내고 나서 알면 이미 늦다."""
    k = key()
    if not k:
        return False, f"{KEY_ENV} 가 비어 있다 — 통보가 꺼져 있다"
    res = await get(client, f"{origin.rstrip('/')}/{k}.txt")
    if res is None:
        return False, f"/{k}.txt 에 연결하지 못했다"
    if res.status_code != 200:
        return False, f"/{k}.txt 이 HTTP {res.status_code} — 통보가 403 으로 거절된다"
    if res.text.strip() != k:
        return False, f"/{k}.txt 내용이 키와 다르다"
    return True, f"/{k}.txt"