AI-Video/pixelle_video/services/publishing/export_publisher.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.

"""
Export Publisher - Format conversion and local export for platforms
without API access (Douyin, Kuaishou).
"""

import asyncio
import subprocess
import shutil
from pathlib import Path
from datetime import datetime
from typing import Optional
from loguru import logger

from pixelle_video.services.publishing import (
    Publisher,
    Platform,
    PublishStatus,
    VideoMetadata,
    PublishResult,
)


class ExportPublisher(Publisher):
    """
    Publisher that converts video to platform-optimized format
    and exports to local filesystem for manual upload.
    """
    
    platform = Platform.EXPORT
    
    def __init__(
        self,
        output_dir: str = "./output/exports",
        target_resolution: tuple = (1080, 1920),  # Portrait 9:16
        target_codec: str = "h264",
        max_file_size_mb: int = 128,
    ):
        self.output_dir = Path(output_dir)
        self.target_resolution = target_resolution
        self.target_codec = target_codec
        self.max_file_size_mb = max_file_size_mb
        
        # Ensure output directory exists
        self.output_dir.mkdir(parents=True, exist_ok=True)
    
    async def publish(
        self,
        video_path: str,
        metadata: VideoMetadata,
        progress_callback: Optional[callable] = None
    ) -> PublishResult:
        """
        Convert video to optimized format and export.
        """
        started_at = datetime.now()
        
        try:
            if progress_callback:
                progress_callback(0.1, "分析视频...")
            
            # Generate output filename
            safe_title = "".join(c if c.isalnum() or c in " -_" else "" for c in metadata.title)
            safe_title = safe_title[:50].strip() or "video"
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            output_filename = f"{safe_title}_{timestamp}.mp4"
            output_path = self.output_dir / output_filename
            
            if progress_callback:
                progress_callback(0.2, "转换格式...")
            
            # Convert video
            success = await self._convert_video(
                video_path,
                str(output_path),
                progress_callback
            )
            
            if not success:
                return PublishResult(
                    success=False,
                    platform=Platform.EXPORT,
                    status=PublishStatus.FAILED,
                    error_message="视频转换失败",
                    started_at=started_at,
                    completed_at=datetime.now(),
                )
            
            if progress_callback:
                progress_callback(1.0, "导出完成")
            
            # Verify file size
            file_size_mb = output_path.stat().st_size / (1024 * 1024)
            
            return PublishResult(
                success=True,
                platform=Platform.EXPORT,
                status=PublishStatus.PUBLISHED,
                export_path=str(output_path),
                started_at=started_at,
                completed_at=datetime.now(),
            )
            
        except Exception as e:
            logger.error(f"Export failed: {e}")
            return PublishResult(
                success=False,
                platform=Platform.EXPORT,
                status=PublishStatus.FAILED,
                error_message=str(e),
                started_at=started_at,
                completed_at=datetime.now(),
            )
    
    async def _convert_video(
        self,
        input_path: str,
        output_path: str,
        progress_callback: Optional[callable] = None
    ) -> bool:
        """Convert video using FFmpeg."""
        
        width, height = self.target_resolution
        
        # FFmpeg command for H.264 conversion with size optimization
        cmd = [
            "ffmpeg", "-y",
            "-i", input_path,
            "-c:v", "libx264",
            "-preset", "medium",
            "-crf", "23",
            "-c:a", "aac",
            "-b:a", "128k",
            "-vf", f"scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2",
            "-movflags", "+faststart",
            output_path
        ]
        
        try:
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await process.communicate()
            
            if process.returncode != 0:
                logger.error(f"FFmpeg error: {stderr.decode()}")
                return False
            
            return True
            
        except FileNotFoundError:
            logger.error("FFmpeg not found. Please install FFmpeg.")
            # Fallback: just copy the file if FFmpeg is not available
            shutil.copy(input_path, output_path)
            return True
    
    async def validate_credentials(self) -> bool:
        """No credentials needed for export."""
        return True
    
    def get_platform_requirements(self):
        return {
            "max_file_size_mb": self.max_file_size_mb,
            "recommended_resolution": self.target_resolution,
            "recommended_codec": self.target_codec,
            "output_format": "mp4",
            "platforms": ["douyin", "kuaishou"],
        }