from __future__ import annotations import hashlib import uuid from datetime import datetime, timezone, timedelta 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, ApiCallLog, ContentScript, ExternalPost, JobStatus, PlatformAccount, PostDraft, Provider, PublishJob, VideoAsset, VideoStatus, WebsiteCompanion, YouTubeUploadAudit, ) from app.security.redaction import redact_json from app.security.tokens import decrypt_token YOUTUBE_UPLOAD_SCOPE = "https://www.googleapis.com/auth/youtube.upload" def youtube_url(video_id: str) -> str: return f"https://www.youtube.com/shorts/{video_id}" def _now() -> datetime: return datetime.now(timezone.utc) def _description_hash(description: str | None) -> str: return hashlib.sha256((description or "").encode("utf-8")).hexdigest() def _contains_synthetic_media(video: VideoAsset, draft: PostDraft) -> bool: disclosure = draft.metadata_json.get("disclosure") if isinstance(draft.metadata_json, dict) else {} if isinstance(disclosure, dict) and "synthetic_media" in disclosure: return bool(disclosure.get("synthetic_media")) return bool(video.is_aigc) def _ensure_existing_file(video: VideoAsset) -> Path: path = Path(video.file_path).expanduser() if not path.is_absolute(): backend_root = Path(__file__).resolve().parents[2] path = backend_root / path if not path.exists(): raise HTTPException(status_code=400, detail="Video file does not exist") return path def validate_private_upload(session: Session, post_draft_id: str, payload: dict[str, Any], dry_run: bool = False) -> tuple[PostDraft, VideoAsset, PlatformAccount, Path]: draft = session.get(PostDraft, post_draft_id) if not draft: raise HTTPException(status_code=404, detail="Draft not found") if draft.provider != Provider.youtube: raise HTTPException(status_code=400, detail="Not a YouTube draft") video = session.get(VideoAsset, draft.video_asset_id) if not video: raise HTTPException(status_code=404, detail="Video asset not found") path = _ensure_existing_file(video) account_id = payload.get("account_id") or draft.platform_account_id if not account_id and dry_run: account = session.exec(select(PlatformAccount).where(PlatformAccount.provider == Provider.youtube, PlatformAccount.external_account_id == "dry-run-youtube-account")).first() if not account: account = PlatformAccount(provider=Provider.youtube, display_name="YouTube Dry-run Account", external_account_id="dry-run-youtube-account", account_type="dry_run", scopes=[YOUTUBE_UPLOAD_SCOPE], status="dry_run") session.add(account); session.commit(); session.refresh(account) account_id = account.id if not account_id: raise HTTPException(status_code=400, detail="YouTube account is required") account = session.get(PlatformAccount, account_id) if not account or account.provider != Provider.youtube: raise HTTPException(status_code=400, detail="Connected YouTube account is required") if account.status != "connected" and not (dry_run and account.account_type == "dry_run"): raise HTTPException(status_code=400, detail=f"YouTube account is not connected: {account.status}") if not payload.get("confirm_private_upload"): raise HTTPException(status_code=400, detail="Private upload confirmation required") if not payload.get("confirm_metadata_reviewed"): raise HTTPException(status_code=400, detail="Metadata review confirmation required") if not payload.get("confirm_disclosure_reviewed"): raise HTTPException(status_code=400, detail="Disclosure review confirmation required") if draft.privacy_status and draft.privacy_status != "private": raise HTTPException(status_code=400, detail="Sprint 8 allows private uploads only") if not draft.title: raise HTTPException(status_code=400, detail="YouTube title is required") if not draft.description: raise HTTPException(status_code=400, detail="YouTube description is required") if not video.is_aigc and "disclosure" not in (draft.metadata_json or {}): raise HTTPException(status_code=400, detail="Disclosure must be reviewed") return draft, video, account, path def _build_real_youtube_client(account: PlatformAccount): from google.oauth2.credentials import Credentials from googleapiclient.discovery import build refresh_token = decrypt_token(account.encrypted_refresh_token) access_token = decrypt_token(account.encrypted_access_token) settings = get_settings() if not refresh_token and not access_token: raise HTTPException(status_code=400, detail="Stored YouTube token is unavailable") creds = Credentials( token=access_token, refresh_token=refresh_token, token_uri="https://oauth2.googleapis.com/token", client_id=settings.google_client_id, client_secret=settings.google_client_secret, scopes=[settings.youtube_oauth_scope], ) return build("youtube", "v3", credentials=creds) def _execute_real_upload(account: PlatformAccount, video_path: Path, body: dict[str, Any], notify_subscribers: bool) -> dict[str, Any]: from googleapiclient.http import MediaFileUpload youtube = _build_real_youtube_client(account) media = MediaFileUpload(str(video_path), chunksize=-1, resumable=True) request = youtube.videos().insert( part="snippet,status", body=body, media_body=media, notifySubscribers=notify_subscribers, ) response = None while response is None: _, response = request.next_chunk() return response or {} def private_upload(session: Session, post_draft_id: str, payload: dict[str, Any]) -> dict[str, Any]: settings = get_settings() dry_run = bool(payload.get("dry_run", settings.youtube_upload_dry_run)) draft, video, account, video_path = validate_private_upload(session, post_draft_id, payload, dry_run=dry_run) category_id = draft.category_id or settings.youtube_default_category_id or "22" notify_subscribers = False contains_synthetic = _contains_synthetic_media(video, draft) tags = draft.tags_json or draft.hashtags_json or [] body = { "snippet": { "title": draft.title, "description": draft.description, "tags": tags, "categoryId": category_id, }, "status": { "privacyStatus": "private", "selfDeclaredMadeForKids": False, "containsSyntheticMedia": contains_synthetic, }, } job = PublishJob( post_draft_id=draft.id, provider=Provider.youtube, job_type="youtube_private_upload_dry_run" if dry_run else "youtube_private_upload", status=JobStatus.queued, attempts=1, ) session.add(job) session.add(ActivityEvent(entity_type="post_draft", entity_id=draft.id, event_type="youtube_upload_queued", label="YouTube private upload queued", payload_json={"privacy": "private", "dry_run": dry_run})) session.commit(); session.refresh(job) audit = YouTubeUploadAudit( video_asset_id=video.id, post_draft_id=draft.id, platform_account_id=account.id, publish_job_id=job.id, privacy_status="private", title=draft.title, description_hash=_description_hash(draft.description), tags_json=tags, category_id=category_id, contains_synthetic_media=contains_synthetic, self_declared_made_for_kids=False, notify_subscribers=notify_subscribers, upload_status="running", api_endpoints_called=[] if dry_run else ["youtube.videos.insert"], oauth_scopes_used=[YOUTUBE_UPLOAD_SCOPE], website_link_allowed=False, request_redacted_json=redact_json({"body": body, "video_path": str(video_path.name), "notifySubscribers": notify_subscribers, "dry_run": dry_run}), ) job.status = JobStatus.running job.started_at = _now() session.add(job); session.add(audit) session.add(ActivityEvent(entity_type="publish_job", entity_id=job.id, event_type="youtube_upload_started", label="YouTube private upload started", payload_json={"dry_run": dry_run})) session.commit(); session.refresh(audit) try: if dry_run: yt_id = "dryrun_" + uuid.uuid4().hex[:16] response = {"id": yt_id, "dry_run": True, "privacyStatus": "private"} else: response = _execute_real_upload(account, video_path, body, notify_subscribers) yt_id = str(response.get("id") or "") if not yt_id: raise RuntimeError("YouTube response did not include video ID") url = youtube_url(yt_id) audit.youtube_video_id = yt_id audit.youtube_url = url audit.upload_status = "dry_run_succeeded" if dry_run else "private_succeeded" audit.response_redacted_json = redact_json(response) audit.updated_at = _now() job.status = JobStatus.succeeded job.external_post_id = yt_id job.external_publish_id = yt_id job.finished_at = _now() draft.external_post_id = yt_id draft.platform_account_id = account.id draft.privacy_status = "private" draft.status = "uploaded_private_dry_run" if dry_run else "uploaded_private" video.external_youtube_id = yt_id video.status = VideoStatus.uploaded_private external = session.exec(select(ExternalPost).where(ExternalPost.provider == Provider.youtube, ExternalPost.external_post_id == yt_id)).first() if not external: external = ExternalPost(provider=Provider.youtube, external_post_id=yt_id) external.account_id = account.id external.video_asset_id = video.id external.title = draft.title external.topic_id = video.topic_id external.series = video.series external.mapping_status = "linked" companion = session.exec(select(WebsiteCompanion).where(WebsiteCompanion.video_asset_id == video.id)).first() if companion: companion.youtube_video_id = yt_id companion.video_url_internal = url companion.website_link_allowed = False companion.status = "waiting_for_public_video" companion.updated_at = _now() session.add(companion) script = session.exec(select(ContentScript).where(ContentScript.video_asset_id == video.id)).first() if script: script.performance_summary_json = {**(script.performance_summary_json or {}), "youtube_video_id": yt_id, "upload_status": "uploaded_private"} script.updated_at = _now() session.add(script) account.last_successful_api_call = _now() if account.account_type == "dry_run": account.status = "dry_run" account.last_error = None account.updated_at = _now() session.add_all([audit, job, draft, video, external, account]) session.add(ApiCallLog(provider="youtube", endpoint="youtube.videos.insert" if not dry_run else "dry-run.youtube.videos.insert", method="POST", status_code=200, request_redacted_json=audit.request_redacted_json, response_redacted_json=audit.response_redacted_json)) session.add(ActivityEvent(entity_type="post_draft", entity_id=draft.id, event_type="youtube_upload_private_succeeded", label="YouTube private upload complete" if not dry_run else "YouTube private upload dry-run complete", payload_json={"youtube_video_id": yt_id, "youtube_url": url, "dry_run": dry_run, "website_link_allowed": False})) session.commit(); session.refresh(audit); session.refresh(job) return {"ok": True, "dry_run": dry_run, "job": job.model_dump(mode="json"), "audit": audit.model_dump(mode="json"), "youtube_video_id": yt_id, "youtube_url": url, "website_link_status": "waiting_for_public_video"} except Exception as exc: message = str(exc) audit.upload_status = "failed" audit.error_code = exc.__class__.__name__ audit.error_message = message audit.updated_at = _now() job.status = JobStatus.failed job.error_code = exc.__class__.__name__ job.error_message = message job.finished_at = _now() account.status = "error" account.last_error = message session.add_all([audit, job, account]) session.add(ActivityEvent(entity_type="post_draft", entity_id=draft.id, event_type="youtube_upload_failed", label="YouTube private upload failed", payload_json={"error": message})) session.commit() raise HTTPException(status_code=502, detail=message) from exc