o2o-site-AEO/solution/backend/services/indexnow.py

"""발행본 URL 을 IndexNow 로 알린다 — 크롤러가 찾아올 때까지 기다리지 않는다.

★ 어디에 닿고 어디에 안 닿는지가 이 파일의 존재 이유다.

  닿는다   네이버(2023-07 부터 지원) · Bing · Yandex · Seznam.
           네이버가 소상공인 검색 트래픽의 주력이라 여기가 핵심이고,
           Bing 은 ChatGPT 검색의 상류라 AEO 로도 값이 있다.

  안 닿는다 **구글**. 구글은 IndexNow 를 지원하지 않는다(2021 년부터 테스트만 하고 채택 안 함).
           구글 색인 요청 API(Indexing API)도 JobPosting·BroadcastEvent 전용이라 우리는 못 쓴다 —
           URL 을 받아 200 을 주지만 그 밖의 타입은 그냥 버린다.
           구글 쪽은 Search Console 사이트맵 제출이 유일한 자동화 경로다.

★ 보낼 URL 을 여기서 다시 계산하지 않는다. 사이트의 `sitemap.xml` 을 읽는다 —
  프리렌더가 실제로 구운 페이지 목록이 거기 있다. 라우트 규칙을 두 군데 두면
  사이트맵에 없는 URL 을 통보하게 되고, 그건 404 통보라 신뢰만 깎는다.

★ 실패해도 발행을 되돌리지 않는다. 색인 통보는 발행의 **부수 효과**다.
  여기서 예외를 올리면 정적 파일이 이미 올라간 뒤에 발행이 실패로 뒤집힌다.
"""

import os
import xml.etree.ElementTree as ET
from pathlib import Path
from urllib.parse import urlsplit

import httpx

from common.logger import LOG

ENDPOINT = "https://api.indexnow.org/indexnow"
SITEMAP_NS = "{http://www.sitemaps.org/schemas/sitemap/0.9}"
TIMEOUT_SEC = 10.0
# 규격 상한은 한 번에 10,000 개다. 사이트 하나는 수십 개라 넉넉하다.
MAX_URLS = 10_000


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


def is_configured() -> bool:
    return bool(key())


def output_dir() -> Path:
    return Path(os.environ.get("SITE_OUTPUT_DIR", "/app/out/sites"))


def site_urls(slug: str) -> list[str]:
    """이 사이트가 실제로 발행한 URL 목록(사이트맵의 `<loc>`)."""
    sitemap = output_dir() / "s" / slug / "sitemap.xml"
    if not sitemap.is_file():
        return []
    try:
        root = ET.parse(sitemap).getroot()
    except ET.ParseError as ex:
        LOG.w(f"[indexnow] 사이트맵을 읽지 못했다 — {sitemap}: {ex}")
        return []
    urls = [(node.text or "").strip() for node in root.iter(f"{SITEMAP_NS}loc")]
    return [url for url in urls if url][:MAX_URLS]


def _payload(urls: list[str]) -> dict | None:
    """IndexNow 요청 본문. 호스트는 URL 에서 뽑는다(커스텀 도메인도 그대로 맞는다).

    한 요청의 URL 은 전부 같은 호스트여야 한다(규격). 섞여 있으면 422 를 받으므로
    첫 URL 의 호스트에 속한 것만 보낸다."""
    host = urlsplit(urls[0]).netloc
    if not host:
        return None
    same_host = [url for url in urls if urlsplit(url).netloc == host]
    return {
        "host": host,
        "key": key(),
        # 키 파일은 오리진 루트에 있다(프리렌더가 굽고 azure_static 이 올린다).
        "keyLocation": f"https://{host}/{key()}.txt",
        "urlList": same_host,
    }


async def submit(slug: str) -> dict | None:
    """설정된 경우에만 통보한다. 실패는 로그로 남기고 삼킨다(발행을 되돌리지 않는다)."""
    if not is_configured():
        return None

    urls = site_urls(slug)
    if not urls:
        LOG.w(f"[indexnow] 보낼 URL 이 없다 — 사이트맵이 없거나 비었다: {slug}")
        return None

    body = _payload(urls)
    if body is None:
        LOG.w(f"[indexnow] URL 에서 호스트를 못 읽었다: {urls[0]}")
        return None

    try:
        async with httpx.AsyncClient(timeout=TIMEOUT_SEC) as client:
            res = await client.post(ENDPOINT, json=body)
    except httpx.HTTPError as ex:
        LOG.w(f"[indexnow] 통보 실패 {slug}: {type(ex).__name__}: {ex}")
        return {"ok": False, "error": f"{type(ex).__name__}: {ex}", "urls": len(body['urlList'])}

    # 200 OK · 202 Accepted 가 정상이다. 그 밖은 규격상 원인이 정해져 있다:
    #   400 형식 · 403 키 불일치 · 422 호스트 불일치 · 429 과다 요청
    ok = res.status_code in (200, 202)
    if ok:
        LOG.i(f"[indexnow] {slug} — URL {len(body['urlList'])}개 통보 (HTTP {res.status_code})")
    else:
        LOG.w(f"[indexnow] {slug} 거절됨 HTTP {res.status_code}: {res.text[:200]}")
    return {"ok": ok, "status": res.status_code, "urls": len(body["urlList"])}