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

import json
import uuid

from fastapi import Depends

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import ai_check_results, places, publish_logs, site_versions, sites
from common.enums import (
    BuildStatus,
    DBWRType,
    ErrorType,
    JobStatus,
    JobType,
    PublishAction,
    PublishResult,
    SiteStatus,
)
from common.logger import LOG
from common.models.gmodel import PageParams, UserInfo
from common.utils.gtime import GTime
from crud.job_crud import JobQueue
from crud.place_crud import PlaceCRUD
from crud.site_crud import ISiteCRUD, SiteCRUD
from router.v1.site.protocol import (
    MySiteData,
    PublishLogData,
    RenderStatusData,
    Req_SiteSlug,
    Req_SiteStatus,
    Req_SiteTemplate,
    Req_SiteTheme,
    Req_StartBuild,
    Res_MySites,
    Res_PublishLogs,
    Res_Site,
    Res_SeoAudit,
    Res_SiteSlug,
    Res_SiteVersions,
    Res_SlugCheck,
    Res_StartBuild,
    SiteData,
    SiteVersionData,
)
from services import render_report, site_payload, site_slug
from services.job_service import enqueue_job

# 주소 저장을 거부할 때 쓰는 결과 코드. 사유(reason)는 응답에 따로 실어 프론트가 문구를 고르게 한다.
_ERROR_BY_SLUG_REASON = {
    site_slug.REASON_LENGTH: ErrorType.INVALID_REQUEST_DATA,
    site_slug.REASON_FORMAT: ErrorType.INVALID_REQUEST_DATA,
    site_slug.REASON_RESERVED: ErrorType.INVALID_REQUEST_DATA,
    site_slug.REASON_TAKEN: ErrorType.DB_ALREADY_SAME_KEY,
    site_slug.REASON_LOCKED: ErrorType.SITE_SLUG_LOCKED,
}

# 템플릿 키 길이 상한. sites.template_id 가 varchar(100) 이라 여기서 막지 않으면 DB 가 트랜잭션째로 튕긴다.
# ★ 값 자체(무슨 템플릿인가)는 검증하지 않는다 — 목록은 프론트가 소유한다(protocol.Req_SiteTemplate 주석).
_TEMPLATE_ID_MAX = 100

# 테마 JSON 직렬화 크기 상한(64KB).
# ★ 왜 상한만 두는가: 테마의 내용(섹션 id·배리에이션 키·색 토큰)은 프론트가 소유하므로 서버가
#   해석하지 않는다. 해석하지 않는다는 건 무엇이든 들어올 수 있다는 뜻이라, 크기까지 놓으면
#   jsonb 컬럼 하나가 DB 를 부풀리고 그대로 site_versions.snapshot 을 타고 빌드까지 번진다.
# ★ 왜 64KB 인가: 섹션은 업종당 10개 남짓이고 한 섹션이 200바이트를 넘기 어렵다 — 실제 테마는 3KB 안쪽이다.
#   64KB 면 정상 값의 20배쯤 되는 여유라 프론트가 항목을 늘려도 걸리지 않고,
#   그걸 넘는 건 '디자인 설정'이 아니라 다른 것이 흘러든 것이다.
# ★ 필드 개수가 아니라 직렬화 바이트로 잰다. 개수로 재면 값 하나가 긴 경우를 못 막는다.
_THEME_MAX_BYTES = 64 * 1024

# ★ theme 안에 들어와도 저장하지 않는 키. sites.template_id 컬럼이 소유하는 값이라
#   theme 에 같이 담기면 어느 쪽이 진짜인지 갈리고, payload 가 읽는 쪽과 화면이 보는 쪽이 어긋난다.
#   화이트리스트가 아니라 '이 한 개만 뺀다'는 블랙리스트다 — 나머지는 전부 그대로 보관한다.
_THEME_OWNED_ELSEWHERE = ("templateId", "template_id")

# '거부' 표식. None 은 '빈 값 → NULL 로 되돌린다'는 정상 결과라서, 거부를 None 으로 표현하면
# 잘못된 요청이 조용히 '디자인 초기화'로 처리된다 — 둘을 반드시 구분한다.
_THEME_REJECTED = object()

