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

import json
import shutil
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


# ── HEVC 自动转码辅助函数（模块级，便于单元测试）─────────────────────────
HEVC_CODECS = ("hevc", "h265", "hvh1")
# 转码目标：长边封顶 1920（只缩不放，与 validate 的 max_long_edge 一致），
# 竖屏/横屏/超宽屏统一按长边等比缩放，短边自动按比例（-2 保证偶数）。
TRANSCODE_MAX_LONG_EDGE = 1920
TRANSCODE_TIMEOUT_SECONDS = 900
# ffmpeg scale 滤镜中 if(...) 表达式内的逗号必须用 \, 转义，
# 否则逗号被当作 filter 分隔符解析，报 "No such filter" / Invalid size。
# subprocess list 传参不经 shell，\ 在 Python 字符串里直接写一个字面反斜杠即可。
# 横屏（iw>=ih）限宽 min(1920,iw)、高 -2 自适应；竖屏（ih>iw）限高、宽自适应；
# min() 保证小视频不放大。与 validate_transcode_output 的"长边 <= 1920"规则对齐，
# 超宽屏（如 4000x1000）短边不触发旧的短边缩放、长边超限被误降级的问题由此消除。
_TRANSCODE_VF = (
    rf"scale=w=if(gte(iw\,ih)\,min({TRANSCODE_MAX_LONG_EDGE}\,iw)\,-2):"
    rf"h=if(gt(ih\,iw)\,min({TRANSCODE_MAX_LONG_EDGE}\,ih)\,-2),format=yuv420p"
)


def is_hevc_codec(codec: str | None) -> bool:
    """判断编码是否为 HEVC（不区分大小写）。"""
    return (codec or "").lower() in HEVC_CODECS


def probe_rotation(path: str) -> int | None:
    """ffprobe 读取视频旋转角度（display matrix side data）。

    返回 0/90/-90/180 等整数；无 side data 或探测失败返回 None。

    注意：旧实现同时请求 side_data 和 stream_tags 且取输出第一行，
    iOS 文件会输出两行（如 "270\\n90"）导致取到错误值，现仅读 side_data。
    """
    try:
        result = subprocess.run(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "side_data=rotation",
                "-of",
                "default=noprint_wrappers=1:nokey=1",
                str(path),
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=60,
        )
        first_line = (result.stdout or "").strip().split("\n")[0].strip()
        if not first_line:
            return None
        return int(float(first_line))
    except (subprocess.TimeoutExpired, ValueError, OSError):
        return None


def is_portrait_rotation(rotation: int | None) -> bool:
    """rotation side data 为 ±90/270 时表示竖屏拍摄。

    注意：这只覆盖"存储横屏 + display matrix 旋转"的 iOS 风格视频；
    物理竖屏视频（Android 常见，存储即 h>w、rotation=None/0）不会命中，
    方向判定请用 is_portrait_video()。
    """
    return rotation in (90, 270, -90)


def is_portrait_video(
    stored_width: int | None,
    stored_height: int | None,
    rotation: int | None,
) -> bool:
    """按显示方向判断是否竖屏（显示高度 > 显示宽度）。

    - rotation 为 90/270/-90 时，显示方向的宽高相对存储维度互换；
    - rotation 为 0/180/None 时，显示方向即存储维度。

    这样两类竖屏都能正确识别：
    - iOS：存储 1920x1080 + rotation=90 → 显示 1080x1920 竖屏
    - Android/物理竖屏：存储 1080x1920、无 rotation → 显示 1080x1920 竖屏
    探测失败（维度为 None）时退回仅看 rotation，保证调用链不中断。
    """
    if not stored_width or not stored_height:
        return is_portrait_rotation(rotation)
    if is_portrait_rotation(rotation):
        return stored_width > stored_height
    return stored_height > stored_width


