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//IMG_2282_h264.MOV， 占位 asset 以原始 key uploads//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"} # ── 执行前状态守卫（#1714）：job 已终态（失败/完成）说明这是作废消息 # （超时清理标记 failed 后旧消息重投、或重复投递），直接丢弃不执行， # 避免重复转码、重复回写。processing 是本任务自己第一次置位前的旧消息 # 极少见，保守起见也丢弃（processing 的占位由恢复流程处理）。 current_status = job.status.value if hasattr(job.status, "value") else str(job.status) if current_status in ("failed", "completed"): logger.warning( "[ingest job_id=%s] 任务状态已为 %s，丢弃作废消息，不执行转码", job_id, current_status, ) return {"status": "discarded", "job_id": job_id, "reason": f"job already terminal: {current_status}"} # 记录原始 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()