几千条视频怎么建档：技术元数据、sidecar 与批量抽取

import json
import subprocess


def probe_full(path: str) -> dict:
    cmd = ['ffprobe', '-v', 'error', '-print_format', 'json',
           '-show_format', '-show_streams', '-show_chapters', path]
    out = subprocess.run(cmd, capture_output=True, text=True, check=True).stdout
    return json.loads(out)

def normalize(info: dict) -> dict:
    fmt = info.get('format', {})
    streams = info.get('streams', [])

    v = next((s for s in streams if s.get('codec_type') == 'video'), None)
    a = next((s for s in streams if s.get('codec_type') == 'audio'), None)
    subs = [s for s in streams if s.get('codec_type') == 'subtitle']

    def fps_of(s):
        r = (s or {}).get('avg_frame_rate') or (s or {}).get('r_frame_rate') or '0/0'
        try:
            n, d = r.split('/')
            return round(float(n) / float(d), 3) if float(d) else None
        except Exception:
            return None

    return {
        'duration': round(float(fmt.get('duration') or 0), 2),
        'size_bytes': int(fmt.get('size') or 0),
        'format_name': fmt.get('format_name'),
        'bitrate_kbps': int(int(fmt.get('bit_rate') or 0) / 1000),
        'video': None if not v else {
            'codec': v.get('codec_name'),
            'profile': v.get('profile'),
            'width': v.get('width'),
            'height': v.get('height'),
            'fps': fps_of(v),
            'pix_fmt': v.get('pix_fmt'),
            'bitrate_kbps': int(int(v.get('bit_rate') or 0) / 1000),
            'color_transfer': v.get('color_transfer'),
            'color_primaries': v.get('color_primaries'),
            'is_hdr': v.get('color_transfer') in ('smpte2084', 'arib-std-b67'),
        },
        'audio': None if not a else {
            'codec': a.get('codec_name'),
            'sample_rate': int(a.get('sample_rate') or 0),
            'channels': a.get('channels'),
            'bitrate_kbps': int(int(a.get('bit_rate') or 0) / 1000),
            'language': (a.get('tags') or {}).get('language'),
        },
        'subtitle_count': len(subs),
        'chapter_count': len(info.get('chapters') or []),
        'has_audio': a is not None,
    }

内嵌：基础信息（title / artist / comment / creation_time）
        ↓ 保证"文件单独发出去时还知道是什么"
sidecar：完整技术元数据 + 内容元数据的 JSON
        ↓ 放在同一个目录/归档包里，跟着文件走
数据库：索引 + 业务字段 + 关系
        ↓ 用于检索和管理

ffmpeg -i in.mp4 -c copy \
  -metadata title="2026年数据安全培训（第一讲）" \
  -metadata artist="技术部" \
  -metadata comment="asset_id=AST-2026-0142" \
  -metadata creation_time="2026-03-11T10:00:00+08:00" \
  out.mp4

{
  "schema": "viddown-asset/1.0",
  "asset_id": "AST-2026-0142",
  "title": "2026年数据安全培训（第一讲）",
  "created_at": "2026-03-11",
  "technical": { ... },
  "content": {
    "summary": "...",
    "tags": ["数据安全", "培训"],
    "people": ["张工"],
    "chapters": [{"start": 0, "end": 450, "title": "开场"}]
  },
  "administrative": {
    "source": "内部录制",
    "owner": "技术部",
    "rights": "内部使用",
    "retention_until": "2029-03-11",
    "checksum": "xxh3:abc123"
  },
  "_mapping": {
     "dc.title": "content.title",
     "dcterms.created": "created_at"
  }
}

class Asset(models.Model):
    asset_id = models.CharField(max_length=64, unique=True)
    title = models.CharField(max_length=200)
    # 管理字段（入库必填）
    source = models.CharField(max_length=100, help_text='来源')
    owner = models.ForeignKey('auth.User', on_delete=models.PROTECT)
    rights = models.CharField(max_length=100, help_text='使用范围')
    retention_until = models.DateField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    # 技术元数据（自动填）
    technical = models.JSONField(default=dict)
    checksum = models.CharField(max_length=128, blank=True)
    path = models.CharField(max_length=512)
    status = models.CharField(max_length=20, default='ACTIVE')

asset_id = "AST-2026-0142"        # 人用的
content_hash = "xxh3:abc123..."   # 机器用的（去重、校验）

import json
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path


def process_one(path: Path):
    try:
        info = probe_full(str(path))
        tech = normalize(info)
        meta = {
            'schema': 'viddown-asset/1.0',
            'asset_id': None,                       # 入库时分配
            'title': path.stem,
            'technical': tech,
            'administrative': {
                'checksum': f'xxh3:{file_hash(str(path))}',
                'file': path.name,
                'indexed_at': time.time(),
            },
        }
        # 写 sidecar
        sidecar = path.with_suffix(path.suffix + '.meta.json')
        sidecar.write_text(json.dumps(meta, ensure_ascii=False, indent=2),
                           encoding='utf-8')
        return {'path': str(path), 'ok': True, 'duration': tech['duration']}
    except Exception as e:
        return {'path': str(path), 'ok': False, 'error': str(e)[:200]}


def index_library(root: str, workers=8):
    files = [p for p in Path(root).rglob('*')
             if p.suffix.lower() in ('.mp4', '.mov', '.mkv', '.ts', '.avi', '.mxf')]
    print(f'发现 {len(files)} 个文件')
    with ProcessPoolExecutor(max_workers=workers) as ex:
        results = list(ex.map(process_one, files))
    ok = [r for r in results if r['ok']]
    bad = [r for r in results if not r['ok']]
    print(f'成功 {len(ok)}，失败 {len(bad)}')
    for b in bad[:10]:
        print(f"  {b['path']}: {b['error']}")
    # 汇总（总时长、总大小、格式分布）
    total_dur = sum(r['duration'] for r in ok)
    print(f'总时长: {total_dur / 3600:.1f} 小时')
    return results


if __name__ == '__main__':
    import sys
    index_library(sys.argv[1])

共 3127 个文件，总时长 2840 小时（约 4.1 TB）
格式分布：MP4 78% / MOV 12% / MKV 6% / 其他 4%
分辨率分布：1080p 54% / 720p 28% / 480p 12% / 4K 6%
无音轨的文件：147 个（需要确认）
HDR 素材：89 个
疑似重复（内容哈希相同）：213 个（可清理，约 280 GB）