merge: feat/mockup-blog-refresh(목업 미니블로그 자동 반영) 을 main 에 병합

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

solution/backend/services/build_service.py

@@ -28,6 +28,7 @@ from services import (
     alert_service,
     azure_static,
     indexnow,
+    mockup_blog,
     publish_gate,
     render_service,
     seo_keywords,
@@ -136,6 +137,19 @@ async def _log(site_id, version_id, action: PublishAction, result: PublishResult
     await DB_SESSION_MNG.execute_lambda_run([site_publish_logs.DBType()], [lambda s: _site_crud.add_log(s, row)])
 
 
+async def _refresh_mockup_blog(place, place_id: str, slug: str) -> dict:
+    snapshot = await build_snapshot(place)
+    try:
+        count = mockup_blog.refresh_posts(slug, snapshot.get("posts") or [])
+    except mockup_blog.MockupBlogError as ex:
+        raise BuildAborted(str(ex)) from ex
+    await DB_SESSION_MNG.execute_lambda_run(
+        [place_posts.DBType()],
+        [lambda s: _post_crud.mark_published(s, uuid.UUID(place_id), None)],
+    )
+    return {"place_id": place_id, "build_status": "BUILT", "mockup": slug, "posts": count}
+
+
 async def run_build(job: dict) -> dict:
     """BUILD 잡 핸들러."""
     payload = job["payload"]
@@ -153,6 +167,10 @@ async def run_build(job: dict) -> dict:
 
     site = await ensure_site(place_id)
 
+    mockup_slug = site_payload.publish_slug(place, site)
+    if mockup_blog.is_mockup(mockup_slug):
+        return await _refresh_mockup_blog(place, place_id, mockup_slug)
+
     # 주변 정보(맛집·관광지·축제·코스)는 빌드 시점에 업장 좌표로 새로 받는다 — 발행본은 정적이라 이때 받은 값이 실린다.
     try:
         synced = await LocalContentService().sync_place(place)

solution/backend/services/mockup_blog.py

@@ -0,0 +1,60 @@
+"""목업(stay · stay2)은 굽지 않고 index.html 의 미니블로그 글만 갈아 끼운다."""
+import json
+import os
+from pathlib import Path
+
+from common.logger import LOG
+from services import render_service
+
+MOCKUP_SLUGS = frozenset({"stay", "stay2"})
+PAYLOAD_MARKER = "window.__SITE_PAYLOAD__"
+BACKUP_NAME = "index.html.bak-blog"
+
+
+class MockupBlogError(Exception):
+    pass
+
+
+def is_mockup(slug: str) -> bool:
+    return slug in MOCKUP_SLUGS
+
+
+def index_path(slug: str) -> Path:
+    return Path(render_service.output_dir()) / "s" / slug / "index.html"
+
+
+def _payload_span(html: str) -> tuple[int, int, dict]:
+    marker = html.find(PAYLOAD_MARKER)
+    if marker < 0:
+        raise MockupBlogError("payload 가 없다")
+    start = html.index("{", marker)
+    payload, end = json.JSONDecoder().raw_decode(html, start)
+    return start, end, payload
+
+
+def merge_posts(existing: list[dict], snapshot_posts: list[dict]) -> list[dict]:
+    merged = {post["postId"]: post for post in existing if post.get("postId")}
+    for row in snapshot_posts:
+        body = (row.get("body") or "").strip()
+        if not body:
+            continue
+        merged[row["post_id"]] = {"postId": row["post_id"], "body": row["body"], "publishedAt": row["published_at"]}
+    return sorted(merged.values(), key=lambda post: post.get("publishedAt") or "", reverse=True)
+
+
+def refresh_posts(slug: str, snapshot_posts: list[dict]) -> int:
+    path = index_path(slug)
+    if not path.is_file():
+        raise MockupBlogError(f"목업 파일이 없다: {path}")
+    html = path.read_text(encoding="utf-8")
+    start, end, payload = _payload_span(html)
+    payload["posts"] = merge_posts(payload.get("posts") or [], snapshot_posts)
+    encoded = json.dumps(payload, ensure_ascii=False).replace("</", "<\\/")
+    updated = html[:start] + encoded + html[end:]
+
+    (path.parent / BACKUP_NAME).write_text(html, encoding="utf-8")
+    tmp = path.with_name(f".{path.name}.tmp")
+    tmp.write_text(updated, encoding="utf-8")
+    os.replace(tmp, path)
+    LOG.i(f"[mockup-blog] {slug} 미니블로그 {len(payload['posts'])}건으로 갱신")
+    return len(payload["posts"])

solution/backend/tests/test_mockup_blog.py

@@ -0,0 +1,80 @@
+import json
+import uuid
+
+from sqlalchemy import text
+
+from common.enums import PostStatus
+from crud.job_crud import JobQueue
+from services import mockup_blog
+from worker.handlers import build_handler
+from worker.runner import Worker
+
+MOCKUP_HTML = (
+    '<html><head></head><body><div id="root">옛 마크업</div>'
+    '<script>window.__SITE_PAYLOAD__ = {"site":{"slug":"stay"},"posts":[{"postId":"old","body":"옛 글","publishedAt":"2026-09-01T00:00:00+00:00"}]};</script>'
+    '</body></html>'
+)
+
+
+def _write_mockup(tmp_path, monkeypatch, slug="stay"):
+    monkeypatch.setenv("SITE_OUTPUT_DIR", str(tmp_path))
+    page = tmp_path / "s" / slug
+    page.mkdir(parents=True)
+    (page / "index.html").write_text(MOCKUP_HTML, encoding="utf-8")
+    return page / "index.html"
+
+
+def _posts(html: str) -> list[dict]:
+    start = html.index("{", html.index(mockup_blog.PAYLOAD_MARKER))
+    return json.JSONDecoder().raw_decode(html, start)[0]["posts"]
+
+
+def test_목업만_대상이다():
+    assert mockup_blog.is_mockup("stay")
+    assert mockup_blog.is_mockup("stay2")
+    assert not mockup_blog.is_mockup("stay3")
+    assert not mockup_blog.is_mockup("euro")
+
+
+def test_새_글을_더하고_옛_글은_남기고_최신순이다(tmp_path, monkeypatch):
+    path = _write_mockup(tmp_path, monkeypatch)
+
+    count = mockup_blog.refresh_posts("stay", [
+        {"post_id": "new", "body": "새 글 </script> 포함", "published_at": "2026-10-01T00:00:00+00:00"},
+        {"post_id": "empty", "body": "  ", "published_at": "2026-10-02T00:00:00+00:00"},
+    ])
+
+    html = path.read_text(encoding="utf-8")
+    assert count == 2
+    assert [p["postId"] for p in _posts(html)] == ["new", "old"]
+    assert "</script> 포함" not in html
+    assert '<div id="root">옛 마크업</div>' in html
+    assert (path.parent / mockup_blog.BACKUP_NAME).read_text(encoding="utf-8") == MOCKUP_HTML
+
+
+async def test_목업_사업장_발행은_굽지_않고_글만_바꾸고_게재로_넘긴다(auth_headers, client, db_engine, tmp_path, monkeypatch):
+    path = _write_mockup(tmp_path, monkeypatch)
+    h = await auth_headers("mockup-owner")
+    pid = (await client.post("/v1/place", headers=h, json={"name": "목업펜션", "category": 1})).json()["place"]["place_id"]
+    post_id = uuid.uuid4()
+    async with db_engine.begin() as c:
+        owner = (await c.execute(text("SELECT owner_user_id FROM places WHERE place_id = :p"), {"p": pid})).scalar()
+        await c.execute(
+            text("INSERT INTO sites (site_id, place_id, status, domain) VALUES (:s, :p, 1, 'stay')"),
+            {"s": uuid.uuid4(), "p": pid},
+        )
+        await c.execute(
+            text("INSERT INTO place_posts (post_id, place_id, body, topic_kind, topic_key, status, approved_at) "
+                 "VALUES (:id, :p, '승인된 새 글입니다', 1, 'fog', :st, now())"),
+            {"id": post_id, "p": pid, "st": PostStatus.APPROVED.value},
+        )
+
+    await JobQueue().enqueue(4, {"place_id": pid, "owner_user_id": str(owner), "publish": True})
+    assert await Worker("w", JobQueue(), build_handler(), job_deadline_sec=60).process_one() is True
+
+    assert [p["postId"] for p in _posts(path.read_text(encoding="utf-8"))][0] == str(post_id)
+    async with db_engine.begin() as c:
+        status = (await c.execute(text("SELECT status FROM place_posts WHERE post_id = :id"), {"id": post_id})).scalar()
+        versions = (await c.execute(text("SELECT count(*) FROM site_versions"))).scalar()
+    assert status == PostStatus.PUBLISHED.value
+    assert versions == 0