Blame for code/scripts/build_channel_dataset.py

#!/usr/bin/env python3
"""解析 Telegram Desktop HTML 导出的频道存档，产出带标签的数据集。

与 build_archive_dataset.py 的区别：那边是 bot 的 JSONL 存档，这边是 TG Desktop 的
分页 HTML（css/js/images 是界面资源，messages*.html 才是正文）。

频道内容整体即广告，因此标签默认取频道先验（label_source=channel_prior）；
但仍要剥掉 service 消息（建频道/置顶/日期分隔）和缩略图，否则训练集会被界面噪音污染。
"""

from __future__ import annotations

import argparse
import collections
import datetime as dt
import hashlib
import html
import json
import os
import re
import sys
from html.parser import HTMLParser

VOID = {"br", "img", "meta", "link", "input", "hr", "source"}
DATE_TITLE_RE = re.compile(r"(\d{2})\.(\d{2})\.(\d{4}) (\d{2}):(\d{2}):(\d{2}) UTC([+-]\d{2}):(\d{2})")
PHOTO_RE = re.compile(r"^photos/(.+?)@(\d{2})-(\d{2})-(\d{4})_(\d{2})-(\d{2})-(\d{2})", re.IGNORECASE)
CONTACT_RE = re.compile(r"(https?://|t\.me/|@[A-Za-z0-9_]{4,}|微信|vx|QQ|电报|飞机|私聊|加我|联系)", re.I)


class Node:
    __slots__ = ("tag", "attrs", "children", "parent")

    def __init__(self, tag: str, attrs: dict[str, str], parent: "Node | None" = None):
        self.tag = tag
        self.attrs = attrs
        self.children: list["Node | str"] = []
        self.parent = parent

    @property
    def classes(self) -> set[str]:
        return set((self.attrs.get("class") or "").split())

    def iter(self):
        yield self
        for ch in self.children:
            if isinstance(ch, Node):
                yield from ch.iter()

    def find_all(self, pred) -> list["Node"]:
        return [n for n in self.iter() if n is not self and pred(n)]

    def text(self, br: str = "\n") -> str:
        out: list[str] = []
        for ch in self.children:
            if isinstance(ch, str):
                out.append(ch)
            elif ch.tag == "br":
                out.append(br)
            else:
                out.append(ch.text(br))
        return "".join(out)


class DomBuilder(HTMLParser):
    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.root = Node("#root", {})
        self.cur = self.root

    def handle_starttag(self, tag, attrs):
        node = Node(tag, {k: (v or "") for k, v in attrs}, self.cur)
        self.cur.children.append(node)
        if tag not in VOID:
            self.cur = node

    def handle_startendtag(self, tag, attrs):
        self.cur.children.append(Node(tag, {k: (v or "") for k, v in attrs}, self.cur))

    def handle_endtag(self, tag):
        node = self.cur
        while node is not self.root and node.tag != tag:
            node = node.parent
        if node is not self.root:
            self.cur = node.parent

    def handle_data(self, data):
        if data:
            self.cur.children.append(data)


def parse_date(title: str) -> int | None:
    """把 '01.04.2023 22:58:27 UTC+08:00' 换成 unix 时间戳。"""
    m = DATE_TITLE_RE.search(title or "")
    if not m:
        return None
    d, mo, y, hh, mm, ss, oh, om = m.groups()
    offset = dt.timezone(dt.timedelta(hours=int(oh), minutes=int(om) * (1 if oh.startswith("+") else -1)))
    try:
        return int(dt.datetime(int(y), int(mo), int(d), int(hh), int(mm), int(ss), tzinfo=offset).timestamp())
    except ValueError:
        return None