# 상태 전이만으로 처리하는 액션 — ★ 물리 삭제 경로는 만들지 않는다.
_STATUS_BY_ACTION = {
    PublishAction.SUSPEND: SiteStatus.SUSPENDED,
    PublishAction.RESUME: SiteStatus.PUBLISHED,
    PublishAction.UNPUBLISH: SiteStatus.UNPUBLISHED,
}


class SiteService:
    """사이트 조회 · 빌드 트리거 · 발행 상태 전이.

    빌드와 발행 판정은 BUILD 잡(services/build_service)이 한다 — 여기는 조회와 상태 전이만.
    """

    def __init__(
        self,
        crud: ISiteCRUD = Depends(SiteCRUD),
        place_crud: PlaceCRUD = Depends(PlaceCRUD),
        queue: JobQueue = Depends(JobQueue),
    ):
        self.crud = crud
        self.place_crud = place_crud
        self.queue = queue

    async def _load_place(self, user_info: UserInfo, place_id: str):
        err_type, place = await DB_SESSION_MNG.execute_lambda(
            places.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.place_crud.get_place(s, uuid.UUID(user_info.company_id), uuid.UUID(place_id)),
        )
        if err_type != ErrorType.SUCCESS:
            return ErrorType.PLACE_NOT_FOUND, None
        return ErrorType.SUCCESS, place

    async def _get_site(self, place_id: str):
        """사업장의 사이트 행. 없으면 None."""
        pid = uuid.UUID(place_id)
        err_type, site = await DB_SESSION_MNG.execute_lambda(
            sites.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.get_site_by_place(s, pid)
        )
        return site if err_type == ErrorType.SUCCESS else None

    async def get_seo_audit(self, user_info: UserInfo, place_id: str) -> Res_SeoAudit:
        """현재 저장값 기준 SEO/AEO 준비도. 조회할 때마다 계산해 오래된 점수를 보여주지 않는다."""
        from sqlalchemy import select
        from services.seo_audit import evaluate
        from services.snapshot import build_snapshot

        res = Res_SeoAudit()
        err_type, place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        site = await self._get_site(place_id)
        version = None
        if site is not None and site.current_version_id:
            v_err, version = await DB_SESSION_MNG.execute_lambda(
                site_versions.DBType(), DBWRType.DB_READ.value,
                lambda s: self.crud.get_version(s, site.site_id, site.current_version_id),
            )
            if v_err != ErrorType.SUCCESS:
                version = None

        _ai_err, ai_rows = await DB_SESSION_MNG.execute_lambda(
            ai_check_results.DBType(), DBWRType.DB_READ.value,
            lambda s: DB_SESSION_MNG.execute(
                s, select(ai_check_results).where(
                    ai_check_results.place_id == uuid.UUID(place_id),
                    ai_check_results.deleted == False,  # noqa: E712
                ).order_by(ai_check_results.checked_at.desc()).limit(20),
                raise_error=False,
            ),
        )
        report = evaluate(
            await build_snapshot(place), verified=place.verified_at is not None,
            site=site, version=version, ai_checks=list(ai_rows or []),
        )
        return Res_SeoAudit(**report)

    # ---- 사이트 주소(네임스페이스) ---------------------------------------
    # 규칙(정규식·예약어)은 services/site_slug 한 곳에만 있다. 확인과 저장이 그것을 같이 쓴다.

    async def _domain_owner(self, slug: str):
        """이 주소를 이미 쓰는 사업장 id. 아무도 안 쓰면 None."""
        err_type, site = await DB_SESSION_MNG.execute_lambda(
            sites.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.get_site_by_domain(s, slug)
        )
        if err_type != ErrorType.SUCCESS or site is None:
            return None
        return str(site.place_id)

    async def _suggest(self, slug: str):
        """`-2`, `-3` … 중 실제로 비어 있는 것 하나. 없으면 None."""
        candidates = site_slug.suggestions(slug)
        if not candidates:
            return None
        err_type, taken = await DB_SESSION_MNG.execute_lambda(
            sites.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.taken_domains(s, candidates)
        )
        if err_type != ErrorType.SUCCESS:
            return None
        return next((c for c in candidates if c not in taken), None)

    async def _judge_slug(self, place_id: str, slug: str):
        """(불가 사유, 대안)을 돌려준다. 쓸 수 있으면 (None, None).

        ★ 확인(check)과 저장(POST)이 이 하나를 같이 쓴다 — 판정이 갈리면
          "된다고 해놓고 저장에서 튕기는" 화면이 나온다."""
        reason = site_slug.validate_slug(slug)
        if reason in (site_slug.REASON_LENGTH, site_slug.REASON_FORMAT):
            # 형식이 깨진 값에 -2 를 붙여 봐야 여전히 못 쓴다. 사유만 돌려준다.
            return reason, None

        value = (slug or "").strip()
        if reason is None:
            owner = await self._domain_owner(value)
            # ★ 이미 자기 주소면 쓸 수 있다 — 저장해 둔 화면을 다시 열었을 때 '중복'이라고 하면 안 된다.
            if owner is None or owner == str(uuid.UUID(place_id)):
                return None, None
            reason = site_slug.REASON_TAKEN
        return reason, await self._suggest(value)

    async def check_slug(self, user_info: UserInfo, place_id: str, slug: str) -> Res_SlugCheck:
        """주소를 쓸 수 있는지 미리 본다.

        ★ 서버가 상호명으로 자동 확정하지 않는다 — 사람이 고르고, 겹치면 고르기 전에 알려 준다."""
        res = Res_SlugCheck()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        reason, suggestion = await self._judge_slug(place_id, slug)
        res.available = reason is None
        res.reason = reason
        res.suggestion = suggestion
        return res

    async def set_slug(self, user_info: UserInfo, place_id: str, req: Req_SiteSlug) -> Res_SiteSlug:
        """주소를 확정해 sites.domain 에 저장한다. 사이트 행이 없으면 만든다."""
        # 사이트 생성은 빌드와 같은 경로를 쓴다(사업장당 1개 보장) — 지역 import 로 빌드 스택을 웹에 얹지 않는다.
        from services.build_service import BuildAborted, ensure_site

        res = Res_SiteSlug()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        value = (req.slug or "").strip()
        site = await self._get_site(place_id)

        # ★ 이미 발행된 사이트의 주소는 바꾸지 않는다.
        #   색인된 주소가 바뀌면 AI 검색이 잡아 둔 페이지가 404 가 되고, 그 자리를 다시 OTA 가 채운다 —
        #   '해지는 상태 전이지 삭제가 아니다' 와 정확히 같은 이유다.
        #   한 번이라도 발행된 적 있으면(published_at) 중지·내림 상태여도 같다. 색인은 그 주소로 남아 있다.
        if site is not None and (site.published_at is not None or site.status == SiteStatus.PUBLISHED.value):
            if (site.domain or "") != value:
                res.result.SetResult(ErrorType.SITE_SLUG_LOCKED)
                res.reason = site_slug.REASON_LOCKED
                return res
            # 같은 값을 다시 보낸 건 변경이 아니다 — 재전송을 에러로 만들지 않는다.
            res.site = SiteData.model_validate(site)
            return res

        # ★ 클라이언트 검증을 믿지 않는다. check 와 같은 규칙으로 서버가 다시 본다.
        reason, suggestion = await self._judge_slug(place_id, value)
        if reason is not None:
            res.result.SetResult(_ERROR_BY_SLUG_REASON.get(reason, ErrorType.INVALID_REQUEST_DATA))
            res.reason = reason
            res.suggestion = suggestion
            return res

        if site is None:
            # 빌드 전에도 주소를 먼저 잡아둘 수 있어야 한다 — 주소를 고르는 건 발행보다 앞선 결정이다.
            try:
                site = await ensure_site(place_id)
            except BuildAborted as ex:
                LOG.e_no_callstack(ex)
                res.result.SetResult(ErrorType.DB_RUN_FAILED)
                return res

        u_err, _rowcount = await DB_SESSION_MNG.execute_lambda_claim(
            sites.DBType(), lambda s: self.crud.update_site(s, site.site_id, {"domain": value})
        )
        if u_err != ErrorType.SUCCESS:
            # uq_sites_domain 이 막았다면 확인과 저장 사이에 남이 먼저 가져간 것이다 — 대안까지 같이 준다.
            res.result.SetResult(u_err)
            if u_err == ErrorType.DB_ALREADY_SAME_KEY:
                res.reason = site_slug.REASON_TAKEN
                res.suggestion = await self._suggest(value)
            return res

        saved = await self._get_site(place_id)
        res.site = SiteData.model_validate(saved if saved is not None else site)
        return res

    # ---- 템플릿(디자인) ---------------------------------------------------

    async def _mark_content_updated(self, place_id: str, ts):
        """★ 발행본과 달라졌다 — 이 사업장만 다시 빌드하면 된다는 표시.
        fact/faq 가 노출값이 바뀔 때 찍는 것과 같은 자리를 같은 이유로 쓴다(services/faq_service).
        부가 효과라 실패해도 본 흐름을 막지 않는다(다음 변경 때 다시 찍힌다)."""
        err = await DB_SESSION_MNG.execute_lambda_run([places.DBType()], [lambda s: self._touch(s, place_id, ts)])
        if err != ErrorType.SUCCESS:
            LOG.e_no_callstack(f"[site] content_updated_at 갱신 실패 place={place_id}")

    async def _touch(self, s, place_id: str, ts):
        from sqlalchemy import update

        query = update(places).where(places.place_id == uuid.UUID(place_id)).values(content_updated_at=ts, updated_at=ts)
        return await DB_SESSION_MNG.add(s, query)

    async def set_template(self, user_info: UserInfo, place_id: str, req: Req_SiteTemplate) -> Res_Site:
        """사장님이 고른 템플릿을 sites.template_id 에 저장한다. 사이트 행이 없으면 만든다.

        ★ 왜 서버에 저장하나: 위저드가 고른 값이 브라우저에만 남으면 발행 잡이 읽을 곳이 없어
          업종 기본 템플릿으로 굽는다 — 고른 디자인과 실제 발행본이 갈린다.
        ★ 슬러그와 달리 발행 뒤에도 바꿀 수 있다. 주소는 AI 검색이 색인한 영구 식별자라 잠그지만,
          디자인은 바뀌어도 URL 이 그대로다(색인이 깨지지 않는다)."""
        # 사이트 생성은 빌드와 같은 경로를 쓴다(사업장당 1개 보장) — 지역 import 로 빌드 스택을 웹에 얹지 않는다.
        from services.build_service import BuildAborted, ensure_site

        res = Res_Site()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        value = (req.template_id or "").strip()
        # 길이만 본다. 모르는 키가 들어와도 발행 잡이 업종 기본으로 떨어뜨리므로 화면이 깨지지 않는다.
        if len(value) > _TEMPLATE_ID_MAX:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res

        site = await self._get_site(place_id)
        if site is None:
            # 빌드 전에도 템플릿을 먼저 잡아둘 수 있어야 한다 — 고르는 건 발행보다 앞선 결정이다.
            try:
                site = await ensure_site(place_id)
            except BuildAborted as ex:
                LOG.e_no_callstack(ex)
                res.result.SetResult(ErrorType.DB_RUN_FAILED)
                return res

        before = site.template_id or None
        # ★ 빈 문자열은 '고르지 않음'이다 — NULL 로 되돌려 업종 기본 템플릿으로 떨어뜨린다.
        #   빈 문자열을 그대로 두면 payload 가 '있는데 이름이 없는 템플릿'을 만나 기본값으로도 못 간다.
        after = value or None
        u_err, _rowcount = await DB_SESSION_MNG.execute_lambda_claim(
            sites.DBType(), lambda s: self.crud.update_site(s, site.site_id, {"template_id": after})
        )
        if u_err != ErrorType.SUCCESS:
            res.result.SetResult(u_err)
            return res

        # 한 번이라도 구워진 사이트에서 디자인이 바뀌었다면 지금 나가 있는 페이지와 달라졌다 —
        # 재빌드 대상으로 표시한다. 값이 그대로면 찍지 않는다(재전송이 재빌드를 만들지 않게).
        if before != after and (site.published_at is not None or site.current_version_id is not None):
            await self._mark_content_updated(place_id, GTime.UTC())

        return await self.get_site(user_info, place_id)

    # ---- 테마(색·서체·섹션) -----------------------------------------------

    async def set_theme(self, user_info: UserInfo, place_id: str, req: Req_SiteTheme) -> Res_Site:
        """에디터가 정한 디자인을 sites.theme 에 저장한다. 사이트 행이 없으면 만든다.

        ★ 왜 서버에 저장하나: template_id 와 정확히 같은 이유다. 섹션 on/off·순서·배리에이션·색·서체가
          브라우저 메모리에만 있으면 새로고침에 사라지고, 발행 잡은 읽을 곳이 없어 업종 기본 모양을 굽는다 —
          사장님이 섹션을 끄고 순서를 바꿔도 발행본은 언제나 업종 기본으로 나갔다.

        ★ 값은 해석하지 않는다. 섹션 목록·배리에이션 키·색 토큰 이름은 프론트가 소유하므로
          검증하면 프론트에 항목이 하나 늘 때마다 백엔드를 같이 고쳐야 한다(protocol.Req_SiteTheme).
          막는 것은 크기 하나뿐이다.

        ★ 템플릿과 마찬가지로 발행 뒤에도 바꿀 수 있다 — 디자인이 바뀌어도 URL 은 그대로라 색인이 안 깨진다."""
        # 사이트 생성은 빌드와 같은 경로를 쓴다(사업장당 1개 보장) — 지역 import 로 빌드 스택을 웹에 얹지 않는다.
        from services.build_service import BuildAborted, ensure_site

        res = Res_Site()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        after = self._clean_theme(req.theme)
        if after is _THEME_REJECTED:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res

        site = await self._get_site(place_id)
        if site is None:
            # 빌드 전에도 디자인을 먼저 잡아둘 수 있어야 한다 — 고르는 건 발행보다 앞선 결정이다.
            try:
                site = await ensure_site(place_id)
            except BuildAborted as ex:
                LOG.e_no_callstack(ex)
                res.result.SetResult(ErrorType.DB_RUN_FAILED)
                return res

        before = site.theme or None
        u_err, _rowcount = await DB_SESSION_MNG.execute_lambda_claim(
            sites.DBType(), lambda s: self.crud.update_site(s, site.site_id, {"theme": after})
        )
        if u_err != ErrorType.SUCCESS:
            res.result.SetResult(u_err)
            return res

        # 한 번이라도 구워진 사이트에서 디자인이 바뀌었다면 지금 나가 있는 페이지와 달라졌다 —
        # 재빌드 대상으로 표시한다. 값이 그대로면 찍지 않는다(재전송이 재빌드를 만들지 않게).
        if before != after and (site.published_at is not None or site.current_version_id is not None):
            await self._mark_content_updated(place_id, GTime.UTC())

        return await self.get_site(user_info, place_id)

    @staticmethod
    def _clean_theme(theme):
        """저장할 theme 값을 만든다. 거부해야 하면 _THEME_REJECTED 를 돌려준다.

        ★ 하는 일이 셋뿐인 게 의도다 — 서버는 이 값을 해석하지 않는다.
          1) 빈 값이면 NULL 로 되돌린다. 빈 dict 를 그대로 두면 payload 가 '있는데 아무것도 없는 테마'를
             만나 업종 기본으로도 못 간다(template_id 의 빈 문자열과 같은 함정이다).
          2) templateId 는 뺀다 — sites.template_id 컬럼이 소유하는 값이다(중복 보관 금지).
          3) 직렬화 크기 상한을 넘으면 거부한다.
          그 외의 키(colorPaletteId 처럼 에디터 복원 전용인 것 포함)는 전부 그대로 보관한다."""
        if not isinstance(theme, dict) or not theme:
            return None

        cleaned = {k: v for k, v in theme.items() if k not in _THEME_OWNED_ELSEWHERE}
        if not cleaned:
            return None

        try:
            size = len(json.dumps(cleaned, ensure_ascii=False).encode("utf-8"))
        except (TypeError, ValueError):
            # jsonb 컬럼에 넣을 수 없는 값이 섞였다는 뜻이다. DB 가 트랜잭션째로 튕기기 전에 여기서 막는다.
            return _THEME_REJECTED
        if size > _THEME_MAX_BYTES:
            return _THEME_REJECTED
        return cleaned

    async def list_my_sites(self, user_info: UserInfo, pg: PageParams) -> Res_MySites:
        """로그인한 계정(회사)이 가진 사이트 전부.

        사업장 목록(/v1/place/list)과 따로 두는 이유: 화면이 알아야 하는 건 '사업장이 있다'가 아니라
        '발행돼 있나 · 주소가 뭔가 · 다시 구워야 하나'다."""
        res = Res_MySites(page=pg.page, size=pg.size)
        cid = uuid.UUID(user_info.company_id)
        err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
            places.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.list_company_sites(s, cid, pg.skip, pg.size),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        res.sites = [self._my_site_row(place, site, built_at) for place, site, built_at in rows]
        res.total = total
        return res

    @staticmethod
    def _my_site_row(place, site, built_at) -> MySiteData:
        # ★ 재빌드 판별은 단건(get_site)과 같은 규칙이어야 한다 — 다르면 목록과 에디터가 다른 답을 한다.
        changed = place.content_updated_at
        return MySiteData(
            place_id=place.place_id,
            name=place.name,
            category=place.category,
            place_status=place.status,
            road_address=place.road_address or place.address,
            created_at=place.created_at,
            site_id=getattr(site, "site_id", None),
            status=getattr(site, "status", None),
            domain=getattr(site, "domain", None),
            template_id=getattr(site, "template_id", None),
            published_at=getattr(site, "published_at", None),
            needs_rebuild=bool(site is not None and changed and (built_at is None or changed > built_at)),
        )

    async def get_site(self, user_info: UserInfo, place_id: str) -> Res_Site:
        res = Res_Site()
        err_type, place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        pid = uuid.UUID(place_id)
        s_err, site = await DB_SESSION_MNG.execute_lambda(
            sites.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.get_site_by_place(s, pid)
        )
        if s_err != ErrorType.SUCCESS or site is None:
            res.result.SetResult(ErrorType.SITE_NOT_FOUND)
            return res
        res.site = SiteData.model_validate(site)

        if site.current_version_id:
            v_err, version = await DB_SESSION_MNG.execute_lambda(
                site_versions.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.crud.get_version(s, site.site_id, site.current_version_id),
            )
            if v_err == ErrorType.SUCCESS and version is not None:
                res.current_version = SiteVersionData.model_validate(version)
                # ★ 개별 재빌드 판별: 노출값이 바뀐 시각이 마지막 빌드보다 나중이면 다시 빌드해야 한다.
                built = version.built_at
                changed = place.content_updated_at
                res.needs_rebuild = bool(changed and (built is None or changed > built))
                res.render = self._render_status(place, site, version)
        else:
            res.needs_rebuild = bool(place.content_updated_at)
            res.render = self._render_status(place, site, None)
        return res

    @staticmethod
    def _render_status(place, site, version) -> RenderStatusData:
        """정적 페이지가 실제로 구워졌는지 + 두 렌더러가 어긋나지 않았는지.

        ★ 빌드(DB)와 렌더(정적 파일)는 다른 단계다. 빌드가 BUILT 라고 페이지가 있는 게 아니다.
          프리렌더가 남긴 보고서를 읽어 그 간극을 드러낸다.
        ★ 버전이 맞는 보고서만 OK 로 본다 — 낡은 렌더를 현재 것처럼 답하면 안 된다."""
        # ★ payload 를 쓸 때와 같은 규칙으로 슬러그를 구한다(site_payload 가 소유).
        slug = site_payload.publish_slug(place, site)
        if not slug:
            return RenderStatusData()

        status = render_report.render_status(slug, getattr(version, "version", None))
        return RenderStatusData(
            state=status["state"],
            rendered_at=status["rendered_at"],
            rendered_version=status["rendered_version"],
            error=status["error"],
        )

    async def start_build(self, user_info: UserInfo, place_id: str, req: Req_StartBuild) -> Res_StartBuild:
        """빌드를 큐에 넣는다. 빌드는 몇 십 초 걸린다 — 동기로 처리하지 않는다."""
        res = Res_StartBuild()
        err_type, place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        # 검증 안 된 사업장은 빌드하지 않는다 — 주소·좌표가 없으면 JSON-LD 가 성립하지 않는다.
        if place.verified_at is None:
            res.result.SetResult(ErrorType.PLACE_NOT_VERIFIED)
            return res

        job_id, created = await enqueue_job(
            self.queue, JobType.BUILD,
            {
                "place_id": place_id, "company_id": user_info.company_id,
                "publish": req.publish, "requested_by": user_info.user_id,
            },
            dedupe_key=f"build:{place_id}",
        )
        if job_id is None:
            res.result.SetResult(ErrorType.COLLECT_ALREADY_RUNNING)
            return res
        res.job_id = uuid.UUID(job_id)
        res.status = JobStatus.PENDING
        res.created = created
        return res

    async def list_versions(self, user_info: UserInfo, place_id: str) -> Res_SiteVersions:
        from sqlalchemy import select

        res = Res_SiteVersions()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        pid = uuid.UUID(place_id)
        s_err, site = await DB_SESSION_MNG.execute_lambda(
            sites.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.get_site_by_place(s, pid)
        )
        if s_err != ErrorType.SUCCESS or site is None:
            res.result.SetResult(ErrorType.SITE_NOT_FOUND)
            return res

        query = (
            select(site_versions)
            .where(site_versions.site_id == site.site_id, site_versions.deleted == False)  # noqa: E712
            .order_by(site_versions.version.desc())
            .limit(50)
        )
        v_err, rows = await DB_SESSION_MNG.execute_lambda(
            site_versions.DBType(),
            DBWRType.DB_READ.value,
            lambda s: DB_SESSION_MNG.execute(s, query),
        )
        if v_err == ErrorType.SUCCESS:
            res.versions = [SiteVersionData.model_validate(r) for r in rows]
        return res

    async def list_logs(self, user_info: UserInfo, place_id: str) -> Res_PublishLogs:
        res = Res_PublishLogs()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        pid = uuid.UUID(place_id)
        s_err, site = await DB_SESSION_MNG.execute_lambda(
            sites.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.get_site_by_place(s, pid)
        )
        if s_err != ErrorType.SUCCESS or site is None:
            res.result.SetResult(ErrorType.SITE_NOT_FOUND)
            return res
        l_err, rows = await DB_SESSION_MNG.execute_lambda(
            publish_logs.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.list_logs(s, site.site_id, 50)
        )
        if l_err == ErrorType.SUCCESS:
            res.logs = [PublishLogData.model_validate(r) for r in rows]
        return res

    async def change_status(self, user_info: UserInfo, place_id: str, req: Req_SiteStatus) -> Res_Site:
        """발행 상태를 전이한다(중지·재개·내림).

        ★ 해지는 물리 삭제가 아니다. 색인된 페이지를 갑자기 404 로 만들면
          AI 검색이 그 자리를 다시 OTA 로 채운다 — 이 서비스가 하려던 것의 정반대가 된다."""
        res = Res_Site()
        err_type, _place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        target = _STATUS_BY_ACTION.get(req.action)
        if target is None:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res

        pid = uuid.UUID(place_id)
        s_err, site = await DB_SESSION_MNG.execute_lambda(
            sites.DBType(), DBWRType.DB_READ.value, lambda s: self.crud.get_site_by_place(s, pid)
        )
        if s_err != ErrorType.SUCCESS or site is None:
            res.result.SetResult(ErrorType.SITE_NOT_FOUND)
            return res
        # 한 번도 발행한 적 없는 사이트는 재개할 것이 없다.
        if req.action == PublishAction.RESUME and site.current_version_id is None:
            res.result.SetResult(ErrorType.SITE_VERSION_NOT_FOUND)
            return res

        now = GTime.UTC()
        await DB_SESSION_MNG.execute_lambda_claim(
            sites.DBType(), lambda s: self.crud.update_site(s, site.site_id, {"status": target.value})
        )
        log = publish_logs(
            site_id=site.site_id,
            site_version_id=site.current_version_id,
            action=req.action.value,
            result=PublishResult.SUCCESS.value,
            actor_user_id=uuid.UUID(user_info.user_id),
        )
        await DB_SESSION_MNG.execute_lambda_run([publish_logs.DBType()], [lambda s: self.crud.add_log(s, log)])
        return await self.get_site(user_info, place_id)