from __future__ import annotations import hashlib import uuid 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, PlatformAccount, PostDraft, Provider, TikTokUploadAttempt, VideoAsset, now_utc from app.security.redaction import redact_json, redact_upload_url TIKTOK_UPLOAD_ENDPOINT = "/v2/post/publish/inbox/video/init/" VIDEO_UPLOAD_SCOPE = "video.upload" FILE_UPLOAD_SOURCE = "FILE_UPLOAD" RECORDING_DEMO_TITLE_PRIORITY = [ "this small fee can expose your card", "if you call this number, the scam starts", "login prompt you didn’t start? don’t approve", "login prompt you didn't start? don't approve", ] def _resolve_media_path(video: VideoAsset) -> Path | None: if not video.file_path: return None path = Path(video.file_path) if not path.is_absolute(): path = (Path.cwd() / path).resolve() return path def _has_existing_media(video: VideoAsset) -> bool: path = _resolve_media_path(video) return bool(path and path.exists() and path.is_file() and path.stat().st_size > 200_000) def _has_caption(draft: PostDraft | None) -> bool: return bool(draft and draft.caption and draft.caption.strip()) def _title_priority(video: VideoAsset) -> int: title = (video.working_title or "").lower() for index, needle in enumerate(RECORDING_DEMO_TITLE_PRIORITY): if needle in title: return index return len(RECORDING_DEMO_TITLE_PRIORITY) def select_recording_demo_video(session: Session) -> tuple[VideoAsset | None, PostDraft | None]: """Pick the cleanest real short for the TikTok recording stage. The recording page should not use the tiny Sprint 8.2 ffmpeg smoke-test clip when real prepared packages exist. Prefer real packages with TikTok captions, meaningful duration, existing media, and known visually useful titles. """ videos = session.exec(select(VideoAsset)).all() drafts = session.exec(select(PostDraft).where(PostDraft.provider == Provider.tiktok)).all() draft_by_video = {draft.video_asset_id: draft for draft in drafts} def score(video: VideoAsset) -> tuple[int, int, int, int, int, int, int, str]: draft = draft_by_video.get(video.id) duration = float(video.duration_sec or 0) size = int(video.file_size or 0) has_media = _has_existing_media(video) has_draft_caption = _has_caption(draft) title_priority = _title_priority(video) real_package = 1 if (video.package_id or "").startswith("package_real_") else 0 long_enough = 1 if duration > 10 else 0 readyish = 1 if video.status in {"ready_for_review", "in_review", "needs_metadata", "uploaded_private", "scheduled"} else 0 # Higher tuple wins; title priority is inverted so the preferred titles sort first. return ( 1 if draft else 0, 1 if has_draft_caption else 0, long_enough, 1 if has_media else 0, real_package, -title_priority, readyish, f"{size:012d}:{video.created_at.isoformat() if video.created_at else ''}", ) candidates = [video for video in videos if _has_existing_media(video)] candidates.sort(key=score, reverse=True) for video in candidates: draft = draft_by_video.get(video.id) if draft and _has_caption(draft) and (video.duration_sec or 0) > 10: return video, draft for video in candidates: draft = draft_by_video.get(video.id) if draft and _has_caption(draft): return video, draft for video in candidates: if video.status in {"ready_for_review", "in_review", "needs_metadata", "uploaded_private", "scheduled"}: return video, draft_by_video.get(video.id) fallback = session.exec( select(VideoAsset) .where(VideoAsset.status.in_(["ready_for_review", "in_review", "needs_metadata", "imported", "uploaded_private", "scheduled"])) .order_by(VideoAsset.created_at.desc()) .limit(1) ).first() return fallback, draft_by_video.get(fallback.id) if fallback else None def _video_size(video: VideoAsset) -> int: if video.file_size: return video.file_size if video.file_path: path = Path(video.file_path) if not path.is_absolute(): # Backend cwd is normally backend/. Dashboard storage paths are repo-relative. path = (Path.cwd() / path).resolve() if path.exists(): return path.stat().st_size return 0 def _chunk_plan(size: int) -> dict[str, int]: chunk_size = max(size, 1) if size and size < 8 * 1024 * 1024 else 8 * 1024 * 1024 total = 1 if size <= chunk_size else (size + chunk_size - 1) // chunk_size return {"video_size": size, "chunk_size": chunk_size, "total_chunk_count": total} def _redacted_request(video: VideoAsset, draft: PostDraft, mode: str) -> dict[str, Any]: plan = _chunk_plan(_video_size(video)) return redact_json({ "endpoint": TIKTOK_UPLOAD_ENDPOINT, "mode": mode, "scope": VIDEO_UPLOAD_SCOPE, "post_info": { "title": draft.caption or draft.title or video.working_title, "privacy_level": "SELF_ONLY_DRAFT", "disable_duet": False, "disable_comment": False, "disable_stitch": False, }, "source_info": { "source": FILE_UPLOAD_SOURCE, **plan, }, "authorization": "[REDACTED]", }) def _attempt_payload(attempt: TikTokUploadAttempt, video: VideoAsset | None = None, draft: PostDraft | None = None) -> dict[str, Any]: data = attempt.model_dump(mode="json") if video: data["video_title"] = video.working_title data["video_candidate_id"] = video.candidate_id if draft: data["caption"] = draft.caption data["hashtags"] = draft.hashtags_json data["endpoint"] = TIKTOK_UPLOAD_ENDPOINT return data def tiktok_config_status(session: Session) -> dict[str, Any]: settings = get_settings() accounts = session.exec(select(PlatformAccount).where(PlatformAccount.provider == Provider.tiktok)).all() sandbox_configured = bool(settings.tiktok_client_id and settings.tiktok_client_secret) last_attempt = session.exec(select(TikTokUploadAttempt).order_by(TikTokUploadAttempt.created_at.desc()).limit(1)).first() last_api = session.exec(select(ApiCallLog).where(ApiCallLog.provider == "tiktok").order_by(ApiCallLog.created_at.desc()).limit(1)).first() recording_video, recording_draft = select_recording_demo_video(session) checklist = [ {"label": "Produktname und Website-Copy sind creator-facing", "ok": True}, {"label": "Nur Scope video.upload sichtbar", "ok": True}, {"label": "Vorbereitetes Video verfügbar", "ok": recording_video is not None}, {"label": "TikTok-Text ist bearbeitbar", "ok": True}, {"label": "Synthetic-Media-Offenlegung sichtbar", "ok": True}, {"label": "Zustimmung ist erforderlich", "ok": True}, {"label": "Upload-Draft-Ablauf funktioniert", "ok": bool(last_attempt)}, {"label": "Protokolle sind geschwärzt", "ok": True}, {"label": "Keine Secrets sichtbar", "ok": True}, {"label": "Website-/Domain-Entscheidung dokumentiert", "ok": Path("../docs/tiktok-verification-domain-plan.md").exists() or Path("docs/tiktok-verification-domain-plan.md").exists()}, {"label": "Sandbox fehlt klar markiert oder konfiguriert", "ok": True}, ] blocking = [] if not sandbox_configured: blocking.append("TikTok Sandbox not configured. Demo can run in simulated API mode for UI recording, but final review video should use sandbox credentials if available.") if not accounts: blocking.append("No TikTok sandbox account is connected in the dashboard.") return { "title": "TikTok Verification Demo", "status": "Ready to record" if last_attempt and last_attempt.mode in {"mock", "sandbox"} else "Not ready", "mode": "Sandbox" if sandbox_configured else "Mock", "real_mode": "Disabled", "requested_scope": VIDEO_UPLOAD_SCOPE, "upload_method": FILE_UPLOAD_SOURCE, "connected_tiktok_account": accounts[0].display_name if accounts else "sandbox / not connected", "sandbox_configured": sandbox_configured, "client_id_present": bool(settings.tiktok_client_id), "client_secret_present": bool(settings.tiktok_client_secret), "last_oauth": accounts[0].last_successful_api_call.isoformat() if accounts and accounts[0].last_successful_api_call else None, "last_upload_init": last_attempt.created_at.isoformat() if last_attempt else None, "last_upload_transfer": last_attempt.updated_at.isoformat() if last_attempt and last_attempt.transfer_status else None, "last_status_check": last_api.created_at.isoformat() if last_api else None, "last_attempt": last_attempt.model_dump(mode="json") if last_attempt else None, "current_test_video": recording_video.model_dump(mode="json") if recording_video else None, "recording_demo_draft": recording_draft.model_dump(mode="json") if recording_draft else None, "checklist": checklist, "blocking_issues": blocking, "readiness": { "ready_to_record": bool(last_attempt and last_attempt.mode in {"mock", "sandbox"}), "ready_for_internal_rehearsal": True, "ready_for_tiktok_submission_video": bool(sandbox_configured and last_attempt and last_attempt.mode == "sandbox"), "checklist": checklist, }, } def create_tiktok_draft_attempt(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: raise HTTPException(404, "Draft not found") if draft.provider != Provider.tiktok: raise HTTPException(400, "Not a TikTok draft") video = session.get(VideoAsset, draft.video_asset_id) if not video: raise HTTPException(404, "Video not found") if not payload.get("consent_confirmed"): raise HTTPException(400, "Explicit creator consent is required before creating a TikTok Upload Draft attempt") requested_mode = payload.get("mode", "mock") if requested_mode not in {"mock", "sandbox", "real_disabled"}: raise HTTPException(400, "Unsupported TikTok demo mode") if requested_mode == "real_disabled": raise HTTPException(400, "Direct Post is not enabled. This verification flow demonstrates Upload Draft only.") settings = get_settings() sandbox_configured = bool(settings.tiktok_client_id and settings.tiktok_client_secret) mode = requested_mode if mode == "sandbox" and not sandbox_configured: raise HTTPException(400, "TikTok Sandbox not configured; use mock mode for internal rehearsal") session.add(ActivityEvent(entity_type="video", entity_id=video.id, event_type="tiktok_demo_started", label="TikTok verification demo started", payload_json={"mode": mode, "scope": VIDEO_UPLOAD_SCOPE, "source": FILE_UPLOAD_SOURCE})) request_json = _redacted_request(video, draft, mode) fake_id = f"mock_publish_{hashlib.sha256((video.id + draft.id + str(uuid.uuid4())).encode()).hexdigest()[:12]}" if mode == "mock": response_json = { "mock": True, "publish_id": fake_id, "upload_url_redacted": redact_upload_url("https://upload.tiktok.mock/inbox/video?upload_token=mock-secret-token"), "status": "mock_uploaded_to_draft", "message": "No external TikTok API call was made.", } attempt = TikTokUploadAttempt( video_asset_id=video.id, post_draft_id=draft.id, platform_account_id=draft.platform_account_id, mode="mock", scope=VIDEO_UPLOAD_SCOPE, source=FILE_UPLOAD_SOURCE, publish_id=fake_id, upload_status="mock_uploaded_to_draft", transfer_status="mock_file_transfer_succeeded", status_fetch_result_json={"mock": True, "status": "draft_created"}, request_redacted_json=request_json, response_redacted_json=response_json, ) api_log = ApiCallLog(provider="tiktok", endpoint=TIKTOK_UPLOAD_ENDPOINT, method="POST", status_code=202, request_redacted_json=request_json, response_redacted_json=response_json) draft.status = "tiktok_mock_draft_created" video.external_tiktok_id = fake_id session.add(attempt); session.add(api_log); session.add(draft); session.add(video) session.add(ActivityEvent(entity_type="tiktok_upload_attempt", entity_id=attempt.id, event_type="tiktok_upload_init_mocked", label="TikTok Upload Draft mocked", payload_json={"publish_id": fake_id, "mock": True})) session.add(ActivityEvent(entity_type="tiktok_upload_attempt", entity_id=attempt.id, event_type="tiktok_file_upload_succeeded", label="Mock file transfer succeeded", payload_json={"mock": True})) session.commit(); session.refresh(attempt) return _attempt_payload(attempt, video, draft) # Sandbox implementation placeholder: honest status, no token exposure, and no real call until credentials flow exists. response_json = {"mock": False, "sandbox": True, "status": "sandbox_configured_but_oauth_not_connected", "message": "Sandbox credentials are present, but token exchange/upload transfer is not implemented in this local verifier yet."} attempt = TikTokUploadAttempt( video_asset_id=video.id, post_draft_id=draft.id, platform_account_id=draft.platform_account_id, mode="sandbox", scope=VIDEO_UPLOAD_SCOPE, source=FILE_UPLOAD_SOURCE, upload_status="sandbox_not_connected", transfer_status="not_started", request_redacted_json=request_json, response_redacted_json=response_json, error_code="sandbox_oauth_not_connected", error_message="Sandbox credentials present, OAuth/token flow not connected.", ) api_log = ApiCallLog(provider="tiktok", endpoint=TIKTOK_UPLOAD_ENDPOINT, method="POST", status_code=409, request_redacted_json=request_json, response_redacted_json=response_json, error_code="sandbox_oauth_not_connected") session.add(attempt); session.add(api_log) session.add(ActivityEvent(entity_type="tiktok_upload_attempt", entity_id=attempt.id, event_type="tiktok_upload_failed", label="TikTok Sandbox upload blocked", payload_json={"reason": "sandbox_oauth_not_connected"})) session.commit(); session.refresh(attempt) return _attempt_payload(attempt, video, draft) def list_tiktok_attempts(session: Session) -> dict[str, Any]: attempts = session.exec(select(TikTokUploadAttempt).order_by(TikTokUploadAttempt.created_at.desc()).limit(100)).all() items = [] for attempt in attempts: video = session.get(VideoAsset, attempt.video_asset_id) draft = session.get(PostDraft, attempt.post_draft_id) items.append(_attempt_payload(attempt, video, draft)) return {"items": items}