john
1aaef71f52
Document-driven MVP with FastAPI backend, Vue H5, WeChat mini shell, product demo, and Docker dev stack. Co-authored-by: Cursor <cursoragent@cursor.com>
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.models import Child
|
|
from app.db.models import Video as VideoModel
|
|
from app.schemas.models import UploadTokenRequest, VideoCreateRequest
|
|
from app.services import oss as oss_service
|
|
|
|
|
|
def verify_child_owner(db: Session, parent_user_id: int, child_id: int) -> Child | None:
|
|
return db.scalar(
|
|
select(Child).where(
|
|
Child.id == child_id,
|
|
Child.parent_user_id == parent_user_id,
|
|
Child.status != "archived",
|
|
)
|
|
)
|
|
|
|
|
|
def create_upload_token(body: UploadTokenRequest) -> dict:
|
|
object_key = oss_service.generate_object_key(body.child_id, body.file_name)
|
|
return oss_service.create_upload_token(object_key, body.content_type, body.size)
|
|
|
|
|
|
def save_direct_upload(token: str, object_key: str, data: bytes) -> None:
|
|
if not oss_service.validate_upload_token(token, object_key):
|
|
raise ValueError("invalid_upload_token")
|
|
oss_service.save_local_upload(object_key, data)
|
|
oss_service.consume_upload_token(token, object_key)
|
|
|
|
|
|
def register_video(
|
|
db: Session, user_id: int, body: VideoCreateRequest
|
|
) -> tuple[VideoModel | None, str | None]:
|
|
child = verify_child_owner(db, user_id, body.child_id)
|
|
if not child:
|
|
return None, "child_not_found"
|
|
|
|
existing = db.scalar(select(VideoModel).where(VideoModel.object_key == body.object_key))
|
|
if existing:
|
|
return existing, "duplicate"
|
|
|
|
video = VideoModel(
|
|
child_id=body.child_id,
|
|
uploaded_by=user_id,
|
|
scene=body.scene.value,
|
|
object_key=body.object_key,
|
|
status="uploaded",
|
|
)
|
|
db.add(video)
|
|
db.commit()
|
|
db.refresh(video)
|
|
return video, None
|