Compare commits

..

No commits in common. "ed0137c9da05069551a119661efdfb5db53a0530" and "acd5383b0f8e9b190177bdd3840f16c915ef77a3" have entirely different histories.

4 changed files with 12 additions and 143 deletions

View File

@ -106,12 +106,7 @@
- 본문: 문구 전문 + 승인 링크 + 수정 링크(`blog_jobs._mail_body`)
- **승인 링크**: `GET /v1/site/post/approve?t=<토큰>` — 로그인 없음, 토큰이 신원. **누르는
즉시 승인된다**(확인 화면 없음, 2026-09-17 사장님 지시). 토큰은 32바이트 랜덤 → DB 엔
sha256 만, **단회용 · 그날 자정(KST) 만료**(`blog_service.issue_token`). 승인 확인
화면은 그 업장의 발행된 사이트(미니 블로그 자리, `#blog`)로 5초 뒤 자동 이동한다
(2026-09-22 사장님 지시 — `router/v1/site/post.py _page`, `PostService._blog_url`).
재발행(BUILD 잡)은 몇 분 걸리므로 5초 뒤에 이 글이 이미 반영돼 있다는 보장은 없다 —
그래도 "어디로 가면 보이는지"는 바로 알려준다. 발행된 사이트가 없으면 자동 이동 없이
확인 문구만 보여준다
sha256 만, **단회용 · 그날 자정(KST) 만료**(`blog_service.issue_token`)
- **수정 링크**: `{origin}/blog?placeId=&postId=&auto=<그날짜리 JWT>` — 로그인 흐름이다.
`CreateDayPassToken`(`router/v1/validator/dependencies.py`)이 자정까지만 사는 접근
토큰을 찍고, 빌더 앱이 그 토큰으로 로그인해 그 글의 편집 모달을 바로 연다

View File

