xiaoxia-saas/apps/worker/worker_app/tasks/ingest.py

import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path

from celery.utils.log import get_task_logger
from video_processing.oss_helpers import download_asset
from worker_app.celery_app import celery_app
from worker_app.core.asset_types import infer_mime_type_from_storage_key
from worker_app.db import SessionLocal

from packages.adapters.sqlalchemy_impl import (
    SQLAlchemyAssetRepository,
    SQLAlchemyIngestJobRepository,
)
from packages.domain import Asset, AssetStatus, IngestJobStatus
from packages.domain.media_validation import is_valid_media as _is_valid_media
from packages.domain.media_validation import safe_parse_fps as _safe_parse_fps

logger = get_task_logger(__name__)


def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]:
    """
    提取媒体文件的元数据。

    Args:
        file_url: 媒体文件 URL 或本地路径
        media_type: 媒体类型 (video, audio, image)

    Returns:
        (metadata_dict, success)
        - metadata_dict: 提取的元数据字典，失败时返回空字典
        - success: 是否成功提取到有效元数据
    """
    metadata = {}
    success = False

    try:
        if media_type == "video":
            # 使用 ffprobe 提取视频元数据
            from video_processing.ffmpeg_utils import run_ffprobe

            cmd = [
                "ffprobe",
                "-v",
                "quiet",
                "-print_format",
                "json",
                "-show_format",
                "-show_streams",
                file_url,
            ]
            try:
                stdout, _ = run_ffprobe(cmd, timeout=30)
                import json as json_lib

                probe_data = json_lib.loads(stdout)

                # 提取视频流信息
                for stream in probe_data.get("streams", []):
                    if stream.get("codec_type") == "video":
                        metadata["width"] = int(stream.get("width", 0))
                        metadata["height"] = int(stream.get("height", 0))
                        metadata["codec"] = stream.get("codec_name", "")
                        metadata["fps"] = (
                            _safe_parse_fps(stream.get("r_frame_rate", "0/1")) if stream.get("r_frame_rate") else 0  # type: ignore[assignment]
                        )
                        break

                # 提取格式信息
                format_info = probe_data.get("format", {})
                metadata["duration"] = float(format_info.get("duration", 0))  # type: ignore[assignment]
                metadata["size_bytes"] = int(format_info.get("size", 0))
                metadata["bitrate"] = int(format_info.get("bit_rate", 0))

                # 有效性判断：有视频流 + duration > 0 + size > 0
                has_video_stream = any(s.get("codec_type") == "video" for s in probe_data.get("streams", []))
                has_audio_stream = any(s.get("codec_type") == "audio" for s in probe_data.get("streams", []))
                if (has_video_stream or has_audio_stream) and metadata.get("duration", 0) > 0:
                    success = True

            except Exception as e:
                logger.warning("视频元数据提取失败: %s", e)

        elif media_type == "audio":
            from video_processing.ffmpeg_utils import run_ffprobe

            cmd = [
                "ffprobe",
                "-v",
                "quiet",
                "-print_format",
                "json",
                "-show_format",
                "-show_streams",
                file_url,
            ]
            try:
                stdout, _ = run_ffprobe(cmd, timeout=30)
                import json as json_lib

                probe_data = json_lib.loads(stdout)

                has_audio_stream = any(s.get("codec_type") == "audio" for s in probe_data.get("streams", []))
                format_info = probe_data.get("format", {})
                metadata["duration"] = float(format_info.get("duration", 0))  # type: ignore[assignment]
                metadata["size_bytes"] = int(format_info.get("size", 0))
                metadata["bitrate"] = int(format_info.get("bit_rate", 0))
                metadata["codec"] = next(
                    (s.get("codec_name", "") for s in probe_data.get("streams", []) if s.get("codec_type") == "audio"),
                    "",
                )

                if has_audio_stream and metadata.get("duration", 0) > 0:
                    success = True

            except Exception as e:
                logger.warning("音频元数据提取失败: %s", e)

        elif media_type == "image":
            # 使用 Pillow 提取图片元数据
            try:
                from PIL import Image

                with Image.open(file_url) as img:
                    img.verify()  # 验证文件完整性
                    # verify后需要重新打开才能读尺寸
                    with Image.open(file_url) as img2:
                        metadata["width"] = img2.width
                        metadata["height"] = img2.height
                        metadata["format"] = img2.format
                        metadata["mode"] = img2.mode
                        if img2.width > 0 and img2.height > 0:
                            success = True

            except ImportError:
                logger.warning("Pillow not available for image metadata extraction")
            except Exception as e:
                logger.warning(f"Failed to extract image metadata: {e}")

    except subprocess.TimeoutExpired:
        logger.warning(f"Timeout extracting metadata from {file_url}")
    except FileNotFoundError:
        logger.warning("ffprobe not found, cannot extract video metadata")
    except Exception as e:
        logger.warning(f"Failed to extract metadata: {e}")

    return metadata, success