def parse_file(path: str) -> list[dict]:
    with open(path, encoding="utf-8", errors="replace") as fh:
        builder = DomBuilder()
        builder.feed(fh.read())
    msgs = []
    for node in builder.root.iter():
        if node.tag != "div" or "message" not in node.classes:
            continue
        cls = node.classes
        raw_id = node.attrs.get("id", "")
        if "service" in cls:
            # 建频道/置顶/日期分隔等系统消息，不是内容
            msgs.append({"kind": "service", "message_id": raw_id,
                         "text": node.text().strip(), "service": True})
            continue
        mid = int(raw_id.replace("message", "")) if raw_id.startswith("message") else None
        date_node = next((n for n in node.iter() if "date" in n.classes and n.attrs.get("title")), None)
        ts = parse_date(date_node.attrs["title"]) if date_node else None
        from_node = next((n for n in node.iter() if "from_name" in n.classes), None)
        text_node = next((n for n in node.iter() if "text" in n.classes and n.tag == "div"), None)
        photos, media_kind = [], "none"
        for n in node.iter():
            if n.tag == "a" and "photo_wrap" in n.classes:
                href = n.attrs.get("href", "")
                if href:
                    photos.append(href)
            for mk in ("media_photo", "media_video", "media_file", "media_music", "media_sticker"):
                if mk in n.classes and media_kind == "none":
                    media_kind = mk.replace("media_", "")
        msgs.append({
            "kind": "message",
            "message_id": mid,
            "source_file": os.path.basename(path),
            "date": ts,
            "from_name": from_node.text().strip() if from_node else None,
            "text": text_node.text().strip() if text_node else "",
            "photos": photos,
            "media_kind": media_kind if not photos else "photo",
        })
    return msgs


def collect_photos(base: str) -> dict[str, dict]:
    """photos/ 下按 photo_id 归组，丢掉 _thumb 缩略图（它们和原图同 id）。"""
    out: dict[str, dict] = {}
    pdir = os.path.join(base, "photos")
    if not os.path.isdir(pdir):
        return out
    for name in sorted(os.listdir(pdir)):
        m = re.match(r"^photo_(\d+)@", name)
        if not m:
            continue
        pid = m.group(1)
        thumb = "_thumb" in name
        rec = out.setdefault(pid, {"pid": pid, "full": [], "thumbs": []})
        (rec["thumbs"] if thumb else rec["full"]).append(name)
    return out


