[feat] solution/backend: 목업(stay·stay2) 사업장 글이 게재되면 index.html 의 미니블로그만 갈아 끼움

stay 는 굽기 보호 주소라 그 사업장 글을 승인하면 발행이 거절되고 글은 승인 상태로 남았다.
목업 화면의 미니블로그는 파일에 박힌 글이라 승인해도 바뀌지 않았다.

- services/mockup_blog.py: window.__SITE_PAYLOAD__.posts 만 교체 — 기존 글 유지 · 새 글 추가 · 최신순, 직전 파일은 index.html.bak-blog
- build_service: 발행 주소가 stay·stay2 면 굽지 않고 글만 갱신 후 승인분을 게재로(버전·Teams 실패 알림 없음)

테스트 3건 추가(test_mockup_blog). 빌드·블로그·롤백·썸네일 105 passed

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Mina Choi 2026-10-01 11:26:54 +09:00
parent 3ecd7b3e18
commit db7c9254f5
3 changed files with 158 additions and 0 deletions

View File

@ -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)

View File

@ -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"])

View File

@ -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