o2o-castad-backend/app/ssulbox/schemas/ssulbox_schema.py

118 lines
4.4 KiB
Python

"""썰박스 API 요청/응답 스키마 (Pydantic v2)."""
from datetime import datetime
from typing import Literal, Optional
from pydantic import BaseModel, Field
from app.ssulbox.constants import SCENARIOS
ScenarioLiteral = Literal["joseon", "samgukji", "greek", "odyssey"]
# =============================================================================
# 장소 검색
# =============================================================================
class SsulPlaceItem(BaseModel):
"""네이버 지도 검색 후보.
castad `/search/accommodation` 응답과 달리 **place_url 을 포함**한다.
generator 가 이 URL 로 place 페이지를 크롤링하기 때문이다.
"""
title: str = Field(..., description="업장명")
category: str = Field(default="", description="업종")
address: str = Field(default="", description="표시용 주소 (동명 업장 구분에 사용)")
roadAddress: str = Field(default="", description="도로명 주소")
place_url: str = Field(..., description="네이버 지도 place URL (생성 파이프라인 입력)")
class SsulPlaceSearchResponse(BaseModel):
query: str
count: int
items: list[SsulPlaceItem]
# =============================================================================
# 생성 요청
# =============================================================================
class SsulCreateRequest(BaseModel):
"""생성 요청.
`scenes`/`seconds` 의 기본값과 허용 범위는 **여기서만** 강제한다.
DB 에 기본값을 두면 진실이 두 곳에 생기므로 모델에는 두지 않았다.
"""
scenario: ScenarioLiteral = Field(..., description="시나리오 코드")
input: str = Field(
...,
min_length=2,
max_length=500,
description="네이버 지도 place URL 또는 업장명",
)
scenes: int = Field(default=9, ge=4, le=20, description="장면 수")
seconds: int = Field(default=30, ge=20, le=90, description="장면당 초 길이")
# 검색으로 업장을 고른 경우 프론트가 함께 보낸다(`/ssul/search/place` 결과).
# 통합 목록의 업장명 표시와 store_name/region 필터가 이 값에 의존한다.
# place URL 을 직접 붙여넣은 경우에는 없으며, 그때 store_name 은 생성 로그의
# `■ 가게:` 마커로 뒤늦게 채운다(region 은 주소가 없어 채울 수 없다).
store_name: str | None = Field(
default=None, max_length=200, description="업장명 (검색 선택 시)"
)
# 도로명·지번을 모두 받는다. castad `/home/crawl` 과 같이 도로명에서 시/군 추출이
# 실패하면 지번으로 재시도해야 지역이 비는 경우를 줄인다.
road_address: str | None = Field(
default=None,
max_length=300,
description="도로명 주소 (검색 선택 시). region 추출에만 쓰고 저장하지 않는다",
)
address: str | None = Field(
default=None,
max_length=300,
description="지번 주소 (검색 선택 시). 도로명 추출 실패 시 폴백",
)
class SsulCreateResponse(BaseModel):
id: int = Field(..., description="생성 잡 ID (폴링·크레딧 원장 앵커)")
status: str = Field(..., description="queued")
poll_interval_seconds: int = Field(
..., description="권장 폴링 간격(초). 클라이언트가 참고한다"
)
# =============================================================================
# 진행 상태 (폴링)
# =============================================================================
class SsulTaskStatus(BaseModel):
"""`GET /ssul/tasks/{id}` 응답. 프론트가 3초마다 폴링한다."""
id: int
scenario: str
status: Literal["queued", "running", "done", "error"]
step: int = Field(..., ge=0, le=4, description="완료한 단계 수 (0=준비, 4=합성 완료)")
error: Optional[str] = None
video_url: Optional[str] = Field(None, description="완료 시에만 채워진다")
created_at: datetime
model_config = {"from_attributes": True}
class SsulActiveTasksResponse(BaseModel):
"""진행 중인 내 잡. 새로고침·새 탭 복구에 쓴다"""
items: list[SsulTaskStatus]
__all__ = [
"SCENARIOS",
"ScenarioLiteral",
"SsulActiveTasksResponse",
"SsulCreateRequest",
"SsulCreateResponse",
"SsulPlaceItem",
"SsulPlaceSearchResponse",
"SsulTaskStatus",
]