SignerX/SignVerse-2M

1#!/usr/bin/env python32import argparse3import csv4import json5from pathlib import Path6from typing import Dict, List, Tuple7 8DEFAULT_COLUMNS = [9    'video_id',10    'sign_language',11    'title',12    'duration_sec',13    'start_sec',14    'end_sec',15    'subtitle_languages',16    'subtitle_dir_path',17    'subtitle_en_source',18    'raw_video_path',19    'raw_metadata_path',20    'metadata_status',21    'subtitle_status',22    'download_status',23    'failure_count',24    'error',25    'processed_at',26    'subtitle_json_path',27    'subtitle_en',28    'subtitle_texts_json',29    'process_status',30    'upload_status',31    'local_cleanup_status',32    'archive_name',33]34 35VIDEO_EXTS = {'.mp4', '.mkv', '.webm', '.mov'}36 37 38from utils.dataset_pool import complete_video_ids39 40 41def read_csv_rows(path: Path) -> Tuple[List[Dict[str, str]], List[str]]:42    text = path.read_text(encoding='utf-8-sig')43    lines = [line for line in text.splitlines() if line.strip()]44    if not lines:45        return [], []46 47    first = next(csv.reader([lines[0]]), [])48    first0 = first[0].strip() if first else ''49    if first0 == 'video_id' or 'download_status' in first or 'process_status' in first:50        with path.open('r', encoding='utf-8-sig', newline='') as f:51            reader = csv.DictReader(f)52            rows = [dict(r) for r in reader]53            fieldnames = list(reader.fieldnames or [])54        return rows, fieldnames55 56    rows: List[Dict[str, str]] = []57    with path.open('r', encoding='utf-8-sig', newline='') as f:58        reader = csv.reader(f)59        for parts in reader:60            if not parts:61                continue62            video_id = (parts[0] or '').strip()63            sign_language = (parts[1] or '').strip() if len(parts) > 1 else ''64            if not video_id:65                continue66            rows.append({'video_id': video_id, 'sign_language': sign_language})67    return rows, ['video_id', 'sign_language']68 69 70def write_csv_rows(path: Path, rows: List[Dict[str, str]], fieldnames: List[str]):71    tmp = path.with_suffix(path.suffix + '.tmp')72    with tmp.open('w', encoding='utf-8', newline='') as f:73        writer = csv.DictWriter(f, fieldnames=fieldnames)74        writer.writeheader()75        for row in rows:76            writer.writerow({k: row.get(k, '') for k in fieldnames})77    tmp.replace(path)78 79 80def load_progress(progress_path: Path):81    if not progress_path.exists():82        return {}, {}83    obj = json.loads(progress_path.read_text())84    return obj.get('uploaded_folders', {}), obj.get('archives', {})85 86 87def load_journal(journal_path: Path):88    updates: Dict[str, Dict[str, str]] = {}89    if not journal_path.exists():90        return updates91    for line in journal_path.read_text(encoding='utf-8').splitlines():92        line = line.strip()93        if not line:94            continue95        try:96            obj = json.loads(line)97        except Exception:98            continue99        row_updates = {k: str(v) for k, v in (obj.get('updates') or {}).items()}100        for vid in obj.get('video_ids') or []:101            updates[str(vid)] = row_updates102    return updates103 104 105def main():106    ap = argparse.ArgumentParser()107    ap.add_argument('--source-metadata-csv', type=Path, required=True)108    ap.add_argument('--output-metadata-csv', type=Path, required=True)109    ap.add_argument('--raw-video-dir', type=Path, required=True)110    ap.add_argument('--scratch-raw-video-dir', type=Path, default=None)111    ap.add_argument('--raw-caption-dir', type=Path, required=True)112    ap.add_argument('--raw-metadata-dir', type=Path, required=True)113    ap.add_argument('--dataset-dir', type=Path, required=True)114    ap.add_argument('--scratch-dataset-dir', type=Path, default=None)115    ap.add_argument('--progress-path', type=Path, required=True)116    ap.add_argument('--status-journal-path', type=Path, required=True)117    args = ap.parse_args()118 119    source_rows, source_fields = read_csv_rows(args.source_metadata_csv)120    output_rows, output_fields = (read_csv_rows(args.output_metadata_csv) if args.output_metadata_csv.exists() else ([], []))121    fields: List[str] = []122    for col in DEFAULT_COLUMNS + source_fields + output_fields:123        if col and col not in fields:124            fields.append(col)125 126    out_by_id = {r.get('video_id', '').strip(): r for r in output_rows if r.get('video_id', '').strip()}127    rows: List[Dict[str, str]] = []128    for src in source_rows:129        vid = (src.get('video_id') or '').strip()130        if not vid:131            continue132        merged = {k: src.get(k, '') for k in fields}133        if vid in out_by_id:134            for k in fields:135                if k in out_by_id[vid]:136                    merged[k] = out_by_id[vid].get(k, '')137        rows.append(merged)138 139    raw_videos = {}140    if args.raw_video_dir.exists():141        raw_videos.update({p.stem: p for p in args.raw_video_dir.iterdir() if p.is_file() and p.suffix.lower() in VIDEO_EXTS})142    if args.scratch_raw_video_dir is not None and args.scratch_raw_video_dir.exists():143        for p in args.scratch_raw_video_dir.iterdir():144            if p.is_file() and p.suffix.lower() in VIDEO_EXTS and p.stem not in raw_videos:145                raw_videos[p.stem] = p146    raw_metadata = {p.stem: p for p in args.raw_metadata_dir.glob('*.json')} if args.raw_metadata_dir.exists() else {}147    complete = complete_video_ids(args.dataset_dir, args.scratch_dataset_dir)148    process_claims_dir = args.dataset_dir.parent / 'slurm' / 'state' / 'claims'149    download_claims_dir = args.dataset_dir.parent / 'slurm' / 'state' / 'download_claims'150    process_claims = {p.stem for p in process_claims_dir.glob('*.claim')} if process_claims_dir.exists() else set()151    download_claims = {p.stem for p in download_claims_dir.glob('*.claim')} if download_claims_dir.exists() else set()152    uploaded_folders, _archives = load_progress(args.progress_path)153    journal_updates = load_journal(args.status_journal_path)154 155    for row in rows:156        vid = (row.get('video_id') or '').strip()157        if not vid:158            continue159        if vid in raw_metadata:160            row['raw_metadata_path'] = str(raw_metadata[vid])161            row['metadata_status'] = 'ok'162        if vid in raw_videos:163            row['raw_video_path'] = str(raw_videos[vid])164            row['download_status'] = 'ok'165        elif vid in download_claims and row.get('download_status', '') not in {'ok', 'skipped'}:166            row['download_status'] = 'running'167        if vid in complete:168            row['process_status'] = 'ok'169        elif vid in process_claims and row.get('process_status', '') != 'ok':170            row['process_status'] = 'running'171        if vid in uploaded_folders:172            row['upload_status'] = 'uploaded'173            row['archive_name'] = uploaded_folders[vid]174            row['local_cleanup_status'] = 'deleted'175            row['process_status'] = 'ok'176            row['download_status'] = 'ok'177            if not row.get('metadata_status'):178                row['metadata_status'] = 'ok'179        elif vid in complete:180            row['upload_status'] = ''181            row['archive_name'] = ''182            row['local_cleanup_status'] = ''183        elif vid in journal_updates:184            for k, v in journal_updates[vid].items():185                if k in {'upload_status', 'archive_name', 'local_cleanup_status'}:186                    row[k] = v187 188    write_csv_rows(args.output_metadata_csv, rows, fields)189    print(f'synced_rows={len(rows)}')190 191if __name__ == '__main__':192    main()193