@ -14,7 +14,6 @@
"바로 발행" — 수정 없이 그대로 승인). PUT(수정)은 저장만 하고 자동으로 승인하지 않는다 —
승인은 이 두 경로 중 하나를 명시적으로 눌러야 한다.
"""
import html
from datetime import date
from uuid import UUID
@ -31,27 +30,18 @@ from services.post_service import PostService
router = APIRouter(prefix="/v1/site/post", tags=["Site"])
owner_router = APIRouter(prefix="/v1/place/{place_id}/post", tags=["Site"])
_REDIRECT_DELAY_SEC = 5
_PAGE = """<!doctype html><html lang="ko"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"><meta name="robots" content="noindex">
{redirect}<title>{title}</title><style>
<title>{title}</title><style>
body{{margin:0;background:#f6f5f1;color:#1b1a15;font:17px/1.7 -apple-system,'Apple SD Gothic Neo','Noto Sans KR',sans-serif}}
.wrap{{max-width:34rem;margin:0 auto;padding:40px 20px}}
h1{{font-size:21px;margin:0 0 6px}} p{{margin:0 0 14px}}
.meta{{color:#6b7269;font-size:14px}} a{{color:#1b1a15}}
.meta{{color:#6b7269;font-size:14px}}
</style></head><body><div class="wrap">{content}</div></body></html>"""
def _page(title: str, content: str, *, redirect_url: str | None = None) -> HTMLResponse:
# ★ redirect_url 은 항상 서버가 site_payload.publish_url() 로 만든 값(고정 오리진 +
# slugify 통과 슬러그)이라 사용자 입력이 아니지만, HTML 속성에 그대로 꽂는 자리라
# escape 를 걸어 둔다 — 이 함수가 나중에 다른 값을 받게 되더라도 안전하게.
redirect = (
f'<meta http-equiv="refresh" content="{_REDIRECT_DELAY_SEC};url={html.escape(redirect_url, quote=True)}">'
if redirect_url else ""
)
return HTMLResponse(_PAGE.format(title=title, content=content, redirect=redirect))
def _page(title: str, content: str) -> HTMLResponse:
return HTMLResponse(_PAGE.format(title=title, content=content))
def _expired_page() -> HTMLResponse:
@ -66,20 +56,7 @@ async def approve_page(t: str = Query(min_length=8, max_length=200), service: Po
result = await service.decide(t, skip=False)
if not result["success"]:
return _expired_page()
redirect_url = result.get("redirect_url")
# ★ 재발행은 몇 분 걸린다(BUILD 잡) — 5초 뒤에 이 글이 이미 반영돼 있다는 보장은 없다.
# 그래도 "어디로 가면 보이는지" 를 알려주는 게 사장님 입장에서 "눌렀는데 어디 갔지" 보다
# 낫다(2026-09-22, 사장님 지시). 링크 자체는 안내 문구에도 남겨 자동 이동을 못 믿어도 되게 한다.
extra = (
f"<p class='meta'>{_REDIRECT_DELAY_SEC}초 뒤 자동으로 이동합니다. "
f"바로 가려면 <a href='{html.escape(redirect_url, quote=True)}'>여기</a>를 눌러주세요.</p>"
if redirect_url else ""
)
return _page(
"올렸습니다",
f"<h1>{result['message']}</h1><p class='meta'>사이트에 반영되기까지 몇 분 걸립니다.</p>{extra}",
redirect_url=redirect_url,
)
return _page("올렸습니다", f"<h1>{result['message']}</h1><p class='meta'>사이트에 반영되기까지 몇 분 걸립니다.</p>")
@owner_router.get(path="", response_model=Res_MyPosts, summary="이번 달(또는 고른 달) 생성된 글 목록")

View File

@ -15,15 +15,15 @@ from crud.job_crud import JobQueue
from crud.place_crud import PlaceCRUD
from crud.post_crud import PostCRUD
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import jobs as jobs_table, place_posts, places, sites
from common.enums import DBWRType, ErrorType, JobStatus, JobType, PostStatus, SiteStatus
from common.database.model.models import jobs as jobs_table, place_posts, places
from common.enums import DBWRType, ErrorType, JobStatus, JobType, PostStatus
from common.logger import LOG
from common.models.gmodel import Res_WebPacketProtocol, UserInfo
from common.utils.gtime import GTime
from router.v1.site.protocol import (
GenerationBatch, PostData, Res_GenerateNow, Res_GenerateOne, Res_GenerationHistory, Res_MyPosts,
)
from services import blog_jobs, blog_service, site_payload, social_service
from services import blog_jobs, blog_service, social_service
from services.job_service import enqueue_job
EXPIRED = "처리할 수 없는 링크입니다."
@ -76,36 +76,7 @@ class PostService:
return {"success": True, "message": SKIPPED}
await self._approve_and_publish(post_id, place_id)
return {
"success": True,
"message": APPROVED,
"redirect_url": await self._blog_url(place_id),
}
async def _blog_url(self, place_id) -> str | None:
"""이 업장의 발행된 사이트에서 미니 블로그가 보이는 자리. 승인 확인 화면이 몇 초
뒤 여기로 자동 연결한다(2026-09-22, 사장님 지시) — 사장님이 승인만 하고 실제로
어디에 올라갔는지 못 찾는 걸 줄인다. 사이트가 없거나 아직 미발행이면 None —
호출부가 자동 연결 없이 확인 문구만 보여준다."""
def query(session):
return session.execute(
select(places, sites)
.join(sites, sites.place_id == places.place_id)
.where(
places.place_id == place_id,
places.deleted == False, # noqa: E712
sites.deleted == False, # noqa: E712
sites.status == SiteStatus.PUBLISHED.value,
)
.limit(1)
)
result = await DB_SESSION_MNG.execute_lambda(places.DBType(), DBWRType.DB_READ.value, query)
row = result.first() if result is not None else None
if not row:
return None
place, site = row
return f"{site_payload.publish_url(place, site)}#blog"
return {"success": True, "message": APPROVED}
async def _load_place(self, user_info: UserInfo, place_id: str):
err_type, place = await DB_SESSION_MNG.execute_lambda(

View File

@ -14,8 +14,8 @@ from datetime import timezone
from sqlalchemy import text
from common.enums import JobType, PlaceCategory, PostStatus, PostTopicKind, SiteStatus
from services import blog_service, site_payload
from common.enums import JobType, PlaceCategory, PostStatus, PostTopicKind
from services import blog_service
BODY = (
"비가 한 차례 지나간 뒤 마당 돌이 검게 젖었습니다. 이런 날에는 대청마루에 앉아 빗소리만 들어도 "
@ -51,53 +51,6 @@ def test_filter_drops_same_topic_twice_in_one_batch():
assert "주제 중복" in dropped[0][1]
def test_build_prompt_falls_back_to_stay_wording_for_unmapped_category():
"""업종별로 분기할 수 있는 구조만 갖춘다 — 숙소 외 업종은 아직 실구현 전이라 숙소 문구로 대체한다."""
stay = blog_service.build_prompt(
place_name="테스트펜션", region="군산", topic_kind=PostTopicKind.WEATHER.value,
material="맑음인 날", used_topics=[], place_category=PlaceCategory.LODGING.value,
)
restaurant = blog_service.build_prompt(
place_name="테스트식당", region="군산", topic_kind=PostTopicKind.WEATHER.value,
material="맑음인 날", used_topics=[], place_category=PlaceCategory.RESTAURANT.value,
)
assert "숙소" in stay
assert "숙소" in restaurant # 음식점 전용 문구가 아직 없어 숙소 문구로 폴백한다
async def test_generate_one_forwards_place_category_into_the_prompt(monkeypatch):
"""generate_one 은 place_category 를 build_prompt 로 그대로 넘긴다 — 링크는 여기서
붙이지 않는다(호출부의 길이 게이트가 이 반환값에 그대로 걸리기 때문에, 붙이면
게이트가 링크까지 세어 정상 문구도 버려진다). 공급자는 provider.active() 를 통해
고르므로(Gemini 로 고정하지 않는다) 가짜 공급자로 검증한다."""
from services.llm import provider
from services.llm.types import LlmResult, Usage
captured = {}
class FakeLlm:
DEFAULT_MODEL = "fake-model"
@staticmethod
def is_configured():
return True
@staticmethod
async def generate(client, model, *, prompt, temperature=0.2, **kwargs):
captured["prompt"] = prompt
return LlmResult(json=None, text=BODY, usage=Usage())
monkeypatch.setattr(provider, "active", lambda: FakeLlm)
result = await blog_service.generate_one(
place_name="테스트식당", region="군산", topic_kind=PostTopicKind.WEATHER.value,
material="맑음인 날", used_topics=[], place_category=PlaceCategory.RESTAURANT.value,
)
assert result == (BODY, "fake-model")
assert "숙소" in captured["prompt"] # 음식점 전용 문구가 아직 없어 숙소 문구로 폴백한다
async def _seed_post(db_engine, owner_id, *, status=PostStatus.SENT, expires_days=14):
place_id, post_id = uuid.uuid4(), uuid.uuid4()
token, token_hash, expires = blog_service.issue_token()
@ -141,33 +94,6 @@ async def test_get_approves_immediately(client, db_engine, owner_id):
assert res.status_code == 200
assert "올렸습니다" in res.text
assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
# 발행된 사이트가 없으면(이 테스트) 자동 이동 없이 확인 문구만 보여준다.
assert "meta http-equiv=\"refresh\"" not in res.text
async def test_approve_page_redirects_to_the_published_blog_section(client, db_engine, owner_id):
"""승인 확인 화면은 5초 뒤 그 업장의 발행된 사이트(미니 블로그 자리)로 자동 이동한다
(2026-09-22, 사장님 지시: "이메일 승인후에 블로그가 올라가는 페이지에 5초 정도
이후에 연결되도록"). 자동 이동을 못 믿어도 되게 같은 주소를 링크로도 남긴다."""
token, post_id = await _seed_post(db_engine, owner_id)
async with db_engine.begin() as conn:
place_id = (
await conn.execute(text("SELECT place_id FROM place_posts WHERE post_id=:p"), {"p": post_id})
).scalar_one()
await conn.execute(
text(
"INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"
),
{"sid": uuid.uuid4(), "pid": place_id, "dom": "blog-redirect-test", "st": SiteStatus.PUBLISHED.value},
)
res = await client.get(f"/v1/site/post/approve?t={token}")
assert res.status_code == 200
assert "올렸습니다" in res.text
expected_url = site_payload.publish_origin() + "/s/blog-redirect-test#blog"
assert f'<meta http-equiv="refresh" content="5;url={expected_url}">' in res.text
assert f"href='{expected_url}'" in res.text
async def test_approve_link_is_single_use(client, db_engine, owner_id):