| |
| """清洗 archive（群消息导出 + 判定流水）→ 微调管线格式。 |
| |
| 输入：data/raw/archive/archive/<chat_id>/{messages,judgments}-YYYY-MM-DD.jsonl |
| - 只取 judgments（自带 text 与 verdict）；messages 用于 join 校验与 kind 统计 |
| - verdict ∈ {ad, clean}；skipped / error 剔除 |
| - 生产兜底口径：verdict=ad 且 ad_kind 缺失/none → clean |
| 标注强度（decider）： |
| 强：llm、systemone+llm、phash、rule:<n> |
| 弱：systemone（旧初判模型自标注，训练默认不用） |
| |
| 去重与裁决（按 text 分组）： |
| 同文本多 sighting → 强标注优先；强 vs 强 标签矛盾 → 丢弃（报告计数） |
| --exclude-against 指定已有语料，剔除已存在的文本（增量净新增） |
| |
| 输出（默认 data/raw/archive/converted/）： |
| archive.strong.binary.jsonl 强标注唯一文本（推荐增补） |
| archive.all.binary.jsonl 含弱标注的全量唯一文本（参考/弱监督） |
| archive.report.json 统计与矛盾明细计数 |
| |
| 用法: |
| python scripts/convert_archive.py |
| python scripts/convert_archive.py --exclude-against data/raw/dataset/converted/gold.binary.jsonl |
| """ |
| import argparse |
| import glob |
| import json |
| import re |
| from collections import Counter |
| from pathlib import Path |
|
|
| STRONG_DECIDERS = {"llm", "systemone+llm", "phash"} |
|
|
|
|
| def is_strong(decider: str) -> bool: |
| return decider in STRONG_DECIDERS or (decider or "").startswith("rule:") |
|
|
|
|
| def load_existing(pattern: str | None) -> set[str]: |
| if not pattern: |
| return set() |
| texts = set() |
| for p in glob.glob(pattern): |
| for line in open(p, encoding="utf-8"): |
| line = line.strip() |
| if line: |
| texts.add(json.loads(line)["text"].strip()) |
| return texts |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--src", default="data/raw/archive/archive") |
| ap.add_argument("--out_dir", default="data/raw/archive/converted") |
| ap.add_argument("--exclude-against", default=None, |
| help="已有语料 glob，剔除其中已出现的文本（增量净新增）") |
| args = ap.parse_args() |
|
|
| existing = load_existing(args.exclude_against) |
| stat = Counter() |
| groups: dict[str, list[dict]] = {} |
|
|
| for p in sorted(glob.glob(f"{args.src}/*/judgments-*.jsonl")): |
| for line in open(p, encoding="utf-8"): |
| line = line.strip() |
| if not line: |
| continue |
| r = json.loads(line) |
| verdict = r.get("verdict") |
| if verdict in ("skipped", "error") or verdict not in ("ad", "clean"): |
| stat[f"剔除:verdict={verdict}"] += 1 |
| continue |
| text = (r.get("text") or "").strip() |
| if not text: |
| stat["剔除:空文本"] += 1 |
| continue |
| kind = r.get("ad_kind") |
| if verdict == "ad" and (not kind or kind == "none"): |
| verdict = "clean" |
| stat["修正:ad+kind_none→clean"] += 1 |
| decider = r.get("decider") or "" |
| groups.setdefault(text, []).append({ |
| "label": "ad" if verdict == "ad" else "normal", |
| "strong": is_strong(decider), |
| "decider": decider, |
| "ad_kind": kind, |
| "action": r.get("action"), |
| "chat_id": r.get("chat_id"), |
| "message_id": r.get("message_id"), |
| "confidence": r.get("confidence"), |
| }) |
|
|
| stat["judgment 行"] = sum(len(v) for v in groups.values()) + \ |
| sum(v for k, v in stat.items() if k.startswith("剔除")) |
| resolved: dict[str, dict] = {} |
| for text, sightings in groups.items(): |
| if len(sightings) > 1: |
| stat["去重:同文本多 sighting"] += 1 |
| strong = [s for s in sightings if s["strong"]] |
| pool = strong or sightings |
| labels = {s["label"] for s in pool} |
| if len(labels) > 1: |
| key = "丢弃:强标注矛盾" if strong else "丢弃:弱标注矛盾" |
| stat[key] += 1 |
| continue |
| best = sorted(pool, key=lambda s: (s["strong"], s["confidence"] or 0), reverse=True)[0] |
| resolved[text] = {**best, "text": text} |
|
|
| strong_rows = {t: v for t, v in resolved.items() if v["strong"]} |
| excluded = 0 |
| if existing: |
| before = len(strong_rows) |
| strong_rows = {t: v for t, v in strong_rows.items() if t not in existing} |
| excluded = before - len(strong_rows) |
| stat["剔除:已在现有语料中"] = excluded |
|
|
| out_dir = Path(args.out_dir) |
| out_dir.mkdir(parents=True, exist_ok=True) |
|
|
| def dump(path: Path, rows: dict): |
| with open(path, "w", encoding="utf-8") as f: |
| for row in sorted(rows.values(), key=lambda r: (r["label"], r["text"])): |
| f.write(json.dumps({ |
| "text": row["text"], "label": row["label"], |
| "source": "archive", "decider": row["decider"], |
| "ad_kind": row["ad_kind"], "action": row["action"], |
| "strong": row["strong"], |
| }, ensure_ascii=False) + "\n") |
|
|
| dump(out_dir / "archive.strong.binary.jsonl", strong_rows) |
| dump(out_dir / "archive.all.binary.jsonl", resolved) |
|
|
| report = { |
| "unique_texts_all": len(resolved), |
| "strong_unique": len({t: v for t, v in resolved.items() if v["strong"]}), |
| "strong_after_exclude": len(strong_rows), |
| "label_dist_strong": dict(Counter(r["label"] for r in strong_rows.values())), |
| "label_dist_all": dict(Counter(r["label"] for r in resolved.values())), |
| "ad_kind_dist_strong": dict(Counter(r["ad_kind"] for r in strong_rows.values()).most_common()), |
| "action_dist_strong": dict(Counter(r["action"] for r in strong_rows.values()).most_common()), |
| "decider_dist_strong": dict(Counter(r["decider"] for r in strong_rows.values()).most_common(10)), |
| "marker_forms": { |
| "has_fullwidth_bracket": sum(1 for t in resolved if "［" in t), |
| "has_xml_tag": sum(1 for t in resolved if re.search(r"<[a-z_]+>", t)), |
| }, |
| "stats": dict(stat), |
| } |
| (out_dir / "archive.report.json").write_text( |
| json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") |
|
|
| print(f"唯一文本 {len(resolved)} | 强标注 {report['strong_unique']} " |
| f"| 剔除已存在 {excluded} → 净新增 {len(strong_rows)}") |
| print("净新增 label:", report["label_dist_strong"]) |
| print("统计:", dict(stat)) |
| print(f"[OK] {out_dir}/archive.strong.binary.jsonl, archive.all.binary.jsonl, archive.report.json") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
