diff --git a/app/core/common.py b/app/core/common.py
index 9fdc41b..08e4bc4 100644
--- a/app/core/common.py
+++ b/app/core/common.py
@@ -54,6 +54,28 @@ async def lifespan(app: FastAPI):
f"{type(e).__name__}: {e}"
)
+ # P2V 고아 잡 스윕 — 검수 방치·크래시로 남은 선차감분을 환불하고,
+ # archiving 에 갇힌 잡을 재시도 가능 상태로 되돌린다 (설계 R-1·R-3).
+ # 스키마는 앱이 만들지 않는다 — migration_2026-08-25_p2v.sql 수동 실행.
+ from config import p2v_settings
+
+ if p2v_settings.P2V_ENABLED:
+ try:
+ from app.database.session import BackgroundSessionLocal
+ from app.p2v.services import f1_service, f2_service
+
+ async with BackgroundSessionLocal() as session:
+ swept = await f1_service.sweep_orphans(session)
+ swept += await f2_service.sweep_orphans(session)
+ await session.commit()
+ if swept:
+ logger.info(f"[p2v] 고아 잡 {swept}건 환불·정리")
+ except Exception as e:
+ logger.error(
+ f"[p2v] 고아 스윕 실패 (다음 기동 시 재시도): "
+ f"{type(e).__name__}: {e}"
+ )
+
await NvMapPwScraper.initiate_scraper()
except asyncio.TimeoutError:
logger.error("Database initialization timed out")
@@ -82,9 +104,11 @@ async def lifespan(app: FastAPI):
# 공유 HTTP 클라이언트 종료
from app.utils.creatomate import close_shared_client
from app.utils.upload_blob_as_request import close_shared_blob_client
+ from app.p2v.services.client import close_client as close_p2v_client
await close_shared_client()
await close_shared_blob_client()
+ await close_p2v_client()
from app.database.like_cache import close_like_cache
await close_like_cache()
diff --git a/app/core/exceptions.py b/app/core/exceptions.py
index fc0fda6..b406a27 100644
--- a/app/core/exceptions.py
+++ b/app/core/exceptions.py
@@ -345,6 +345,24 @@ def add_exception_handlers(app: FastAPI):
},
)
+ # P2vException 핸들러 추가
+ # (SsulboxException 과 같은 (message, status_code, code) 형태 — 수동 등록)
+ from app.p2v.exceptions import P2vException
+
+ @app.exception_handler(P2vException)
+ def p2v_exception_handler(request: Request, exc: P2vException) -> Response:
+ if exc.status_code < 500:
+ logger.warning(f"Handled P2vException: {exc.__class__.__name__} - {exc.message}")
+ else:
+ logger.error(f"Handled P2vException: {exc.__class__.__name__} - {exc.message}")
+ return JSONResponse(
+ status_code=exc.status_code,
+ content={
+ "detail": exc.message,
+ "code": exc.code,
+ },
+ )
+
@app.exception_handler(status.HTTP_500_INTERNAL_SERVER_ERROR)
def internal_server_error_handler(request, exception):
# 에러 메시지 로깅 (한글 포함 가능)
diff --git a/app/database/session.py b/app/database/session.py
index 839bf62..703ea79 100644
--- a/app/database/session.py
+++ b/app/database/session.py
@@ -90,6 +90,7 @@ async def create_db_tables():
from app.ssulbox.models import ( # noqa: F401
SsulContent,
)
+ from app.p2v.models import P2vF1Job, P2vF2Job # noqa: F401
# 생성할 테이블 목록 (FK 순서: 참조 대상 먼저)
tables_to_create = [
@@ -114,6 +115,9 @@ async def create_db_tables():
CreditTransaction.__table__,
# 썰박스 (FK 순서: ssul_content 를 나머지가 참조)
SsulContent.__table__,
+ # P2V (FK 순서: p2v_video 를 p2v_poster 가 참조)
+ P2vF1Job.__table__,
+ P2vF2Job.__table__,
]
logger.info("Creating database tables...")
diff --git a/app/p2v/__init__.py b/app/p2v/__init__.py
new file mode 100644
index 0000000..b9aa8c4
--- /dev/null
+++ b/app/p2v/__init__.py
@@ -0,0 +1,12 @@
+# -*- coding: utf-8 -*-
+"""P2V — 무빙 포스터(F1) / 포스터 스타일링(F2) 프록시 모듈.
+
+P2V 서버(:8010, o2o-ado2-poster-to-video)는 사용자 개념이 없는 별도 프로세스다.
+이 모듈은 그 앞에서 castad 가 책임지는 것들을 얹는다:
+
+- 인증(JWT)과 잡 소유권 (p2v_video / p2v_poster 테이블)
+- 크레딧 선차감·환불 (credit_transaction 원장 재사용, job_type='p2v_f1'|'p2v_f2')
+- 완성 결과물의 Azure Blob 아카이빙 (P2V 컨테이너가 재생성돼도 결과물이 살아남게)
+
+설계 문서: docs/design/p2v-credit-integration.md
+"""
diff --git a/app/p2v/api/__init__.py b/app/p2v/api/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/p2v/api/routers/__init__.py b/app/p2v/api/routers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/p2v/api/routers/v1/__init__.py b/app/p2v/api/routers/v1/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/p2v/api/routers/v1/f1.py b/app/p2v/api/routers/v1/f1.py
new file mode 100644
index 0000000..db5fc98
--- /dev/null
+++ b/app/p2v/api/routers/v1/f1.py
@@ -0,0 +1,429 @@
+# -*- coding: utf-8 -*-
+"""P2V F1(무빙 포스터) 프록시 라우터.
+
+생성 → 폴링 → (검수: 나레이션/메타 수정 → 승인) → 폴링 → done 의 2단 게이트.
+크레딧은 생성 시점 선차감. 실패 시 즉시 환불하지 않는다 — 재시도 버튼이 있어
+"환불 + 공짜 재시도"가 되기 때문이다. 환불은 삭제·고아 스윕에서 일어난다
+(f1_service 모듈 docstring 참조).
+"""
+
+from typing import Optional
+
+from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, UploadFile
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.credit.exceptions import InsufficientCreditError
+from app.database.session import AsyncSessionLocal, get_session
+from app.p2v.exceptions import (
+ P2vDisabledError,
+ P2vException,
+ P2vInvalidStateError,
+ P2vUploadTooLargeError,
+ P2vUpstreamError,
+)
+from app.p2v.models import P2vF1Job
+from app.p2v.schemas.p2v_schema import (
+ ApproveRequest,
+ MetadataUpdateRequest,
+ NarrationUpdateRequest,
+ P2vDeleteResponse,
+ P2vF1StatusResponse,
+ P2vJobCreateResponse,
+ P2vJobError,
+ P2vMetadata,
+)
+from app.p2v.services import archive_service, client, f1_service
+from app.user.dependencies.auth import get_current_user
+from app.user.models import User
+from app.utils.logger import get_logger
+from config import p2v_settings
+
+logger = get_logger("p2v")
+
+router = APIRouter(prefix="/p2v/f1", tags=["P2V"])
+
+
+def _guard_enabled() -> None:
+ if not p2v_settings.P2V_ENABLED:
+ raise P2vDisabledError()
+
+
+@router.post(
+ "/jobs",
+ response_model=P2vJobCreateResponse,
+ summary="무빙 포스터 생성 요청",
+ description="**이 시점에 크레딧이 선차감됩니다.** 검수 승인 없이 방치하면 "
+ "일정 시간 후 자동 환불·실패 처리됩니다.",
+ responses={
+ 401: {"description": "인증 실패"},
+ 402: {"description": "크레딧 부족"},
+ 413: {"description": "용량 초과"},
+ 503: {"description": "P2V 서버 연결 실패"},
+ },
+)
+async def create_job(
+ background_tasks: BackgroundTasks,
+ poster: UploadFile = File(...),
+ name: str = Form(""),
+ current_user: User = Depends(get_current_user),
+) -> P2vJobCreateResponse:
+ _guard_enabled()
+ raw = await poster.read()
+ if len(raw) > p2v_settings.P2V_MAX_UPLOAD_BYTES:
+ raise P2vUploadTooLargeError(p2v_settings.P2V_MAX_UPLOAD_BYTES)
+
+ # 행 삽입 + 크레딧 선차감을 한 트랜잭션으로 묶는다 (ssulbox create_ssul 패턴)
+ async with AsyncSessionLocal() as session:
+ try:
+ row = await f1_service.create_job(
+ session, user_uuid=current_user.user_uuid, name=name.strip()
+ )
+ await session.commit()
+ job_id = row.id
+ except InsufficientCreditError:
+ await session.rollback()
+ logger.info(f"[f1] INSUFFICIENT CREDIT user={current_user.user_uuid}")
+ raise
+ except Exception:
+ await session.rollback()
+ raise
+
+ # 차감이 확정된 뒤에야 P2V 를 부른다
+ try:
+ created = await client.request_json(
+ "POST",
+ "/api/f1/jobs",
+ data={"name": name},
+ files={
+ "poster": (
+ poster.filename or "poster.png",
+ raw,
+ poster.content_type or "image/png",
+ )
+ },
+ )
+ except P2vException as e:
+ # P2V 를 태우지 못했다(실비용 없음) — 환불하고 실패로 남긴다
+ async with AsyncSessionLocal() as session:
+ failed_row = await session.get(P2vF1Job, job_id)
+ if failed_row is not None:
+ await f1_service.fail_job(
+ session, failed_row, f"P2V 요청 실패: {e.message}", refund=True
+ )
+ await session.commit()
+ raise
+
+ async with AsyncSessionLocal() as session:
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+ row.p2v_job_id = str(created["id"])
+ await session.commit()
+
+ background_tasks.add_task(
+ archive_service.upload_source_image,
+ "f1",
+ job_id,
+ current_user.user_uuid,
+ raw,
+ poster.filename or "poster.png",
+ )
+ return P2vJobCreateResponse(
+ id=job_id,
+ status="queued",
+ poll_interval_seconds=p2v_settings.P2V_POLL_HINT_SECONDS,
+ )
+
+
+def _error_of(row: P2vF1Job) -> Optional[P2vJobError]:
+ if row.status != "failed" or not row.error:
+ return None
+ stage, _, detail = row.error.partition(": ")
+ return P2vJobError(stage=stage or "?", detail=detail or row.error)
+
+
+def _metadata_of(row: P2vF1Job) -> Optional[P2vMetadata]:
+ if not (row.event_name or row.place or row.date_text):
+ return None
+ return P2vMetadata(
+ event_name=row.event_name or "",
+ date_text=row.date_text or "",
+ place=row.place or "",
+ keywords=row.keywords,
+ )
+
+
+def _db_response(row: P2vF1Job) -> P2vF1StatusResponse:
+ """P2V 를 부르지 않고 castad DB 미러만으로 응답 (archived/failed 종결 상태)."""
+ artifacts: dict[str, str] = {}
+ if row.p2v_video_url:
+ artifacts["video"] = row.p2v_video_url
+ if row.poster_url:
+ artifacts["thumbnail"] = row.poster_url
+ return P2vF1StatusResponse(
+ id=row.id,
+ status=row.status, # type: ignore[arg-type]
+ name=row.name,
+ narration=row.narration,
+ metadata=_metadata_of(row),
+ duration=float(row.duration) if row.duration is not None else None,
+ error=_error_of(row),
+ artifacts=artifacts,
+ )
+
+
+@router.get(
+ "/jobs/{job_id}",
+ response_model=P2vF1StatusResponse,
+ summary="생성 진행 상태 (폴링)",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "잡 없음"}},
+)
+async def get_job(
+ job_id: int,
+ background_tasks: BackgroundTasks,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> P2vF1StatusResponse:
+ _guard_enabled()
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+
+ # 종결 상태 — P2V 를 부르지 않는다 (컨테이너 재기동으로 잡이 지워졌을 수 있다)
+ if row.archived_at is not None or row.status == "failed" or not row.p2v_job_id:
+ return _db_response(row)
+
+ try:
+ p2v_job = await client.request_json("GET", f"/api/f1/jobs/{row.p2v_job_id}")
+ except P2vUpstreamError as e:
+ if e.status_code == 404:
+ # P2V 재기동으로 잡 소실 — 재시도가 불가능하므로 환불하고 실패 처리
+ await f1_service.fail_job(
+ session,
+ row,
+ "서버 재시작으로 작업이 유실되었습니다. 크레딧을 환불했습니다",
+ refund=True,
+ )
+ await session.commit()
+ return _db_response(row)
+ raise
+
+ f1_service.sync_from_p2v(row, p2v_job)
+
+ status = row.status
+ if p2v_job.get("status") == "done" and row.archived_at is None:
+ # Blob 업로드 전까지는 진행 중으로 보인다 — URL 없는 done 창을 만들지 않는다
+ status = "archiving"
+ background_tasks.add_task(archive_service.archive_f1, row.id)
+ await session.commit()
+
+ # 검수용 영역분석 이미지는 프록시 경로로 재작성한다 (regions/ 는 인증 프록시)
+ artifacts: dict[str, str] = {}
+ check = (p2v_job.get("artifacts") or {}).get("check_jpg")
+ if check and check.startswith("/files/"):
+ artifacts["check_jpg"] = "/p2v" + check
+
+ return P2vF1StatusResponse(
+ id=row.id,
+ status=status, # type: ignore[arg-type]
+ stage=p2v_job.get("stage"),
+ stages=p2v_job.get("stages"),
+ name=row.name,
+ narration=row.narration,
+ metadata=_metadata_of(row),
+ motion_elements=p2v_job.get("motion_elements"),
+ duration=float(row.duration) if row.duration is not None else None,
+ queue_size=p2v_job.get("queue_size"),
+ error=_error_of(row),
+ artifacts=artifacts,
+ )
+
+
+# =============================================================================
+# 검수 게이트
+# =============================================================================
+
+
+@router.put(
+ "/jobs/{job_id}/narration",
+ summary="검수 중 나레이션 수정",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "잡 없음"}},
+)
+async def update_narration(
+ job_id: int,
+ body: NarrationUpdateRequest,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> dict:
+ _guard_enabled()
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+ if not row.p2v_job_id:
+ raise P2vInvalidStateError("아직 서버에 접수되지 않은 작업입니다")
+ await client.request_json(
+ "PUT",
+ f"/api/f1/jobs/{row.p2v_job_id}/narration",
+ json_body={"narration": body.narration},
+ )
+ row.narration = body.narration
+ await session.commit()
+ return {"ok": True}
+
+
+@router.put(
+ "/jobs/{job_id}/metadata",
+ summary="검수 중 행사 메타데이터 수정",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "잡 없음"}},
+)
+async def update_metadata(
+ job_id: int,
+ body: MetadataUpdateRequest,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> dict:
+ _guard_enabled()
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+ if not row.p2v_job_id:
+ raise P2vInvalidStateError("아직 서버에 접수되지 않은 작업입니다")
+ await client.request_json(
+ "PUT",
+ f"/api/f1/jobs/{row.p2v_job_id}/metadata",
+ json_body=body.model_dump(),
+ )
+ row.event_name = body.event_name[:200]
+ row.date_text = body.date_text[:100] or None
+ row.place = body.place[:200]
+ await session.commit()
+ return {"ok": True}
+
+
+@router.post(
+ "/jobs/{job_id}/approve",
+ summary="검수 승인 — TTS·BGM·애니메이션·렌더 시작",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "잡 없음"}},
+)
+async def approve_job(
+ job_id: int,
+ body: Optional[ApproveRequest] = None,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> dict:
+ _guard_enabled()
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+ if not row.p2v_job_id:
+ raise P2vInvalidStateError("아직 서버에 접수되지 않은 작업입니다")
+ json_body = (
+ {"motions": body.motions} if body is not None and body.motions is not None else None
+ )
+ await client.request_json(
+ "POST", f"/api/f1/jobs/{row.p2v_job_id}/approve", json_body=json_body
+ )
+ row.status = "running"
+ await session.commit()
+ return {"ok": True}
+
+
+@router.post(
+ "/jobs/{job_id}/retry",
+ summary="실패한 스테이지부터 재시도",
+ description="선차감분이 그대로 유효하므로 재시도에 크레딧이 추가 차감되지 않는다. "
+ "(재과금 정책은 D-2 로 보류 — 도입 시 attempt 컬럼 추가)",
+ responses={
+ 401: {"description": "인증 실패"},
+ 404: {"description": "잡 없음"},
+ 409: {"description": "실패 상태가 아님"},
+ },
+)
+async def retry_job(
+ job_id: int,
+ body: Optional[ApproveRequest] = None,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> dict:
+ _guard_enabled()
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+ if row.status != "failed":
+ raise P2vInvalidStateError(f"실패 상태가 아닙니다: {row.status}")
+ if not row.p2v_job_id:
+ raise P2vInvalidStateError("서버에 접수되지 못한 작업은 재시도할 수 없습니다")
+ json_body = (
+ {"motions": body.motions} if body is not None and body.motions is not None else None
+ )
+ await client.request_json(
+ "POST", f"/api/f1/jobs/{row.p2v_job_id}/retry", json_body=json_body
+ )
+ row.status = "queued"
+ row.error = None
+ await session.commit()
+ return {"ok": True}
+
+
+@router.post(
+ "/jobs/{job_id}/force",
+ summary="실패를 무시하고 진행 (i2v 게이트 실패 전용)",
+ description="제목 훼손 게이트 등 클립 생성 **이후의** 실패를 사람이 확인하고 "
+ "이미 만들어진 클립으로 렌더를 잇는다. 재시도와 달리 Higgsfield 재생성이 없다. "
+ "클립이 없는 실패는 P2V 가 409 로 거절한다.",
+ responses={
+ 401: {"description": "인증 실패"},
+ 404: {"description": "잡 없음"},
+ 409: {"description": "실패 상태가 아니거나 무시할 수 없는 실패"},
+ },
+)
+async def force_job(
+ job_id: int,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> dict:
+ _guard_enabled()
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+ if row.status != "failed":
+ raise P2vInvalidStateError(f"실패 상태가 아닙니다: {row.status}")
+ if not row.p2v_job_id:
+ raise P2vInvalidStateError("서버에 접수되지 못한 작업은 진행할 수 없습니다")
+ await client.request_json("POST", f"/api/f1/jobs/{row.p2v_job_id}/force")
+ row.status = "queued"
+ row.error = None
+ await session.commit()
+ return {"ok": True}
+
+
+@router.delete(
+ "/jobs/{job_id}",
+ response_model=P2vDeleteResponse,
+ summary="잡과 산출물 삭제 — 되돌릴 수 없다",
+ description="완성(done) 전에 버리는 잡은 이 시점에 크레딧이 환불된다.",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "잡 없음"}},
+)
+async def delete_job(
+ job_id: int,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> P2vDeleteResponse:
+ _guard_enabled()
+ row = await f1_service.get_owned(session, job_id, current_user.user_uuid)
+
+ # P2V 쪽 산출물 먼저 정리. 이미 사라진 잡(404)은 정상 — 지울 게 없다.
+ if row.p2v_job_id:
+ try:
+ await client.request_json("DELETE", f"/api/f1/jobs/{row.p2v_job_id}")
+ except P2vUpstreamError as e:
+ if e.status_code != 404:
+ raise # 진행 중(409) 등은 그대로 중계 — 반쪽 삭제를 만들지 않는다
+
+ # 환불 판정은 잠금 재조회로 한다 (2026-08-26 리뷰 TOCTOU 반영) —
+ # 처음 읽은 스냅샷과 달리, 그 사이 백그라운드 아카이빙이 done 으로 전이를
+ # 커밋했을 수 있다. FOR UPDATE 로 최신 상태를 고정한 뒤 판정해야
+ # "완성본도 받고 환불도 받는" 창이 닫힌다.
+ result = await session.execute(
+ select(P2vF1Job)
+ .where(P2vF1Job.id == job_id)
+ .with_for_update()
+ )
+ row = result.scalar_one_or_none()
+ if row is None:
+ return P2vDeleteResponse(id=job_id, removed=True)
+
+ # done(완성)·archiving(렌더 완료, 업로드 중)은 결과물이 이미 만들어졌다 — 환불 없음
+ if row.archived_at is None and row.status not in ("done", "archiving"):
+ await f1_service.refund_job(session, row, reason="무빙 포스터 삭제(미완성)")
+
+ await session.delete(row)
+ await session.commit()
+ return P2vDeleteResponse(id=job_id, removed=True)
diff --git a/app/p2v/api/routers/v1/f2.py b/app/p2v/api/routers/v1/f2.py
new file mode 100644
index 0000000..b5de115
--- /dev/null
+++ b/app/p2v/api/routers/v1/f2.py
@@ -0,0 +1,335 @@
+# -*- coding: utf-8 -*-
+"""P2V F2(포스터 스타일링) 프록시 라우터.
+
+브라우저 → castad(인증·크레딧) → P2V 서버(:8010) 중계.
+템플릿 목록 등 조회는 순수 중계, 잡 생성만 크레딧 선차감이 붙는다.
+"""
+
+from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, UploadFile
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.credit.exceptions import InsufficientCreditError
+from app.database.session import AsyncSessionLocal, get_session
+from app.p2v.exceptions import (
+ P2vDisabledError,
+ P2vException,
+ P2vUploadTooLargeError,
+ P2vUpstreamError,
+)
+from app.p2v.models import P2vF2Job
+from app.p2v.schemas.p2v_schema import (
+ F2CategoryResponse,
+ F2FormatResponse,
+ F2TemplateResponse,
+ F2UploadHintResponse,
+ P2vF2StatusResponse,
+ P2vJobCreateResponse,
+ P2vJobError,
+)
+from app.p2v.services import archive_service, client, f2_service
+from app.user.dependencies.auth import get_current_user
+from app.user.models import User
+from app.utils.logger import get_logger
+from config import p2v_settings
+
+logger = get_logger("p2v")
+
+router = APIRouter(prefix="/p2v/f2", tags=["P2V"])
+
+
+def _guard_enabled() -> None:
+ if not p2v_settings.P2V_ENABLED:
+ raise P2vDisabledError()
+
+
+def _proxy_thumb(url: str) -> str:
+ """P2V 의 /files/... 경로를 castad 프록시 경로로 재작성한다."""
+ if url.startswith("/files/"):
+ return "/p2v" + url
+ return url
+
+
+def _rewrite_template(t: dict) -> dict:
+ return {**t, "thumb_url": _proxy_thumb(t.get("thumb_url", ""))}
+
+
+# =============================================================================
+# 템플릿 · 메타 (순수 중계)
+# =============================================================================
+
+
+@router.get(
+ "/templates",
+ response_model=list[F2TemplateResponse],
+ summary="스타일 템플릿 목록",
+ responses={401: {"description": "인증 실패"}, 503: {"description": "P2V 서버 연결 실패"}},
+)
+async def list_templates(
+ current_user: User = Depends(get_current_user),
+) -> list[dict]:
+ _guard_enabled()
+ items = await client.request_json("GET", "/api/f2/templates")
+ return [_rewrite_template(t) for t in items]
+
+
+@router.post(
+ "/templates",
+ response_model=F2TemplateResponse,
+ summary="사용자 레퍼런스 업로드",
+ description="저장 + 화풍 분석까지 P2V 가 동기로 끝낸다(10초 안팎).",
+ responses={401: {"description": "인증 실패"}, 413: {"description": "용량 초과"}},
+)
+async def create_template(
+ reference: UploadFile = File(...),
+ name: str = Form(""),
+ current_user: User = Depends(get_current_user),
+) -> dict:
+ _guard_enabled()
+ raw = await reference.read()
+ if len(raw) > p2v_settings.P2V_MAX_UPLOAD_BYTES:
+ raise P2vUploadTooLargeError(p2v_settings.P2V_MAX_UPLOAD_BYTES)
+ created = await client.request_json(
+ "POST",
+ "/api/f2/templates",
+ data={"name": name},
+ files={
+ "reference": (
+ reference.filename or "reference.png",
+ raw,
+ reference.content_type or "image/png",
+ )
+ },
+ )
+ return _rewrite_template(created)
+
+
+@router.delete(
+ "/templates/{template_id}",
+ summary="사용자 레퍼런스 삭제",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "삭제 불가 템플릿"}},
+)
+async def delete_template(
+ template_id: str,
+ current_user: User = Depends(get_current_user),
+) -> dict:
+ _guard_enabled()
+ return await client.request_json("DELETE", f"/api/f2/templates/{template_id}")
+
+
+@router.get("/categories", response_model=list[F2CategoryResponse], summary="템플릿 카테고리")
+async def list_categories(current_user: User = Depends(get_current_user)) -> list[dict]:
+ _guard_enabled()
+ return await client.request_json("GET", "/api/f2/categories")
+
+
+@router.get("/formats", response_model=list[F2FormatResponse], summary="출력 포맷 목록")
+async def list_formats(current_user: User = Depends(get_current_user)) -> list[dict]:
+ _guard_enabled()
+ return await client.request_json("GET", "/api/f2/formats")
+
+
+@router.get("/upload-hint", response_model=F2UploadHintResponse, summary="레퍼런스 업로드 안내")
+async def upload_hint(current_user: User = Depends(get_current_user)) -> dict:
+ _guard_enabled()
+ return await client.request_json("GET", "/api/f2/upload-hint")
+
+
+# =============================================================================
+# 잡
+# =============================================================================
+
+
+@router.post(
+ "/jobs",
+ response_model=P2vJobCreateResponse,
+ summary="스타일 변환 요청",
+ description="**이 시점에 크레딧이 선차감됩니다.** 실패 시 자동 환불됩니다.",
+ responses={
+ 401: {"description": "인증 실패"},
+ 402: {"description": "크레딧 부족"},
+ 413: {"description": "용량 초과"},
+ 503: {"description": "P2V 서버 연결 실패"},
+ },
+)
+async def create_job(
+ background_tasks: BackgroundTasks,
+ poster: UploadFile = File(...),
+ template_id: str = Form(...),
+ format: str = Form("poster"),
+ current_user: User = Depends(get_current_user),
+) -> P2vJobCreateResponse:
+ _guard_enabled()
+ raw = await poster.read()
+ if len(raw) > p2v_settings.P2V_MAX_UPLOAD_BYTES:
+ raise P2vUploadTooLargeError(p2v_settings.P2V_MAX_UPLOAD_BYTES)
+
+ # 템플릿 스냅샷은 차감 **전에** 확보한다 — 없는 템플릿이면 크레딧을 건드리지 않는다
+ templates = await client.request_json("GET", "/api/f2/templates")
+ tpl = next((t for t in templates if t.get("id") == template_id), None)
+ if tpl is None:
+ raise P2vUpstreamError("존재하지 않는 템플릿입니다", 404)
+
+ # 포스터 이름은 업로드 파일명에서 딴다 (F2 에는 F1 의 행사명 같은 입력이 없다).
+ # 템플릿 이름을 이름 자리에 쓰면 "별이 빛나는 밤"이 포스터 이름처럼 보인다.
+ poster_name = (poster.filename or "").rsplit(".", 1)[0].strip()
+
+ # 행 삽입 + 크레딧 선차감을 한 트랜잭션으로 묶는다 (ssulbox create_ssul 패턴)
+ async with AsyncSessionLocal() as session:
+ try:
+ row = await f2_service.create_job(
+ session,
+ user_uuid=current_user.user_uuid,
+ name=poster_name,
+ template_id=template_id,
+ template_name=tpl.get("name_ko"),
+ license=tpl.get("license"),
+ format=format,
+ )
+ await session.commit()
+ job_id = row.id
+ except InsufficientCreditError:
+ await session.rollback()
+ logger.info(f"[f2] INSUFFICIENT CREDIT user={current_user.user_uuid}")
+ raise
+ except Exception:
+ await session.rollback()
+ raise
+
+ # 차감이 확정된 뒤에야 P2V 를 부른다 — 잔액 없는 사용자가 실비용을 태우지 않게
+ try:
+ created = await client.request_json(
+ "POST",
+ "/api/f2/jobs",
+ data={"template_id": template_id, "format": format},
+ files={
+ "poster": (
+ poster.filename or "poster.png",
+ raw,
+ poster.content_type or "image/png",
+ )
+ },
+ )
+ except P2vException as e:
+ # 차감은 이미 확정됐다 — 새 트랜잭션에서 실패 처리(환불)한다
+ async with AsyncSessionLocal() as session:
+ failed_row = await session.get(P2vF2Job, job_id)
+ if failed_row is not None:
+ await f2_service.fail_job(
+ session, failed_row, f"P2V 요청 실패: {e.message}"
+ )
+ await session.commit()
+ raise
+
+ async with AsyncSessionLocal() as session:
+ row = await f2_service.get_owned(session, job_id, current_user.user_uuid)
+ row.p2v_job_id = str(created["id"])
+ await session.commit()
+
+ background_tasks.add_task(
+ archive_service.upload_source_image,
+ "f2",
+ job_id,
+ current_user.user_uuid,
+ raw,
+ poster.filename or "poster.png",
+ )
+ return P2vJobCreateResponse(
+ id=job_id,
+ status="queued",
+ poll_interval_seconds=p2v_settings.P2V_POLL_HINT_SECONDS,
+ )
+
+
+@router.delete(
+ "/jobs/{job_id}",
+ summary="스타일링 잡 삭제 (내 콘텐츠 목록에서 제거)",
+ description="완성(done) 전에 버리는 잡은 이 시점에 크레딧이 환불된다. "
+ "P2V 서버에는 F2 잡 삭제 API 가 없어 castad 기록만 지운다 "
+ "(P2V 쪽 산출물은 서버 자체 보존 주기를 따른다).",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "잡 없음"}},
+)
+async def delete_job(
+ job_id: int,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> dict:
+ _guard_enabled()
+ row = await f2_service.get_owned(session, job_id, current_user.user_uuid)
+ # 완성본을 이미 받았다면 환불 대상이 아니다 (실패분은 fail_job 이 이미 환불했고,
+ # refund 는 멱등이라 중복 호출돼도 안전하다)
+ if row.archived_at is None and row.status not in ("done", "archiving"):
+ await f2_service.refund_job(session, row, reason="포스터 스타일링 삭제(미완성)")
+ await session.delete(row)
+ await session.commit()
+ return {"id": job_id, "removed": True}
+
+
+def _error_of(row) -> P2vJobError | None:
+ if row.status != "failed" or not row.error:
+ return None
+ stage, _, detail = row.error.partition(": ")
+ return P2vJobError(stage=stage or "?", detail=detail or row.error)
+
+
+@router.get(
+ "/jobs/{job_id}",
+ response_model=P2vF2StatusResponse,
+ summary="스타일 변환 상태 (폴링)",
+ responses={401: {"description": "인증 실패"}, 404: {"description": "잡 없음"}},
+)
+async def get_job(
+ job_id: int,
+ background_tasks: BackgroundTasks,
+ current_user: User = Depends(get_current_user),
+ session: AsyncSession = Depends(get_session),
+) -> P2vF2StatusResponse:
+ _guard_enabled()
+ row = await f2_service.get_owned(session, job_id, current_user.user_uuid)
+
+ # 아카이브 완료 또는 종결 상태 — P2V 를 부를 필요가 없다 (잡이 지워졌을 수도 있다)
+ if row.archived_at is not None or (row.status == "failed") or not row.p2v_job_id:
+ artifacts = {}
+ if row.p2v_poster_url:
+ artifacts["image"] = row.p2v_poster_url
+ return P2vF2StatusResponse(
+ id=row.id,
+ status=row.status, # type: ignore[arg-type]
+ template_id=row.template_id,
+ error=_error_of(row),
+ artifacts=artifacts,
+ )
+
+ try:
+ p2v_job = await client.request_json("GET", f"/api/f2/jobs/{row.p2v_job_id}")
+ except P2vUpstreamError as e:
+ if e.status_code == 404:
+ # P2V 재기동 등으로 잡 소실 — 결과를 받을 길이 없으니 실패·환불
+ await f2_service.fail_job(
+ session, row, "서버 재시작으로 작업이 유실되었습니다. 크레딧을 환불했습니다"
+ )
+ await session.commit()
+ return P2vF2StatusResponse(
+ id=row.id, status="failed", template_id=row.template_id,
+ error=_error_of(row), artifacts={},
+ )
+ raise
+
+ await f2_service.sync_from_p2v(session, row, p2v_job)
+
+ status = row.status
+ if p2v_job.get("status") == "done" and row.archived_at is None:
+ # Blob 업로드 전까지는 진행 중으로 보인다 — URL 없는 done 창을 만들지 않는다
+ status = "archiving"
+ background_tasks.add_task(archive_service.archive_f2, row.id)
+ await session.commit()
+
+ return P2vF2StatusResponse(
+ id=row.id,
+ status=status, # type: ignore[arg-type]
+ stage=p2v_job.get("stage"),
+ stages=p2v_job.get("stages"),
+ template_id=row.template_id,
+ queue_size=p2v_job.get("queue_size"),
+ error=_error_of(row),
+ artifacts={}, # 결과 이미지는 아카이브 완료 후에만 내려간다
+ )
diff --git a/app/p2v/api/routers/v1/files.py b/app/p2v/api/routers/v1/files.py
new file mode 100644
index 0000000..6435207
--- /dev/null
+++ b/app/p2v/api/routers/v1/files.py
@@ -0,0 +1,81 @@
+# -*- coding: utf-8 -*-
+"""P2V 정적 파일 경량 프록시.
+
+결과물(영상·결과 이미지)은 Azure Blob 공개 URL 로 나가므로 여기를 지나지 않는다.
+이 프록시는 **작은 이미지 두 종류**만 중계한다:
+
+- regions/ — 검수용 영역분석 이미지. 사용자 포스터가 담기므로 로그인 필수.
+ (프론트는 authenticatedFetch → blob URL 로 ![]()
에 넣는다)
+- templates/, user_templates/ — 템플릿 썸네일. `![]()
` 는 인증 헤더를 못
+ 붙이고 내용도 공개 명화(퍼블릭 도메인)라 인증 없이 중계한다.
+
+인증 라우트를 분리한 이유(2026-08-26 리뷰 Critical 반영): 401 응답을 직접 만들면
+프로젝트 표준 포맷(detail={"code", "message"})과 어긋나 authenticatedFetch 의
+토큰 자동 갱신이 무력화되고 강제 로그아웃이 난다. get_current_user 의존성에
+맡기면 만료/누락 케이스 모두 표준 예외 포맷으로 나간다.
+
+render/·uploads/·f2/ 는 어느 라우트에도 없다 — 원본·결과물은 Blob 으로만 나간다.
+Range 미지원(대용량 스트리밍 용도가 아니다).
+"""
+
+from fastapi import APIRouter, Depends, Response
+
+from app.p2v.constants import PUBLIC_FILE_PREFIXES
+from app.p2v.exceptions import P2vDisabledError, P2vFileNotFoundError
+from app.p2v.services import client
+from app.user.dependencies.auth import get_current_user
+from app.user.models import User
+from app.utils.logger import get_logger
+from config import p2v_settings
+
+logger = get_logger("p2v")
+
+router = APIRouter(prefix="/p2v", tags=["P2V"])
+
+
+def _validate_path(path: str) -> None:
+ """경로 탈출 방어. uvicorn 이 라우팅 전에 percent-decoding 하므로
+ %2e%2e 우회도 여기서 걸린다 (2026-08-26 리뷰에서 검증됨)."""
+ if not p2v_settings.P2V_ENABLED:
+ raise P2vDisabledError()
+ if ".." in path or path.startswith("/") or "\\" in path:
+ raise P2vFileNotFoundError()
+
+
+async def _relay(path: str, cache: str) -> Response:
+ content, content_type = await client.download(f"/files/{path}")
+ return Response(
+ content=content,
+ media_type=content_type,
+ headers={"Cache-Control": cache},
+ )
+
+
+# 정적 경로가 더 구체적이므로 catch-all 보다 먼저 선언한다 (ssulbox 라우트 순서 관례)
+@router.get(
+ "/files/regions/{path:path}",
+ summary="검수용 영역분석 이미지 중계 (로그인 필수)",
+ responses={
+ 401: {"description": "인증 실패 (표준 인증 오류 포맷 — 토큰 갱신 대상)"},
+ 404: {"description": "없는 파일"},
+ },
+)
+async def proxy_region_file(
+ path: str,
+ current_user: User = Depends(get_current_user),
+) -> Response:
+ _validate_path(path)
+ return await _relay(f"regions/{path}", cache="private, max-age=300")
+
+
+@router.get(
+ "/files/{path:path}",
+ summary="P2V 공개 파일 중계 (템플릿 썸네일 전용)",
+ responses={404: {"description": "화이트리스트 밖이거나 없는 파일"}},
+)
+async def proxy_public_file(path: str) -> Response:
+ _validate_path(path)
+ if not path.startswith(PUBLIC_FILE_PREFIXES):
+ # render/·uploads/ 등 — 존재 여부를 노출하지 않는다
+ raise P2vFileNotFoundError()
+ return await _relay(path, cache="public, max-age=3600")
diff --git a/app/p2v/constants.py b/app/p2v/constants.py
new file mode 100644
index 0000000..49bd1b9
--- /dev/null
+++ b/app/p2v/constants.py
@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+"""P2V 모듈 상수."""
+
+from typing import Final
+
+# =============================================================================
+# 크레딧 원장 멱등 키
+# =============================================================================
+#: credit_transaction.job_type 값. (job_type, job_ref, type) 유니크로 중복 차감을 막는다.
+#: job_ref 는 p2v_video.id / p2v_poster.id 의 문자열이다 (P2V 서버 잡 id 가 아니다).
+JOB_TYPE_P2V_F1: Final[str] = "p2v_f1"
+JOB_TYPE_P2V_F2: Final[str] = "p2v_f2"
+
+
+# =============================================================================
+# 잡 상태
+# =============================================================================
+# P2V 서버가 보고하는 상태에 castad 전용 'archiving'(Blob 업로드 중)을 더한 축.
+STATUS_QUEUED: Final[str] = "queued"
+STATUS_RUNNING: Final[str] = "running"
+STATUS_AWAITING_REVIEW: Final[str] = "awaiting_review"
+STATUS_ARCHIVING: Final[str] = "archiving"
+STATUS_DONE: Final[str] = "done"
+STATUS_FAILED: Final[str] = "failed"
+
+#: 고아 스윕 대상 — 이 상태로 P2V_ORPHAN_TIMEOUT_HOURS 를 넘기면 환불·실패 처리한다.
+ORPHAN_STATUSES: Final[frozenset[str]] = frozenset(
+ {STATUS_QUEUED, STATUS_RUNNING, STATUS_AWAITING_REVIEW}
+)
+
+
+# =============================================================================
+# 정적 파일 프록시 화이트리스트
+# =============================================================================
+# P2V 의 /files/* 마운트 중 castad 프록시가 인증 없이 중계해도 되는 것만 나열한다.
+# render/·uploads/·f2/ 는 결과물·원본이라 Blob 으로만 나간다 — 프록시 금지.
+# 검수용 regions/ 는 별도 라우트(/p2v/files/regions/*)가 로그인 필수로 중계한다.
+#: 인증 없이 중계 (템플릿 썸네일 — ![]()
는 헤더를 못 붙이고, 내용도 공개 명화다)
+PUBLIC_FILE_PREFIXES: Final[tuple[str, ...]] = ("templates/", "user_templates/")
diff --git a/app/p2v/exceptions.py b/app/p2v/exceptions.py
new file mode 100644
index 0000000..3b606b5
--- /dev/null
+++ b/app/p2v/exceptions.py
@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+"""P2V 모듈 도메인 예외.
+
+SsulboxException 과 같은 (message, status_code, code) 형태다.
+`code` 는 프론트엔드가 분기에 쓰므로 값을 바꾸면 안 된다.
+전역 핸들러 등록: app/core/exceptions.py add_exception_handlers().
+"""
+
+from fastapi import status
+
+
+class P2vException(Exception):
+ """P2V 기본 예외"""
+
+ def __init__(
+ self,
+ message: str,
+ status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
+ code: str = "P2V_ERROR",
+ ):
+ self.message = message
+ self.status_code = status_code
+ self.code = code
+ super().__init__(self.message)
+
+
+# =============================================================================
+# 기능/연결 상태
+# =============================================================================
+
+
+class P2vDisabledError(P2vException):
+ """P2V 기능이 설정으로 꺼져 있음"""
+
+ def __init__(self):
+ super().__init__(
+ message="포스터 생성 기능이 현재 비활성화되어 있습니다.",
+ status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+ code="P2V_DISABLED",
+ )
+
+
+class P2vUnavailableError(P2vException):
+ """P2V 서버 연결 실패 (다운, 타임아웃, 접근 키 불일치)"""
+
+ def __init__(self, detail: str = ""):
+ message = "포스터 생성 서버에 연결할 수 없습니다."
+ if detail:
+ message += f" ({detail})"
+ super().__init__(
+ message=message,
+ status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+ code="P2V_UNAVAILABLE",
+ )
+
+
+class P2vUpstreamError(P2vException):
+ """P2V 서버가 4xx/5xx 를 돌려줌 — detail 과 상태 코드를 그대로 중계한다"""
+
+ def __init__(self, message: str, status_code: int):
+ super().__init__(
+ message=message,
+ status_code=status_code,
+ code="P2V_UPSTREAM_ERROR",
+ )
+
+
+# =============================================================================
+# 잡 관련
+# =============================================================================
+
+
+class P2vJobNotFoundError(P2vException):
+ """잡을 찾을 수 없음 (없거나 남의 것 — 존재 여부를 노출하지 않는다)"""
+
+ def __init__(self):
+ super().__init__(
+ message="생성 요청을 찾을 수 없습니다.",
+ status_code=status.HTTP_404_NOT_FOUND,
+ code="P2V_JOB_NOT_FOUND",
+ )
+
+
+class P2vInvalidStateError(P2vException):
+ """현재 상태에서 허용되지 않는 조작 (예: 실패하지 않은 잡의 재시도)"""
+
+ def __init__(self, detail: str):
+ super().__init__(
+ message=detail,
+ status_code=status.HTTP_409_CONFLICT,
+ code="P2V_INVALID_STATE",
+ )
+
+
+class P2vFileNotFoundError(P2vException):
+ """프록시 화이트리스트 밖이거나 존재하지 않는 파일"""
+
+ def __init__(self):
+ super().__init__(
+ message="파일을 찾을 수 없습니다.",
+ status_code=status.HTTP_404_NOT_FOUND,
+ code="P2V_FILE_NOT_FOUND",
+ )
+
+
+class P2vUploadTooLargeError(P2vException):
+ """업로드 파일이 상한을 초과"""
+
+ def __init__(self, limit_bytes: int):
+ super().__init__(
+ message=f"{limit_bytes // (1024 * 1024)}MB 이하만 업로드할 수 있습니다.",
+ status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+ code="P2V_UPLOAD_TOO_LARGE",
+ )
diff --git a/app/p2v/models.py b/app/p2v/models.py
new file mode 100644
index 0000000..3aa42e4
--- /dev/null
+++ b/app/p2v/models.py
@@ -0,0 +1,265 @@
+# -*- coding: utf-8 -*-
+"""P2V 잡 모델.
+
+⚠️ 이 프로젝트는 Alembic 마이그레이션을 쓰지 않는다. 스키마 변경은
+docs/database-schema/migration_2026-08-25_p2v.sql 을 수동 실행한다.
+
+테이블명은 산출물 종류를 따른다(p2v_video / p2v_poster). 클래스명의 F1/F2 는
+P2V 서버 API 경로(/api/f1/*, /api/f2/*)와의 대응 관계를 가리킨다.
+
+여기에는 P2V 서버가 못 가진 것만 둔다 — 소유권, 크레딧 앵커, Blob URL,
+그리고 "P2V 없이도 이력이 완결"되기 위한 확정 메타데이터 스냅샷.
+진행 중 잡의 상세(stages, 검수 중간값)는 P2V 가 진실 공급원이다.
+"""
+
+from datetime import datetime
+from decimal import Decimal
+from typing import Optional
+
+from sqlalchemy import (
+ JSON,
+ BigInteger,
+ DateTime,
+ ForeignKey,
+ Index,
+ Integer,
+ Numeric,
+ String,
+ UniqueConstraint,
+ func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database.session import Base
+
+
+class P2vF1Job(Base):
+ """무빙 포스터(포스터→영상) 잡 — 테이블 p2v_video"""
+
+ __tablename__ = "p2v_video"
+ __table_args__ = (
+ UniqueConstraint("p2v_job_id", name="uq_p2v_video_ref"),
+ Index("idx_p2v_video_user_created", "user_uuid", "created_at"),
+ Index("idx_p2v_video_status", "status"),
+ {
+ "mysql_engine": "InnoDB",
+ "mysql_charset": "utf8mb4",
+ "mysql_collate": "utf8mb4_unicode_ci",
+ },
+ )
+
+ id: Mapped[int] = mapped_column(
+ BigInteger,
+ primary_key=True,
+ autoincrement=True,
+ comment="고유 식별자 (크레딧 원장 job_ref 앵커)",
+ )
+ user_uuid: Mapped[str] = mapped_column(
+ String(36),
+ ForeignKey("user.user_uuid", ondelete="CASCADE"),
+ nullable=False,
+ comment="생성 요청한 사용자 UUID",
+ )
+ p2v_job_id: Mapped[Optional[str]] = mapped_column(
+ String(64),
+ nullable=True,
+ comment="P2V 서버가 발급한 잡 id(F1). 서버 호출 성공 후 채워진다",
+ )
+ name: Mapped[Optional[str]] = mapped_column(
+ String(100),
+ nullable=True,
+ comment="사용자가 입력한 행사명 (비우면 서버가 포스터에서 추출)",
+ )
+ status: Mapped[str] = mapped_column(
+ String(20),
+ nullable=False,
+ default="queued",
+ server_default="queued",
+ comment="상태 (queued/running/awaiting_review/archiving/done/failed)",
+ )
+ credit_amount: Mapped[int] = mapped_column(
+ Integer,
+ nullable=False,
+ default=0,
+ server_default="0",
+ comment="차감한 크레딧 수량 (환불 금액 결정)",
+ )
+
+ # ==========================================================================
+ # 산출물 (Blob 업로드 후 채워짐). 명명은 ssul_content 관례:
+ # poster_url = 커버/썸네일(og:image 역할), 원본은 source_*, 결과물은 p2v_ 접두.
+ # ==========================================================================
+ p2v_video_url: Mapped[Optional[str]] = mapped_column(
+ String(500), nullable=True, comment="완성 영상 Blob URL"
+ )
+ poster_url: Mapped[Optional[str]] = mapped_column(
+ String(500),
+ nullable=True,
+ comment="썸네일 Blob URL (SNS 공유 og:image 역할, ssul_content.poster_url 관례)",
+ )
+ source_image_url: Mapped[Optional[str]] = mapped_column(
+ String(500), nullable=True, comment="원본 업로드 포스터 Blob URL"
+ )
+
+ # ==========================================================================
+ # 확정 메타데이터 (검수에서 사용자가 수정한 최종본만 보관)
+ # ==========================================================================
+ event_name: Mapped[Optional[str]] = mapped_column(
+ String(200), nullable=True, comment="행사명"
+ )
+ date_text: Mapped[Optional[str]] = mapped_column(
+ String(100), nullable=True, comment="일시 표기"
+ )
+ place: Mapped[Optional[str]] = mapped_column(
+ String(200), nullable=True, comment="장소"
+ )
+ keywords: Mapped[Optional[list]] = mapped_column(
+ JSON, nullable=True, comment="키워드 목록"
+ )
+ narration: Mapped[Optional[list]] = mapped_column(
+ JSON, nullable=True, comment="나레이션 3문장"
+ )
+ duration: Mapped[Optional[Decimal]] = mapped_column(
+ Numeric(6, 2), nullable=True, comment="완성 영상 길이(초)"
+ )
+
+ error: Mapped[Optional[str]] = mapped_column(
+ String(1000), nullable=True, comment="실패 사유 (스테이지 + detail)"
+ )
+ archived_at: Mapped[Optional[datetime]] = mapped_column(
+ DateTime,
+ nullable=True,
+ comment="Blob 업로드 완료 일시. NULL 이면 아직 P2V 에만 있음",
+ )
+ created_at: Mapped[datetime] = mapped_column(
+ DateTime, nullable=False, server_default=func.now(), comment="생성 일시"
+ )
+ updated_at: Mapped[datetime] = mapped_column(
+ DateTime,
+ nullable=False,
+ server_default=func.now(),
+ onupdate=func.now(),
+ comment="수정 일시",
+ )
+
+ def __repr__(self) -> str:
+ return (
+ f""
+ )
+
+
+class P2vF2Job(Base):
+ """포스터 스타일링 잡 — 테이블 p2v_poster"""
+
+ __tablename__ = "p2v_poster"
+ __table_args__ = (
+ UniqueConstraint("p2v_job_id", name="uq_p2v_poster_ref"),
+ Index("idx_p2v_poster_user_created", "user_uuid", "created_at"),
+ Index("idx_p2v_poster_status", "status"),
+ Index("idx_p2v_poster_source", "source_video_id"),
+ {
+ "mysql_engine": "InnoDB",
+ "mysql_charset": "utf8mb4",
+ "mysql_collate": "utf8mb4_unicode_ci",
+ },
+ )
+
+ id: Mapped[int] = mapped_column(
+ BigInteger,
+ primary_key=True,
+ autoincrement=True,
+ comment="고유 식별자 (크레딧 원장 job_ref 앵커)",
+ )
+ user_uuid: Mapped[str] = mapped_column(
+ String(36),
+ ForeignKey("user.user_uuid", ondelete="CASCADE"),
+ nullable=False,
+ comment="생성 요청한 사용자 UUID",
+ )
+ p2v_job_id: Mapped[Optional[str]] = mapped_column(
+ String(64), nullable=True, comment="P2V 서버가 발급한 잡 id(F2)"
+ )
+ name: Mapped[Optional[str]] = mapped_column(
+ String(100),
+ nullable=True,
+ comment="포스터 이름 (업로드 파일명에서 추출). 내 콘텐츠 카드 제목에 쓴다",
+ )
+ source_video_id: Mapped[Optional[int]] = mapped_column(
+ BigInteger,
+ ForeignKey("p2v_video.id", ondelete="SET NULL"),
+ nullable=True,
+ comment="원본 p2v_video 잡 (P2V source_slug 대응). 독립 잡이면 NULL",
+ )
+ status: Mapped[str] = mapped_column(
+ String(20),
+ nullable=False,
+ default="queued",
+ server_default="queued",
+ comment="상태 (queued/running/archiving/done/failed)",
+ )
+ credit_amount: Mapped[int] = mapped_column(
+ Integer,
+ nullable=False,
+ default=0,
+ server_default="0",
+ comment="차감한 크레딧 수량",
+ )
+
+ # ==========================================================================
+ # 입력 스냅샷 (템플릿이 삭제돼도 이력이 남아야 한다)
+ # ==========================================================================
+ template_id: Mapped[str] = mapped_column(
+ String(64), nullable=False, comment="사용한 스타일 템플릿 id"
+ )
+ template_name: Mapped[Optional[str]] = mapped_column(
+ String(100), nullable=True, comment="템플릿 이름 스냅샷"
+ )
+ license: Mapped[Optional[str]] = mapped_column(
+ String(20),
+ nullable=True,
+ comment=(
+ "템플릿 배포 등급 (public-domain/internal-only/user-uploaded). "
+ "외부 공개 가부 판단"
+ ),
+ )
+ format: Mapped[str] = mapped_column(
+ String(16),
+ nullable=False,
+ default="poster",
+ server_default="poster",
+ comment="출력 포맷 (poster/story/feed/square)",
+ )
+
+ # ==========================================================================
+ # 산출물 (명명은 p2v_video 와 동일 관례)
+ # ==========================================================================
+ p2v_poster_url: Mapped[Optional[str]] = mapped_column(
+ String(500), nullable=True, comment="스타일 변환 결과 Blob URL"
+ )
+ source_image_url: Mapped[Optional[str]] = mapped_column(
+ String(500), nullable=True, comment="원본 업로드 포스터 Blob URL"
+ )
+
+ error: Mapped[Optional[str]] = mapped_column(
+ String(1000), nullable=True, comment="실패 사유"
+ )
+ archived_at: Mapped[Optional[datetime]] = mapped_column(
+ DateTime, nullable=True, comment="Blob 업로드 완료 일시"
+ )
+ created_at: Mapped[datetime] = mapped_column(
+ DateTime, nullable=False, server_default=func.now(), comment="생성 일시"
+ )
+ updated_at: Mapped[datetime] = mapped_column(
+ DateTime,
+ nullable=False,
+ server_default=func.now(),
+ onupdate=func.now(),
+ comment="수정 일시",
+ )
+
+ def __repr__(self) -> str:
+ return (
+ f""
+ )
diff --git a/app/p2v/schemas/__init__.py b/app/p2v/schemas/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/p2v/schemas/p2v_schema.py b/app/p2v/schemas/p2v_schema.py
new file mode 100644
index 0000000..f888ece
--- /dev/null
+++ b/app/p2v/schemas/p2v_schema.py
@@ -0,0 +1,190 @@
+# -*- coding: utf-8 -*-
+"""P2V 요청/응답 스키마.
+
+응답 형태는 P2V 서버의 잡 JSON 과 최대한 호환되게 유지한다 — 프론트엔드
+(o2o-castad-frontend p2vApi.ts) 가 이미 그 형태(artifacts 딕셔너리, stages 등)에
+맞춰 작성돼 있어, 필드명을 바꾸면 P2V 직접 호출 → castad 프록시 전환 때
+갈아엎어야 할 코드가 배로 늘기 때문이다. 단 `id` 는 castad 잡 id(정수)다 —
+P2V 잡 id 는 절대 밖으로 내보내지 않는다.
+
+검증 범위·기본값은 여기서만 강제한다. DB 모델에 두면 진실이 두 곳에 생긴다.
+"""
+
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field
+
+F1StatusLiteral = Literal[
+ "queued", "running", "awaiting_review", "archiving", "done", "failed"
+]
+F2StatusLiteral = Literal["queued", "running", "archiving", "done", "failed"]
+FormatLiteral = Literal["poster", "story", "feed", "square"]
+
+
+# =============================================================================
+# 공통
+# =============================================================================
+
+
+class P2vJobCreateResponse(BaseModel):
+ """생성 요청 접수. `id` 로 폴링한다."""
+
+ id: int = Field(..., description="castad 잡 ID (P2V 서버 식별자가 아니다)")
+ status: F1StatusLiteral = Field(..., description="접수 직후 상태")
+ poll_interval_seconds: int = Field(..., description="권장 폴링 간격(초)")
+
+
+class P2vJobError(BaseModel):
+ """실패 정보 — P2V 의 {stage, detail} 형태를 그대로 중계한다"""
+
+ stage: str = Field(..., description="실패한 스테이지")
+ detail: str = Field(..., description="실패 사유")
+
+
+class P2vMetadata(BaseModel):
+ """행사 메타데이터 (검수 대상)"""
+
+ event_name: str = Field(default="", description="행사명")
+ date_text: str = Field(default="", description="일시 표기")
+ place: str = Field(default="", description="장소")
+ category: Optional[str] = Field(None, description="분류")
+ keywords: Optional[list[str]] = Field(None, description="키워드 목록")
+ region_guess: Optional[str] = Field(None, description="추정 지역")
+
+
+class P2vDeleteResponse(BaseModel):
+ """삭제 결과"""
+
+ id: int = Field(..., description="삭제한 castad 잡 ID")
+ removed: bool = Field(..., description="삭제 여부")
+
+
+# =============================================================================
+# F1 — 무빙 포스터
+# =============================================================================
+
+
+class P2vF1StatusResponse(BaseModel):
+ """F1 잡 상태 (폴링 응답).
+
+ 진행 중에는 P2V 실시간 값을, archived 이후에는 castad DB 미러를 내려준다.
+ artifacts 값은 전부 Azure Blob 공개 URL 또는 castad 프록시 경로(/p2v/files/...)다.
+ """
+
+ id: int = Field(..., description="castad 잡 ID")
+ status: F1StatusLiteral = Field(..., description="진행 상태")
+ stage: Optional[str] = Field(None, description="현재 스테이지 (진행 중에만)")
+ stages: Optional[dict] = Field(
+ None, description="스테이지별 상태 (P2V 원형 그대로 — 진행 중에만)"
+ )
+ name: Optional[str] = Field(None, description="행사명")
+ narration: Optional[list[str]] = Field(None, description="나레이션 3문장 (검수 대상)")
+ metadata: Optional[P2vMetadata] = Field(None, description="행사 메타데이터 (검수 대상)")
+ motion_elements: Optional[list[str]] = Field(None, description="채택된 모션 요소")
+ duration: Optional[float] = Field(None, description="완성 영상 길이(초)")
+ queue_size: Optional[int] = Field(None, description="P2V 대기열 크기 (진행 중에만)")
+ error: Optional[P2vJobError] = Field(None, description="실패 정보")
+ artifacts: dict[str, str] = Field(
+ default_factory=dict,
+ description="산출물 URL — video/thumbnail 은 Blob, check_jpg 는 프록시 경로",
+ )
+
+
+class NarrationUpdateRequest(BaseModel):
+ """검수 중 나레이션 수정"""
+
+ narration: list[str] = Field(
+ ..., min_length=1, max_length=5, description="수정한 나레이션 문장 목록"
+ )
+
+
+class MetadataUpdateRequest(BaseModel):
+ """검수 중 행사 메타데이터 수정"""
+
+ event_name: str = Field(..., max_length=200, description="행사명")
+ date_text: str = Field(default="", max_length=100, description="일시 표기")
+ place: str = Field(..., max_length=200, description="장소")
+
+
+class ApproveRequest(BaseModel):
+ """검수 승인/재시도. motions 를 주면 모션 선정을 덮어쓴다"""
+
+ motions: Optional[list[str]] = Field(
+ None, description="모션 키 목록 (None 이면 서버 선정 유지)"
+ )
+
+
+# =============================================================================
+# F2 — 포스터 스타일링
+# =============================================================================
+
+
+class P2vF2StatusResponse(BaseModel):
+ """F2 잡 상태 (폴링 응답)"""
+
+ id: int = Field(..., description="castad 잡 ID")
+ status: F2StatusLiteral = Field(..., description="진행 상태")
+ stage: Optional[str] = Field(None, description="현재 스테이지 (진행 중에만)")
+ stages: Optional[dict] = Field(None, description="스테이지별 상태 (진행 중에만)")
+ template_id: Optional[str] = Field(None, description="사용한 템플릿 id")
+ queue_size: Optional[int] = Field(None, description="P2V 대기열 크기 (진행 중에만)")
+ error: Optional[P2vJobError] = Field(None, description="실패 정보")
+ artifacts: dict[str, str] = Field(
+ default_factory=dict, description="산출물 URL — image 는 Blob 공개 URL"
+ )
+
+
+class F2TemplateResponse(BaseModel):
+ """스타일 템플릿 (P2V 목록 중계 — thumb_url 은 castad 프록시 경로로 재작성됨)"""
+
+ id: str = Field(..., description="템플릿 id")
+ name_ko: str = Field(..., description="한국어 이름")
+ thumb_url: str = Field(..., description="썸네일 (castad 프록시 경로 /p2v/files/...)")
+ category: str = Field(..., description="카테고리 id")
+ license: str = Field(..., description="배포 등급 (public-domain/internal-only/user-uploaded)")
+ attribution: str = Field(default="", description="출처 표기")
+ license_note: str = Field(default="", description="등급 설명")
+ removable: bool = Field(default=False, description="사용자 레퍼런스 여부 (삭제 가능)")
+ small_ref: bool = Field(default=False, description="저해상 레퍼런스 경고 여부")
+
+
+class F2CategoryResponse(BaseModel):
+ """템플릿 카테고리"""
+
+ id: str = Field(..., description="카테고리 id")
+ label: str = Field(..., description="표시 이름")
+
+
+class F2FormatResponse(BaseModel):
+ """출력 포맷"""
+
+ id: str = Field(..., description="포맷 id")
+ label: str = Field(..., description="표시 이름")
+
+
+class F2UploadHintResponse(BaseModel):
+ """사용자 레퍼런스 업로드 안내"""
+
+ enabled: bool = Field(..., description="업로드 허용 여부")
+ min_long_edge: int = Field(..., description="권장 최소 긴 변(px)")
+ ref_long_edge: Optional[int] = Field(None, description="분석 시 리사이즈 기준(px)")
+
+
+__all__ = [
+ "F1StatusLiteral",
+ "F2StatusLiteral",
+ "FormatLiteral",
+ "P2vJobCreateResponse",
+ "P2vJobError",
+ "P2vMetadata",
+ "P2vDeleteResponse",
+ "P2vF1StatusResponse",
+ "NarrationUpdateRequest",
+ "MetadataUpdateRequest",
+ "ApproveRequest",
+ "P2vF2StatusResponse",
+ "F2TemplateResponse",
+ "F2CategoryResponse",
+ "F2FormatResponse",
+ "F2UploadHintResponse",
+]
diff --git a/app/p2v/services/__init__.py b/app/p2v/services/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/p2v/services/archive_service.py b/app/p2v/services/archive_service.py
new file mode 100644
index 0000000..3a3b3be
--- /dev/null
+++ b/app/p2v/services/archive_service.py
@@ -0,0 +1,171 @@
+# -*- coding: utf-8 -*-
+"""P2V 산출물 Azure Blob 아카이빙.
+
+P2V 서버는 콜백이 없으므로, 폴링 핸들러가 'done' 을 처음 관측했을 때
+FastAPI BackgroundTasks 로 이 모듈을 태운다 (짧은 async I/O — 스레드 워커는 과함).
+
+업로더는 castad `AzureBlobUploader` 를 그대로 재사용한다 (ssulbox blob_service 와
+같은 이유 — 같은 Azure 계정에 업로더를 두 벌 둘 이유가 없다).
+경로: {user_uuid}/{P2V_BLOB_PREFIX}-f1-{id}/video/... — castad·썰박스와 분리된다.
+
+중복 실행 방지: try_claim_archiving 낙관적 잠금이 첫 태스크에게만 착수권을 준다.
+실패 복구: 예외 시 status 를 'done'(archived_at NULL)으로 되돌린다 — 다음 폴링이
+다시 선점해 재시도한다. 크래시로 'archiving' 에 갇힌 행은 기동 시 스윕이 되돌린다.
+"""
+
+from app.database.session import BackgroundSessionLocal
+from app.p2v.constants import STATUS_DONE
+from app.p2v.models import P2vF1Job, P2vF2Job
+from app.p2v.services import client, f1_service, f2_service
+from app.utils.logger import get_logger
+from app.utils.upload_blob_as_request import AzureBlobUploader
+from config import azure_blob_settings, p2v_settings
+
+logger = get_logger("p2v")
+
+#: 설정되지 않았을 때의 플레이스홀더 (ssulbox blob_service 와 동일 판정)
+_PLACEHOLDER_SAS = {"", "your-sas-token", "none"}
+
+
+def blob_enabled() -> bool:
+ """Blob 업로드가 가능한 상태인지. 비활성이면 URL 없이 done 처리된다(개발 환경)."""
+ token = (azure_blob_settings.AZURE_BLOB_SAS_TOKEN or "").strip()
+ return token.lower() not in _PLACEHOLDER_SAS
+
+
+def _uploader(user_uuid: str, kind: str, row_id: int) -> AzureBlobUploader:
+ return AzureBlobUploader(
+ user_uuid=user_uuid,
+ task_id=f"{p2v_settings.P2V_BLOB_PREFIX}-{kind}-{row_id}",
+ )
+
+
+async def upload_source_image(
+ kind: str, row_id: int, user_uuid: str, content: bytes, filename: str
+) -> None:
+ """원본 업로드 포스터를 Blob 에 보관한다 (잡 생성 직후 백그라운드).
+
+ 생성 요청이 이미 파일 바이트를 들고 있으므로, 나중에 P2V 에서 되받아오는
+ 대신 이 시점에 바로 올린다. 실패해도 잡 진행에는 영향을 주지 않는다(로그만).
+ """
+ if not blob_enabled():
+ return
+ try:
+ uploader = _uploader(user_uuid, kind, row_id)
+ ok = await uploader.upload_image_bytes(content, filename or "source.png")
+ if not ok:
+ logger.warning(f"[archive] 원본 업로드 실패 {kind} id={row_id}")
+ return
+ url = uploader.public_url
+ model = P2vF1Job if kind == "f1" else P2vF2Job
+ async with BackgroundSessionLocal() as session:
+ row = await session.get(model, row_id)
+ if row is not None:
+ row.source_image_url = url
+ await session.commit()
+ except Exception as e:
+ # 원본 보관은 부수 기능 — 실패가 잡을 죽이면 안 된다
+ logger.warning(f"[archive] 원본 업로드 예외 {kind} id={row_id}: {e}")
+
+
+async def _revert_claim(kind: str, row_id: int) -> None:
+ """아카이빙 실패 → 재시도 가능 상태('done', archived_at NULL)로 복구."""
+ model = P2vF1Job if kind == "f1" else P2vF2Job
+ try:
+ async with BackgroundSessionLocal() as session:
+ row = await session.get(model, row_id)
+ if row is not None and row.archived_at is None:
+ row.status = STATUS_DONE
+ await session.commit()
+ except Exception as e:
+ logger.error(f"[archive] 복구 실패 {kind} id={row_id}: {e} — 기동 스윕이 처리한다")
+
+
+async def archive_f1(row_id: int) -> None:
+ """F1 완성 영상·썸네일을 Blob 으로 옮기고 done 전이한다."""
+ async with BackgroundSessionLocal() as session:
+ if not await f1_service.try_claim_archiving(session, row_id):
+ return # 다른 폴링이 이미 집어갔다
+ await session.commit()
+
+ try:
+ async with BackgroundSessionLocal() as session:
+ row = await session.get(P2vF1Job, row_id)
+ if row is None or not row.p2v_job_id:
+ return
+
+ video_url = None
+ thumb_url = None
+ if blob_enabled():
+ p2v_job = await client.request_json(
+ "GET", f"/api/f1/jobs/{row.p2v_job_id}"
+ )
+ artifacts = p2v_job.get("artifacts") or {}
+ uploader = _uploader(row.user_uuid, "f1", row_id)
+
+ if artifacts.get("video"):
+ content, _ = await client.download(artifacts["video"])
+ if await uploader.upload_video_bytes(content, "video.mp4"):
+ video_url = uploader.public_url
+ else:
+ raise RuntimeError("영상 Blob 업로드 실패")
+ if artifacts.get("thumbnail"):
+ content, _ = await client.download(artifacts["thumbnail"])
+ if await uploader.upload_image_bytes(content, "thumbnail.jpg"):
+ thumb_url = uploader.public_url
+ else:
+ # 조용히 넘기면 poster_url(og:image)이 영구 결손된다 —
+ # 영상과 같이 실패시켜 다음 폴링이 아카이빙을 재시도하게 한다
+ raise RuntimeError("썸네일 Blob 업로드 실패")
+ # 확정 메타데이터도 이 시점에 마지막으로 미러링해 둔다
+ f1_service.sync_from_p2v(row, p2v_job)
+ else:
+ logger.warning(
+ f"[archive] Blob 비활성 — f1 id={row_id} URL 없이 done 처리 (개발 환경)"
+ )
+
+ await f1_service.finish_archive(
+ session, row, video_url=video_url, thumbnail_url=thumb_url
+ )
+ await session.commit()
+ except Exception as e:
+ logger.error(f"[archive] f1 id={row_id} 아카이빙 실패: {type(e).__name__}: {e}")
+ await _revert_claim("f1", row_id)
+
+
+async def archive_f2(row_id: int) -> None:
+ """F2 결과 이미지를 Blob 으로 옮기고 done 전이한다."""
+ async with BackgroundSessionLocal() as session:
+ if not await f2_service.try_claim_archiving(session, row_id):
+ return
+ await session.commit()
+
+ try:
+ async with BackgroundSessionLocal() as session:
+ row = await session.get(P2vF2Job, row_id)
+ if row is None or not row.p2v_job_id:
+ return
+
+ image_url = None
+ if blob_enabled():
+ p2v_job = await client.request_json(
+ "GET", f"/api/f2/jobs/{row.p2v_job_id}"
+ )
+ artifacts = p2v_job.get("artifacts") or {}
+ if artifacts.get("image"):
+ content, _ = await client.download(artifacts["image"])
+ uploader = _uploader(row.user_uuid, "f2", row_id)
+ if await uploader.upload_image_bytes(content, "result.png"):
+ image_url = uploader.public_url
+ else:
+ raise RuntimeError("결과 이미지 Blob 업로드 실패")
+ else:
+ logger.warning(
+ f"[archive] Blob 비활성 — f2 id={row_id} URL 없이 done 처리 (개발 환경)"
+ )
+
+ await f2_service.finish_archive(session, row, image_url=image_url)
+ await session.commit()
+ except Exception as e:
+ logger.error(f"[archive] f2 id={row_id} 아카이빙 실패: {type(e).__name__}: {e}")
+ await _revert_claim("f2", row_id)
diff --git a/app/p2v/services/client.py b/app/p2v/services/client.py
new file mode 100644
index 0000000..df5c2bf
--- /dev/null
+++ b/app/p2v/services/client.py
@@ -0,0 +1,118 @@
+# -*- coding: utf-8 -*-
+"""P2V 서버 HTTP 클라이언트.
+
+httpx.AsyncClient 를 모듈 싱글턴으로 유지한다 (creatomate.py 의 _shared_client 패턴).
+모든 요청에 X-P2V-Key 를 실어 보내고, 연결 실패·업스트림 오류를 도메인 예외로 바꾼다.
+"""
+
+from typing import Any, Optional
+
+import httpx
+
+from app.p2v.exceptions import P2vUnavailableError, P2vUpstreamError
+from app.utils.logger import get_logger
+from config import p2v_settings
+
+logger = get_logger("p2v")
+
+#: 기본 타임아웃. 잡 생성(파일 업로드)·레퍼런스 분석(동기 10초 안팎)을 견디도록
+#: read 를 넉넉히 잡는다. 폴링 GET 은 서버가 즉시 답하므로 문제되지 않는다.
+_DEFAULT_TIMEOUT = httpx.Timeout(10.0, read=120.0)
+#: 결과물 다운로드용 (영상 수십 MB)
+_DOWNLOAD_TIMEOUT = httpx.Timeout(10.0, read=300.0)
+
+_client: Optional[httpx.AsyncClient] = None
+
+
+def _headers() -> dict[str, str]:
+ key = (p2v_settings.P2V_ACCESS_KEY or "").strip()
+ return {"X-P2V-Key": key} if key else {}
+
+
+def get_client() -> httpx.AsyncClient:
+ """싱글턴 AsyncClient. 커넥션 풀을 프로세스에서 공유한다."""
+ global _client
+ if _client is None or _client.is_closed:
+ _client = httpx.AsyncClient(
+ base_url=p2v_settings.P2V_BASE_URL.rstrip("/"),
+ headers=_headers(),
+ timeout=_DEFAULT_TIMEOUT,
+ limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
+ )
+ return _client
+
+
+async def close_client() -> None:
+ """앱 종료 시 커넥션 풀 정리 (lifespan shutdown 에서 호출)"""
+ global _client
+ if _client is not None and not _client.is_closed:
+ await _client.aclose()
+ _client = None
+
+
+def _raise_for_status(resp: httpx.Response) -> None:
+ """업스트림 4xx/5xx 를 도메인 예외로 변환한다.
+
+ - 401: castad 의 접근 키 문제다. 프론트에 401 을 흘리면 "castad 로그인 만료"로
+ 오독되므로 503 으로 바꾼다.
+ - 그 외: P2V 의 detail 과 상태 코드를 그대로 중계한다 (조용한 fallback 금지).
+ """
+ if resp.status_code < 400:
+ return
+ if resp.status_code == 401:
+ logger.error("[p2v] 접근 키 거부 — P2V_ACCESS_KEY 설정을 확인할 것")
+ raise P2vUnavailableError("접근 키가 유효하지 않습니다")
+ try:
+ detail = resp.json().get("detail", "")
+ except Exception:
+ detail = resp.text[:300]
+ raise P2vUpstreamError(str(detail) or f"P2V 오류 {resp.status_code}", resp.status_code)
+
+
+async def request_json(
+ method: str,
+ path: str,
+ *,
+ data: Optional[dict] = None,
+ files: Optional[dict] = None,
+ json_body: Optional[dict] = None,
+) -> Any:
+ """P2V API 호출 → JSON 응답.
+
+ Args:
+ method: HTTP 메서드
+ path: /api/... 경로
+ data: form 필드 (multipart 일 때 files 와 함께)
+ files: httpx files 딕셔너리 {"poster": (name, bytes, content_type)}
+ json_body: JSON 바디
+
+ Raises:
+ P2vUnavailableError: 연결 실패·타임아웃·접근 키 거부
+ P2vUpstreamError: P2V 가 4xx/5xx 를 반환
+ """
+ try:
+ resp = await get_client().request(
+ method, path, data=data, files=files, json=json_body
+ )
+ except httpx.HTTPError as e:
+ logger.warning(f"[p2v] {method} {path} 연결 실패: {type(e).__name__}: {e}")
+ raise P2vUnavailableError(type(e).__name__)
+ _raise_for_status(resp)
+ if not resp.content:
+ return None
+ return resp.json()
+
+
+async def download(path: str) -> tuple[bytes, str]:
+ """P2V 정적 파일(/files/...)을 받아온다.
+
+ Returns:
+ (바이트, Content-Type)
+ """
+ try:
+ resp = await get_client().get(path, timeout=_DOWNLOAD_TIMEOUT)
+ except httpx.HTTPError as e:
+ logger.warning(f"[p2v] GET {path} 다운로드 실패: {type(e).__name__}: {e}")
+ raise P2vUnavailableError(type(e).__name__)
+ _raise_for_status(resp)
+ return resp.content, resp.headers.get("content-type", "application/octet-stream")
diff --git a/app/p2v/services/f1_service.py b/app/p2v/services/f1_service.py
new file mode 100644
index 0000000..a6c7566
--- /dev/null
+++ b/app/p2v/services/f1_service.py
@@ -0,0 +1,218 @@
+# -*- coding: utf-8 -*-
+"""F1(무빙 포스터) 잡 라이프사이클 + 크레딧.
+
+모든 함수는 **자체 commit 하지 않는다.** 트랜잭션 소유는 라우터/백그라운드 태스크다.
+
+환불 정책 (F2 와 다르다):
+ F1 실패는 대부분 재시도 가능하다(제목 게이트, 모델 큐 등 — 프론트에 재시도
+ 버튼이 있다). 실패 관측 즉시 환불하면 크레딧 원장의 (job_type, job_ref, type)
+ 유니크 제약 때문에 재시도 시 재차감이 불가능해져 "환불받고 공짜 재시도"가 된다.
+ 그래서 실패 시에는 환불하지 않고, **잡을 버릴 때(삭제)와 고아 스윕에서만** 환불한다.
+"""
+
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import func, select, text, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.credit.services.credit_service import (
+ deduct_credit_for_job,
+ refund_credit_for_job,
+)
+from app.p2v.constants import (
+ JOB_TYPE_P2V_F1,
+ ORPHAN_STATUSES,
+ STATUS_ARCHIVING,
+ STATUS_DONE,
+ STATUS_FAILED,
+)
+from app.p2v.exceptions import P2vJobNotFoundError
+from app.p2v.models import P2vF1Job
+from app.utils.logger import get_logger
+from config import p2v_settings
+
+logger = get_logger("p2v")
+
+
+async def create_job(
+ session: AsyncSession, *, user_uuid: str, name: str
+) -> P2vF1Job:
+ """잡 행 생성 + 크레딧 선차감 (한 트랜잭션 — 호출부가 commit).
+
+ 행을 먼저 flush 해 id 를 확보한 뒤 그 id 를 멱등 키(job_ref)로 차감한다.
+ 크레딧이 부족하면 InsufficientCreditError 가 올라가고 행도 롤백된다 —
+ P2V 는 아직 호출되지 않았으므로 실비용(OpenAI 등)이 나가지 않는다.
+
+ Raises:
+ InsufficientCreditError: 잔액 부족 (전역 핸들러가 402 로 변환)
+ """
+ row = P2vF1Job(
+ user_uuid=user_uuid,
+ name=name or None,
+ status="queued",
+ credit_amount=p2v_settings.P2V_CREDITS_PER_F1,
+ )
+ session.add(row)
+ await session.flush() # id 확보
+
+ await deduct_credit_for_job(
+ session=session,
+ user_uuid=user_uuid,
+ amount=p2v_settings.P2V_CREDITS_PER_F1,
+ job_type=JOB_TYPE_P2V_F1,
+ job_ref=str(row.id),
+ reason="무빙 포스터 생성",
+ )
+ logger.info(f"[f1] 잡 생성 id={row.id} user={user_uuid}")
+ return row
+
+
+async def get_owned(
+ session: AsyncSession, job_id: int, user_uuid: str
+) -> P2vF1Job:
+ """소유권 검사 포함 조회. 남의 잡이면 존재를 알리지 않고 동일하게 404."""
+ row = await session.get(P2vF1Job, job_id)
+ if row is None or row.user_uuid != user_uuid:
+ raise P2vJobNotFoundError()
+ return row
+
+
+async def refund_job(
+ session: AsyncSession, row: P2vF1Job, reason: str
+) -> None:
+ """선차감분 환불 (멱등 — 이미 환불했거나 차감 기록이 없으면 조용히 넘어간다)."""
+ await refund_credit_for_job(
+ session=session,
+ user_uuid=row.user_uuid,
+ amount=row.credit_amount,
+ job_type=JOB_TYPE_P2V_F1,
+ job_ref=str(row.id),
+ reason=reason,
+ )
+
+
+async def fail_job(
+ session: AsyncSession, row: P2vF1Job, detail: str, *, refund: bool
+) -> None:
+ """실패 처리.
+
+ Args:
+ refund: True 면 환불까지. P2V 호출 자체가 실패했거나(실비용 없음)
+ 잡이 P2V 에서 사라져 재시도가 불가능할 때만 True 로 준다.
+ 일반 스테이지 실패는 재시도 가능하므로 False (모듈 docstring 참조).
+ """
+ row.status = STATUS_FAILED
+ row.error = detail[:1000]
+ if refund:
+ await refund_job(session, row, reason=f"무빙 포스터 실패: {detail[:100]}")
+ logger.info(f"[f1] 실패 처리 id={row.id} refund={refund} detail={detail[:200]}")
+
+
+def sync_from_p2v(row: P2vF1Job, p2v_job: dict) -> None:
+ """P2V 잡 JSON → castad 행 미러링 (순수 상태 복사 — DB 호출 없음).
+
+ 'done' 은 여기서 반영하지 않는다 — Blob 아카이빙이 끝나야 done 이다
+ (archive_service 가 archiving → done 전이를 소유한다).
+ 실패도 상태만 미러하고 환불하지 않는다 (재시도 가능 — 모듈 docstring 참조).
+ """
+ p2v_status = p2v_job.get("status", "")
+ if p2v_status in ("queued", "running", "awaiting_review"):
+ row.status = p2v_status
+ elif p2v_status == "failed":
+ row.status = STATUS_FAILED
+ err = p2v_job.get("error") or {}
+ row.error = f"{err.get('stage', '?')}: {err.get('detail', '')}"[:1000]
+
+ if p2v_job.get("name"):
+ row.name = str(p2v_job["name"])[:100]
+ if p2v_job.get("narration") is not None:
+ row.narration = p2v_job["narration"]
+ md = p2v_job.get("metadata")
+ if md:
+ row.event_name = (md.get("event_name") or "")[:200] or None
+ row.date_text = (md.get("date_text") or "")[:100] or None
+ row.place = (md.get("place") or "")[:200] or None
+ row.keywords = md.get("keywords")
+ if p2v_job.get("duration") is not None:
+ row.duration = p2v_job["duration"]
+
+
+async def try_claim_archiving(session: AsyncSession, job_id: int) -> bool:
+ """Blob 아카이빙 착수권 선점 (낙관적 잠금).
+
+ 폴링이 겹쳐도 UPDATE 는 한 요청만 성공한다. rowcount 0 이면 다른 폴링이
+ 이미 집어갔거나 이미 아카이브된 것이므로 조용히 물러난다.
+ """
+ result = await session.execute(
+ update(P2vF1Job)
+ .where(
+ P2vF1Job.id == job_id,
+ P2vF1Job.status != STATUS_ARCHIVING,
+ P2vF1Job.archived_at.is_(None),
+ )
+ .values(status=STATUS_ARCHIVING)
+ .execution_options(synchronize_session=False)
+ )
+ return result.rowcount > 0
+
+
+async def finish_archive(
+ session: AsyncSession,
+ row: P2vF1Job,
+ *,
+ video_url: Optional[str],
+ thumbnail_url: Optional[str],
+) -> None:
+ """아카이빙 완료 — Blob URL 기록 + done 전이."""
+ row.p2v_video_url = video_url
+ row.poster_url = thumbnail_url
+ row.archived_at = datetime.now()
+ row.status = STATUS_DONE
+ logger.info(f"[f1] 아카이브 완료 id={row.id} video={bool(video_url)}")
+
+
+async def sweep_orphans(session: AsyncSession) -> int:
+ """방치된 잡 정리 (앱 기동 시 lifespan 훅에서 호출).
+
+ 1) 진행/검수 상태로 P2V_ORPHAN_TIMEOUT_HOURS 를 넘긴 잡 → 환불 + 실패
+ (검수 화면에서 이탈한 사용자의 선차감분이 증발하는 것을 막는다 — 설계 R-1)
+ 2) archiving 으로 P2V_ARCHIVE_STALE_MINUTES 넘게 방치된 잡 → 'done'(archived_at
+ NULL) 으로 되돌린다. 다음 폴링이 아카이빙을 다시 선점한다 (설계 R-3 크래시 복구)
+
+ 시각 비교는 DB 시계(func.now())로 한다 — 앱과 DB 의 타임존이 다를 수 있다.
+ """
+ swept = 0
+
+ orphan_cutoff = func.date_sub(
+ func.now(), text(f"INTERVAL {int(p2v_settings.P2V_ORPHAN_TIMEOUT_HOURS)} HOUR")
+ )
+ result = await session.execute(
+ select(P2vF1Job).where(
+ P2vF1Job.status.in_(ORPHAN_STATUSES),
+ P2vF1Job.created_at < orphan_cutoff,
+ )
+ )
+ for row in result.scalars().all():
+ await fail_job(
+ session, row, "시간 초과 — 크레딧을 자동 환불했습니다", refund=True
+ )
+ swept += 1
+
+ stale_cutoff = func.date_sub(
+ func.now(),
+ text(f"INTERVAL {int(p2v_settings.P2V_ARCHIVE_STALE_MINUTES)} MINUTE"),
+ )
+ result = await session.execute(
+ update(P2vF1Job)
+ .where(
+ P2vF1Job.status == STATUS_ARCHIVING,
+ P2vF1Job.updated_at < stale_cutoff,
+ )
+ .values(status=STATUS_DONE)
+ .execution_options(synchronize_session=False)
+ )
+ if result.rowcount:
+ logger.info(f"[f1] 정체된 archiving {result.rowcount}건 재시도 가능으로 복구")
+
+ return swept
diff --git a/app/p2v/services/f2_service.py b/app/p2v/services/f2_service.py
new file mode 100644
index 0000000..cea5d97
--- /dev/null
+++ b/app/p2v/services/f2_service.py
@@ -0,0 +1,190 @@
+# -*- coding: utf-8 -*-
+"""F2(포스터 스타일링) 잡 라이프사이클 + 크레딧.
+
+모든 함수는 **자체 commit 하지 않는다.** 트랜잭션 소유는 라우터/백그라운드 태스크다.
+
+F1 과 달리 F2 에는 재시도 경로가 없다(P2V API 에 retry 엔드포인트 자체가 없음).
+따라서 실패를 관측하는 즉시 환불한다.
+"""
+
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import func, select, text, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.credit.services.credit_service import (
+ deduct_credit_for_job,
+ refund_credit_for_job,
+)
+from app.p2v.constants import (
+ JOB_TYPE_P2V_F2,
+ STATUS_ARCHIVING,
+ STATUS_DONE,
+ STATUS_FAILED,
+)
+from app.p2v.exceptions import P2vJobNotFoundError
+from app.p2v.models import P2vF2Job
+from app.utils.logger import get_logger
+from config import p2v_settings
+
+logger = get_logger("p2v")
+
+
+async def create_job(
+ session: AsyncSession,
+ *,
+ user_uuid: str,
+ name: Optional[str],
+ template_id: str,
+ template_name: Optional[str],
+ license: Optional[str],
+ format: str,
+) -> P2vF2Job:
+ """잡 행 생성 + 크레딧 선차감 (한 트랜잭션 — 호출부가 commit).
+
+ 템플릿 이름·라이선스는 이 시점의 스냅샷이다 — 템플릿이 나중에 삭제되거나
+ 등급이 바뀌어도 "이 결과물이 어떤 등급 레퍼런스로 만들어졌는지"가 남는다.
+
+ Raises:
+ InsufficientCreditError: 잔액 부족 (전역 핸들러가 402 로 변환)
+ """
+ row = P2vF2Job(
+ user_uuid=user_uuid,
+ name=(name or "")[:100] or None,
+ template_id=template_id,
+ template_name=(template_name or "")[:100] or None,
+ license=(license or "")[:20] or None,
+ format=format,
+ status="queued",
+ credit_amount=p2v_settings.P2V_CREDITS_PER_F2,
+ )
+ session.add(row)
+ await session.flush() # id 확보
+
+ await deduct_credit_for_job(
+ session=session,
+ user_uuid=user_uuid,
+ amount=p2v_settings.P2V_CREDITS_PER_F2,
+ job_type=JOB_TYPE_P2V_F2,
+ job_ref=str(row.id),
+ reason="포스터 스타일링",
+ )
+ logger.info(f"[f2] 잡 생성 id={row.id} user={user_uuid} template={template_id}")
+ return row
+
+
+async def get_owned(
+ session: AsyncSession, job_id: int, user_uuid: str
+) -> P2vF2Job:
+ """소유권 검사 포함 조회. 남의 잡이면 존재를 알리지 않고 동일하게 404."""
+ row = await session.get(P2vF2Job, job_id)
+ if row is None or row.user_uuid != user_uuid:
+ raise P2vJobNotFoundError()
+ return row
+
+
+async def refund_job(session: AsyncSession, row: P2vF2Job, reason: str) -> None:
+ """선차감분 환불 (멱등 — 이미 환불했거나 차감 기록이 없으면 조용히 넘어간다)."""
+ await refund_credit_for_job(
+ session=session,
+ user_uuid=row.user_uuid,
+ amount=row.credit_amount,
+ job_type=JOB_TYPE_P2V_F2,
+ job_ref=str(row.id),
+ reason=reason,
+ )
+
+
+async def fail_job(session: AsyncSession, row: P2vF2Job, detail: str) -> None:
+ """실패 처리 + 즉시 환불 (F2 는 재시도 경로가 없다)."""
+ row.status = STATUS_FAILED
+ row.error = detail[:1000]
+ await refund_credit_for_job(
+ session=session,
+ user_uuid=row.user_uuid,
+ amount=row.credit_amount,
+ job_type=JOB_TYPE_P2V_F2,
+ job_ref=str(row.id),
+ reason=f"포스터 스타일링 실패: {detail[:100]}",
+ )
+ logger.info(f"[f2] 실패 처리(환불) id={row.id} detail={detail[:200]}")
+
+
+async def sync_from_p2v(
+ session: AsyncSession, row: P2vF2Job, p2v_job: dict
+) -> None:
+ """P2V 잡 JSON → castad 행 미러링.
+
+ 'done' 은 반영하지 않는다 — Blob 아카이빙 완료가 done 이다.
+ 'failed' 는 즉시 환불한다 (F1 과 다른 지점).
+ """
+ p2v_status = p2v_job.get("status", "")
+ if p2v_status in ("queued", "running"):
+ row.status = p2v_status
+ elif p2v_status == "failed" and row.status != STATUS_FAILED:
+ err = p2v_job.get("error") or {}
+ await fail_job(
+ session, row, f"{err.get('stage', '?')}: {err.get('detail', '')}"
+ )
+
+
+async def try_claim_archiving(session: AsyncSession, job_id: int) -> bool:
+ """Blob 아카이빙 착수권 선점 (낙관적 잠금 — f1_service 와 동일 방식)."""
+ result = await session.execute(
+ update(P2vF2Job)
+ .where(
+ P2vF2Job.id == job_id,
+ P2vF2Job.status != STATUS_ARCHIVING,
+ P2vF2Job.archived_at.is_(None),
+ )
+ .values(status=STATUS_ARCHIVING)
+ .execution_options(synchronize_session=False)
+ )
+ return result.rowcount > 0
+
+
+async def finish_archive(
+ session: AsyncSession, row: P2vF2Job, *, image_url: Optional[str]
+) -> None:
+ """아카이빙 완료 — Blob URL 기록 + done 전이."""
+ row.p2v_poster_url = image_url
+ row.archived_at = datetime.now()
+ row.status = STATUS_DONE
+ logger.info(f"[f2] 아카이브 완료 id={row.id} image={bool(image_url)}")
+
+
+async def sweep_orphans(session: AsyncSession) -> int:
+ """방치된 잡 정리 — f1_service.sweep_orphans 와 같은 규칙 (검수 단계만 없다)."""
+ swept = 0
+
+ orphan_cutoff = func.date_sub(
+ func.now(), text(f"INTERVAL {int(p2v_settings.P2V_ORPHAN_TIMEOUT_HOURS)} HOUR")
+ )
+ result = await session.execute(
+ select(P2vF2Job).where(
+ P2vF2Job.status.in_(("queued", "running")),
+ P2vF2Job.created_at < orphan_cutoff,
+ )
+ )
+ for row in result.scalars().all():
+ await fail_job(session, row, "시간 초과 — 크레딧을 자동 환불했습니다")
+ swept += 1
+
+ stale_cutoff = func.date_sub(
+ func.now(),
+ text(f"INTERVAL {int(p2v_settings.P2V_ARCHIVE_STALE_MINUTES)} MINUTE"),
+ )
+ result = await session.execute(
+ update(P2vF2Job)
+ .where(
+ P2vF2Job.status == STATUS_ARCHIVING,
+ P2vF2Job.updated_at < stale_cutoff,
+ )
+ .values(status=STATUS_DONE)
+ .execution_options(synchronize_session=False)
+ )
+ if result.rowcount:
+ logger.info(f"[f2] 정체된 archiving {result.rowcount}건 재시도 가능으로 복구")
+
+ return swept
diff --git a/app/video/schemas/video_schema.py b/app/video/schemas/video_schema.py
index a878dd3..791ea4c 100644
--- a/app/video/schemas/video_schema.py
+++ b/app/video/schemas/video_schema.py
@@ -158,9 +158,10 @@ class VideoListItem(BaseModel):
# **`(type, video_id)` 쌍**으로 다뤄야 한다.
# 특히 `DELETE /archive/videos/{id}` 는 `Video.id` 로 지우므로,
# 썰박스 항목의 id 를 그대로 넘기면 **엉뚱한 ADO2 영상이 삭제된다.**
- type: Literal["video", "ssul"] = Field(
+ type: Literal["video", "ssul", "p2v_video", "p2v_poster"] = Field(
default="video",
- description="콘텐츠 종류 (video: ADO2 영상, ssul: 썰박스)",
+ description="콘텐츠 종류 (video: ADO2 영상, ssul: 썰박스, "
+ "p2v_video: 무빙 포스터, p2v_poster: 포스터 스타일링 — 이미지라 ![]()
로 그릴 것)",
)
video_id: int = Field(..., description="콘텐츠 고유 ID (type 안에서만 유일)")
store_name: Optional[str] = Field(None, description="업체명")
@@ -193,13 +194,14 @@ class VideoThumbnailItem(BaseModel):
# ⚠️ `video_id` 는 종류 안에서만 유일하다. `video.id` 와 `ssul_content.id` 가
# **둘 다 1부터 시작**하므로 식별자는 반드시 `(type, video_id)` 쌍으로 다뤄야 한다.
# 한 곳이라도 id 만 쓰면 다른 종류의 콘텐츠가 열린다.
- type: Literal["video", "ssul"] = Field(
+ type: Literal["video", "ssul", "p2v_video", "p2v_poster"] = Field(
default="video",
- description="콘텐츠 종류 (video: ADO2 영상, ssul: 썰박스). video_id 와 쌍으로 식별한다",
+ description="콘텐츠 종류 (video: ADO2 영상, ssul: 썰박스, p2v_video: 무빙 포스터, "
+ "p2v_poster: 포스터 스타일링 — 이미지라 ![]()
로 그릴 것). video_id 와 쌍으로 식별한다",
)
video_id: int = Field(..., description="콘텐츠 고유 ID (type 안에서만 유일)")
- store_name: str = Field(..., description="업체명")
- result_movie_url: str = Field(..., description="영상 URL")
+ store_name: str = Field(..., description="업체명 (P2V 는 행사명/포스터명)")
+ result_movie_url: str = Field(..., description="영상 URL (p2v_poster 는 이미지 URL)")
poster_url: Optional[str] = Field(None, description="영상 첫 프레임 포스터 이미지 URL (썸네일 표시용)")
title: Optional[str] = Field(None, description="SNS 업로드 제목")
description: Optional[str] = Field(None, description="SNS 업로드 설명")
diff --git a/app/video/services/unified_list.py b/app/video/services/unified_list.py
index 9ddaae7..0187d8e 100644
--- a/app/video/services/unified_list.py
+++ b/app/video/services/unified_list.py
@@ -27,7 +27,7 @@ from dataclasses import dataclass
from datetime import datetime
from typing import Literal, Optional
-from sqlalchemy import Select, func, literal, or_, select, union_all
+from sqlalchemy import Select, func, literal, null, or_, select, union_all
from sqlalchemy.ext.asyncio import AsyncSession
from app.comment.models import Comment
@@ -40,11 +40,18 @@ from app.database.like_cache import (
mset_like_counts,
)
from app.home.models import Project
+from app.p2v.models import P2vF1Job, P2vF2Job
from app.ssulbox.models import SsulContent
from app.utils.address_parser import SIDO_CITIES, SIDO_SEARCH_ALIASES
from app.video.models import Video, VideoReaction
-ContentType = Literal["video", "ssul"]
+#: p2v_video(무빙 포스터 영상)·p2v_poster(스타일링 이미지). 내 콘텐츠와 갤러리
+#: 양쪽에 나오지만(2026-08-26 갤러리 합류) **좋아요·댓글 축은 없다** — enrich 는
+#: CT_VIDEO/CT_SSUL 만 돌아 0 으로 남고, 프론트도 P2V 카드에는 소셜 액션을 숨긴다.
+#: 갤러리의 F2 는 퍼블릭 도메인 템플릿 결과물만 나간다(fetch_gallery 참조).
+ContentType = Literal["video", "ssul", "p2v_video", "p2v_poster"]
+CT_P2V_VIDEO = "p2v_video"
+CT_P2V_POSTER = "p2v_poster"
#: 정렬 가능한 키. 그 외 값은 created_at 으로 떨어진다.
SORT_CREATED = "created_at"
@@ -228,6 +235,58 @@ def _ssul_branch(where: list, sort_by: str) -> Select:
return select(*cols).where(*where)
+def _p2v_f1_branch(where: list, sort_by: str = SORT_CREATED) -> Select:
+ """무빙 포스터(F1) 브랜치.
+
+ 통합 목록의 축과 P2V 필드 대응: store_name←행사명, region←장소,
+ hashtags←키워드. P2V 에는 좋아요·댓글 축이 없으므로 해당 정렬에서는
+ sort_value 0 으로 뒤에 깔린다 (UNION 컬럼 수 맞춤용).
+ """
+ cols = [
+ literal(CT_P2V_VIDEO).label("ctype"),
+ P2vF1Job.id.label("cid"),
+ func.coalesce(P2vF1Job.event_name, P2vF1Job.name).label("store_name"),
+ P2vF1Job.place.label("region"),
+ P2vF1Job.p2v_video_url.label("movie_url"),
+ P2vF1Job.created_at.label("created_at"),
+ literal("").label("task_id"),
+ P2vF1Job.poster_url.label("poster_url"),
+ P2vF1Job.name.label("title"),
+ null().label("description"),
+ P2vF1Job.keywords.label("hashtags"),
+ ]
+ if sort_by in (SORT_LIKE, SORT_COMMENT):
+ cols.append(literal(0).label("sort_value"))
+ return select(*cols).where(*where)
+
+
+def _p2v_f2_branch(where: list, sort_by: str = SORT_CREATED) -> Select:
+ """포스터 스타일링(F2) 브랜치 — 결과물이 영상이 아니라 이미지다.
+
+ movie_url 자리에 결과 이미지 URL 을 싣는다. **프론트는 ctype 으로 분기해
+ ![]()
로 그려야 한다** (ssul/video 처럼