def build_transcode_vf() -> str:
    """构建转码视频滤镜（竖屏/横屏统一，按显示长边封顶 1920、只缩不放）。

    依赖 ffmpeg 内置 autorotate（默认开启）按 display matrix 物理旋转画面，
    输出自动剥离 rotation side data；滤镜只做等比缩放，方向无关：
    横屏限宽、竖屏限高，短边 -2 自适应偶数，min() 保证小视频不放大。

    旧实现的问题：
    - 显式 transpose=1 与 autorotate 叠加，竖屏被二次旋转成横屏；
    - 竖屏沿用按高缩放表达式，1080x1920 被错误缩成 608x1080；
    - 仅按短边 1080 触发缩放，超宽屏（如 4000x1000）长边超 1920 会被
      validate 拦截误降级，用户拿到浏览器无法播放的 HEVC 原文件。
    """
    return _TRANSCODE_VF


def probe_dimensions(path: str) -> tuple[int | None, int | None]:
    """ffprobe 读取视频宽高（像素维度）。"""
    try:
        result = subprocess.run(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=width,height",
                "-of",
                "csv=p=0:s=x",
                str(path),
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=60,
        )
        text = (result.stdout or "").strip().split("\n")[0].strip()
        width_str, height_str = text.split("x")
        return int(width_str), int(height_str)
    except (subprocess.TimeoutExpired, ValueError, OSError):
        return None, None


def probe_video_info(path: str) -> tuple[int | None, int | None, int | None]:
    """一次 ffprobe 同时读取视频宽高与旋转角度（display matrix side data）。

    返回 (width, height, rotation)；探测失败对应位置为 None。
    合并维度/角度两次探测，减少大文件、高并发下的 ffprobe 进程开销。
    rotation 仅取 stream side_data_list 的 Display Matrix（不读 tags.rotate，
    避免 iOS 文件 tag 值与 side data 双来源取错）。用 -show_streams 全量 JSON
    输出解析，兼容 ffmpeg 4.x/7.x（show_entries 嵌套 section 名跨版本不一致）。
    """
    try:
        result = subprocess.run(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_streams",
                "-of",
                "json",
                str(path),
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=60,
        )
        data = json.loads(result.stdout or "{}")
        streams = data.get("streams") or []
        if not streams:
            return None, None, None
        stream = streams[0]
        width = int(stream["width"]) if stream.get("width") else None
        height = int(stream["height"]) if stream.get("height") else None
        rotation = None
        for side in stream.get("side_data_list") or []:
            if side.get("side_data_type") == "Display Matrix" and side.get("rotation") is not None:
                deg = int(round(float(side["rotation"]))) % 360
                # ffprobe：顺时针 90 拍摄输出 90，逆时针 90 输出 -90（归一为 270）
                rotation = {0: 0, 90: 90, 180: 180, 270: -90}.get(deg, deg if deg in (90, 180) else None)
                break
        return width, height, rotation
    except (subprocess.TimeoutExpired, ValueError, OSError, json.JSONDecodeError, KeyError, TypeError):
        return None, None, None


def validate_transcode_output(
    output_path: str,
    expected_portrait: bool,
    max_long_edge: int = TRANSCODE_MAX_LONG_EDGE,
) -> bool:
    """校验转码产物方向与维度。

    - 竖屏源：产物必须 height > width，且仍有 rotation side data 视为失败
      （播放器会二次旋转成横屏）
    - 横屏源：产物必须 width >= height
    - 长边不得超过 max_long_edge（只缩不放）
    校验失败时调用方应降级使用原始文件，不允许产出方向错误的文件覆盖。
    """
    width, height = probe_dimensions(output_path)
    if not width or not height:
        return False
    if expected_portrait and height <= width:
        return False
    if not expected_portrait and width < height:
        return False
    if max(width, height) > max_long_edge:
        return False
    # 产物仍带 rotation side data 说明方向没有物理固化，播放器会再次旋转
    if probe_rotation(output_path) is not None:
        return False
    return True


