[fix] solution/backend: Teams 카드 필수 필드 보완

웹훅 요청의 contentUrl 및 스키마 선언 누락을 공식 형식에 맞춤. HTTP 202와 채널 게시 성공을 구분하도록 검증 한계 기록.

검증: Teams 요청 형식 회귀 테스트 1건 통과. 워크플로 실제 수신은 별도 확인 필요.

.env.example

@@ -79,6 +79,14 @@ PUBLIC_WEB_BASE_URL=http://localhost
 # SITE_PUBLIC_HOST=web4ai.o2osolution.ai
 # 비우면 색인 통보를 건너뛴다(발행은 정상)
 INDEXNOW_KEY=
+
+# Google Search Console — 최초 소유권/서비스 계정 권한 설정 후 켠다 (docs/SEARCH_CONSOLE.md).
+GSC_ENABLED=0
+GSC_PROPERTY_URL=
+GSC_CREDENTIALS_FILE=
+GSC_CREDENTIALS_HOST_FILE=
+GSC_ALERT_DAYS=7
+GSC_ALERT_WEBHOOK_URL=
 # 비우면 로컬 발행만 한다
 AZURE_STORAGE_CONNECTION_STRING=
 AZURE_STORAGE_CONTAINER=

docker-compose.search-console.yml

@@ -0,0 +1,13 @@
+# 선택 연동. 키 파일은 저장소 밖에 두고 기존 API 스케줄러에만 읽기 전용으로 전달한다.
+# docker compose -f docker-compose.yml -f docker-compose.search-console.yml up -d solution-backend
+services:
+  solution-backend:
+    environment:
+      GSC_CREDENTIALS_FILE: /run/secrets/search-console.json
+    volumes:
+      - type: bind
+        source: ${GSC_CREDENTIALS_HOST_FILE:?Search Console 키 파일 절대경로 필요}
+        target: /run/secrets/search-console.json
+        read_only: true
+        bind:
+          create_host_path: false

docs/ARCHITECTURE.md

@@ -69,6 +69,10 @@ BUILD 잡 (worker)  ─ services/build_service.py:99 run_build()
 
 ## 3. 서빙 — 테스트 서버가 정적 파일을 직접 서빙한다
 
+Google 추적은 발행 잡 밖에서 실행한다. 기존 API 스케줄러가 발행 완료 DB를 감지해
+사이트맵 제출·색인 조회·알림을 수행하고 `site_search_status`에 저장한다.
+선택 설정/인증/재시도 경계는 [SEARCH_CONSOLE.md](SEARCH_CONSOLE.md)가 단일 출처다.
+
 **결정 (2026-08-31).** 발행 사이트는 **서버 안에서 nginx 가 정적 파일로 서빙한다.**
 Azure Blob 업로드 경로(`azure_static.py`)는 코드에 있고 동작하지만 **지금은 켜지 않는다** —
 `AZURE_STORAGE_CONNECTION_STRING` 을 비워 두면 발행 잡이 업로드 단계를 건너뛴다.

docs/DATA_MODEL.md

@@ -7,6 +7,8 @@
   여기서는 그 앞뒤를 잇는다.
 - 수집이 **무엇을 어디서 가져오는지**는 [COLLECTION_SEO_AEO_FLOW.md](COLLECTION_SEO_AEO_FLOW.md).
 - 표를 고치는 절차는 [postgres-init/migrations/README.md](../postgres-init/migrations/README.md).
+- Google 제출/색인 관측은 `site_search_status`의 별도 상태다. 발행 상태와 섞지 않는다.
+  컬럼 의미·조회·설정은 [SEARCH_CONSOLE.md](SEARCH_CONSOLE.md).
 
 정의는 두 곳이고 **둘 다 최신이어야 한다** — ORM(`solution/backend/common/database/model/models.py`)
 과 DDL(`postgres-init/init-data/init.sql` + `migrations/`). 컬럼 주석은 ORM 이 더 자세하다.

docs/DEVLOG.md

@@ -5,6 +5,15 @@
 
 ---
 
+## 2026-09-15 — Google 사이트맵 자동 제출·색인 관측
+
+- 기존 스케줄러에서 발행 완료 DB 감지 → 사이트맵 제출 → 색인 조회 → 지연/실패 알림.
+- 관측값·재시도·알림 시각은 `site_search_status`에 보관. 발행 잡/상태는 건드리지 않는다.
+- API 인증/호출과 DB·배치·알림 모듈 분리. Google·Teams 실호출은 설정 전까지 꺼진다.
+- 설정/적용/관측 의미: [SEARCH_CONSOLE.md](SEARCH_CONSOLE.md). 운영 배포·권한 부여는 미실행.
+
+**검증** — 관련 59건 통과. 추가 회귀 23건 통과·기존 발행 검수 실패 1건(변경 전 코드에서도 재현).
+
 ## 2026-09-14 — 엽서 쓰기를 발행본에도 넣는다 (사진이 남의 도메인이면 저장·공유는 막힌다)
 
 **무슨 일** — 시연본에만 주입 스크립트로 있던 '엽서 쓰기'(사진 고르기 + 한 마디 + 캔버스 엽서)를

docs/SEARCH_CONSOLE.md

