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.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.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 노출 점검 이력은 읽지 않는다. `place_ai_checks` 는 한 번도 쓰지 않아 # 마이그레이션 0006 이 뗐다(JobType.AI_CHECK 도 아직 미배선이다). # 여기서 그 표를 계속 부르면 SEO 진단이 통째로 죽는다 — 지금 그랬다. # reports 모듈이 붙는 날 표와 함께 되살린다. 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**. DB 도 파일도 건드리지 않는다. ★ 왜 필요한가 (2026-09-09) 미리보기와 발행본이 **렌더러를 두 벌** 쓰고 있었다 — 캔버스는 `frontend/features/builder/canvas/variants/*`, 발행본은 `site/src/sections/*`. 둘이 공유하는 건 타입과 CSS 토큰뿐이라 같은 데이터로도 다른 그림이 나왔다. 실측(2026-09-09): 캔버스는 소개 섹션을 **설명 문구를 자리표시로** 그리는데 발행본은 데이터가 0자면 섹션째 뺀다 — 사장님은 채워진 화면을 보고 발행해 절반이 사라진 페이지를 받는다. `shared/lib/section-data.ts` 가 경고해 둔 "빌더에서는 보이는데 발행하면 없다"가 파서가 아니라 **렌더러**에서 났다. ★ 그래서 미리보기도 이 payload 하나만 먹는다. 발행이 굽는 것과 같은 함수 (`snapshot.build_snapshot` → `site_payload.to_site_payload`)를 그대로 거치므로, 여기서 갈릴 자리가 없다. 버전은 아직 없으니 0 으로 넘긴다 — 화면에 안 쓰인다. """ from services.build_service import ensure_site, _load_links from services.site_payload import to_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_links(place_id) snapshot = await build_snapshot(place) # ★ version 은 None 이다. 발행 전이라 버전 행이 없고, payload 의 site.version 은 # 캐시 무효화 키라 미리보기에서는 뜻이 없다(to_site_payload 가 0 으로 떨어뜨린다). return to_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): """(불가 사유, 대안)을 돌려준다. 쓸 수 있으면 (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) 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) 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), thumbnail_url=getattr(site, "thumbnail_url", 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, "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 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: """발행 상태를 전이한다(중지·재개·내림). ★ 해지는 물리 삭제가 아니다. 색인된 페이지를 갑자기 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 = 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)