70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
从 数据.jsonl 提取【最终状态是错误/无字幕】的视频 COS 路径，写到 txt(可当任务列表重跑测试)。
|
||
按每个视频的【最后一条记录】判定最终状态(忽略中间重试)。
|
||
|
||
用法:
|
||
python temp/extract_errors.py 2号机器数据.jsonl error_videos.txt
|
||
python temp/extract_errors.py 2号机器数据.jsonl out.txt error,no_subtitle # 也要无字幕的
|
||
"""
|
||
import json
|
||
import sys
|
||
from collections import Counter, defaultdict
|
||
|
||
JSONL = sys.argv[1] if len(sys.argv) > 1 else "数据.jsonl"
|
||
OUT = sys.argv[2] if len(sys.argv) > 2 else "error_videos.txt"
|
||
STATUSES = sys.argv[3].split(",") if len(sys.argv) > 3 else ["error"]
|
||
|
||
|
||
def main():
|
||
try:
|
||
f = open(JSONL, encoding="utf-8")
|
||
except FileNotFoundError:
|
||
print(f"找不到文件: {JSONL}")
|
||
return
|
||
|
||
latest = {} # video_cos_path -> (status, error_reason)
|
||
with f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
r = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
vid = r.get("video_cos_path")
|
||
if not vid:
|
||
continue
|
||
latest[vid] = (r.get("status"), r.get("error_reason", ""))
|
||
|
||
# 最终状态计数
|
||
status_count = Counter(s for s, _ in latest.values())
|
||
# 选中的(最终状态在 STATUSES 里)
|
||
selected = [(v, er) for v, (s, er) in latest.items() if s in STATUSES]
|
||
|
||
# 写 txt(每行一个 COS 路径)
|
||
with open(OUT, "w", encoding="utf-8") as w:
|
||
for v, _ in selected:
|
||
w.write(v + "\n")
|
||
|
||
print(f"文件: {JSONL}")
|
||
print(f"去重后视频数: {len(latest)}")
|
||
print(f"最终状态分布: {dict(status_count)}")
|
||
print(f"提取状态 {STATUSES}: {len(selected)} 个 → 写入 {OUT}")
|
||
|
||
# 错误原因 Top（帮你看主要错在哪）
|
||
reasons = Counter()
|
||
for _, er in selected:
|
||
if er:
|
||
key = er.split(":")[0][:60] if ":" in er else er[:60]
|
||
reasons[key] += 1
|
||
if reasons:
|
||
print("\n错误原因 Top:")
|
||
for k, c in reasons.most_common(10):
|
||
print(f" {c:>5} {k}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|