def _original_key_from_storage_key(storage_key: str) -> str:
    """从可能被 HEVC 转码改写的 storage_key 还原原始 key。

    转码成功后 key 形如 uploads/<id>/IMG_2282_h264.MOV，
    占位 asset 以原始 key uploads/<id>/IMG_2282.MOV 创建。
    """
    if not storage_key:
        return storage_key
    _p = Path(storage_key)
    if _p.stem.endswith("_h264"):
        return str(_p.parent / (_p.stem[: -len("_h264")] + _p.suffix))
    return storage_key


def _resolve_placeholder_asset(asset_repo, job, original_storage_key):
    """找到 complete 阶段创建的 PROCESSING 占位 asset（Issue #1714）。

    HEVC 转码成功后 job.storage_key 会被改写为 *_h264 新 key，旧实现用新 key
    回查占位必然落空，进而兜底新建一条 READY 记录，导致原占位永久卡 processing。

    查找优先级：
    1. job.asset_id（complete 派单时透传的占位 id，最可靠，不依赖 key）；
    2. 原始 storage_key（占位记录以原始 key 创建）；
    3. 当前 job.storage_key（未转码/降级场景与原始 key 相同）。

    找不到返回 None（旧链路兼容，由调用方兜底新建并告警）。
    """
    asset_id = getattr(job, "asset_id", "") or ""
    if asset_id:
        try:
            found = asset_repo.find_by_id(asset_id)
            if found is not None:
                return found
        except Exception as find_err:
            logger.warning("占位 asset 按 id 查询失败 asset_id=%s: %s", asset_id, find_err)
    for key in (original_storage_key, getattr(job, "storage_key", "")):
        if not key:
            continue
        try:
            found = asset_repo.find_by_storage_key(key)
        except Exception:
            logger.warning("find_by_storage_key not available, trying fallback lookup")
            found = None
        if found is not None:
            return found
    return None


