From 5de5c7507dbeb340a97eb3cff80d57653f03d4a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=84=B1=EA=B2=BD?= Date: Wed, 26 Aug 2026 13:52:18 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20p2v=20=EC=97=B0=EB=8F=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/common.py | 24 + app/core/exceptions.py | 18 + app/database/session.py | 4 + app/p2v/__init__.py | 12 + app/p2v/api/__init__.py | 0 app/p2v/api/routers/__init__.py | 0 app/p2v/api/routers/v1/__init__.py | 0 app/p2v/api/routers/v1/f1.py | 429 ++++++++++++++++++ app/p2v/api/routers/v1/f2.py | 335 ++++++++++++++ app/p2v/api/routers/v1/files.py | 81 ++++ app/p2v/constants.py | 39 ++ app/p2v/exceptions.py | 114 +++++ app/p2v/models.py | 265 +++++++++++ app/p2v/schemas/__init__.py | 0 app/p2v/schemas/p2v_schema.py | 190 ++++++++ app/p2v/services/__init__.py | 0 app/p2v/services/archive_service.py | 171 +++++++ app/p2v/services/client.py | 118 +++++ app/p2v/services/f1_service.py | 218 +++++++++ app/p2v/services/f2_service.py | 190 ++++++++ app/video/schemas/video_schema.py | 14 +- app/video/services/unified_list.py | 126 ++++- config.py | 66 +++ .../migration_2026-08-25_p2v.sql | 88 ++++ docs/design/p2v-credit-integration.md | 414 +++++++++++++++++ main.py | 37 +- 26 files changed, 2944 insertions(+), 9 deletions(-) create mode 100644 app/p2v/__init__.py create mode 100644 app/p2v/api/__init__.py create mode 100644 app/p2v/api/routers/__init__.py create mode 100644 app/p2v/api/routers/v1/__init__.py create mode 100644 app/p2v/api/routers/v1/f1.py create mode 100644 app/p2v/api/routers/v1/f2.py create mode 100644 app/p2v/api/routers/v1/files.py create mode 100644 app/p2v/constants.py create mode 100644 app/p2v/exceptions.py create mode 100644 app/p2v/models.py create mode 100644 app/p2v/schemas/__init__.py create mode 100644 app/p2v/schemas/p2v_schema.py create mode 100644 app/p2v/services/__init__.py create mode 100644 app/p2v/services/archive_service.py create mode 100644 app/p2v/services/client.py create mode 100644 app/p2v/services/f1_service.py create mode 100644 app/p2v/services/f2_service.py create mode 100644 docs/database-schema/migration_2026-08-25_p2v.sql create mode 100644 docs/design/p2v-credit-integration.md 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 처럼