File size: 6,707 Bytes
71dabc7 | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | #!/usr/bin/env python3
"""清洗 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()
|