@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"}

        # 记录原始 storage_key：HEVC 转码成功后 job.storage_key 会改写为 *_h264，
        # 而 complete 阶段的占位 asset 始终以原始 key 创建，关联回写必须保留它。
        original_storage_key = job.storage_key

        # 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"
                    upload_ok = upload_to_oss(frame_path, thumb_storage_key)
                    if upload_ok:
                        # 存储 storage_key（非公开 URL），API 层通过 get_download_url 生成签名 URL
                        thumbnail_url = thumb_storage_key
                        logger.info(
                            "素材缩略图生成成功: job_id=%s key=%s",
                            job_id,
                            thumb_storage_key[:80],
                        )
                    else:
                        logger.warning(
                            "素材缩略图上传 OSS 失败: job_id=%s",
                            job_id,
                        )
                    # frame_path 是临时文件，及时清理
                    if frame_path:
                        try:
                            Path(frame_path).unlink(missing_ok=True)
                        except Exception:
                            pass
                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 local_file and local_file.exists():
                if is_hevc_codec(metadata.get("codec")):
                    logger.info(
                        "检测到 HEVC 编码 (codec=%s)，启动转码: job_id=%s",
                        metadata.get("codec"),
                        job_id,
                    )
                    _tc_tmp = None

                    # ── Step 1: 磁盘空间检查（独立 try/except，失败仍尝试转码）──
                    try:
                        _disk_usage = shutil.disk_usage("/tmp")
                        _free_gb = _disk_usage.free / (1024**3)
                        if _free_gb < 2:
                            raise RuntimeError(f"磁盘空间不足 ({_free_gb:.1f}GB < 2GB)")
                    except Exception as _disk_err:
                        logger.warning("磁盘检查失败，仍尝试转码: job_id=%s err=%s", job_id, _disk_err)

                    # ── Step 2: 方向检测（按显示方向判定竖/横屏）──────────────
                    # 不能只看 rotation side data：Android 等设备的物理竖屏视频
                    # 存储维度已是 h>w 且 rotation=0/None，只看 rotation 会误判横屏、
                    # 套用横屏滤镜把 1080x1920 压成 608x1080，转码产物校验失败降级，
                    # 用户拿到 HEVC 原文件浏览器仍黑帧。
                    _src_w, _src_h, _rotation = probe_video_info(str(local_file))
                    _is_portrait = is_portrait_video(_src_w, _src_h, _rotation)
                    logger.info(
                        "视频方向检测: stored=%sx%s rotation=%s portrait=%s: job_id=%s",
                        _src_w,
                        _src_h,
                        _rotation,
                        _is_portrait,
                        job_id,
                    )

                    # ── Step 3: ffmpeg 转码（独立 try/except）──
                    try:
                        _tc_tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix="_h264.mp4")
                        _tc_tmp = Path(_tc_tmp_file.name)
                        _tc_tmp_file.close()  # 关闭文件描述符，ffmpeg 会自己打开

                        # 旋转交给 ffmpeg 内置 autorotate（按 display matrix 物理旋转，
                        # 输出自动剥离 side data）；滤镜只做 1080p 等比"只缩不放"。
                        # 注意不能再加 transpose：旧逻辑 autorotate + transpose 双重旋转，
                        # 竖屏被转成横屏；scale 表达式内逗号必须 \, 转义（见 build_transcode_vf）。
                        _vf = build_transcode_vf()

                        _cmd = [
                            "ffmpeg",
                            "-y",
                            "-i",
                            str(local_file),
                            "-c:v",
                            "libx264",
                            "-preset",
                            "fast",
                            "-crf",
                            "18",
                            "-vf",
                            _vf,
                            "-colorspace",
                            "bt709",
                            "-color_primaries",
                            "bt709",
                            "-color_trc",
                            "bt709",
                            "-pix_fmt",
                            "yuv420p",
                            "-level",
                            "4.2",
                            "-c:a",
                            "aac",
                            "-b:a",
                            "128k",
                            "-movflags",
                            "+faststart",
                            str(_tc_tmp),
                        ]
                        _proc = subprocess.run(
                            _cmd,
                            stdout=subprocess.DEVNULL,
                            stderr=subprocess.PIPE,
                            text=True,
                            timeout=TRANSCODE_TIMEOUT_SECONDS,
                        )
                        if _proc.returncode == 0 and _tc_tmp.exists() and _tc_tmp.stat().st_size > 0:
                            # ── Step 4: 方向/维度校验，不符则降级，杜绝横屏文件覆盖 ──
                            if not validate_transcode_output(str(_tc_tmp), _is_portrait):
                                _w, _h = probe_dimensions(str(_tc_tmp))
                                _rot = probe_rotation(str(_tc_tmp))
                                logger.error(
                                    "转码产物方向/维度校验失败，降级使用原始文件: "
                                    "job_id=%s source_rotation=%s portrait=%s out=%sx%s out_rotation=%s",
                                    job_id,
                                    _rotation,
                                    _is_portrait,
                                    _w,
                                    _h,
                                    _rot,
                                )
                            else:
                                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 stderr=%s: job_id=%s",
                                _proc.returncode,
                                _tail,
                                job_id,
                            )
                    except subprocess.TimeoutExpired:
                        logger.warning(
                            "FFmpeg 转码超时(900s)，降级原始文件: 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,
            )

            placeholder = _resolve_placeholder_asset(asset_repo, job, original_storage_key)
            if placeholder is not None:
                # 回写占位记录：标 ERROR（Issue #1714：禁止新建第二条导致占位孤儿）
                asset = placeholder
                asset.mime_type = mime_type
                asset.metadata = {"source": "upload", "ingest_error": error_reason}
                asset.file_size = int(metadata.get("size_bytes", 0))
                asset.duration = float(metadata.get("duration", 0)) or None
                asset.width = int(metadata.get("width", 0)) or None
                asset.height = int(metadata.get("height", 0)) or None
                codec_val = metadata.get("codec")
                if codec_val:
                    asset.codec = str(codec_val)
                asset.status = AssetStatus.ERROR
                asset.updated_at = datetime.now(timezone.utc)
                asset_repo.update(asset)
            else:
                # 旧链路兜底：无占位记录（如历史 job 重跑）才新建
                logger.warning("无效素材且未找到占位记录，兜底新建 ERROR asset: job_id=%s", job_id)
                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={"source": "upload", "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,
            }

        # 查找 complete 阶段创建的占位 Asset 记录（Issue #1714）。
        # 必须用原始 storage_key / job.asset_id 关联——HEVC 转码后 job.storage_key
        # 已改写为 *_h264，用新 key 回查占位必然落空，旧实现因此兜底新建 READY 记录，
        # 导致原 PROCESSING 占位永久卡住（每个 HEVC 视频产生两条记录）。
        existing_asset = _resolve_placeholder_asset(asset_repo, job, original_storage_key)

        if existing_asset is None:
            # 兜底：仅当确实没有占位记录（旧版本 API / 历史 job 重跑）才新建。
            logger.warning(
                "No placeholder asset found for job_id=%s original_key=%s, creating new",
                job_id,
                original_storage_key,
            )
            metadata["source"] = "upload"
            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)
        else:
            # 更新占位记录：补充元数据、置 READY。转码成功时 storage_key 同步改写为
            # *_h264（播放/下载走转码产物），原始 key 记入 metadata 可溯源。
            asset = existing_asset
            if job.storage_key != asset.storage_key:
                metadata["original_storage_key"] = asset.storage_key
                metadata["hevc_transcoded"] = True
                asset.storage_key = job.storage_key
            asset.mime_type = mime_type
            metadata["source"] = "upload"
            asset.metadata = metadata
            asset.file_size = int(metadata.get("size_bytes", 0))
            asset.duration = float(metadata.get("duration", 0))
            asset.width = int(metadata.get("width", 0))
            asset.height = int(metadata.get("height", 0))
            codec_val = metadata.get("codec")
            if codec_val:
                asset.codec = str(codec_val)
            fps_val = metadata.get("fps")
            if fps_val:
                try:
                    asset.fps = float(fps_val)
                except (ValueError, TypeError):
                    pass
            asset.status = AssetStatus.READY
            asset.thumbnail_url = thumbnail_url
            asset.updated_at = datetime.now(timezone.utc)
            asset_repo.update(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 and mark pre-created Asset as ERROR
        try:
            job_repo = SQLAlchemyIngestJobRepository(db)
            asset_repo = SQLAlchemyAssetRepository(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)

                # 将上传时创建的占位 Asset（PROCESSING/UPLOADING）标记为 ERROR，
                # 避免素材永远卡在中间状态。转码可能已把 job.storage_key 改写为
                # *_h264，需用 asset_id / 原始 key 多路径关联占位（Issue #1714）。
                try:
                    existing = None
                    _asset_id = getattr(job, "asset_id", "") or ""
                    if _asset_id:
                        try:
                            existing = asset_repo.find_by_id(_asset_id)
                        except Exception:
                            existing = None
                    if existing is None:
                        _candidate_keys = [
                            _original_key_from_storage_key(job.storage_key),
                            job.storage_key,
                        ]
                        for _key in _candidate_keys:
                            try:
                                existing = asset_repo.find_by_storage_key(_key)
                            except Exception:
                                existing = None
                            if existing is not None:
                                break
                    if existing and existing.status in (
                        AssetStatus.PROCESSING,
                        AssetStatus.UPLOADING,
                    ):
                        existing.status = AssetStatus.ERROR
                        existing.metadata = {**(existing.metadata or {}), "ingest_error": str(e)}
                        existing.updated_at = datetime.now(timezone.utc)
                        asset_repo.update(existing)
                        logger.info(
                            "Marked asset as ERROR due to ingest failure: asset_id=%s job_id=%s",
                            existing.id,
                            job_id,
                        )
                except Exception as asset_err:
                    logger.warning("Failed to mark asset as ERROR: %s", asset_err)

                db.commit()
        except Exception:
            db.rollback()

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