def validate_image(path: str) -> tuple[int, int, str] | None:
    from PIL import Image
    try:
        with Image.open(path) as im:
            im.load()
            return im.size[0], im.size[1], im.format or ""
    except Exception:
        return None


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--src", default="存档数据-1")
    ap.add_argument("--out", default="data/processed/channel")
    ap.add_argument("--default-kind", default="porn_bait", help="频道先验对应的 ad_kind")
    args = ap.parse_args()

    src = os.path.abspath(args.src)
    htmls = sorted(p for p in os.listdir(src) if re.match(r"^messages\d*\.html$", p))
    msgs: list[dict] = []
    for name in htmls:
        msgs.extend(parse_file(os.path.join(src, name)))
    service = [m for m in msgs if m.get("service")]
    msgs = [m for m in msgs if not m.get("service")]
    print(f"[parse] {len(htmls)} 个 HTML，内容消息 {len(msgs)}，service {len(service)}")

    photos = collect_photos(src)
    print(f"[photo] {len(photos)} 个 photo_id，"
          f"原图 {sum(len(p['full']) for p in photos.values())}，"
          f"缩略图 {sum(len(p['thumbs']) for p in photos.values())}")

    # 去重：同一 message_id 只保留一条
    by_id: dict[int, dict] = {}
    dup = 0
    for m in msgs:
        if m["message_id"] is None:
            continue
        if m["message_id"] in by_id:
            dup += 1
            continue
        by_id[m["message_id"]] = m
    print(f"[dedup] 消息去重 {dup}，保留 {len(by_id)}")

    rows, image_rows, stats = [], [], collections.Counter()
    seen_hash: dict[str, str] = {}
    for mid, m in sorted(by_id.items()):
        text = m["text"]
        # photo_wrap 的 href 指向 photos/photo_X@...; 映射回 pid
        pids = []
        for href in m["photos"]:
            mm = re.match(r"^photos/photo_(\d+)@", href)
            if mm:
                pids.append(mm.group(1))
        imgs = []
        for pid in pids:
            rec = photos.get(pid)
            if not rec or not rec["full"]:
                stats["photo_missing_file"] += 1
                continue
            fname = rec["full"][0]
            path = os.path.join(src, "photos", fname)
            checked = validate_image(path)
            if not checked:
                stats["photo_corrupt"] += 1
                continue
            w, h, fmt = checked
            if w < 64 or h < 64:
                stats["photo_too_small"] += 1
                continue
            with open(path, "rb") as fh:
                digest = hashlib.sha256(fh.read()).hexdigest()
            if digest in seen_hash:
                stats["photo_exact_dup"] += 1
                continue
            seen_hash[digest] = fname
            imgs.append({"path": os.path.relpath(path, os.path.dirname(src)).replace("\\", "/"),
                         "sha256": digest, "width": w, "height": h, "format": fmt,
                         "bytes": os.path.getsize(path)})
            stats["photo_kept"] += 1

        # 频道先验：所有内容帖视为广告；仅把"纯运营/求助"类标记待复核
        low_info = len(text) < 4
        needs_review = low_info or ("置顶" in text and len(text) < 20)
        rows.append({
            "chat_id": "channel",
            "message_id": mid,
            "date": m["date"] or 0,
            "iso_time": dt.datetime.fromtimestamp(m["date"]).isoformat(timespec="seconds") if m["date"] else None,
            "from_name": m["from_name"],
            "text": text,
            "text_len": len(text),
            "has_link": bool(CONTACT_RE.search(text)),
            "media_kind": m["media_kind"],
            "images": [i["path"] for i in imgs],
            "label": "ad",
            "ad_kind": args.default_kind,
            "label_source": "channel_prior",
            "label_trust": "prior",
            "needs_review": needs_review,
            "low_info": low_info,
            "source_file": m["source_file"],
        })
        for i in imgs:
            image_rows.append({**i, "chat_id": "channel", "message_id": mid,
                               "date": m["date"] or 0, "label": "ad",
                               "ad_kind": args.default_kind, "label_source": "channel_prior",
                               "text": text})

    # 时间切分
    rows.sort(key=lambda r: r["date"])
    img_rows = sorted(image_rows, key=lambda r: r["date"])
    for group in (rows, img_rows):
        dates = [r["date"] for r in group if r["date"]]
        if not dates:
            continue
        cutoff = dates[int(len(dates) * 0.9)]
        for r in group:
            r["split"] = "val" if r["date"] >= cutoff else "train"

    out = args.out
    os.makedirs(out, exist_ok=True)
    for name, data in (("messages.jsonl", rows), ("images.jsonl", img_rows)):
        with open(os.path.join(out, name), "w", encoding="utf-8") as fh:
            for r in data:
                fh.write(json.dumps(r, ensure_ascii=False) + "\n")

    report = {
        "src": src,
        "html_files": htmls,
        "service_messages": len(service),
        "service_samples": [s["text"][:60] for s in service[:5]],
        "messages": len(rows),
        "images": len(img_rows),
        "image_join_stats": dict(stats),
        "date_range": [dt.datetime.fromtimestamp(rows[0]["date"]).isoformat(timespec="seconds"),
                       dt.datetime.fromtimestamp(rows[-1]["date"]).isoformat(timespec="seconds")] if rows else None,
        "with_text": sum(1 for r in rows if r["text"]),
        "with_link": sum(1 for r in rows if r["has_link"]),
        "with_image": sum(1 for r in rows if r["images"]),
        "needs_review": sum(1 for r in rows if r["needs_review"]),
        "split": dict(collections.Counter(r["split"] for r in rows)),
    }
    with open(os.path.join(out, "report.json"), "w", encoding="utf-8") as fh:
        json.dump(report, fh, ensure_ascii=False, indent=2)
    print(json.dumps(report, ensure_ascii=False, indent=2))
    print(f"[out] {os.path.abspath(out)}")


if __name__ == "__main__":
    sys.exit(main())