#!/usr/bin/env python3 """把 jsonl 清洗并导出为 xlsx。 同时认两种输入： records.jsonl 流水线产物，字段全（标题/作者/标签/时长/COS 地址/raw） manifest.jsonl 老格式，只有 cos_mp4 / cos_mp3 / duration records.jsonl 内嵌了 raw，单行可能几十 KB、整个文件几百 MB，所以这里 一律逐行流式读，任何时候都不把整个文件读进内存。选中的列也只留需要的 那几个字段，raw 不会进内存驻留。 清洗动作（每一项都会报告剔除了多少）： 1. 丢弃无法解析的行（被 Ctrl-C 打断可能留下半行） 2. 丢弃缺字段 / 字段类型不对的行 3. 丢弃时长非正数的行 4. 丢弃 URL 前缀不符合预期的行 5. 按 mp4 地址去重，保留首次出现 用法: python jsonl_to_xlsx.py python jsonl_to_xlsx.py -i records.jsonl -o "抖音-四川话.xlsx" python jsonl_to_xlsx.py -i records.jsonl --columns cos_mp4,cos_mp3,duration_sec python jsonl_to_xlsx.py --list-columns """ from __future__ import annotations # 移到 scripts/ tests/ 后，直接运行时脚本目录成了 sys.path[0]， # 把项目根加回来，import douyin / pipeline 才找得到。 import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent.parent)) import argparse import json import sys from pathlib import Path from typing import Any, Dict, Iterator, List EXPECTED_PREFIX = "cos://asr-dig-data-1253181912/0005/zh/" # 列名 -> (表头, 取值函数)。取值函数只碰需要的字段，raw 不会被带进来。 COLUMNS: Dict[str, tuple] = { "aweme_id": ("视频ID", lambda r: str(r.get("aweme_id") or "")), "title": ("标题", lambda r: r.get("title") or ""), "desc": ("原始描述", lambda r: r.get("desc") or ""), "tags": ("标签", lambda r: " ".join(r.get("tags") or [])), "author": ("作者", lambda r: (r.get("author") or {}).get("nickname", "")), "author_uid": ("作者UID", lambda r: (r.get("author") or {}).get("uid", "")), "create_time": ("发布时间", lambda r: r.get("create_time_str") or ""), "duration_sec": ("时长(秒)", _dur := (lambda r: r.get("duration_sec", r.get("duration")))), "duration": ("时长(秒)", _dur), "cos_mp4": ("cosmp4地址", lambda r: r.get("cos_mp4") or ""), "cos_mp3": ("cosmp3地址", lambda r: r.get("cos_mp3") or ""), "size_mp4": ("mp4字节数", lambda r: r.get("size_mp4") or 0), "size_mp3": ("mp3字节数", lambda r: r.get("size_mp3") or 0), "play": ("播放数", lambda r: (r.get("stats") or {}).get("play", 0)), "digg": ("点赞数", lambda r: (r.get("stats") or {}).get("digg", 0)), "comment": ("评论数", lambda r: (r.get("stats") or {}).get("comment", 0)), "source_url": ("原始链接", lambda r: r.get("source_url") or ""), } DEFAULT_COLUMNS = ["cos_mp4", "cos_mp3", "duration_sec"] FULL_COLUMNS = ["aweme_id", "title", "author", "create_time", "tags", "duration_sec", "cos_mp4", "cos_mp3"] NUMERIC = {"duration_sec", "duration", "size_mp4", "size_mp3", "play", "digg", "comment"} def iter_lines(path: Path) -> Iterator[str]: with open(path, "r", encoding="utf-8", errors="replace") as f: for line in f: line = line.strip() if line: yield line def clean(path: Path, prefix: str, columns: List[str]): """流式清洗，只保留选中的列。""" rows: List[List[Any]] = [] seen: set = set() dropped = {"解析失败": 0, "字段缺失": 0, "时长非法": 0, "地址异常": 0, "重复": 0} total = 0 dur_idx = next((i for i, c in enumerate(columns) if c in ("duration_sec", "duration")), None) for line in iter_lines(path): total += 1 try: r = json.loads(line) except ValueError: dropped["解析失败"] += 1 continue mp4, mp3 = r.get("cos_mp4"), r.get("cos_mp3") dur = r.get("duration_sec", r.get("duration")) if not isinstance(mp4, str) or not isinstance(mp3, str) or dur is None: dropped["字段缺失"] += 1 continue if not isinstance(dur, (int, float)) or isinstance(dur, bool) or dur <= 0: dropped["时长非法"] += 1 continue if not mp4.startswith(prefix) or not mp3.startswith(prefix): dropped["地址异常"] += 1 continue if mp4 in seen: dropped["重复"] += 1 continue seen.add(mp4) row = [COLUMNS[c][1](r) for c in columns] if dur_idx is not None: row[dur_idx] = float(row[dur_idx]) rows.append(row) return rows, dropped, total, dur_idx def write_xlsx(rows, headers, out: Path, sheet: str, dur_idx) -> None: from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter wb = Workbook() ws = wb.active ws.title = sheet ws.append(headers) head_font = Font(bold=True, color="FFFFFF") head_fill = PatternFill("solid", fgColor="4472C4") for c in range(1, len(headers) + 1): cell = ws.cell(row=1, column=c) cell.font = head_font cell.fill = head_fill cell.alignment = Alignment(horizontal="center", vertical="center") for r in rows: ws.append(r) # 时长保留 3 位小数，与 ffprobe 输出精度一致 if dur_idx is not None: for row in ws.iter_rows(min_row=2, min_col=dur_idx + 1, max_col=dur_idx + 1): for cell in row: cell.number_format = "0.000" for i, head in enumerate(headers): widest = max((len(str(r[i])) for r in rows), default=10) width = min(max(len(head) + 4, widest + 2), 60) ws.column_dimensions[get_column_letter(i + 1)].width = width ws.freeze_panes = "A2" # 表头固定 ws.auto_filter.ref = ws.dimensions # 开筛选 wb.save(out) def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description="jsonl -> xlsx") ap.add_argument("-i", "--input", default="records/records.jsonl") ap.add_argument("-o", "--output", default="exports/抖音-四川话.xlsx") ap.add_argument("--sheet", default="抖音-四川话") ap.add_argument("--prefix", default=EXPECTED_PREFIX) ap.add_argument("--columns", default="", help=f"逗号分隔的列，默认 {','.join(DEFAULT_COLUMNS)}；" "用 full 展开为常用全字段") ap.add_argument("--list-columns", action="store_true", help="列出可用列名后退出") args = ap.parse_args(argv) if args.list_columns: print("可用列：") for k, (head, _) in COLUMNS.items(): print(f" {k:<14} {head}") return 0 if not args.columns: columns = list(DEFAULT_COLUMNS) elif args.columns.strip() == "full": columns = list(FULL_COLUMNS) else: columns = [c.strip() for c in args.columns.split(",") if c.strip()] bad = [c for c in columns if c not in COLUMNS] if bad: print(f"[错误] 未知列名 {bad}，用 --list-columns 看可选项", file=sys.stderr) return 2 src = Path(args.input) if not src.exists(): print(f"[错误] 找不到 {src}", file=sys.stderr) return 2 rows, dropped, total, dur_idx = clean(src, args.prefix, columns) print(f"读入 {total} 行") if sum(dropped.values()): for k, v in dropped.items(): if v: print(f" 剔除 {k}: {v}") else: print(" 清洗检查全部通过，无需剔除") print(f"保留 {len(rows)} 行") if not rows: print("[错误] 清洗后没有数据", file=sys.stderr) return 3 headers = [COLUMNS[c][0] for c in columns] out = Path(args.output) write_xlsx(rows, headers, out, args.sheet, dur_idx) print(f"\n已写入 {out}") print(f" {len(rows)} 行 × {len(headers)} 列：{' | '.join(headers)}") if dur_idx is not None: ds = [r[dur_idx] for r in rows] print(f" 总时长 {sum(ds) / 3600:.2f} 小时，平均 {sum(ds) / len(ds):.1f} 秒，" f"最短 {min(ds):.3f}，最长 {max(ds):.3f}") return 0 if __name__ == "__main__": raise SystemExit(main())