@@ -0,0 +1,99 @@
+# Google Search Console 자동 추적
+
+`발행 DB 감지 → 공개 사이트맵 확인/제출 → 색인 조회 → 상태 저장·Teams 알림`
+
+## 경계
+
+- 기존 API의 스케줄러에서 10분마다 실행한다. 컨테이너 추가 없음.
+- `sites.status=PUBLISHED`인 사이트만 등록하므로 초안/목업 디렉토리 나열을 작업 원장으로 쓰지 않는다.
+- 발행 DB에서 재발견한다. 발행 순간 별도 큐 적재가 실패하는 틈이 없고 재시작해도 이어진다.
+- 발행 트랜잭션/잡과 독립적이다. Google 실패가 사이트 발행을 실패로 바꾸지 않는다.
+- 한 번에 신규 발행 100개 등록, 조회는 오래 기다린 5개 처리. 정상 조회는 24시간 후 반복.
+- 현재 렌더러의 단일 루트 urlset만 지원하고 읽기 상한은 5MB다. 향후 sitemap index 분할 시 확장한다.
+- 오류는 1·2·4·8·16·24시간 간격 재시도. 기본 주기 기준 하루 최대 720회 검사이며,
+  다른 도구의 같은 속성 사용량도 Google 할당량에 포함된다. 대량 백로그는 여러 날에 걸쳐 소진한다.
+- PostgreSQL transaction advisory lock으로 다중 API 프로세스의 동시 배치를 막는다.
+  단일 배치는 외부 호출 동안 트랜잭션/연결 1개를 점유한다(검사 1건 최대 90초, 최대 5건).
+- 사이트맵 제출 성공과 URL 색인 성공은 별개다. `first_indexed_at`은 **우리가 처음 PASS를 관측한 시각**이다.
+  Google 내부 색인 시각이나 최신 발행 버전 반영 시각이 아니다. 원본 `lastCrawlTime`도 함께 보관한다.
+- 재발행 시 해당 발행의 관측 상태를 초기화한다. 지난 관측 이력 전체를 누적하는 이벤트 저장소는 아니다.
+- `SITE_PUBLIC_HOST` 변경은 기존 지침대로 재발행이 필요하다. 사이트 주소의 단일 출처는 `site_payload`다.
+
+## 최초 설정 (운영자)
+
+1. Search Console에서 발행 도메인의 소유권 확인. URL-prefix 속성이면
+   `https://web4ai.o2osolution.ai/`, 도메인 속성이면 `sc-domain:web4ai.o2osolution.ai` 형태.
+2. Google Cloud에서 Search Console API 활성화, 전용 서비스 계정 생성.
+3. Search Console 속성 설정 → 사용자 및 권한에서 그 서비스 계정 이메일에 전체 사용자 권한 부여.
+   Google 로그인용 `GOOGLE_CLIENT_ID`와는 다른 인증이다.
+4. 서비스 계정 JSON 키는 **저장소 밖**에 보관한다. 권한을 최소화하고 git/이미지/로그에 넣지 않는다.
+5. 루트 `.env` 설정:
+
+```dotenv
+GSC_ENABLED=1
+GSC_PROPERTY_URL=https://web4ai.o2osolution.ai/
+GSC_CREDENTIALS_HOST_FILE=/secure/location/search-console.json
+GSC_ALERT_DAYS=7
+GSC_ALERT_WEBHOOK_URL=
+```
+
+키 생성/권한 부여/실제 알림 전송은 구현 검증 중 자동 수행하지 않는다.
+
+## 배포
+
+먼저 새 이미지에 requirements를 설치하고 `0014_search_console.sql`을 기존 마이그레이션 도구로 적용한다.
+프로젝트 전체 마이그레이션 순서를 확인한 뒤 실행한다. 아래는 운영자가 실행할 명령이며 자동 배포하지 않았다.
+
+```bash
+docker compose exec -T solution-backend python scripts/migrate.py
+docker compose -f docker-compose.yml -f docker-compose.search-console.yml up -d --build solution-backend
+```
+
+선택 compose 파일은 API에만 키를 읽기 전용 마운트하고 `GSC_CREDENTIALS_FILE`을 설정한다.
+없는 파일을 디렉토리로 자동 생성하지 않는다. 이후 배포에서도 이 override를 함께 사용해야 한다.
+로컬 Python 실행은 `GSC_CREDENTIALS_FILE`에 로컬 키 파일 경로를 지정한다.
+켜진 스케줄러는 첫 10분 주기부터 기존 발행 사이트도 등록한다. `GSC_ENABLED=0`이면 DB/Google 호출 모두 생략한다.
+
+## 알림
+
+Teams Workflows의 webhook 수신 → 채널에 Adaptive Card 게시 흐름 URL을
+`GSC_ALERT_WEBHOOK_URL`에 넣는다. 비우면 외부 전송 없이 경고 로그/DB만 남는다.
+API/사이트맵 오류 또는 발행 후 기본 7일 미색인 시 알린다. 성공한 알림은 사이트별 24시간 중복 억제.
+전송 실패는 `alerted_at`을 갱신하지 않아 다음 검사 때 재시도한다.
+외부 전송 후 DB commit 전에 죽으면 중복 알림이 가능하다(at-least-once).
+키·토큰·webhook URL·Google 오류 본문은 알림에 포함하지 않는다.
+
+## 결과 확인
+
+```bash
+docker compose exec -T solution-backend python scripts/search_console_status.py
+```
+
+읽기 전용이며 Google API를 추가 호출하지 않는다. 프론트 화면/API 계약은 변경하지 않았다.
+
+| 파일 | 책임 |
+|---|---|
+| `services/search_console_client.py` | 인증·Google HTTP·오류 정규화 |
+| `services/search_console_settings.py` | 선택 설정·속성 URL 범위 |
+| `services/search_console_service.py` | 배치 흐름·재시도·관측 결과 |
+| `crud/search_console_crud.py` | 발행 감지·등록·조회 순서·동시 실행 잠금 |
+| `services/search_console_alerts.py` | 알림 조건·Teams 전송 |
+
+## 구글 지원 범위 / 남은 운영 작업
+
+- [사이트맵 제출 API](https://developers.google.com/webmaster-tools/v1/sitemaps/submit)는 지원된다.
+- [URL Inspection API](https://developers.google.com/webmaster-tools/v1/urlInspection.index/inspect)는
+  Google이 이미 알고 있는 상태 조회용이며 실시간 페이지 테스트나 색인 요청 API가 아니다.
+- 일반 숙박 사이트는 [Indexing API](https://developers.google.com/search/apis/indexing-api/v3/using-api) 대상이 아니다.
+- [검사 할당량](https://developers.google.com/webmaster-tools/limits)은 속성당 하루 2,000회다.
+- [Teams webhook 형식](https://learn.microsoft.com/en-us/microsoftteams/platform/webhooks-and-connectors/how-to/add-incoming-webhook).
+- 실제 서비스 계정 권한/사이트맵 제출/색인 관측/Teams 수신은 설정 후 운영 검증이 필요하다.
+- 기존 루트 사이트맵의 백업 URL 정리와 IndexNow 개별 사이트맵 참조 문제는 이 기능과 별도다.
+  이 기능은 기존 공개 사이트맵을 제출하며 내용을 다시 만들거나 목업을 삭제하지 않는다.
+
+## 구현 검증 (2026-09-15)
+
+- 격리 PostgreSQL에서 클라이언트·배치·스키마·IndexNow 관련 59건 통과.
+- 발행·설정·사이트 목록 회귀검사: 23건 통과, `test_unverified_fact_blocks_publish` 1건 실패.
+  해당 실패는 변경 전 HEAD `9773bc0`의 발행 코드에서도 동일 재현됨(GSC 비활성).
+- Google/Teams 실호출 없음. 서비스 계정 권한·실제 제출·채널 수신은 운영 설정 후 검증 대상.

docs/SEARCH_CONSOLE_CLIENT.md

@@ -0,0 +1,71 @@
+# Search Console 클라이언트
+
+`solution/backend/services/search_console_client.py` — Google Search Console 에
+사이트맵을 제출하고 URL 색인 상태를 조회하는 REST 클라이언트만 다룬다.
+DB 저장·스케줄링·발행 감지·환경 설정은 [SEARCH_CONSOLE.md](SEARCH_CONSOLE.md)를 따른다.
+
+공식 문서: [Sitemaps.submit](https://developers.google.com/webmaster-tools/v1/sitemaps/submit) ·
+[urlInspection.index.inspect](https://developers.google.com/webmaster-tools/v1/urlInspection.index/inspect)
+
+## 1. 계약
+
+```python
+class SearchConsoleClient:
+    def __init__(self, credentials_file: str, *, transport: httpx.AsyncBaseTransport | None = None): ...
+    async def submit_sitemap(self, property_url: str, sitemap_url: str) -> None: ...
+    async def inspect_url(self, property_url: str, page_url: str) -> dict: ...  # indexStatusResult 만
+    async def aclose(self) -> None: ...
+    # async with SearchConsoleClient(...) as client: ...
+```
+
+- `credentials_file`: 서비스 계정 JSON 키 파일 경로.
+- `transport`: 테스트에서 `httpx.MockTransport` 를 꽂는 자리 — 실제 Google 호출 없이 검증한다.
+- `inspect_url` 은 응답의 `inspectionResult.indexStatusResult` 만 돌려준다. 그 경로가
+  없거나(검사 실패) 빈 dict 면(실제 검사가 안 된 응답) `SearchConsoleError` 를 올린다 —
+  **"미색인"으로 넘겨짚지 않는다.**
+
+## 2. 인증
+
+서비스 계정 JSON 키 파일 + scope `https://www.googleapis.com/auth/webmasters`.
+`google.oauth2.service_account` · `google.auth.transport.requests.Request` · `requests` 는
+전부 함수 안에서 import 한다. 두 패키지는 백엔드 `requirements.txt`에 포함되어 있다.
+
+- 토큰은 클라이언트 인스턴스에 캐시된다(`credentials.valid` 인 동안 재사용, 매 호출
+  갱신하지 않는다). 동시 호출은 `asyncio.Lock` 으로 갱신을 한 번만 태운다.
+- 갱신은 `asyncio.to_thread` 로 별도 스레드에서 돈다. 내부 `requests.Session` 요청에는
+  타임아웃을 강제로 20초로 덮어씌운다(`Request.__call__` 기본값 120초를 무시) — 만료된
+  키·막힌 네트워크에서 무한정 걸리는 것을 막는다.
+
+## 3. 오류 — `SearchConsoleError(code)`
+
+`code` 문자열 하나만 들고 다닌다. **Google 응답 본문·액세스 토큰·키 파일 내용·원본 예외
+메시지는 절대 담지 않는다** — 로그·잡 상태·관리 화면 어디로 흘러도 안전하다.
+
+| code | 뜻 |
+|---|---|
+| `invalid_credentials_file` | 키 파일을 못 읽거나 형식이 잘못됨 |
+| `auth_failed` | 토큰 갱신 실패, 또는 갱신 후에도 토큰이 비어 있음 |
+| `unauthorized` | HTTP 401 |
+| `forbidden` | HTTP 403 |
+| `rate_limited` | HTTP 429 |
+| `server_error` | HTTP 5xx |
+| `http_<code>` | 그 외 실패 상태코드 |
+| `timeout` | 요청 타임아웃 |
+| `transport_error` | 그 외 전송 실패(연결 끊김 등) |
+| `invalid_json` | 200 인데 본문이 JSON 이 아님 |
+| `missing_inspection_result` | 응답에 `inspectionResult` 가 없음 |
+| `missing_index_status_result` | `inspectionResult` 는 있는데 `indexStatusResult` 가 없거나 빈 dict |
+
+## 4. 테스트
+
+```bash
+cd solution/backend
+APP_ENV=test .venv/bin/python -m pytest tests/test_search_console_client.py --confcutdir=tests
+```
+
+`--confcutdir=tests` 가 필요한 이유: 저장소 루트 `conftest.py` 의 세션 스코프 autouse
+픽스처가 실 Postgres 연결을 요구한다(`solution/backend/conftest.py`). 이 클라이언트
+테스트는 DB 를 전혀 쓰지 않으므로 그 픽스처를 건너뛴다 — `--confcutdir=tests` 로 상위
+`conftest.py` 탐색을 끊는다. (통합 후 전체 스위트를 돌릴 때는 이 플래그 없이 실행한다.)
+
+Google 실 API 는 전부 `httpx.MockTransport` 로 막았다 — 네트워크 호출도, 과금도 없다.

docs/TEAMS_WEBHOOK.md

@@ -0,0 +1,11 @@
+# Teams 웹훅 확인 (2026-09-15)
+
+- Adaptive Card 요청의 `contentUrl: null` 및 `$schema`를 공식 예제에 맞춰 보완했다.
+- HTTP 202는 워크플로의 요청 접수다. Teams 채널 게시 성공을 뜻하지 않는다.
+- 실제 전송 2건은 202였지만 사용자가 확인한 워크플로 실행은 실패였다.
+  상세 오류를 확인하지 못했으므로 누락 필드를 실제 실패 원인으로 단정하지 않는다.
+- 운영 자동 알림 활성화 전, 채널 수신 또는 워크플로의 최종 게시 단계 성공을 확인해야 한다.
+- 웹훅은 `.env`에만 보관하고 커밋하지 않는다.
+
+검증: 백엔드에서 `APP_ENV=test PYTHONPATH=. .venv/bin/pytest tests/test_search_console_alerts.py --confcutdir=tests`.
+공식 형식: https://learn.microsoft.com/en-us/connectors/teams/#adaptivecarditemschema

postgres-init/init-data/init.sql

@@ -317,6 +317,25 @@ CREATE TABLE IF NOT EXISTS public.place_area_refs (
 --   ★ 정적 빌드 — DB 는 빌드 시점에만 읽고 방문자와 만나지 않는다.
 --   ★ 해지는 물리 삭제가 아니라 상태 전이다. 색인된 페이지를 갑자기 404 로 만들지 않는다.
 -- ============================================================
+CREATE TABLE IF NOT EXISTS public.site_search_status (
+    site_id uuid PRIMARY KEY,
+    site_version_id uuid NOT NULL,
+    property_url TEXT NOT NULL,
+    page_url TEXT NOT NULL,
+    published_at TIMESTAMPTZ NOT NULL,
+    sitemap_submitted_at TIMESTAMPTZ NULL,
+    inspected_at TIMESTAMPTZ NULL,
+    first_indexed_at TIMESTAMPTZ NULL,
+    inspection JSONB NULL,
+    error_code VARCHAR(100) NULL,
+    failures INTEGER NOT NULL DEFAULT 0,
+    next_check_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    alerted_at TIMESTAMPTZ NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted BOOLEAN NOT NULL DEFAULT false
+);
+
 CREATE TABLE IF NOT EXISTS public.sites (
     site_id            uuid         PRIMARY KEY DEFAULT gen_random_uuid(),
     place_id           uuid         NOT NULL,                                                     -- 사업장과 1:1

postgres-init/migrations/0014_search_console.sql

@@ -0,0 +1,19 @@
+-- 발행 상태를 바꾸지 않고 Google 제출/검사 결과를 추적한다.
+CREATE TABLE IF NOT EXISTS public.site_search_status (
+    site_id uuid PRIMARY KEY,
+    site_version_id uuid NOT NULL,
+    property_url TEXT NOT NULL,
+    page_url TEXT NOT NULL,
+    published_at TIMESTAMPTZ NOT NULL,
+    sitemap_submitted_at TIMESTAMPTZ NULL,
+    inspected_at TIMESTAMPTZ NULL,
+    first_indexed_at TIMESTAMPTZ NULL,
+    inspection JSONB NULL,
+    error_code VARCHAR(100) NULL,
+    failures INTEGER NOT NULL DEFAULT 0,
+    next_check_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    alerted_at TIMESTAMPTZ NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted BOOLEAN NOT NULL DEFAULT false
+);

solution/backend/common/database/model/models.py

@@ -454,6 +454,25 @@ class sites(MainTableMixin, MAIN_BASE):
     thumbnail_url = Column(String(500), nullable=True)
 
 
+class site_search_status(MainTableMixin, MAIN_BASE):
+    """발행 성공과 Google 색인 성공은 다른 사건이라 별도 보관한다."""
+
+    __tablename__ = "site_search_status"
+    site_id = Column(UUID(as_uuid=True), primary_key=True)
+    site_version_id = Column(UUID(as_uuid=True), nullable=False)
+    property_url = Column(Text, nullable=False)
+    page_url = Column(Text, nullable=False)
+    published_at = Column(DateTime(timezone=True), nullable=False)
+    sitemap_submitted_at = Column(DateTime(timezone=True), nullable=True)
+    inspected_at = Column(DateTime(timezone=True), nullable=True)
+    first_indexed_at = Column(DateTime(timezone=True), nullable=True)
+    inspection = Column(JSONB, nullable=True)
+    error_code = Column(String(100), nullable=True)
+    failures = Column(Integer, nullable=False, server_default=text("0"))
+    next_check_at = Column(DateTime(timezone=True), nullable=False, server_default=_utc_now_sql())
+    alerted_at = Column(DateTime(timezone=True), nullable=True)
+
+
 class site_sections(MainTableMixin, MAIN_BASE):
     """섹션 하나의 콘텐츠. **JSON import/export 의 단위**다.
 

solution/backend/crud/search_console_crud.py

@@ -0,0 +1,57 @@
+"""발행 DB를 작업 원장으로 사용해 알림 적재 실패/재시작에도 대상을 다시 찾는다."""
+from sqlalchemy import select, or_, text
+from sqlalchemy.dialects.postgresql import insert
+
+from common.database.model.models import sites, places, site_versions, site_search_status as Status
+from common.enums import SiteStatus
+
+
+def published_conditions():
+    return (
+        sites.deleted.is_(False), places.deleted.is_(False),
+        sites.status == SiteStatus.PUBLISHED.value,
+        sites.published_at.is_not(None), sites.current_version_id.is_not(None),
+    )
+
+
+async def lock_batch(session) -> bool:
+    # 여러 API 프로세스가 같은 크론을 등록해도 외부 호출은 한 곳만 수행한다.
+    return bool(await session.scalar(text("SELECT pg_try_advisory_xact_lock(734920151)")))
+
+
+async def new_publications(session, property_url: str):
+    result = await session.execute(
+        select(sites, places, site_versions)
+        .join(places, places.place_id == sites.place_id)
+        .join(site_versions, site_versions.site_version_id == sites.current_version_id)
+        .outerjoin(Status, Status.site_id == sites.site_id)
+        .where(*published_conditions(), site_versions.deleted.is_(False), or_(
+            Status.site_id.is_(None), Status.site_version_id != sites.current_version_id,
+            Status.property_url != property_url, Status.published_at != sites.published_at,
+        ))
+        .order_by(sites.published_at).limit(100)
+    )
+    return result.all()
+
+
+async def register(session, values: dict):
+    reset = dict(values, sitemap_submitted_at=None, inspected_at=None, first_indexed_at=None,
+                 inspection=None, error_code=None, failures=0, alerted_at=None,
+                 next_check_at=text("now()"), updated_at=text("now()"), deleted=False)
+    await session.execute(insert(Status).values(**values).on_conflict_do_update(
+        index_elements=[Status.site_id], set_=reset,
+    ))
+
+
+async def due_sites(session, property_url: str):
+    result = await session.execute(
+        select(Status).join(sites, sites.site_id == Status.site_id)
+        .join(places, places.place_id == sites.place_id)
+        .where(*published_conditions(), Status.deleted.is_(False),
+               Status.site_version_id == sites.current_version_id,
+               Status.published_at == sites.published_at, Status.property_url == property_url,
+               Status.next_check_at <= text("now()"))
+        .order_by(Status.next_check_at).limit(5)
+        .with_for_update(of=Status)
+    )
+    return result.scalars().all()

solution/backend/requirements.txt

@@ -9,6 +9,8 @@ orjson
 pydantic>=2.0
 python-multipart
 httpx
+google-auth>=2.0                 # Search Console 서비스 계정 인증 (선택 기능)
+requests>=2.31                  # google-auth 토큰 갱신 transport
 apscheduler>=3.10
 pydantic-settings                # 환경변수·.env 로드 (FastAPI 공식 설정 방식)
 azure-storage-blob>=12.19

solution/backend/scheduler/__init__.py

@@ -2,7 +2,7 @@
 
 다중 워커(운영)에서 잡이 워커마다 중복 실행되면 안 되므로 SCHEDULER_ENABLED=1 인 프로세스에서만 등록한다.
 
-등록된 잡: 없음(스켈레톤). 붙을 잡 —
+등록된 잡: Search Console (GSC_ENABLED=1, 10분마다). 붙을 잡 —
   · 지역정보 갱신 : 축제 주 1회 / 관광정보 월 1회 / 날씨 시간 단위 — 행정구역 코드 단위 캐시 갱신
   · 수집 재시도   : 실패한 수집 작업 재시도 (외부 API 실패 시 직전 값 유지 + 내부 알림)
   · 사이트 재빌드 : 검증 상태가 바뀐 place 만 개별 재빌드 (전체 재빌드 금지)
@@ -33,8 +33,12 @@ def start_scheduler():
 
     # 한국시간 기준. 잡은 scheduler/jobs.py 에 정의하고 여기서 add_job 으로 등록한다.
     _scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
+    if os.environ.get("GSC_ENABLED") == "1":
+        from services.search_console_service import run_scheduled_check
+        _scheduler.add_job(run_scheduled_check, "interval", minutes=10,
+                           id="search-console", max_instances=1, coalesce=True)
     _scheduler.start()
-    LOG.i("[scheduler] started (KST: 등록된 잡 없음)")
+    LOG.i(f"[scheduler] started (KST: {len(_scheduler.get_jobs())}개 잡)")
 
 
 def shutdown_scheduler():

solution/backend/scripts/search_console_status.py

@@ -0,0 +1,32 @@
+"""저장된 관측값만 출력한다. Google 호출/색인 요청은 하지 않는다."""
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from sqlalchemy import select
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import site_search_status as Status
+from common.enums import DBType
+
+
+async def main():
+    async def read(session):
+        rows = (await session.execute(select(Status).where(Status.deleted.is_(False))
+                                     .order_by(Status.published_at.desc()))).scalars().all()
+        return [{"url": row.page_url, "published_at": row.published_at,
+                 "sitemap_submitted_at": row.sitemap_submitted_at,
+                 "inspected_at": row.inspected_at, "first_indexed_at": row.first_indexed_at,
+                 "inspection": row.inspection, "error_code": row.error_code,
+                 "next_check_at": row.next_check_at, "alerted_at": row.alerted_at} for row in rows]
+    try:
+        rows = await DB_SESSION_MNG.execute_lambda_write(DBType.MAIN.value, read)
+        print(json.dumps(rows, ensure_ascii=False, indent=2, default=str))
+    finally:
+        await DB_SESSION_MNG.dispose_all()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

solution/backend/services/search_console_alerts.py

@@ -0,0 +1,45 @@
+"""Teams Workflow 수신용. 원문 오류/인증 정보는 메시지에 싣지 않는다."""
+from datetime import timedelta
+
+import httpx
+
+from common.logger import LOG
+
+
+def alert_reason(row, now, days: int) -> str | None:
+    if row.alerted_at and now - row.alerted_at < timedelta(days=1):
+        return None
+    if row.error_code:
+        return f"조회/제출 실패: {row.error_code}"
+    if (row.inspection or {}).get("verdict") == "PASS":
+        return None
+    if now - row.published_at >= timedelta(days=days):
+        return f"발행 후 {days}일 이상 색인 미확인"
+    return None
+
+
+async def send_alert(webhook_url: str, page_url: str, reason: str) -> bool:
+    if not webhook_url:
+        return False
+    if not webhook_url.startswith("https://"):
+        LOG.w("[search-console] ALERT_URL_INVALID")
+        return False
+    payload = {"type": "message", "attachments": [{
+        "contentType": "application/vnd.microsoft.card.adaptive",
+        "contentUrl": None,
+        "content": {"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+                    "type": "AdaptiveCard", "version": "1.2", "body": [
+            {"type": "TextBlock", "text": "Google 색인 확인 필요", "weight": "Bolder"},
+            {"type": "TextBlock", "text": page_url, "wrap": True},
+            {"type": "TextBlock", "text": reason, "wrap": True},
+        ]},
+    }]}
+    try:
+        async with httpx.AsyncClient(timeout=10, follow_redirects=False) as client:
+            response = await client.post(webhook_url, json=payload)
+            response.raise_for_status()
+        return True
+    except httpx.HTTPError:
+        # webhook 주소 자체가 인증 수단이므로 예외 문자열도 로그에 남기지 않는다.
+        LOG.w("[search-console] ALERT_DELIVERY_FAILED")
+        return False

solution/backend/services/search_console_client.py

@@ -0,0 +1,164 @@
+"""Google Search Console REST 클라이언트 — 사이트맵 제출 · URL 색인 상태 조회.
+
+  PUT  https://www.googleapis.com/webmasters/v3/sites/{siteUrl}/sitemaps/{feedpath}
+  POST https://searchconsole.googleapis.com/v1/urlInspection/index:inspect
+
+자세한 계약은 docs/SEARCH_CONSOLE_CLIENT.md.
+
+★ `SearchConsoleError` 는 `code` 문자열만 담는다 — Google 응답 본문·토큰·키·원본
+  예외 메시지는 절대 싣지 않는다.
+★ `inspectionResult`·`indexStatusResult` 가 없거나 빈 응답은 "미색인"으로
+  넘겨짚지 않고 예외로 끊는다 — 검사 실패와 색인 결과를 구분 못하면 오판이 된다.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+from urllib.parse import quote
+
+import httpx
+
+SITEMAPS_BASE = "https://www.googleapis.com/webmasters/v3/sites"
+INSPECT_URL = "https://searchconsole.googleapis.com/v1/urlInspection/index:inspect"
+SCOPES = ["https://www.googleapis.com/auth/webmasters"]
+TIMEOUT_SEC = 20.0
+AUTH_REFRESH_TIMEOUT_SEC = 20.0
+
+
+class SearchConsoleError(RuntimeError):
+    """code 만 담는다 — Google 응답 본문·토큰·키는 여기 실으면 안 된다."""
+
+    def __init__(self, code: str):
+        self.code = code
+        super().__init__(code)
+
+
+def _service_account_credentials(credentials_file: str):
+    """lazy import — 미설치여도 이 모듈의 import 자체는 죽지 않는다."""
+    from google.oauth2 import service_account
+
+    return service_account.Credentials.from_service_account_file(credentials_file, scopes=SCOPES)
+
+
+def _refresh_sync(credentials) -> None:
+    """동기 토큰 갱신. `asyncio.to_thread` 로 감싸 부른다.
+
+    `Request.__call__` 기본 타임아웃(120s)을 그대로 두면 만료된 키·막힌 네트워크에서
+    오래 걸릴 수 있다 — 매 요청에 상한을 강제로 덮어씌운다(호출측이 넘긴 값 포함)."""
+    import requests
+    from google.auth.transport.requests import Request
+
+    with requests.Session() as session:
+        original_request = session.request
+
+        def _request_with_timeout(*args, **kwargs):
+            kwargs["timeout"] = AUTH_REFRESH_TIMEOUT_SEC
+            return original_request(*args, **kwargs)
+
+        session.request = _request_with_timeout
+        credentials.refresh(Request(session=session))
+
+
+def _status_error_code(status_code: int) -> str:
+    if status_code == 401:
+        return "unauthorized"
+    if status_code == 403:
+        return "forbidden"
+    if status_code == 429:
+        return "rate_limited"
+    if 500 <= status_code < 600:
+        return "server_error"
+    return f"http_{status_code}"
+
+
+def _transport_error_code(ex: httpx.HTTPError) -> str:
+    if isinstance(ex, httpx.TimeoutException):
+        return "timeout"
+    return "transport_error"
+
+
+class SearchConsoleClient:
+    def __init__(self, credentials_file: str, *, transport: httpx.AsyncBaseTransport | None = None):
+        self._credentials_file = credentials_file
+        self._client = httpx.AsyncClient(timeout=TIMEOUT_SEC, transport=transport)
+        self._credentials: Any = None
+        self._auth_lock = asyncio.Lock()
+
+    async def __aenter__(self) -> "SearchConsoleClient":
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        await self.aclose()
+
+    async def aclose(self) -> None:
+        await self._client.aclose()
+
+    async def submit_sitemap(self, property_url: str, sitemap_url: str) -> None:
+        url = f"{SITEMAPS_BASE}/{quote(property_url, safe='')}/sitemaps/{quote(sitemap_url, safe='')}"
+        headers = await self._auth_headers()
+        res = await self._put(url, headers)
+        self._raise_for_status(res)
+
+    async def inspect_url(self, property_url: str, page_url: str) -> dict[str, Any]:
+        headers = await self._auth_headers()
+        body = {"inspectionUrl": page_url, "siteUrl": property_url}
+        res = await self._post(INSPECT_URL, headers, body)
+        self._raise_for_status(res)
+        return _index_status_result(res)
+
+    async def _put(self, url: str, headers: dict[str, str]) -> httpx.Response:
+        try:
+            return await self._client.put(url, headers=headers)
+        except httpx.HTTPError as ex:
+            raise SearchConsoleError(_transport_error_code(ex)) from None
+
+    async def _post(self, url: str, headers: dict[str, str], body: dict[str, Any]) -> httpx.Response:
+        try:
+            return await self._client.post(url, headers=headers, json=body)
+        except httpx.HTTPError as ex:
+            raise SearchConsoleError(_transport_error_code(ex)) from None
+
+    def _raise_for_status(self, res: httpx.Response) -> None:
+        if not res.is_success:
+            raise SearchConsoleError(_status_error_code(res.status_code))
+
+    async def _auth_headers(self) -> dict[str, str]:
+        token = await self._access_token()
+        return {"Authorization": f"Bearer {token}"}
+
+    async def _access_token(self) -> str:
+        credentials = self._load_credentials()
+        if not getattr(credentials, "valid", False):
+            async with self._auth_lock:
+                if not getattr(credentials, "valid", False):
+                    try:
+                        await asyncio.to_thread(_refresh_sync, credentials)
+                    except Exception:
+                        raise SearchConsoleError("auth_failed") from None
+        token = getattr(credentials, "token", None)
+        if not token:
+            raise SearchConsoleError("auth_failed")
+        return token
+
+    def _load_credentials(self) -> Any:
+        if self._credentials is None:
+            try:
+                self._credentials = _service_account_credentials(self._credentials_file)
+            except Exception:
+                raise SearchConsoleError("invalid_credentials_file") from None
+        return self._credentials
+
+
+def _index_status_result(res: httpx.Response) -> dict[str, Any]:
+    try:
+        data = res.json()
+    except ValueError:
+        raise SearchConsoleError("invalid_json") from None
+    inspection_result = data.get("inspectionResult") if isinstance(data, dict) else None
+    if not isinstance(inspection_result, dict):
+        raise SearchConsoleError("missing_inspection_result")
+    index_status_result = inspection_result.get("indexStatusResult")
+    if not isinstance(index_status_result, dict) or not index_status_result:
+        raise SearchConsoleError("missing_index_status_result")
+    return index_status_result

solution/backend/services/search_console_service.py

@@ -0,0 +1,129 @@
+"""발행 감지 → 사이트맵 제출 → 색인 조회 → 알림. 발행 잡과 별도 트랜잭션이다."""
+import asyncio
+import xml.etree.ElementTree as ET
+from datetime import datetime, timedelta, timezone
+
+import httpx
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.enums import DBType
+from common.logger import LOG
+from crud import search_console_crud as store
+from services import site_payload
+from services.search_console_alerts import alert_reason, send_alert
+from services.search_console_client import SearchConsoleClient, SearchConsoleError
+from services.search_console_settings import load_settings, belongs_to_property
+
+
+async def run_scheduled_check():
+    try:
+        settings = load_settings()
+        if settings is None:
+            return
+        await DB_SESSION_MNG.execute_lambda_write(
+            DBType.MAIN.value, lambda session: run_batch(session, settings),
+        )
+    except Exception:
+        # 크론 실패가 API/발행에 전파되지 않고 다음 주기에 재시도된다.
+        LOG.w("[search-console] BATCH_FAILED — 설정 및 DB 마이그레이션 확인 필요")
+
+
+async def run_batch(session, settings):
+    if not await store.lock_batch(session):
+        return
+    if not belongs_to_property(site_payload.publish_origin() + "/sitemap.xml", settings.property_url):
+        raise SearchConsoleError("PROPERTY_MISMATCH")
+    await register_publications(session, settings)
+    rows = await store.due_sites(session, settings.property_url)
+    if not rows:
+        return
+    async with SearchConsoleClient(settings.credentials_file) as client:
+        submitted = set()
+        for row in rows:
+            await check_site(client, row, settings, submitted)
+    await session.flush()
+
+
+async def register_publications(session, settings):
+    for site, place, version in await store.new_publications(session, settings.property_url):
+        slug = site_payload.publish_slug(place, site)
+        page_url = f"{site_payload.publish_origin()}/s/{slug}"
+        if not belongs_to_property(page_url, settings.property_url):
+            LOG.w("[search-console] PROPERTY_MISMATCH")
+            continue
+        await store.register(session, {
+            "site_id": site.site_id, "site_version_id": version.site_version_id,
+            "property_url": settings.property_url, "page_url": page_url,
+            "published_at": site.published_at,
+        })
+
+
+async def read_sitemap(sitemap_url: str) -> set[str]:
+    try:
+        async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client:
+            async with client.stream("GET", sitemap_url) as response:
+                response.raise_for_status()
+                body = bytearray()
+                async for chunk in response.aiter_bytes():
+                    body.extend(chunk)
+                    if len(body) > 5_000_000:
+                        raise SearchConsoleError("SITEMAP_TOO_LARGE")
+        root = ET.fromstring(body)
+        return {(item.text or "").strip() for item in root.iter(
+            "{http://www.sitemaps.org/schemas/sitemap/0.9}loc"
+        )}
+    except (httpx.HTTPError, ET.ParseError):
+        raise SearchConsoleError("SITEMAP_UNAVAILABLE") from None
+
+
+async def submit_sitemap(client, row, submitted: set):
+    if row.sitemap_submitted_at:
+        return
+    sitemap_url = site_payload.publish_origin() + "/sitemap.xml"
+    # 디스크가 아니라 실제 공개 URL을 확인한다. 업로드 지연 시 Google에 먼저 알리지 않는다.
+    urls = await read_sitemap(sitemap_url)
+    if row.page_url not in urls:
+        raise SearchConsoleError("URL_NOT_IN_SITEMAP")
+    if sitemap_url not in submitted:
+        await client.submit_sitemap(row.property_url, sitemap_url)
+        submitted.add(sitemap_url)
+    row.sitemap_submitted_at = datetime.now(timezone.utc)
+
+
+def record_inspection(row, inspection: dict, now):
+    row.inspection = inspection
+    row.inspected_at = now
+    row.error_code = None
+    row.failures = 0
+    if inspection.get("verdict") == "PASS":
+        row.first_indexed_at = row.first_indexed_at or now
+    row.next_check_at = now + timedelta(days=1)
+
+
+def record_error(row, code: str, now):
+    row.error_code = code
+    row.failures = (row.failures or 0) + 1
+    hours = min(24, 2 ** min(row.failures - 1, 5))
+    row.next_check_at = now + timedelta(hours=hours)
+
+
+async def check_site(client, row, settings, submitted: set):
+    now = datetime.now(timezone.utc)
+    try:
+        if not belongs_to_property(row.page_url, settings.property_url):
+            raise SearchConsoleError("PROPERTY_MISMATCH")
+        async with asyncio.timeout(90):
+            await submit_sitemap(client, row, submitted)
+            inspection = await client.inspect_url(row.property_url, row.page_url)
+        record_inspection(row, inspection, now)
+    except SearchConsoleError as ex:
+        record_error(row, ex.code, now)
+    except TimeoutError:
+        record_error(row, "CHECK_TIMEOUT", now)
+    except Exception:
+        record_error(row, "CHECK_FAILED", now)
+    reason = alert_reason(row, now, settings.alert_days)
+    if reason:
+        LOG.w(f"[search-console] site={row.site_id} {reason}")
+        if await send_alert(settings.alert_url, row.page_url, reason):
+            row.alerted_at = now

solution/backend/services/search_console_settings.py

@@ -0,0 +1,43 @@
+"""Google 로그인 설정과 분리한다. 서비스 계정 키는 서버 파일로만 읽는다."""
+import os
+from dataclasses import dataclass
+from urllib.parse import urlsplit
+
+
+@dataclass(frozen=True)
+class SearchConsoleSettings:
+    property_url: str
+    credentials_file: str
+    alert_url: str = ""
+    alert_days: int = 7
+
+
+def load_settings() -> SearchConsoleSettings | None:
+    if os.environ.get("GSC_ENABLED") != "1":
+        return None
+    prop = os.environ.get("GSC_PROPERTY_URL", "").strip()
+    key_file = os.environ.get("GSC_CREDENTIALS_FILE", "").strip()
+    if not prop or not key_file:
+        raise ValueError("GSC_CONFIG_MISSING")
+    parsed = urlsplit(prop)
+    domain_property = prop.startswith("sc-domain:") and bool(prop.removeprefix("sc-domain:"))
+    root_property = (parsed.scheme == "https" and parsed.hostname and parsed.path == "/"
+                     and not parsed.query and not parsed.fragment and not parsed.username)
+    if not domain_property and not root_property:
+        raise ValueError("GSC_PROPERTY_INVALID")
+    days = int(os.environ.get("GSC_ALERT_DAYS", "7"))
+    if days < 1:
+        raise ValueError("GSC_ALERT_DAYS_INVALID")
+    return SearchConsoleSettings(prop, key_file, os.environ.get("GSC_ALERT_WEBHOOK_URL", ""), days)
+
+
+def belongs_to_property(page_url: str, property_url: str) -> bool:
+    parsed = urlsplit(page_url)
+    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
+        return False
+    if parsed.query or parsed.fragment:
+        return False
+    if property_url.startswith("sc-domain:"):
+        domain = property_url.removeprefix("sc-domain:").lower()
+        return bool(domain) and (parsed.hostname == domain or parsed.hostname.endswith("." + domain))
+    return page_url.startswith(property_url)

solution/backend/tests/test_search_console_alerts.py

@@ -0,0 +1,21 @@
+import httpx
+from services import search_console_alerts as alerts
+
+
+async def test_teams_payload_has_required_attachment_fields(monkeypatch):
+    requests = []
+    real_client = httpx.AsyncClient
+
+    def receive(request):
+        import json
+        requests.append(json.loads(request.content))
+        return httpx.Response(202)
+
+    monkeypatch.setattr(alerts.httpx, "AsyncClient", lambda **kw: real_client(
+        transport=httpx.MockTransport(receive), **kw,
+    ))
+    assert await alerts.send_alert("https://example.test/webhook", "https://example.test/s/a", "test")
+    card = requests[0]["attachments"][0]
+    assert card["contentType"] == "application/vnd.microsoft.card.adaptive"
+    assert "contentUrl" in card and card["contentUrl"] is None
+    assert card["content"]["$schema"] == "http://adaptivecards.io/schemas/adaptive-card.json"

solution/backend/tests/test_search_console_client.py

@@ -0,0 +1,342 @@
+"""Search Console 클라이언트 — 사이트맵 제출 · URL 검사.
+
+★ 실제 Google API 를 호출하지 않는다(httpx.MockTransport). google-auth 도 이
+저장소의 의존성이 아니라서(Codex 가 추가) 인증은 모듈 함수
+`_service_account_credentials` / `_refresh_sync` 를 monkeypatch 해서 흉내 낸다 —
+설치 여부와 무관하게 이 테스트가 돌아야 한다.
+
+여기서 고정하는 것:
+  1. `inspect_url` 은 `inspectionResult.indexStatusResult` 만 돌려준다
+  2. ★ 그 필드가 없는 응답을 "미색인"으로 짐작하지 않고 예외를 올린다
+  3. 오류는 전부 `SearchConsoleError(code)` 로 정규화되고, 원본 예외 메시지·응답
+     본문이 `code` 에 섞여 나오지 않는다
+  4. 사이트맵 제출 URL 은 property/sitemap 주소를 완전 percent-encode 한다
+"""
+import asyncio
+
+import httpx
+import pytest
+
+from services import search_console_client as scc
+from services.search_console_client import SearchConsoleClient, SearchConsoleError
+
+PROPERTY = "https://example.com/"
+SITEMAP = "https://example.com/sitemap.xml"
+PAGE = "https://example.com/s/butter/"
+
+
+def _auth(monkeypatch, *, token="test-token", load_error=None, refresh_error=None):
+    """`_service_account_credentials`/`_refresh_sync` 를 흉내내 google-auth 없이 인증을 통과시킨다."""
+    state = {"loads": 0, "refreshes": 0}
+
+    class _FakeCredentials:
+        def __init__(self):
+            self.token = None
+            self.valid = False
+
+    def fake_load(path):
+        state["loads"] += 1
+        if load_error is not None:
+            raise load_error
+        return _FakeCredentials()
+
+    def fake_refresh(creds):
+        state["refreshes"] += 1
+        if refresh_error is not None:
+            raise refresh_error
+        creds.token = token
+        creds.valid = bool(token)
+
+    monkeypatch.setattr(scc, "_service_account_credentials", fake_load)
+    monkeypatch.setattr(scc, "_refresh_sync", fake_refresh)
+    return state
+
+
+def _client(handler, monkeypatch, **auth_kwargs) -> SearchConsoleClient:
+    _auth(monkeypatch, **auth_kwargs)
+    return SearchConsoleClient("dummy-credentials.json", transport=httpx.MockTransport(handler))
+
+
+# ── 1) 사이트맵 제출 ──────────────────────────────────────────────────────
+async def test_사이트맵_제출은_URL을_완전_퍼센트인코딩한다(monkeypatch):
+    """검증: property/sitemap 주소를 PUT 경로에 싣는다.
+    기대결과: 두 주소 모두 `/` `:` 까지 percent-encode 되어 경로 세그먼트 하나로 들어간다."""
+    captured = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["method"] = request.method
+        captured["url"] = str(request.url)
+        captured["auth"] = request.headers.get("Authorization")
+        return httpx.Response(200)
+
+    client = _client(handler, monkeypatch, token="tok-abc")
+    async with client:
+        await client.submit_sitemap(PROPERTY, SITEMAP)
+
+    assert captured["method"] == "PUT"
+    assert captured["url"] == (
+        f"{scc.SITEMAPS_BASE}/https%3A%2F%2Fexample.com%2F"
+        "/sitemaps/https%3A%2F%2Fexample.com%2Fsitemap.xml"
+    )
+    assert captured["auth"] == "Bearer tok-abc"
+
+
+async def test_사이트맵_제출_성공은_아무것도_돌려주지_않는다(monkeypatch):
+    """검증: 2xx 응답(빈 본문).
+    기대결과: 예외 없이 끝난다 — 반환값은 None."""
+    def handler(request):
+        return httpx.Response(200)
+
+    client = _client(handler, monkeypatch)
+    async with client:
+        assert await client.submit_sitemap(PROPERTY, SITEMAP) is None
+
+
+# ── 2) URL 검사 ───────────────────────────────────────────────────────────
+async def test_URL_검사는_indexStatusResult만_돌려준다(monkeypatch):
+    """검증: 응답에 indexStatusResult 외에 mobileUsabilityResult 등 다른 필드도 있다.
+    기대결과: indexStatusResult 만 뽑아 돌려준다."""
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert scc.INSPECT_URL in str(request.url)
+        return httpx.Response(200, json={
+            "inspectionResult": {
+                "indexStatusResult": {"verdict": "PASS", "coverageState": "Submitted and indexed"},
+                "mobileUsabilityResult": {"verdict": "PASS"},
+            }
+        })
+
+    client = _client(handler, monkeypatch)
+    async with client:
+        result = await client.inspect_url(PROPERTY, PAGE)
+
+    assert result == {"verdict": "PASS", "coverageState": "Submitted and indexed"}
+
+
+async def test_URL_검사_요청_본문은_inspectionUrl과_siteUrl이다(monkeypatch):
+    """검증: 검사 요청 본문.
+    기대결과: {"inspectionUrl": 검사할 페이지, "siteUrl": 프로퍼티} 그대로."""
+    captured = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["body"] = request.content
+        return httpx.Response(200, json={"inspectionResult": {"indexStatusResult": {"verdict": "PASS"}}})
+
+    import json as _json
+
+    client = _client(handler, monkeypatch)
+    async with client:
+        await client.inspect_url(PROPERTY, PAGE)
+
+    assert _json.loads(captured["body"]) == {"inspectionUrl": PAGE, "siteUrl": PROPERTY}
+
+
+async def test_inspectionResult가_없으면_미색인으로_넘겨짚지_않는다(monkeypatch):
+    """검증: ★ 응답 본문이 비어 있다({}).
+    기대결과: SearchConsoleError("missing_inspection_result") — 빈 dict 를 돌려주지 않는다.
+    (호출측이 빈 dict 를 "미색인"으로 오판할 수 있으므로 여기서 예외로 끊는다.)"""
+    def handler(request):
+        return httpx.Response(200, json={})
+
+    client = _client(handler, monkeypatch)
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.inspect_url(PROPERTY, PAGE)
+
+    assert exc.value.code == "missing_inspection_result"
+
+
+async def test_indexStatusResult가_없으면_예외(monkeypatch):
+    """검증: inspectionResult 는 있는데 indexStatusResult 가 없다(다른 결과만 있음).
+    기대결과: SearchConsoleError("missing_index_status_result")."""
+    def handler(request):
+        return httpx.Response(200, json={"inspectionResult": {"mobileUsabilityResult": {}}})
+
+    client = _client(handler, monkeypatch)
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.inspect_url(PROPERTY, PAGE)
+
+    assert exc.value.code == "missing_index_status_result"
+
+
+async def test_indexStatusResult가_빈_객체면_예외(monkeypatch):
+    """검증: indexStatusResult 는 있지만 빈 dict({}) — 실제 검사가 안 된 응답이다.
+    기대결과: SearchConsoleError("missing_index_status_result") — 빈 결과를 색인 상태로 돌려주지 않는다."""
+    def handler(request):
+        return httpx.Response(200, json={"inspectionResult": {"indexStatusResult": {}}})
+
+    client = _client(handler, monkeypatch)
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.inspect_url(PROPERTY, PAGE)
+
+    assert exc.value.code == "missing_index_status_result"
+
+
+async def test_비정상_JSON_본문은_invalid_json(monkeypatch):
+    """검증: 200 인데 본문이 JSON 이 아니다.
+    기대결과: SearchConsoleError("invalid_json") — 파싱 실패가 조용히 넘어가지 않는다."""
+    def handler(request):
+        return httpx.Response(200, text="<html>not json</html>")
+
+    client = _client(handler, monkeypatch)
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.inspect_url(PROPERTY, PAGE)
+
+    assert exc.value.code == "invalid_json"
+
+
+# ── 3) HTTP 상태 코드 정규화 ─────────────────────────────────────────────
+@pytest.mark.parametrize(
+    ("status", "code"),
+    [(401, "unauthorized"), (403, "forbidden"), (429, "rate_limited"), (500, "server_error"), (503, "server_error")],
+)
+async def test_HTTP_오류_상태코드는_code로_정규화된다(monkeypatch, status, code):
+    """검증: 401/403/429/5xx 응답.
+    기대결과: SearchConsoleError.code 가 상태별로 정규화되고, Google 응답 본문은 code 에 섞이지 않는다."""
+    def handler(request):
+        return httpx.Response(status, text="google 응답 본문(비밀은 아니지만 새면 안 된다)")
+
+    client = _client(handler, monkeypatch)
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.submit_sitemap(PROPERTY, SITEMAP)
+
+    assert exc.value.code == code
+    assert "google 응답 본문" not in str(exc.value)
+
+
+# ── 4) 전송 오류 ─────────────────────────────────────────────────────────
+async def test_타임아웃은_timeout으로_정규화(monkeypatch):
+    """검증: 요청이 타임아웃된다.
+    기대결과: SearchConsoleError("timeout")."""
+    def handler(request):
+        raise httpx.ReadTimeout("timed out", request=request)
+
+    client = _client(handler, monkeypatch)
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.inspect_url(PROPERTY, PAGE)
+
+    assert exc.value.code == "timeout"
+
+
+async def test_연결_실패는_transport_error로_정규화(monkeypatch):
+    """검증: 연결 자체가 끊긴다(DNS 실패 등).
+    기대결과: SearchConsoleError("transport_error")."""
+    def handler(request):
+        raise httpx.ConnectError("연결 실패", request=request)
+
+    client = _client(handler, monkeypatch)
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.submit_sitemap(PROPERTY, SITEMAP)
+
+    assert exc.value.code == "transport_error"
+
+
+# ── 5) 인증 실패 ─────────────────────────────────────────────────────────
+async def test_키파일_로딩_실패는_invalid_credentials_file(monkeypatch):
+    """검증: 서비스 계정 키 파일을 못 읽는다(없음·손상).
+    기대결과: SearchConsoleError("invalid_credentials_file") — 원본 예외 메시지는 안 실린다."""
+    def handler(request):
+        raise AssertionError("인증에 실패했으면 네트워크를 타면 안 된다")
+
+    client = _client(handler, monkeypatch, load_error=ValueError("키 파일 내용: super-secret-key-material"))
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.submit_sitemap(PROPERTY, SITEMAP)
+
+    assert exc.value.code == "invalid_credentials_file"
+    assert "super-secret-key-material" not in str(exc.value)
+
+
+async def test_토큰_갱신_실패는_auth_failed(monkeypatch):
+    """검증: 키 파일은 읽었지만 Google 토큰 발급이 거부된다(폐기된 키 등).
+    기대결과: SearchConsoleError("auth_failed") — 원본 예외 메시지는 안 실린다."""
+    def handler(request):
+        raise AssertionError("인증에 실패했으면 네트워크를 타면 안 된다")
+
+    client = _client(handler, monkeypatch, refresh_error=RuntimeError("invalid_grant: token revoked=xyz"))
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.inspect_url(PROPERTY, PAGE)
+
+    assert exc.value.code == "auth_failed"
+    assert "revoked=xyz" not in str(exc.value)
+
+
+async def test_토큰이_비면_auth_failed(monkeypatch):
+    """검증: 갱신은 예외 없이 끝났지만 credentials.token 이 비어 있다.
+    기대결과: SearchConsoleError("auth_failed") — 빈 토큰으로 요청을 보내지 않는다."""
+    def handler(request):
+        raise AssertionError("빈 토큰으로 네트워크를 타면 안 된다")
+
+    client = _client(handler, monkeypatch, token="")
+    with pytest.raises(SearchConsoleError) as exc:
+        async with client:
+            await client.submit_sitemap(PROPERTY, SITEMAP)
+
+    assert exc.value.code == "auth_failed"
+
+
+async def test_인증정보_로딩과_유효한_토큰_갱신은_한_번만_한다(monkeypatch):
+    """검증: 같은 클라이언트로 두 번 호출한다.
+    기대결과: 키 파일 로딩 1회 · 토큰 갱신 1회 — 두 번째 호출은 여전히 유효한(`credentials.valid`)
+    토큰을 그대로 재사용한다."""
+    def handler(request):
+        return httpx.Response(200)
+
+    state = _auth(monkeypatch)
+    client = SearchConsoleClient("dummy-credentials.json", transport=httpx.MockTransport(handler))
+    async with client:
+        await client.submit_sitemap(PROPERTY, SITEMAP)
+        await client.submit_sitemap(PROPERTY, SITEMAP)
+
+    assert state["loads"] == 1
+    assert state["refreshes"] == 1
+
+
+async def test_동시_호출은_토큰_갱신을_한_번만_한다(monkeypatch):
+    """검증: 아직 토큰이 없는 클라이언트를 두 요청이 동시에 부른다.
+    기대결과: 갱신 1회 — 잠금 없이 두 번째 요청이 갱신 중인 자격을 다시 갱신하면
+    Google 토큰 엔드포인트를 요청 수만큼 때리게 된다."""
+    import time as _time
+
+    def handler(request):
+        return httpx.Response(200)
+
+    state = _auth(monkeypatch)
+
+    def slow_refresh(creds):
+        _time.sleep(0.05)
+        state["refreshes"] += 1
+        creds.token = "tok"
+        creds.valid = True
+
+    monkeypatch.setattr(scc, "_refresh_sync", slow_refresh)
+
+    client = SearchConsoleClient("dummy-credentials.json", transport=httpx.MockTransport(handler))
+    async with client:
+        await asyncio.gather(
+            client.submit_sitemap(PROPERTY, SITEMAP),
+            client.submit_sitemap(PROPERTY, SITEMAP),
+        )
+
+    assert state["refreshes"] == 1
+
+
+# ── 6) 닫기 ──────────────────────────────────────────────────────────────
+async def test_async_컨텍스트매니저는_내부_httpx_클라이언트를_닫는다(monkeypatch):
+    """검증: `async with` 블록을 빠져나간다.
+    기대결과: 내부 httpx.AsyncClient 가 닫힌다 — 커넥션을 남겨두지 않는다."""
+    def handler(request):
+        return httpx.Response(200)
+
+    client = _client(handler, monkeypatch)
+    async with client:
+        pass
+
+    assert client._client.is_closed is True

solution/backend/tests/test_search_console_service.py

@@ -0,0 +1,273 @@
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+import uuid
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.database.model.models import places, sites, site_versions, site_search_status
+from crud import search_console_crud as store
+from services import search_console_service as service
+from services.search_console_client import SearchConsoleError
+from services.search_console_alerts import alert_reason, send_alert
+from services.search_console_settings import SearchConsoleSettings, belongs_to_property, load_settings
+
+NOW = datetime.now(timezone.utc)
+SETTINGS = SearchConsoleSettings("https://example.com/", "/test/key.json")
+
+
+def status(**changes):
+    return SimpleNamespace(site_id=uuid.uuid4(), page_url="https://example.com/s/stay",
+                           property_url=SETTINGS.property_url, published_at=NOW - timedelta(days=8),
+                           sitemap_submitted_at=None, inspected_at=None, inspection=None,
+                           first_indexed_at=None, error_code=None, failures=0, alerted_at=None,
+                           next_check_at=NOW, **changes)
+
+
+@pytest.mark.parametrize("url,prop,expected", [
+    ("https://example.com/s/a", "https://example.com/", True),
+    ("https://example.com.evil.test/s/a", "https://example.com/", False),
+    ("https://sub.example.com/s/a", "sc-domain:example.com", True),
+    ("https://notexample.com/s/a", "sc-domain:example.com", False),
+    ("http://example.com/s/a", "sc-domain:example.com", False),
+    ("https://example.com/s/a?secret=x", "sc-domain:example.com", False),
+])
+def test_property_scope(url, prop, expected):
+    assert belongs_to_property(url, prop) is expected
+
+
+def test_disabled_needs_no_credentials(monkeypatch):
+    monkeypatch.delenv("GSC_ENABLED", raising=False)
+    assert load_settings() is None
+    monkeypatch.setenv("GSC_ENABLED", "1")
+    monkeypatch.delenv("GSC_PROPERTY_URL", raising=False)
+    with pytest.raises(ValueError):
+        load_settings()
+
+
+async def test_submit_and_inspect_are_separate(monkeypatch):
+    row = status()
+    client = SimpleNamespace(submit_sitemap=AsyncMock(), inspect_url=AsyncMock(return_value={"verdict": "NEUTRAL"}))
+    monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value={row.page_url}))
+    monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com")
+    await service.check_site(client, row, SETTINGS, set())
+    assert row.sitemap_submitted_at
+    assert row.inspected_at
+    assert row.first_indexed_at is None
+    assert row.next_check_at > NOW
+    assert row.error_code is None
+
+
+async def test_missing_sitemap_url_retries_without_google_call(monkeypatch):
+    row = status()
+    client = SimpleNamespace(submit_sitemap=AsyncMock(), inspect_url=AsyncMock())
+    monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value=set()))
+    await service.check_site(client, row, SETTINGS, set())
+    assert row.error_code == "URL_NOT_IN_SITEMAP"
+    assert row.sitemap_submitted_at is None
+    assert row.failures == 1
+    client.submit_sitemap.assert_not_awaited()
+    client.inspect_url.assert_not_awaited()
+
+
+async def test_one_sitemap_submission_per_batch(monkeypatch):
+    client = SimpleNamespace(submit_sitemap=AsyncMock())
+    monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value={status().page_url}))
+    seen = set()
+    await service.submit_sitemap(client, status(), seen)
+    await service.submit_sitemap(client, status(), seen)
+    client.submit_sitemap.assert_awaited_once()
+
+
+async def test_inspection_failure_preserves_previous_observation():
+    row = status()
+    row.sitemap_submitted_at = NOW
+    row.inspection = {"verdict": "PASS"}
+    row.inspected_at = NOW - timedelta(days=1)
+    client = SimpleNamespace(inspect_url=AsyncMock(side_effect=SearchConsoleError("HTTP_429")))
+    await service.check_site(client, row, SETTINGS, set())
+    assert row.inspection == {"verdict": "PASS"}
+    assert row.inspected_at < NOW
+    assert row.error_code == "HTTP_429"
+    assert row.failures == 1
+
+
+def test_first_indexed_is_observation_time_not_google_crawl_time():
+    row = status()
+    service.record_inspection(row, {"verdict": "PASS", "lastCrawlTime": "2020-01-01T00:00:00Z"}, NOW)
+    service.record_inspection(row, {"verdict": "PASS"}, NOW + timedelta(days=1))
+    assert row.first_indexed_at == NOW
+
+
+def test_alert_cooldown_and_backoff():
+    row = status()
+    assert alert_reason(row, NOW, 7)
+    row.alerted_at = NOW
+    assert alert_reason(row, NOW, 7) is None
+    for _ in range(30):
+        service.record_error(row, "HTTP_503", NOW)
+    assert row.next_check_at == NOW + timedelta(days=1)
+
+
+async def test_failed_alert_does_not_mark_delivered(monkeypatch):
+    row = status()
+    row.sitemap_submitted_at = NOW
+    client = SimpleNamespace(inspect_url=AsyncMock(return_value={"verdict": "NEUTRAL"}))
+    send = AsyncMock(return_value=False)
+    monkeypatch.setattr(service, "send_alert", send)
+    await service.check_site(client, row, SETTINGS, set())
+    send.assert_awaited_once()
+    assert row.alerted_at is None
+
+
+async def test_empty_alert_url_never_calls_network():
+    assert not await send_alert("", "https://example.com/s/a", "delayed")
+
+
+async def test_disabled_cron_never_opens_database(monkeypatch):
+    monkeypatch.delenv("GSC_ENABLED", raising=False)
+    call = AsyncMock()
+    monkeypatch.setattr(service.DB_SESSION_MNG, "execute_lambda_write", call)
+    await service.run_scheduled_check()
+    call.assert_not_awaited()
+
+
+async def seed_site(session, *, state=3):
+    place = places(place_id=uuid.uuid4(), owner_user_id=uuid.uuid4(), name="숙소", category=1)
+    site = sites(site_id=uuid.uuid4(), place_id=place.place_id, status=state,
+                 domain="stay", published_at=NOW)
+    version = site_versions(site_version_id=uuid.uuid4(), site_id=site.site_id, version=1, build_status=2)
+    site.current_version_id = version.site_version_id
+    session.add_all([place, site, version])
+    await session.flush()
+    return site, place, version
+
+
+async def test_registration_survives_restart_and_resets_on_publish(db_engine, monkeypatch):
+    monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com")
+    async with AsyncSession(db_engine, expire_on_commit=False) as session:
+        site, _, _ = await seed_site(session)
+        await service.register_publications(session, SETTINGS)
+        await session.commit()
+        row = await session.get(site_search_status, site.site_id)
+        row.sitemap_submitted_at = NOW
+        row.first_indexed_at = NOW
+        await session.commit()
+        await service.register_publications(session, SETTINGS)
+        assert row.first_indexed_at == NOW
+        site.published_at = NOW + timedelta(seconds=1)
+        await session.commit()
+        await service.register_publications(session, SETTINGS)
+        await session.commit()
+        await session.refresh(row)
+        assert row.first_indexed_at is None
+        assert row.sitemap_submitted_at is None
+
+
+async def test_unpublished_and_deleted_places_are_not_checked(db_engine, monkeypatch):
+    monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com")
+    async with AsyncSession(db_engine) as session:
+        site, place, _ = await seed_site(session)
+        await service.register_publications(session, SETTINGS)
+        assert len(await store.due_sites(session, SETTINGS.property_url)) == 1
+        site.status = 5
+        await session.flush()
+        assert not await store.due_sites(session, SETTINGS.property_url)
+        site.status = 3
+        place.deleted = True
+        await session.flush()
+        assert not await store.due_sites(session, SETTINGS.property_url)
+
+
+async def test_batch_lock_excludes_second_process(db_engine):
+    async with AsyncSession(db_engine) as first, AsyncSession(db_engine) as second:
+        assert await store.lock_batch(first)
+        assert not await store.lock_batch(second)
+        await first.rollback()
+        assert await store.lock_batch(second)
+
+
+async def test_full_batch_commits_and_does_not_repeat_same_day(db_engine, monkeypatch):
+    monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com")
+    monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value={status().page_url}))
+    client = SimpleNamespace(submit_sitemap=AsyncMock(), inspect_url=AsyncMock(return_value={"verdict": "PASS"}))
+
+    class FakeClient:
+        def __init__(self, *_):
+            pass
+
+        async def __aenter__(self):
+            return client
+
+        async def __aexit__(self, *_):
+            pass
+
+    monkeypatch.setattr(service, "SearchConsoleClient", FakeClient)
+    async with AsyncSession(db_engine) as session:
+        site, _, _ = await seed_site(session)
+        sid = site.site_id
+        await session.commit()
+        await service.run_batch(session, SETTINGS)
+        await session.commit()
+    async with AsyncSession(db_engine) as session:
+        row = await session.get(site_search_status, sid)
+        assert row.sitemap_submitted_at
+        assert row.first_indexed_at
+        await service.run_batch(session, SETTINGS)
+        await session.commit()
+    client.submit_sitemap.assert_awaited_once()
+    client.inspect_url.assert_awaited_once()
+
+
+async def test_cron_failure_is_isolated(monkeypatch):
+    monkeypatch.setattr(service, "load_settings", lambda: SETTINGS)
+    monkeypatch.setattr(service.DB_SESSION_MNG, "execute_lambda_write", AsyncMock(side_effect=RuntimeError("db")))
+    await service.run_scheduled_check()
+
+
+async def test_sitemap_http_parser(monkeypatch):
+    import httpx
+
+    real_client = httpx.AsyncClient
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, text='''
+      <urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
+      <url><loc>https://example.com/s/stay</loc></url></urlset>'''))
+    monkeypatch.setattr(service.httpx, "AsyncClient", lambda **kw: real_client(transport=transport, **kw))
+    assert await service.read_sitemap("https://example.com/sitemap.xml") == {status().page_url}
+
+
+async def test_sitemap_redirect_is_not_followed(monkeypatch):
+    import httpx
+
+    real_client = httpx.AsyncClient
+    transport = httpx.MockTransport(lambda request: httpx.Response(302, headers={"location": "http://internal/"}))
+    monkeypatch.setattr(service.httpx, "AsyncClient", lambda **kw: real_client(transport=transport, **kw))
+    with pytest.raises(SearchConsoleError, match="SITEMAP_UNAVAILABLE"):
+        await service.read_sitemap("https://example.com/sitemap.xml")
+
+
+@pytest.mark.parametrize("enabled,count", [("0", 0), ("1", 1)])
+def test_existing_scheduler_registers_optional_job(monkeypatch, enabled, count):
+    import scheduler
+
+    instance = SimpleNamespace(add_job=lambda *a, **kw: jobs.append((a, kw)),
+                               start=lambda: None, get_jobs=lambda: jobs)
+    jobs = []
+    monkeypatch.setattr(scheduler, "_scheduler", None)
+    monkeypatch.setattr(scheduler, "AsyncIOScheduler", lambda **kw: instance)
+    monkeypatch.setenv("SCHEDULER_ENABLED", "1")
+    monkeypatch.setenv("GSC_ENABLED", enabled)
+    scheduler.start_scheduler()
+    assert len(jobs) == count
+    if jobs:
+        assert jobs[0][1]["minutes"] == 10
+        assert jobs[0][1]["max_instances"] == 1
+
+
+def test_migration_matches_fresh_database_schema():
+    from pathlib import Path
+    root = Path(__file__).resolve().parents[3]
+    migration = (root / "postgres-init/migrations/0014_search_console.sql").read_text()
+    ddl = migration[migration.index("CREATE TABLE"):].strip()
+    assert ddl in (root / "postgres-init/init-data/init.sql").read_text()