AI-Video/api/routers/publish.py

# Copyright (C) 2025 AIDC-AI
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#     http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Publish API router for multi-platform video distribution.

Endpoints:
- POST /publish/export - Format conversion and export
- POST /publish/bilibili - Publish to Bilibili (TODO)
- POST /publish/youtube - Publish to YouTube (TODO)
- GET /publish/tasks/{id} - Get task status
"""

import uuid
from datetime import datetime
from fastapi import APIRouter, HTTPException, Path, BackgroundTasks
from loguru import logger

from api.schemas.publish import (
    PublishRequest,
    PublishResultSchema,
    PublishTaskSchema,
    PublishStatusEnum,
    PlatformRequirementsSchema,
    VideoMetadataSchema,
)
from pixelle_video.services.publishing import (
    VideoMetadata,
    PublishStatus,
    PublishTask,
    Platform,
)
from pixelle_video.services.publishing.export_publisher import ExportPublisher


router = APIRouter(prefix="/publish", tags=["Publish"])


# In-memory task storage (use Redis in production)
_publish_tasks: dict = {}

# Publisher instances
_export_publisher = ExportPublisher()


@router.post("/export", response_model=PublishResultSchema)
async def export_video(
    request: PublishRequest,
    background_tasks: BackgroundTasks
):
    """
    Convert video to platform-optimized format and export.
    
    Optimizes for:
    - Portrait 9:16 aspect ratio (1080x1920)
    - H.264 codec
    - ≤128MB file size
    
    For manual upload to Douyin/Kuaishou.
    """
    # Create task
    task_id = str(uuid.uuid4())[:8]
    
    metadata = VideoMetadata(
        title=request.metadata.title,
        description=request.metadata.description,
        tags=request.metadata.tags,
        category=request.metadata.category,
        cover_path=request.metadata.cover_path,
        privacy=request.metadata.privacy,
        platform_options=request.metadata.platform_options,
    )
    
    task = PublishTask(
        id=task_id,
        video_path=request.video_path,
        platform=Platform.EXPORT,
        metadata=metadata,
        status=PublishStatus.PENDING,
    )
    _publish_tasks[task_id] = task
    
    logger.info(f"📤 Starting export task {task_id} for: {metadata.title}")
    
    # Execute synchronously for now (can be moved to background)
    result = await _export_publisher.publish(
        request.video_path,
        metadata,
        progress_callback=lambda p, m: logger.info(f"Export {task_id}: {p:.0%} - {m}")
    )
    
    # Update task
    task.status = PublishStatus(result.status.value)
    task.result = result
    task.updated_at = datetime.now()
    
    return PublishResultSchema(
        success=result.success,
        platform="export",
        status=PublishStatusEnum(result.status.value),
        export_path=result.export_path,
        error_message=result.error_message,
    )


@router.get("/tasks/{task_id}", response_model=PublishTaskSchema)
async def get_publish_task(task_id: str = Path(..., description="Task ID")):
    """Get publishing task status"""
    if task_id not in _publish_tasks:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    
    task = _publish_tasks[task_id]
    
    return PublishTaskSchema(
        id=task.id,
        platform=task.platform.value,
        status=PublishStatusEnum(task.status.value),
        result=PublishResultSchema(
            success=task.result.success,
            platform=task.result.platform.value,
            status=PublishStatusEnum(task.result.status.value),
            export_path=task.result.export_path,
            error_message=task.result.error_message,
        ) if task.result else None,
        created_at=task.created_at.isoformat(),
        updated_at=task.updated_at.isoformat() if task.updated_at else None,
    )


@router.get("/requirements/{platform}", response_model=PlatformRequirementsSchema)
async def get_platform_requirements(platform: str = Path(..., description="Platform name")):
    """Get platform-specific requirements"""
    
    requirements = {
        "export": {
            "max_file_size_mb": 128,
            "max_duration_seconds": 900,
            "supported_formats": ["mp4"],
            "recommended_resolution": (1080, 1920),
            "recommended_codec": "h264",
        },
        "bilibili": {
            "max_file_size_mb": 4096,
            "max_duration_seconds": 14400,  # 4 hours
            "supported_formats": ["mp4", "flv", "webm"],
            "recommended_resolution": (1920, 1080),  # Landscape
            "recommended_codec": "h264",
        },
        "youtube": {
            "max_file_size_mb": 256000,  # 256GB
            "max_duration_seconds": 43200,  # 12 hours
            "supported_formats": ["mp4", "mov", "avi", "webm"],
            "recommended_resolution": (1920, 1080),
            "recommended_codec": "h264",
        },
    }
    
    if platform not in requirements:
        raise HTTPException(status_code=404, detail=f"Platform {platform} not supported")
    
    return requirements[platform]


from pixelle_video.services.publishing.bilibili_publisher import BilibiliPublisher

# Bilibili publisher instance
_bilibili_publisher = BilibiliPublisher()


@router.post("/bilibili", response_model=PublishResultSchema)
async def publish_to_bilibili(request: PublishRequest):
    """
    Publish video to Bilibili.
    
    Requires environment variables:
    - BILIBILI_ACCESS_TOKEN or
    - BILIBILI_SESSDATA + BILIBILI_BILI_JCT
    """
    if not await _bilibili_publisher.validate_credentials():
        raise HTTPException(
            status_code=400,
            detail="B站凭证未配置。请设置 BILIBILI_ACCESS_TOKEN 或 BILIBILI_SESSDATA 环境变量"
        )
    
    metadata = VideoMetadata(
        title=request.metadata.title,
        description=request.metadata.description,
        tags=request.metadata.tags,
        category=request.metadata.category,
        cover_path=request.metadata.cover_path,
        privacy=request.metadata.privacy,
        platform_options=request.metadata.platform_options,
    )
    
    logger.info(f"📤 Starting Bilibili upload: {metadata.title}")
    
    result = await _bilibili_publisher.publish(
        request.video_path,
        metadata,
        progress_callback=lambda p, m: logger.info(f"Bilibili: {p:.0%} - {m}")
    )
    
    return PublishResultSchema(
        success=result.success,
        platform="bilibili",
        status=PublishStatusEnum(result.status.value),
        video_url=result.video_url,
        platform_video_id=result.platform_video_id,
        error_message=result.error_message,
    )

from pixelle_video.services.publishing.youtube_publisher import YouTubePublisher

# YouTube publisher instance
_youtube_publisher = YouTubePublisher()


@router.post("/youtube", response_model=PublishResultSchema)
async def publish_to_youtube(request: PublishRequest):
    """
    Publish video to YouTube.
    
    Requires:
    - config/youtube_client_secrets.json (OAuth 2.0 credentials)
    - First-time auth will open browser for authorization
    """
    if not await _youtube_publisher.validate_credentials():
        raise HTTPException(
            status_code=400,
            detail="YouTube 凭证未配置。请添加 config/youtube_client_secrets.json"
        )
    
    metadata = VideoMetadata(
        title=request.metadata.title,
        description=request.metadata.description,
        tags=request.metadata.tags,
        category=request.metadata.category,
        cover_path=request.metadata.cover_path,
        privacy=request.metadata.privacy,
        platform_options=request.metadata.platform_options,
    )
    
    logger.info(f"📤 Starting YouTube upload: {metadata.title}")
    
    result = await _youtube_publisher.publish(
        request.video_path,
        metadata,
        progress_callback=lambda p, m: logger.info(f"YouTube: {p:.0%} - {m}")
    )
    
    return PublishResultSchema(
        success=result.success,
        platform="youtube",
        status=PublishStatusEnum(result.status.value),
        video_url=result.video_url,
        platform_video_id=result.platform_video_id,
        error_message=result.error_message,
    )


# ============================================================
# Async Task Queue Endpoints
# ============================================================

from pixelle_video.services.publishing.task_manager import get_publish_manager, TaskPriority
from pydantic import BaseModel
from typing import List


class AsyncPublishRequest(BaseModel):
    """Request for async publishing"""
    video_path: str
    platform: str  # export, bilibili, youtube
    metadata: VideoMetadataSchema
    priority: str = "normal"  # low, normal, high


class QueuedTaskSchema(BaseModel):
    """Schema for queued task"""
    id: str
    platform: str
    status: str
    progress: float
    progress_message: str
    retries: int
    created_at: str
    started_at: str = None
    completed_at: str = None


class QueueStatusSchema(BaseModel):
    """Queue status overview"""
    pending: int
    active: int
    completed: int
    failed: int
    workers: int


@router.post("/async", response_model=dict)
async def publish_async(request: AsyncPublishRequest):
    """
    Submit a publish task to the background queue.
    
    Returns immediately with task ID for tracking.
    """
    manager = get_publish_manager()
    
    # Ensure manager is running
    if not manager._running:
        await manager.start()
    
    # Map platform string to enum
    platform_map = {
        "export": Platform.EXPORT,
        "bilibili": Platform.BILIBILI,
        "youtube": Platform.YOUTUBE,
    }
    
    platform = platform_map.get(request.platform.lower())
    if not platform:
        raise HTTPException(status_code=400, detail=f"Invalid platform: {request.platform}")
    
    # Map priority
    priority_map = {
        "low": TaskPriority.LOW,
        "normal": TaskPriority.NORMAL,
        "high": TaskPriority.HIGH,
    }
    priority = priority_map.get(request.priority.lower(), TaskPriority.NORMAL)
    
    metadata = VideoMetadata(
        title=request.metadata.title,
        description=request.metadata.description,
        tags=request.metadata.tags,
        category=request.metadata.category,
        cover_path=request.metadata.cover_path,
        privacy=request.metadata.privacy,
        platform_options=request.metadata.platform_options,
    )
    
    task_id = await manager.enqueue(
        video_path=request.video_path,
        platform=platform,
        metadata=metadata,
        priority=priority,
    )
    
    return {
        "task_id": task_id,
        "status": "queued",
        "message": f"Task queued for {request.platform}",
    }


@router.get("/queue/status", response_model=QueueStatusSchema)
async def get_queue_status():
    """Get queue status overview."""
    manager = get_publish_manager()
    
    all_tasks = manager.get_all_tasks()
    
    pending = sum(1 for t in all_tasks if t.task.status == PublishStatus.PENDING)
    active = len(manager.get_active_tasks())
    completed = sum(1 for t in all_tasks if t.task.status == PublishStatus.PUBLISHED)
    failed = sum(1 for t in all_tasks if t.task.status == PublishStatus.FAILED)
    
    return QueueStatusSchema(
        pending=pending,
        active=active,
        completed=completed,
        failed=failed,
        workers=manager.max_workers,
    )


@router.get("/queue/tasks", response_model=List[QueuedTaskSchema])
async def list_queued_tasks():
    """List all tasks in queue."""
    manager = get_publish_manager()
    
    result = []
    for qt in manager.get_all_tasks():
        result.append(QueuedTaskSchema(
            id=qt.task.id,
            platform=qt.task.platform.value,
            status=qt.task.status.value,
            progress=qt.progress,
            progress_message=qt.progress_message,
            retries=qt.retries,
            created_at=qt.created_at.isoformat(),
            started_at=qt.started_at.isoformat() if qt.started_at else None,
            completed_at=qt.completed_at.isoformat() if qt.completed_at else None,
        ))
    
    return result


@router.get("/queue/tasks/{task_id}", response_model=QueuedTaskSchema)
async def get_queued_task(task_id: str = Path(..., description="Task ID")):
    """Get specific queued task status."""
    manager = get_publish_manager()
    
    qt = manager.get_task(task_id)
    if not qt:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    
    return QueuedTaskSchema(
        id=qt.task.id,
        platform=qt.task.platform.value,
        status=qt.task.status.value,
        progress=qt.progress,
        progress_message=qt.progress_message,
        retries=qt.retries,
        created_at=qt.created_at.isoformat(),
        started_at=qt.started_at.isoformat() if qt.started_at else None,
        completed_at=qt.completed_at.isoformat() if qt.completed_at else None,
    )