@celery_app.task(name="worker.ingest_asset")
def ingest_asset(job_id: str) -> dict:
    """
    Ingest asset task.

    Steps:
    1. Fetch IngestJob from repository
    2. Extract metadata from storage_key
    3. Create Asset entity
    4. Update IngestJob status to COMPLETED
    5. Return result
    """
    db = SessionLocal()
    try:
        job_repo = SQLAlchemyIngestJobRepository(db)
        asset_repo = SQLAlchemyAssetRepository(db)

        job = job_repo.get(job_id)
        if job is None:
            return {"status": "failed", "error": "job not found"}

        # Update job status to PROCESSING
        job.status = IngestJobStatus.PROCESSING
        job.updated_at = datetime.now(timezone.utc)
        job_repo.update(job)
        db.commit()

        # Extract real metadata from media file
        filename = job.storage_key.split("/")[-1]
        mime_type = infer_mime_type_from_storage_key(job.storage_key)

        # Determine media type from mime_type
        media_type = "video"
        if mime_type.startswith("image/"):
            media_type = "image"
        elif mime_type.startswith("audio/"):
            media_type = "audio"

        # 先从 OSS 下载文件到本地临时目录，再提取元数据
        # （storage_key 是 OSS 内部路径，不能直接传给 ffprobe/Pillow）
        local_file = None
        thumbnail_url = None
        try:
            suffix = Path(job.storage_key).suffix or ".bin"
            with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
                local_file = Path(tmp.name)

            download_ok = download_asset(job.storage_key, local_file)
            if not download_ok:
                logger.warning("素材下载失败，无法提取元数据: job_id=%s storage_key=%s", job_id, job.storage_key)
                metadata, extract_success = {}, False
            else:
                metadata, extract_success = extract_media_metadata(str(local_file), media_type)

            # 视频类型：生成缩略图（文件还在的时候生成）
            thumbnail_url = None
            if media_type == "video" and extract_success:
                frame_path = None
                try:
                    from video_processing.oss_helpers import upload_to_oss
                    from video_processing.thumbnail_generator import extract_first_frame

                    frame_path = extract_first_frame(str(local_file), width=640)
                    thumb_storage_key = f"assets/{job.project_id}/thumbnails/{job_id}.jpg"
                    try:
                        thumbnail_url = upload_to_oss(frame_path, thumb_storage_key)
                    finally:
                        if frame_path:
                            try:
                                Path(frame_path).unlink(missing_ok=True)
                            except Exception:
                                pass
                    if thumbnail_url:
                        logger.info(
                            "素材缩略图生成成功: job_id=%s url=%s",
                            job_id,
                            thumbnail_url[:80],
                        )
                except Exception as thumb_err:
                    logger.warning(
                        "素材缩略图生成失败（不影响主流程）: job_id=%s error=%s",
                        job_id,
                        thumb_err,
                    )

            # ── HEVC 自动转码为 1080p H.264 ──────────────────────────────
            # 浏览器 WebCodecs 硬件解码 HEVC 输出黑帧，上传时自动转码
            # 失败时降级使用原始文件，不阻塞上传流程
            if media_type == "video" and extract_success and local_file and local_file.exists():
                codec = (metadata.get("codec") or "").lower()
                if codec in ("hevc", "h265", "hvh1"):
                    logger.info(
                        "检测到 HEVC 编码 (codec=%s)，启动转码: job_id=%s",
                        codec,
                        job_id,
                    )
                    _tc_tmp = None
                    try:
                        _tc_tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix="_h264.mp4")
                        _tc_tmp = Path(_tc_tmp_file.name)
                        _tc_tmp_file.close()  # 关闭文件描述符，ffmpeg 会自己打开
                        _cmd = [
                            "ffmpeg",
                            "-y",
                            "-i",
                            str(local_file),
                            "-c:v",
                            "libx264",
                            "-preset",
                            "fast",
                            "-crf",
                            "18",
                            "-vf",
                            "scale='if(gt(ih,1080),-2,iw)':'if(gt(ih,1080),1080,ih)'",
                            "-pix_fmt",
                            "yuv420p",
                            "-c:a",
                            "aac",
                            "-b:a",
                            "128k",
                            "-movflags",
                            "+faststart",
                            str(_tc_tmp),
                        ]
                        _proc = subprocess.run(
                            _cmd,
                            stdout=subprocess.DEVNULL,
                            stderr=subprocess.PIPE,
                            text=True,
                            timeout=300,
                        )
                        if _proc.returncode == 0 and _tc_tmp.exists() and _tc_tmp.stat().st_size > 0:
                            from video_processing.oss_helpers import upload_to_oss

                            _p = Path(job.storage_key)
                            _new_key = str(_p.parent / (_p.stem + "_h264" + _p.suffix))
                            _url = upload_to_oss(_tc_tmp, _new_key)
                            if _url:
                                # 先提取元数据，确认成功后再更新 storage_key（避免脏数据）
                                _new_metadata, _new_extract_success = extract_media_metadata(
                                    str(_tc_tmp),
                                    media_type,
                                )
                                if _new_extract_success:
                                    job.storage_key = _new_key
                                    metadata = _new_metadata
                                    extract_success = _new_extract_success
                                logger.info(
                                    "HEVC→H.264 转码完成: job_id=%s key=%s",
                                    job_id,
                                    _new_key[:80],
                                )
                            else:
                                logger.warning(
                                    "转码文件上传 OSS 失败，使用原始文件: job_id=%s",
                                    job_id,
                                )
                        else:
                            _tail = _proc.stderr[-300:] if _proc.stderr else ""
                            logger.warning(
                                "FFmpeg 转码失败 rc=%s，降级原始文件: job_id=%s",
                                _proc.returncode,
                                job_id,
                            )
                    except subprocess.TimeoutExpired:
                        logger.warning(
                            "FFmpeg 转码超时 (300s)，降级原始文件: job_id=%s",
                            job_id,
                        )
                    except Exception as _e:
                        logger.warning(
                            "HEVC 转码异常（降级原始文件）: job_id=%s err=%s",
                            job_id,
                            _e,
                        )
                    finally:
                        if _tc_tmp and _tc_tmp.exists():
                            try:
                                _tc_tmp.unlink()
                            except OSError:
                                pass
        finally:
            if local_file and local_file.exists():
                try:
                    local_file.unlink()
                except OSError:
                    pass

        # 有效性校验：ffprobe/Pillow 必须成功，且文件大小/时长/尺寸满足最小要求
        is_valid = extract_success and _is_valid_media(metadata, media_type)

        if not is_valid:
            # 文件无效，创建 ERROR 状态的 asset 并标记 job 失败
            error_reason = "metadata extraction failed" if not extract_success else "media validation failed"
            logger.warning(
                "素材有效性校验失败，标记为ERROR: job_id=%s storage_key=%s media_type=%s reason=%s",
                job_id,
                job.storage_key,
                media_type,
                error_reason,
            )

            asset = Asset.create(
                project_id=job.project_id,
                library_id=job.library_id,
                name=filename,
                storage_key=job.storage_key,
                mime_type=mime_type,
                metadata={"ingest_error": error_reason},
                file_size=int(metadata.get("size_bytes", 0)),
                duration=float(metadata.get("duration", 0)),
                width=int(metadata.get("width", 0)),
                height=int(metadata.get("height", 0)),
                codec=metadata.get("codec") or None,
                status=AssetStatus.ERROR,
                file_hash=job.file_hash,
            )
            asset_repo.create(asset)

            # Update job status to FAILED
            job.status = IngestJobStatus.FAILED
            job.error_message = f"Invalid media file: {error_reason}"
            job.result_asset_id = asset.id
            job.updated_at = datetime.now(timezone.utc)
            job_repo.update(job)
            db.commit()

            return {
                "status": "failed",
                "job_id": job.id,
                "asset_id": asset.id,
                "error": error_reason,
            }

        # Create Asset
        asset = Asset.create(
            project_id=job.project_id,
            library_id=job.library_id,
            name=filename,
            storage_key=job.storage_key,
            mime_type=mime_type,
            metadata=metadata,
            file_size=int(metadata.get("size_bytes", 0)),
            duration=float(metadata.get("duration", 0)),
            width=int(metadata.get("width", 0)),
            height=int(metadata.get("height", 0)),
            codec=metadata.get("codec") or None,
            status=AssetStatus.READY,
            file_hash=job.file_hash,
            thumbnail_url=thumbnail_url,
        )
        asset_repo.create(asset)

        # Update job status to COMPLETED
        job.status = IngestJobStatus.COMPLETED
        job.result_asset_id = asset.id
        job.updated_at = datetime.now(timezone.utc)
        job_repo.update(job)

        db.commit()

        return {
            "status": "completed",
            "job_id": job.id,
            "asset_id": asset.id,
        }
    except Exception as e:
        db.rollback()
        logger.error(f"Failed to ingest asset {job_id}: {e}")

        # Update job status to FAILED
        try:
            job_repo = SQLAlchemyIngestJobRepository(db)
            job = job_repo.get(job_id)
            if job:
                job.status = IngestJobStatus.FAILED
                job.error_message = str(e)
                job.updated_at = datetime.now(timezone.utc)
                job_repo.update(job)
                db.commit()
        except Exception:
            db.rollback()

        return {
            "status": "failed",
            "job_id": job_id,
            "error": str(e),
        }
    finally:
        db.close()