from __future__ import annotations import json from pathlib import Path from typing import Any from fastapi import HTTPException from sqlmodel import Session, select from app.core.config import get_settings from app.models.core import ActivityEvent, PostDraft, Provider, UploadExecutor, UploadRequest, VideoAsset, WebsiteCompanion, now_utc def serialize_executor(executor: UploadExecutor) -> dict[str, Any]: return { "id": executor.id, "name": executor.name, "type": executor.type, "status": executor.status, "description": executor.description, "config_json": executor.config_json, "credentials": (executor.config_json or {}).get("credentials") if executor.config_json else ("available" if executor.type == "existing_jarvis_uploader" and executor.status == "ready" else "missing" if executor.type == "dashboard_oauth" else None), "mode": (executor.config_json or {}).get("mode") if executor.config_json else ("private_upload" if executor.type == "existing_jarvis_uploader" else None), "created_at": executor.created_at.isoformat() if executor.created_at else None, "updated_at": executor.updated_at.isoformat() if executor.updated_at else None, } def serialize_upload_request(request: UploadRequest) -> dict[str, Any]: return { "id": request.id, "video_asset_id": request.video_asset_id, "post_draft_id": request.post_draft_id, "executor_type": request.executor_type, "status": request.status, "workflow_status": "waiting_for_executor" if request.status == "approved" and request.executor_type == "existing_jarvis_uploader" else request.status, "approval_text": request.approval_text, "approval_command": request.approval_command, "expected_candidate_id": request.expected_candidate_id, "expected_video_sha": request.expected_video_sha, "expected_privacy": request.expected_privacy, "metadata_snapshot_json": request.metadata_snapshot_json, "outbox_path": request.outbox_path, "upload_audit_id": request.upload_audit_id, "youtube_video_id": request.youtube_video_id, "created_at": request.created_at.isoformat() if request.created_at else None, "approved_at": request.approved_at.isoformat() if request.approved_at else None, "sent_at": request.sent_at.isoformat() if request.sent_at else None, "completed_at": request.completed_at.isoformat() if request.completed_at else None, "updated_at": request.updated_at.isoformat() if request.updated_at else None, "error": request.error, } def ensure_default_executors(session: Session) -> list[UploadExecutor]: defaults = [ ("existing_jarvis_uploader", "Existing Jarvis YouTube uploader", "Primary real private-upload route. Uses the existing Jarvis upload workflow after creator approval; no tokens are written to the dashboard repo."), ("dashboard_oauth", "Dashboard OAuth uploader", "Optional direct Dashboard-managed Google OAuth. Disabled until account is connected."), ("dry_run", "Dry-run uploader", "Does not call YouTube; exercises audit and dashboard state."), ("imported_existing", "Imported existing upload", "Represents uploads found in existing Jarvis audit logs."), ] rows = [] for typ, name, desc in defaults: row = session.exec(select(UploadExecutor).where(UploadExecutor.type == typ)).first() if not row: status = "ready" if typ == "existing_jarvis_uploader" else "active" if typ in {"dry_run", "imported_existing"} else "configured_required" row = UploadExecutor(name=name, type=typ, description=desc, status=status) session.add(row) session.commit() session.refresh(row) else: row.name = name row.description = desc if typ == "existing_jarvis_uploader": row.status = "ready" row.config_json = {"credentials": "available", "mode": "private_upload", "handoff": "outbox_or_agent_polling", "tokens_in_repo": False} row.updated_at = now_utc() session.add(row) session.commit() session.refresh(row) rows.append(row) return rows def _sha_prefix(video: VideoAsset) -> str: return (video.checksum or video.candidate_id or video.id)[:12] def _outbox_root() -> Path: return get_settings().storage_root / "outbox" / "upload_requests" def create_existing_upload_request(session: Session, post_draft_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: payload = payload or {} draft = session.get(PostDraft, post_draft_id) if not draft or draft.provider != Provider.youtube: raise HTTPException(status_code=404, detail="YouTube draft not found") video = session.get(VideoAsset, draft.video_asset_id) if not video: raise HTTPException(status_code=404, detail="Video not found") blocked_statuses = {"rejected", "needs_changes", "quality_failed", "needs_internal_fix", "archived", "failed"} video_status = getattr(video.status, "value", video.status) if video_status in blocked_statuses: raise HTTPException(status_code=409, detail="Video is not eligible for upload/export") expected_privacy = payload.get("expected_privacy") or draft.privacy_status or "private" if expected_privacy != "private": raise HTTPException(status_code=400, detail="Sprint 8.1 only allows private upload requests") candidate_id = video.candidate_id sha_prefix = _sha_prefix(video) command = payload.get("approval_command") or f"APPROVE_UPLOAD {candidate_id} {sha_prefix}" request = UploadRequest( video_asset_id=video.id, post_draft_id=draft.id, executor_type="existing_jarvis_uploader", status="approved", approval_text=payload.get("approval_text") or "Creator approved private upload via Dashboard using existing Jarvis uploader.", approval_command=command, expected_candidate_id=candidate_id, expected_video_sha=sha_prefix, expected_privacy="private", approved_at=now_utc(), metadata_snapshot_json={ "candidate_id": candidate_id, "video_asset_id": video.id, "post_draft_id": draft.id, "title": draft.title, "description": draft.description, "tags": draft.tags_json, "privacy_status": expected_privacy, "executor_type": "existing_jarvis_uploader", }, ) session.add(request) session.commit() session.refresh(request) outbox = _outbox_root() outbox.mkdir(parents=True, exist_ok=True) outbox_payload = { "request_id": request.id, "executor_type": request.executor_type, "status": request.status, "approval_command": request.approval_command, "candidate_id": candidate_id, "video_asset_id": video.id, "post_draft_id": draft.id, "expected_video_sha": sha_prefix, "expected_privacy": "private", "metadata": request.metadata_snapshot_json, } outbox_path = outbox / f"{request.id}.json" outbox_path.write_text(json.dumps(outbox_payload, indent=2), encoding="utf-8") request.outbox_path = str(outbox_path) request.updated_at = now_utc() session.add(request) session.add(ActivityEvent(entity_type="upload_request", entity_id=request.id, event_type="upload_request_approved", label="Existing Jarvis uploader request approved", payload_json={"candidate_id": candidate_id, "approval_command": command, "outbox_path": str(outbox_path)})) session.commit() session.refresh(request) return serialize_upload_request(request) def pending_agent_requests(session: Session) -> list[dict[str, Any]]: rows = session.exec(select(UploadRequest).where(UploadRequest.status == "approved").order_by(UploadRequest.created_at)).all() return [serialize_upload_request(r) for r in rows] def mark_sent(session: Session, request_id: str) -> dict[str, Any]: req = session.get(UploadRequest, request_id) if not req: raise HTTPException(status_code=404, detail="UploadRequest not found") req.status = "sent_to_executor" req.sent_at = now_utc() req.updated_at = now_utc() session.add(req); session.commit(); session.refresh(req) return serialize_upload_request(req) def complete_request(session: Session, request_id: str, payload: dict[str, Any]) -> dict[str, Any]: req = session.get(UploadRequest, request_id) if not req: raise HTTPException(status_code=404, detail="UploadRequest not found") req.status = "completed" req.youtube_video_id = payload.get("youtube_video_id", req.youtube_video_id) req.upload_audit_id = payload.get("upload_audit_id", req.upload_audit_id) if req.youtube_video_id: video = session.get(VideoAsset, req.video_asset_id) draft = session.get(PostDraft, req.post_draft_id) if video: video.external_youtube_id = req.youtube_video_id video.updated_at = now_utc() session.add(video) if draft: draft.external_post_id = req.youtube_video_id draft.updated_at = now_utc() session.add(draft) comp = session.exec(select(WebsiteCompanion).where(WebsiteCompanion.video_asset_id == req.video_asset_id)).first() if not comp: comp = WebsiteCompanion(video_asset_id=req.video_asset_id, post_draft_id=req.post_draft_id) comp.youtube_video_id = req.youtube_video_id comp.video_url_internal = f"https://www.youtube.com/shorts/{req.youtube_video_id}" comp.website_link_allowed = bool(payload.get("website_link_allowed", False)) comp.status = "ready_with_public_video" if comp.website_link_allowed else "ready_without_public_video" comp.updated_at = now_utc() session.add(comp) session.add(ActivityEvent(entity_type="website_companion", entity_id=comp.id, event_type="private_upload_companion_ready", label="Website guide can be prepared without public video link", payload_json={"video_asset_id": req.video_asset_id, "youtube_video_id": req.youtube_video_id, "website_link_allowed": comp.website_link_allowed})) req.completed_at = now_utc() req.updated_at = now_utc() session.add(req); session.commit(); session.refresh(req) return serialize_upload_request(req) def fail_request(session: Session, request_id: str, payload: dict[str, Any]) -> dict[str, Any]: req = session.get(UploadRequest, request_id) if not req: raise HTTPException(status_code=404, detail="UploadRequest not found") req.status = "failed" req.error = payload.get("error") or "Executor reported failure" req.updated_at = now_utc() session.add(req); session.commit(); session.refresh(req) return serialize_upload_request(req)