model/utils.py

import os, re, json, unicodedata
from typing import Dict, List, Any

def norm_key(s: str) -> str:
    s = s.strip().lower()
    s = unicodedata.normalize('NFKC', s)
    s = re.sub(r'[^a-z0-9]+', '_', s)
    return s.strip('_')

# 这里扩充了列名同义词集合，兼容你的 CSV 表头：
# - source_video_path -> video_path
# - chinese_instruction -> edit_instruction
# - videoid -> uid
LOGICAL_KEYS = {
    'video_path': {
        'video','path','video_path','video_file','filepath','file_path','input','src',
        'source_video_path', 'source_path', 'source'  # 新增
    },
    'edit_instruction': {
        'instruction','edit','edit_instruction','prompt','command','text',
        'chinese_instruction', 'cn_instruction', 'zh_instruction'  # 新增
    },
    'sampling_mode': {'sampling','mode','sampling_mode','sample_mode'},
    'fps': {'fps'},
    'n_frames': {'n','nframes','num_frames','n_frames'},
    'uid': {
        'uid','id','task_id','clip_id','name','key',
        'videoid'  # 新增
    },
}

def map_header(cols: List[str]) -> Dict[str,str]:
    mapped = {}
    for c in cols:
        nc = norm_key(c)
        for logical, cand in LOGICAL_KEYS.items():
            if nc in cand:
                # 若同一 logical 命中多个候选，保留第一个（按列顺序）
                mapped.setdefault(logical, c)
    return mapped

def ensure_dir(p: str):
    os.makedirs(p, exist_ok=True)

def load_json_if_exists(p: str):
    return json.load(open(p,'r',encoding='utf-8')) if os.path.exists(p) else None