김성경
36dbe26c91
예전에는 소재 목록을 순서대로 뽑아 날짜에 차례로 붙여서, 9월 날짜에 '한겨울' 글이나 끝난 축제 글이 붙을 수 있었다. 이제 날짜를 먼저 정하고 그 날짜에 맞는 소재를 고른다. - 축제: 시작 14일 전 ~ 종료일 사이만 / 계절: 게시일 절기 하나 / 날씨: 그 달에 있을 법한 것만(눈 12~2월, 소나기 6~8월) / 주변: 늘 후보 - 프롬프트에 게시일을 넣고, 날씨를 단정하지 않게 문구를 고쳤다 - 계절·날씨·축제 주제 키에 연(월)을 넣어 해마다 다시 쓸 수 있게 했다 - 자동·구간·개별 세 경로가 _compose_for_dates 하나로 만든다 같이 고친 버그: materials 가 스냅샷에 없는 local.festivals·attractions 를 읽어 축제·주변 소재가 늘 비어 있었다. 원문 행(local.contents)을 읽는다.
258 lines
12 KiB
Python
258 lines
12 KiB
Python
"""미니 블로그 — 문구 필터와 메일 승인.
|
|
|
|
★ 이 파일이 지키는 것:
|
|
- 확인되지 않은 주장(요금·시간·인원·전화)이 든 문구는 화면에 못 간다
|
|
- 같은 링크를 두 번 눌러도 두 번 올라가지 않는다
|
|
- 만료된 링크는 오류가 아니라 안내로 끝난다
|
|
|
|
★ 2026-09-17 이후 GET 이 곧 승인이다(사장님 지시: "승인은 바로 승인 되게 그 링크만
|
|
클릭하면"). 메일 클라이언트의 링크 미리 열기에 그대로 노출된다는 걸 알고도 택한 것 —
|
|
예전처럼 "GET 은 확인 화면만" 을 기대하는 테스트를 두지 않는다.
|
|
"""
|
|
import uuid
|
|
from datetime import date, timezone
|
|
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import JobType, LocalContentType, PlaceCategory, PostStatus, PostTopicKind, SiteStatus
|
|
from services import blog_service, site_payload
|
|
|
|
BODY = (
|
|
"비가 한 차례 지나간 뒤 마당 돌이 검게 젖었습니다. 이런 날에는 대청마루에 앉아 빗소리만 들어도 "
|
|
"하루가 지나갑니다. 우산은 현관에 넉넉히 두었으니 편하게 다녀오세요. 젖은 길은 미끄러우니 "
|
|
"편한 신발을 권합니다. 마당 평상은 비가 그치면 금세 마릅니다."
|
|
)
|
|
|
|
|
|
def test_filter_rejects_unverifiable_claims():
|
|
for bad, word in [
|
|
(BODY[:120] + " 주중 198,000원입니다.", "198,000원"),
|
|
(BODY[:120] + " 입실은 15:00 입니다.", "15:00"),
|
|
(BODY[:120] + " 문의는 010-1234-5678 로.", "010-1234-5678"),
|
|
(BODY[:120] + " 군산 최고의 숙소입니다.", "최고"),
|
|
]:
|
|
ok, reason = blog_service.is_publishable_body(bad)
|
|
assert ok is False
|
|
assert word in reason
|
|
|
|
|
|
def test_filter_accepts_a_plain_scene():
|
|
ok, reason = blog_service.is_publishable_body(BODY)
|
|
assert ok is True, reason
|
|
|
|
|
|
def test_filter_drops_same_topic_twice_in_one_batch():
|
|
rows = [
|
|
{"body": BODY, "topic_key": "festival:야행", "topic_kind": PostTopicKind.FESTIVAL.value},
|
|
{"body": BODY, "topic_key": "festival:야행", "topic_kind": PostTopicKind.FESTIVAL.value},
|
|
]
|
|
kept, dropped = blog_service.filter_drafts(rows)
|
|
assert len(kept) == 1
|
|
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 # 음식점 전용 문구가 아직 없어 숙소 문구로 폴백한다
|
|
|
|
|
|
def _snapshot_with(*rows):
|
|
return {"local": {"region_code": "52130", "contents": list(rows)}}
|
|
|
|
|
|
def _festival_row(name, start, end):
|
|
return {"content_type": LocalContentType.FESTIVAL.value, "title": name,
|
|
"body": {"name": name, "eventstartdate": start, "eventenddate": end, "location": "군산 원도심"}}
|
|
|
|
|
|
def test_materials_follow_the_post_date():
|
|
"""게시일에 맞는 소재만 나온다(2026-09-23) — 9월 글에 '한겨울'·'눈'·끝난 축제가 붙지 않는다."""
|
|
snapshot = _snapshot_with(
|
|
_festival_row("시간여행축제", "20261002", "20261004"), # 곧 열린다(9일 뒤)
|
|
_festival_row("벚꽃축제", "20260401", "20260405"), # 이미 끝났다
|
|
_festival_row("겨울빛축제", "20261220", "20261231"), # 아직 한참 멀다
|
|
{"content_type": LocalContentType.ATTRACTION.value, "title": "경암동 철길마을",
|
|
"body": {"name": "경암동 철길마을", "description": "철길 옆 골목"}},
|
|
)
|
|
|
|
out = blog_service.materials(snapshot, date(2026, 9, 23))
|
|
keys = [key for _kind, key, _material in out]
|
|
|
|
assert keys[0] == "festival:2026:시간여행축제"
|
|
assert "2026.10.02 ~ 2026.10.04" in out[0][2]
|
|
assert not any("벚꽃" in k or "겨울빛" in k for k in keys)
|
|
assert "season:2026:초가을" in keys
|
|
assert not any(k.startswith("season:") and not k.endswith("초가을") for k in keys)
|
|
assert "nearby:경암동 철길마을" in keys
|
|
assert "weather:2026-09:비" in keys
|
|
assert not any(k.endswith(":눈") or k.endswith(":소나기") for k in keys)
|
|
|
|
|
|
def test_materials_allow_snow_in_winter_and_reuse_seasons_each_year():
|
|
winter = [key for _k, key, _m in blog_service.materials(_snapshot_with(), date(2027, 1, 10))]
|
|
|
|
assert "weather:2027-01:눈" in winter
|
|
assert "season:2027:한겨울" in winter
|
|
|
|
|
|
def test_build_prompt_tells_the_post_date():
|
|
prompt = blog_service.build_prompt(
|
|
place_name="테스트펜션", region="군산", topic_kind=PostTopicKind.SEASON.value,
|
|
material="초가을", used_topics=[], post_date=date(2026, 9, 23),
|
|
)
|
|
|
|
assert "게시일: 2026년 9월 23일(수) · 초가을" in prompt
|
|
|
|
|
|
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()
|
|
if expires_days < 0:
|
|
from common.utils.gtime import GTime
|
|
expires = GTime.AddDays(expires_days)
|
|
# ★ 함정: 이 시각은 naive-UTC 다. raw text() 로 바인딩하면 asyncpg 가 컬럼 타입 정보 없이
|
|
# 드라이버 기본값(로컬 시스템 시간대)으로 해석해 KST(UTC+9) 서버에서는 9시간 밀린다 —
|
|
# 14일짜리 옛 만료값에서는 안 드러났는데(그래도 미래), 자정 단위(issue_token 2026-09-17
|
|
# 개편)로 정밀해지자 자정 만료가 "9시간 전에 이미 만료"로 뒤집혔다. ORM 경로(update()/
|
|
# insert())는 컬럼의 DateTime(timezone=True) 프로세서를 타서 이 문제가 없다 — raw text()
|
|
# 로 timestamptz 컬럼에 naive datetime 을 바인딩할 때만 명시적으로 UTC 를 달아야 한다.
|
|
aware_expires = expires.replace(tzinfo=timezone.utc)
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("INSERT INTO places (place_id, owner_user_id, name, category, status) "
|
|
"VALUES (:pid, :uid, :name, :cat, 1)"),
|
|
{"pid": place_id, "uid": owner_id, "name": "스테이,머뭄", "cat": PlaceCategory.LODGING.value},
|
|
)
|
|
await conn.execute(
|
|
text("INSERT INTO place_posts (post_id, place_id, body, topic_kind, topic_key, status, "
|
|
"approve_token_hash, token_expires_at) "
|
|
"VALUES (:id, :pid, :body, :kind, :key, :st, :hash, :exp)"),
|
|
{"id": post_id, "pid": place_id, "body": BODY, "kind": PostTopicKind.WEATHER.value,
|
|
"key": f"weather:{post_id.hex[:6]}", "st": status.value, "hash": token_hash, "exp": aware_expires},
|
|
)
|
|
return token, post_id
|
|
|
|
|
|
async def _status(db_engine, post_id) -> int:
|
|
async with db_engine.begin() as conn:
|
|
row = await conn.execute(text("SELECT status FROM place_posts WHERE post_id = :id"), {"id": post_id})
|
|
return row.scalar()
|
|
|
|
|
|
async def test_get_approves_immediately(client, db_engine, owner_id):
|
|
token, post_id = await _seed_post(db_engine, owner_id)
|
|
|
|
res = await client.get(f"/v1/site/post/approve?t={token}")
|
|
|
|
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):
|
|
token, post_id = await _seed_post(db_engine, owner_id)
|
|
|
|
first = await client.get(f"/v1/site/post/approve?t={token}")
|
|
second = await client.get(f"/v1/site/post/approve?t={token}")
|
|
|
|
assert "올렸습니다" in first.text
|
|
assert "처리할 수 없는 링크" in second.text
|
|
assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
|
|
|
|
|
|
async def test_expired_link_is_an_answer_not_an_error(client, db_engine, owner_id):
|
|
token, post_id = await _seed_post(db_engine, owner_id, expires_days=-1)
|
|
|
|
res = await client.get(f"/v1/site/post/approve?t={token}")
|
|
|
|
assert res.status_code == 200
|
|
assert "처리할 수 없는 링크" in res.text
|
|
assert await _status(db_engine, post_id) == PostStatus.SENT.value
|
|
|
|
|
|
async def test_approve_enqueues_build_with_owner_user_id(client, db_engine, owner_id):
|
|
"""회귀: BUILD 잡 payload 에 owner_user_id 가 없으면 build_service.run_build 가
|
|
payload["owner_user_id"] 에서 KeyError 로 죽는다 — 승인 클릭이 실제로는 사이트를
|
|
재발행하지 않는 채로 '올렸습니다' 만 보여주고 있었다."""
|
|
token, post_id = await _seed_post(db_engine, owner_id)
|
|
|
|
res = await client.get(f"/v1/site/post/approve?t={token}")
|
|
assert "올렸습니다" in res.text
|
|
|
|
async with db_engine.begin() as conn:
|
|
row = await conn.execute(
|
|
text("SELECT payload FROM jobs WHERE job_type = :jt ORDER BY created_at DESC LIMIT 1"),
|
|
{"jt": JobType.BUILD.value},
|
|
)
|
|
payload = row.scalar()
|
|
assert payload["owner_user_id"] == owner_id
|