hbyang
c8dc68536e
연결 버튼이 발행 모달 안에 있었다. 그런데 계정은 `user × provider` 하나다(표도 그렇게 생겼다)
— 버튼이 사업장 화면에 있으면 사장님은 **업장마다 연결해야 하는 줄 안다.** 사장님 지적:
"여기 내 사이트에 sns 연동 하나 두고 연동해두고, 사이트 발행 후 게재하는 형식으로".
화면이 데이터 모양을 그대로 말해야 한다 — 연결은 한 번, 게재는 사이트마다다.
- `GET /v1/social/account` 신설: 연결 상태만 준다. 사업장을 고르지 않아도 답할 수 있어야 하는
값인데, 지금까지는 `/place/{id}` 안에만 있어서 사이트를 하나 고르기 전에는 물어볼 수 없었다
- features/social/SocialConnectionCard: '내 사이트' 목록 위의 연동 카드
(@핸들 · [연결] · [다시 연결] · [연결 해제]). 앱 자격증명이 없으면 **아무것도 안 그린다** —
누를 수 없는 버튼을 세우면 사장님에게는 고장난 화면이고 우리에게는 문의가 된다
- SocialPanel(발행 화면)에서 연결·해제 버튼 제거. 대신 계정이 없으면 "막다른 문구"가 아니라
**갈 곳**을 알린다 — [내 사이트] 로 보낸다. 연결 전에도 소개글 복사는 된다
- docs/SOCIAL.md: 사장님 흐름을 '연결(한 번) / 게재(사이트마다)' 로 다시 씀
검증: SNS 테스트 14건 통과 · frontend tsc·eslint 통과. 로컬에서 임시 자격증명으로 카드가
켜지는 것과 인가 URL 조립을 확인하고 값을 되돌렸다(지금은 connection_enabled=false 로 안 뜬다)
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
563 lines
21 KiB
Python
563 lines
21 KiB
Python
"""SNS 초안·승인·게시. 일반 발행과 분리해 사장님의 명시적인 요청만 처리한다."""
|
||
|
||
import hashlib
|
||
import json
|
||
from config import social_config as config
|
||
import secrets
|
||
from datetime import datetime, timedelta, timezone
|
||
from uuid import UUID
|
||
|
||
import httpx
|
||
from fastapi import HTTPException
|
||
from sqlalchemy import select, update, text
|
||
from sqlalchemy.dialects.postgresql import insert
|
||
|
||
from common.database.model.models import (
|
||
place_social_posts as Post,
|
||
sites,
|
||
places,
|
||
place_facts,
|
||
owner_social_accounts as Account,
|
||
)
|
||
from common.enums import SiteStatus, ErrorType, PUBLISHABLE_FACT_STATUSES
|
||
from crud.place_crud import PlaceCRUD
|
||
from crud import social_crud as db
|
||
from services import site_payload, social_account_service as accounts
|
||
from services.external import gemini_text
|
||
from services.external.social import (
|
||
SocialError,
|
||
SocialOutcomeUnknown,
|
||
adapter,
|
||
weighted_length,
|
||
)
|
||
|
||
|
||
def sha(token):
|
||
return hashlib.sha256(token.encode()).hexdigest()
|
||
|
||
|
||
def public_post(row):
|
||
return {
|
||
"account_bound": bool(row.account_id),
|
||
**{
|
||
key: (str(value) if isinstance(value, UUID) else value)
|
||
for key in (
|
||
"post_id",
|
||
"provider",
|
||
"body",
|
||
"link_url",
|
||
"status",
|
||
"approval_expires_at",
|
||
"approval_channel",
|
||
"permalink",
|
||
"posted_at",
|
||
"last_error",
|
||
)
|
||
if (value := getattr(row, key)) is not None
|
||
},
|
||
}
|
||
|
||
|
||
async def owned_place(s, user_id, place_id):
|
||
err, place = await PlaceCRUD().get_place(s, UUID(str(user_id)), UUID(str(place_id)))
|
||
if err != ErrorType.SUCCESS or not place:
|
||
raise HTTPException(404, "PLACE_NOT_FOUND")
|
||
return place
|
||
|
||
|
||
async def target(s, user_id, place_id):
|
||
place = await owned_place(s, user_id, place_id)
|
||
site = (
|
||
await s.execute(
|
||
select(sites)
|
||
.where(sites.place_id == place.place_id, sites.deleted == False)
|
||
.with_for_update()
|
||
)
|
||
).scalar_one_or_none() # noqa: E712
|
||
if (
|
||
not site
|
||
or site.status != SiteStatus.PUBLISHED.value
|
||
or not site.current_version_id
|
||
or not site.domain
|
||
):
|
||
raise HTTPException(409, "SOCIAL_PUBLISHED_FIXED_URL_REQUIRED")
|
||
return (
|
||
place,
|
||
site,
|
||
site_payload.publish_origin() + "/s/" + site_payload.publish_slug(place, site),
|
||
)
|
||
|
||
|
||
async def list_posts(user_id, place_id):
|
||
async def run(s):
|
||
await owned_place(s, user_id, place_id)
|
||
rows = (
|
||
(
|
||
await s.execute(
|
||
select(Post)
|
||
.where(
|
||
Post.place_id == place_id,
|
||
Post.user_id == user_id,
|
||
Post.deleted == False,
|
||
)
|
||
.order_by(Post.created_at.desc())
|
||
.limit(20)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
) # noqa: E712
|
||
account = await accounts.account(s, user_id, 2)
|
||
return {
|
||
"posts": [public_post(r) for r in rows],
|
||
"connection_enabled": accounts.configured(),
|
||
"posting_enabled": posting_enabled(),
|
||
"account": (
|
||
{
|
||
"handle": account.handle,
|
||
"profile_url": account.profile_url,
|
||
"status": account.status,
|
||
}
|
||
if account
|
||
else None
|
||
),
|
||
}
|
||
|
||
return await db.transaction(run)
|
||
|
||
|
||
async def account_state(user_id, provider=2):
|
||
"""사장님의 SNS 계정 연결 상태. **사업장과 무관하다.**
|
||
|
||
★ 왜 place 경로와 따로 두나 — 계정은 `user × provider` 단위라(표도 그렇게 생겼다)
|
||
사이트마다 물어볼 값이 아니다. 연결 화면이 사업장 안에 있으면 사장님은 업장 수만큼
|
||
연결해야 하는 줄 안다. 연결은 한 번, 게재는 사이트마다다.
|
||
"""
|
||
async def run(s):
|
||
row = await accounts.account(s, user_id, provider)
|
||
return {
|
||
"connection_enabled": accounts.configured(provider),
|
||
"posting_enabled": posting_enabled(),
|
||
"account": ({"handle": row.handle, "profile_url": row.profile_url,
|
||
"status": row.status} if row else None),
|
||
}
|
||
return await db.transaction(run)
|
||
|
||
|
||
def posting_enabled():
|
||
# 플랫폼 계약과 해지 안내 페이지 정책을 운영에서 확인한 뒤 명시적으로 연다.
|
||
return accounts.configured() and config.get("SOCIAL_POSTING_ENABLED") == "1"
|
||
|
||
|
||
async def create_draft(user_id, place_id, provider=2):
|
||
async def run(s):
|
||
place, site, url = await target(s, user_id, place_id)
|
||
rows = (
|
||
(
|
||
await s.execute(
|
||
select(place_facts).where(
|
||
place_facts.place_id == place.place_id,
|
||
place_facts.deleted == False, # noqa: E712
|
||
place_facts.status.in_(
|
||
[v.value for v in PUBLISHABLE_FACT_STATUSES]
|
||
),
|
||
(
|
||
place_facts.expires_at.is_(None)
|
||
| (place_facts.expires_at > datetime.now(timezone.utc))
|
||
),
|
||
)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
facts = [
|
||
{
|
||
"key": r.key,
|
||
"value": r.value,
|
||
"fact_id": str(r.fact_id),
|
||
"unit": r.unit,
|
||
"source_url": r.source_url,
|
||
}
|
||
for r in rows
|
||
if r.value and r.value.strip() and r.key != "meta_description"
|
||
]
|
||
if not facts:
|
||
raise HTTPException(409, "NO_GROUNDED_FACTS")
|
||
row_id = (
|
||
await s.execute(
|
||
insert(Post)
|
||
.values(
|
||
place_id=place_id,
|
||
user_id=user_id,
|
||
site_version_id=site.current_version_id,
|
||
provider=provider,
|
||
link_url=url,
|
||
grounded_facts=facts,
|
||
)
|
||
.on_conflict_do_nothing(
|
||
index_elements=["place_id", "site_version_id"],
|
||
index_where=text("deleted=false"),
|
||
)
|
||
.returning(Post.post_id)
|
||
)
|
||
).scalar_one_or_none()
|
||
row = (
|
||
await s.execute(
|
||
select(Post).where(
|
||
Post.place_id == place_id,
|
||
Post.site_version_id == site.current_version_id,
|
||
Post.deleted == False,
|
||
)
|
||
)
|
||
).scalar_one() # noqa: E712
|
||
# 생성 실패는 같은 원고 행을 재시도한다. 이미 쓴 글은 유료 재생성하지 않는다.
|
||
if not row_id and row.status == "FAILED" and not row.body:
|
||
row.status, row.last_error = "DRAFTING", None
|
||
row.updated_at = datetime.now(timezone.utc)
|
||
await s.flush()
|
||
row_id = row.post_id
|
||
if row_id:
|
||
await db.enqueue(s, row.post_id, 8)
|
||
return public_post(row)
|
||
|
||
return await db.transaction(run)
|
||
|
||
|
||
async def run_draft(job):
|
||
post_id = UUID(job["payload"]["post_id"])
|
||
|
||
async def load(s):
|
||
row = await s.get(Post, post_id)
|
||
if not row or row.deleted or row.status != "DRAFTING":
|
||
return None
|
||
place, _, url = await target(s, row.user_id, row.place_id)
|
||
if row.link_url != url:
|
||
raise SocialError("PUBLISH_URL_CHANGED")
|
||
return place.name, row.provider, row.link_url, row.grounded_facts
|
||
|
||
try:
|
||
data = await db.transaction(load)
|
||
if not data:
|
||
return {"skipped": True}
|
||
name, provider, url, facts = data
|
||
body = await gemini_text.generate_social_post(
|
||
name,
|
||
[
|
||
gemini_text.FactInput(
|
||
key=f["key"], label=f["key"], value=f["value"], unit=f.get("unit")
|
||
)
|
||
for f in facts
|
||
],
|
||
url,
|
||
provider,
|
||
)
|
||
|
||
async def save(s):
|
||
await s.execute(
|
||
update(Post)
|
||
.where(Post.post_id == post_id, Post.status == "DRAFTING")
|
||
.values(
|
||
body=body,
|
||
status="DRAFT",
|
||
last_error=None,
|
||
updated_at=datetime.now(timezone.utc),
|
||
)
|
||
)
|
||
|
||
await db.transaction(save)
|
||
return {"post_id": str(post_id)}
|
||
except Exception:
|
||
await set_failure(post_id, "DRAFT_FAILED", expected="DRAFTING")
|
||
raise SocialError("DRAFT_FAILED") from None
|
||
|
||
|
||
async def set_failure(post_id, code, *, expected="POSTING", status="FAILED"):
|
||
async def run(s):
|
||
await s.execute(
|
||
update(Post)
|
||
.where(Post.post_id == post_id, Post.status == expected)
|
||
.values(
|
||
status=status, last_error=code, updated_at=datetime.now(timezone.utc)
|
||
)
|
||
)
|
||
|
||
await db.transaction(run)
|
||
|
||
|
||
async def request_approval(user_id, post_id):
|
||
token = secrets.token_urlsafe(32)
|
||
|
||
async def run(s):
|
||
initial = await s.get(Post, post_id)
|
||
if not initial or initial.deleted or initial.user_id != user_id:
|
||
raise HTTPException(404, "PLACE_NOT_FOUND")
|
||
_, _, url = await target(s, user_id, initial.place_id)
|
||
# 생성/승인/게시 모두 site → post 순서로 잠근다. 반대면 동시 재요청이 교착된다.
|
||
row = (
|
||
await s.execute(
|
||
select(Post)
|
||
.where(
|
||
Post.post_id == post_id,
|
||
Post.user_id == user_id,
|
||
Post.deleted == False,
|
||
)
|
||
.with_for_update()
|
||
.execution_options(populate_existing=True)
|
||
)
|
||
).scalar_one() # noqa: E712
|
||
if row.link_url != url:
|
||
raise HTTPException(409, "PUBLISH_URL_CHANGED")
|
||
expired = row.approval_expires_at and row.approval_expires_at <= datetime.now(
|
||
timezone.utc
|
||
)
|
||
resendable = row.status == "PENDING_APPROVAL" and (expired or row.last_error)
|
||
reconnectable = row.status == "APPROVED" and row.account_id is None
|
||
if (
|
||
row.status not in ("DRAFT", "EXPIRED", "DECLINED", "FAILED")
|
||
and not resendable
|
||
and not reconnectable
|
||
) or not row.body:
|
||
return {"post": public_post(row), "already_processed": True}
|
||
account = (
|
||
await accounts.account(s, user_id, row.provider)
|
||
if posting_enabled()
|
||
else None
|
||
)
|
||
if posting_enabled() and (not account or account.status != "linked"):
|
||
raise HTTPException(409, "ACCOUNT_CONNECTION_REQUIRED")
|
||
row.status = "PENDING_APPROVAL"
|
||
row.account_id = account.account_id if account else None
|
||
row.approval_token_sha = sha(token)
|
||
row.approval_channel = "screen"
|
||
row.approval_sent_at = None
|
||
row.approval_expires_at = datetime.now(timezone.utc) + timedelta(
|
||
hours=max(1, min(168, int(config.get("SOCIAL_APPROVAL_HOURS", "24"))))
|
||
)
|
||
row.updated_at = datetime.now(timezone.utc)
|
||
row.last_error = None
|
||
await s.flush()
|
||
return {
|
||
"post": public_post(row),
|
||
"approval_path": f"/approve/{post_id}?t={token}",
|
||
}
|
||
|
||
result = await db.transaction(run)
|
||
if result.get("approval_path"):
|
||
from services.notify_service import request_approval as notify
|
||
|
||
await notify(post_id, user_id, token, result["approval_path"])
|
||
return result
|
||
|
||
|
||
async def approval(post_id, token, *, approve=None):
|
||
async def run(s):
|
||
row = (
|
||
await s.execute(
|
||
select(Post).where(
|
||
Post.post_id == post_id,
|
||
Post.deleted == False,
|
||
Post.approval_token_sha == sha(token),
|
||
)
|
||
)
|
||
).scalar_one_or_none() # noqa: E712
|
||
if not row:
|
||
raise HTTPException(404, "APPROVAL_NOT_FOUND")
|
||
await owned_place(s, row.user_id, row.place_id)
|
||
account = await s.get(Account, row.account_id) if row.account_id else None
|
||
if approve is None:
|
||
result = public_post(row)
|
||
result["account_handle"] = account.handle if account else None
|
||
return result
|
||
if approve and row.status == "PENDING_APPROVAL":
|
||
_, _, url = await target(s, row.user_id, row.place_id)
|
||
if row.link_url != url or (
|
||
row.account_id
|
||
and (not account or account.deleted or account.status != "linked")
|
||
):
|
||
raise HTTPException(409, "APPROVAL_TARGET_CHANGED")
|
||
applied = await db.decide(s, post_id, sha(token), approve, "link")
|
||
return {
|
||
"applied": applied,
|
||
"message": "승인했습니다"
|
||
if applied and approve
|
||
else "게재하지 않습니다"
|
||
if applied
|
||
else "이미 처리됐거나 만료된 요청입니다",
|
||
}
|
||
|
||
return await db.transaction(run)
|
||
|
||
|
||
async def owner_decision(user_id, post_id, approve):
|
||
# 화면은 비밀 링크를 저장하지 않아도 승인할 수 있다. 신원 범위만 다르고 CAS는 같다.
|
||
async def load(s):
|
||
row = await s.get(Post, post_id)
|
||
if not row or row.deleted or row.user_id != user_id:
|
||
raise HTTPException(404, "PLACE_NOT_FOUND")
|
||
await owned_place(s, user_id, row.place_id)
|
||
if approve:
|
||
_, _, url = await target(s, user_id, row.place_id)
|
||
account = await s.get(Account, row.account_id) if row.account_id else None
|
||
if url != row.link_url or (
|
||
row.account_id
|
||
and (not account or account.deleted or account.status != "linked")
|
||
):
|
||
raise HTTPException(409, "APPROVAL_TARGET_CHANGED")
|
||
return {
|
||
"applied": await db.decide(
|
||
s, post_id, row.approval_token_sha, approve, "builder"
|
||
)
|
||
}
|
||
|
||
return await db.transaction(load)
|
||
|
||
|
||
async def run_post(job):
|
||
post_id = UUID(job["payload"]["post_id"])
|
||
if not posting_enabled():
|
||
await set_failure(post_id, "SOCIAL_POSTING_DISABLED", expected="APPROVED")
|
||
raise SocialError("SOCIAL_POSTING_DISABLED")
|
||
|
||
async def claim(s):
|
||
return (
|
||
await s.execute(
|
||
update(Post)
|
||
.where(
|
||
Post.post_id == post_id,
|
||
Post.deleted == False, # noqa: E712
|
||
Post.status == "APPROVED",
|
||
)
|
||
.values(status="POSTING", updated_at=datetime.now(timezone.utc))
|
||
.returning(Post.user_id, Post.provider, Post.account_id)
|
||
)
|
||
).first()
|
||
|
||
claimed = await db.transaction(claim)
|
||
if not claimed:
|
||
return {"skipped": True}
|
||
user_id, provider, account_id = claimed
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
try:
|
||
|
||
async def refresh(s):
|
||
await accounts.lock_user(s, user_id, provider)
|
||
row = await s.get(Account, account_id)
|
||
if (
|
||
not row
|
||
or row.deleted
|
||
or row.user_id != user_id
|
||
or row.status != "linked"
|
||
):
|
||
raise SocialError("ACCOUNT_CONNECTION_REQUIRED")
|
||
return await accounts.get_usable_token(s, row, client)
|
||
|
||
try:
|
||
token = await db.transaction(refresh)
|
||
except Exception:
|
||
|
||
async def invalidate(s):
|
||
await s.execute(
|
||
update(Account)
|
||
.where(
|
||
Account.account_id == account_id, Account.status == "linked"
|
||
)
|
||
.values(
|
||
status="needs_reauth", last_error="TOKEN_REFRESH_FAILED"
|
||
)
|
||
)
|
||
|
||
await db.transaction(invalidate)
|
||
raise SocialError("ACCOUNT_NEEDS_REAUTH", reauth=True) from None
|
||
|
||
async def publish(s):
|
||
await accounts.lock_user(s, user_id, provider)
|
||
initial = await s.get(Post, post_id)
|
||
_, _, url = await target(s, user_id, initial.place_id)
|
||
row = (
|
||
await s.execute(
|
||
select(Post)
|
||
.where(Post.post_id == post_id)
|
||
.with_for_update()
|
||
.execution_options(populate_existing=True)
|
||
)
|
||
).scalar_one()
|
||
if row.status != "POSTING":
|
||
return {"skipped": True}
|
||
account = await s.get(Account, account_id)
|
||
if not account or account.status != "linked" or account.deleted:
|
||
raise SocialError("ACCOUNT_CONNECTION_REQUIRED")
|
||
# 마지막까지 사이트 잠금을 유지한다: 게시 도중 사이트를 내리는 경합을 직렬화한다.
|
||
if url != row.link_url:
|
||
raise SocialError("PUBLISH_URL_CHANGED")
|
||
if (
|
||
weighted_length(row.body, provider)
|
||
> adapter(provider).weighted_limit()
|
||
):
|
||
raise SocialError("TEXT_TOO_LONG")
|
||
identity = await adapter(provider).me(token, client=client)
|
||
if str(identity["id"]) != account.provider_user_id:
|
||
raise SocialError("ACCOUNT_IDENTITY_CHANGED", reauth=True)
|
||
result = await adapter(provider).publish(row.body, token, client=client)
|
||
row.status, row.provider_post_id, row.permalink = (
|
||
"POSTED",
|
||
result["id"],
|
||
result["permalink"],
|
||
)
|
||
row.posted_at = row.updated_at = datetime.now(timezone.utc)
|
||
row.last_error = None
|
||
await s.execute(
|
||
update(places)
|
||
.where(places.place_id == row.place_id)
|
||
.values(content_updated_at=row.posted_at)
|
||
)
|
||
# 재빌드는 사이트당 큐 키를 쓰되 여기서 외부 게시를 다시 호출하지 않는다.
|
||
await s.execute(
|
||
text("""INSERT INTO jobs(job_type,payload,dedupe_key) VALUES
|
||
(4,CAST(:payload AS jsonb),:key) ON CONFLICT (dedupe_key)
|
||
WHERE status IN (1,2) AND dedupe_key IS NOT NULL DO NOTHING"""),
|
||
{
|
||
"payload": json.dumps(
|
||
{
|
||
"place_id": str(row.place_id),
|
||
"owner_user_id": str(user_id),
|
||
"publish": True,
|
||
}
|
||
),
|
||
"key": f"social-build:{post_id}",
|
||
},
|
||
)
|
||
await s.execute(text("SELECT pg_notify('web4ai_job', '')"))
|
||
return {"post_id": str(post_id), "permalink": result["permalink"]}
|
||
|
||
# API 성공 뒤 DB 저장 실패도 결과 불명이다: FAILED로 떨어뜨려 재게시시키지 않는다.
|
||
return await db.transaction(publish)
|
||
except SocialOutcomeUnknown:
|
||
await set_failure(post_id, "POST_RESULT_UNKNOWN", status="UNKNOWN")
|
||
raise SocialOutcomeUnknown("POST_RESULT_UNKNOWN") from None
|
||
except (SocialError, HTTPException) as ex:
|
||
if getattr(ex, "reauth", False):
|
||
|
||
async def reauth(s):
|
||
await s.execute(
|
||
update(Account)
|
||
.where(
|
||
Account.account_id == account_id, Account.status == "linked"
|
||
)
|
||
.values(
|
||
status="needs_reauth", last_error="ACCOUNT_NEEDS_REAUTH"
|
||
)
|
||
)
|
||
|
||
await db.transaction(reauth)
|
||
await set_failure(
|
||
post_id,
|
||
str(ex)
|
||
if isinstance(ex, SocialError)
|
||
else "PUBLISH_TARGET_UNAVAILABLE",
|
||
)
|
||
raise SocialError("POST_FAILED") from None
|
||
except Exception:
|
||
await set_failure(post_id, "POST_RESULT_UNKNOWN", status="UNKNOWN")
|
||
raise SocialOutcomeUnknown("POST_RESULT_UNKNOWN") from None
|