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 places, site_publish_logs, site_versions, sites
from common.enums import (
    BuildStatus,
    DBWRType,
    ErrorType,
    JobStatus,
    JobType,
    PublishAction,
    PublishResult,
    SiteStatus,
)
from common.logger import LOG
from common.template_catalog import is_allowed
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_Rollback,
    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

# 주소 저장을 거부할 때 쓰는 결과 코드.
_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,
}

# 테마 JSON 직렬화 크기 상한(64KB).
_THEME_MAX_BYTES = 64 * 1024

# theme 안에 들어와도 저장하지 않는 키.
_THEME_OWNED_ELSEWHERE = ("templateId", "template_id")

# '거부' 표식.
_THEME_REJECTED = object()

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


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

    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.user_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 노출 점검 이력은 읽지 않는다.
        report = evaluate(
            await build_snapshot(place), verified=place.verified_at is not None,
            site=site, version=version, ai_checks=[],
        )
        return Res_SeoAudit(**report)

    async def preview_payload(self, user_info: UserInfo, place_id: str) -> dict | None:
        """에디터 미리보기가 쓰는 **발행본과 똑같은 payload**."""
        from services.build_service import ensure_site, load_channel_links
        from services.site_payload import prepare_site_payload
        from services.snapshot import build_snapshot

        err_type, place = await self._load_place(user_info, place_id)
        if err_type != ErrorType.SUCCESS or place is None:
            return None

        site = await ensure_site(place_id)
        links = await load_channel_links(place_id)
        snapshot = await build_snapshot(place)
        # version 은 None 이다.
        return await prepare_site_payload(place, snapshot, site, None, links)

    # ---- 사이트 주소(네임스페이스) --------------------------------------- 규칙(정규식·예약어)은 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):
        """(불가 사유, 대안)을 돌려준다."""
        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()
        # 주인 확인이 예약어보다 먼저다.
        owner = await self._domain_owner(value)
        if owner is not None and owner == str(uuid.UUID(place_id)):
            return None, None
        if reason is None:
            # 이미 자기 주소면 쓸 수 있다 — 저장해 둔 화면을 다시 열었을 때 '중복'이라고 하면 안 된다.
            if owner is None:
                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

        site = await self._get_site(place_id)
        if site is not None and (site.published_at is not None or site.status == SiteStatus.PUBLISHED.value):
            if (site.domain or "") != (slug or "").strip():
                res.available = False
                res.reason = site_slug.REASON_LOCKED
                return res
            res.available = True
            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)

        # 이미 발행된 사이트의 주소는 바꾸지 않는다.
        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

        # 클라이언트 검증을 믿지 않는다.
        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):
        """발행본과 달라졌다 — 이 사업장만 다시 빌드하면 된다는 표시."""
        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에 저장한다."""
        # 사이트 생성은 빌드와 같은 경로를 쓴다(사업장당 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 value and not is_allowed(int(_place.category), value):
            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 로 되돌려 업종 기본 템플릿으로 떨어뜨린다.
        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에 저장한다."""
        # 사이트 생성은 빌드와 같은 경로를 쓴다(사업장당 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 값을 만든다."""
        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):
            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:
        """로그인한 사장님이 가진 사이트 전부."""
        res = Res_MySites(page=pg.page, size=pg.size)
        uid = uuid.UUID(user_info.user_id)
        err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
            places.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.crud.list_owner_sites(s, uid, 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, primary_photo_url)
            for place, site, built_at, primary_photo_url in rows
        ]
        res.total = total
        return res

    @staticmethod
    def _my_site_row(place, site, built_at, primary_photo_url) -> MySiteData:
        # 재빌드 판별은 단건(get_site)과 같은 규칙이어야 한다 — 다르면 목록과 에디터가 다른 답을 한다.
        changed = place.content_updated_at
        # sites.thumbnail_url 은 발행할 때 Azure 에 대표 사진을 재호스팅해야 채워진다 (services/site_thumbnail.store) — 저장소가 안 꺼져 있으면(로컬 개발 등) 늘 비어 있다.
        ever_published = site is not None and getattr(site, "published_at", None) is not None
        thumbnail_url = getattr(site, "thumbnail_url", None) or (primary_photo_url if ever_published else None)
        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),
            thumbnail_url=thumbnail_url,
            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:
        """정적 페이지가 실제로 구워졌는지 + 두 렌더러가 어긋나지 않았는지."""
        # 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

        # publish=False 는 **이미 나가 있는 사이트를 다시 굽는** 경로다.
        site = await self._get_site(place_id)
        if not req.publish:
            if site is None or site.published_at is None:
                res.result.SetResult(ErrorType.SITE_VERSION_NOT_FOUND)
                return res

        # 주소를 안 골랐어도 막지 않는다.

        job_id, created = await enqueue_job(
            self.queue, JobType.BUILD,
            {
                "place_id": place_id, "owner_user_id": user_info.user_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 start_rollback(self, user_info: UserInfo, place_id: str, req: Req_Rollback) -> Res_StartBuild:
        """dedupe_key 를 빌드와 **같은 것**(`build:{place_id}`)을 쓴다 — 롤백 도중에 새 발행이 끼어들거나 그 반대가 되면 어느 쪽이 이겼는지 알 수 없는 상태가 된다."""
        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

        job_id, created = await enqueue_job(
            self.queue, JobType.ROLLBACK,
            {
                "place_id": place_id, "owner_user_id": user_info.user_id,
                "target_version": req.target_version, "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(
            site_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:
        """발행 상태를 전이한다(중지·재개·내림)."""
        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 = site_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([site_publish_logs.DBType()], [lambda s: self.crud.add_log(s, log)])
        return await self.get_site(user_info, place_id)