File size: 1,544 Bytes
aa975a2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | 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
|