scripts/canonicalize_duplicates.py

Rama main · réplica de lectura del repositorio original.

#!/usr/bin/env python3
"""
Canonicalize duplicate source_url entries in search_index.json.

Strategy:
- group by source_url
- skip invalid / empty source_url
- score candidates by quality
- keep best entry as canonical
- mark the rest as excluded duplicates

Usage:
    python3 scripts/canonicalize_duplicates.py
    python3 scripts/canonicalize_duplicates.py --dry-run
"""

from __future__ import annotations

import argparse
import json
import os
import re
from collections import defaultdict
import sys
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))
import xkb_paths
import xkb_frontmatter
import xkb_index

WORKSPACE = xkb_paths.WORKSPACE
BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR
INDEX_FILE = xkb_paths.INDEX_FILE

from normalize_index_quality import LOW_SIGNAL_SUMMARIES, clean_summary, is_valid_source_url


def entry_score(item: dict[str, Any]) -> tuple:
    title = (item.get("title") or "").strip()
    summary = clean_summary(item.get("summary") or "")
    rel_path = (item.get("relative_path") or item.get("path") or "").strip()

    enriched = 1 if item.get("enriched") else 0
    has_good_summary = 1 if summary not in LOW_SIGNAL_SUMMARIES and len(summary) >= 18 else 0
    has_human_title = 1
    if not title or re.fullmatch(r"\d{15,20}", title) or re.fullmatch(r"tweet\s+\d{15,20}", title.lower()):
        has_human_title = 0
    in_cards = 1 if rel_path.startswith(("cards/", "memory/cards/")) else 0
    not_excluded = 1 if not item.get("excluded") else 0
    non_legacy = 1 if "legacy-" not in rel_path else 0
    title_len = min(len(title), 120)
    summary_len = min(len(summary), 300)

    return (
        not_excluded,
        enriched,
        in_cards,
        has_good_summary,
        has_human_title,
        non_legacy,
        summary_len,
        title_len,
    )


def main() -> int:
    parser = argparse.ArgumentParser(description="Canonicalize duplicate source_url entries")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    raw = json.loads(INDEX_FILE.read_text(encoding="utf-8"))
    items = raw.get("items", raw) if isinstance(raw, dict) else raw
    is_dict = isinstance(raw, dict)

    groups: dict[str, list[int]] = defaultdict(list)
    for idx, item in enumerate(items):
        source_url = (item.get("source_url") or "").strip()
        if source_url and is_valid_source_url(source_url):
            groups[source_url].append(idx)

    duplicate_groups = {url: idxs for url, idxs in groups.items() if len(idxs) > 1}
    changed = 0
    excluded_dupes = 0
    examples = []
    unmarkable: list[str] = []

    for source_url, idxs in duplicate_groups.items():
        ranked = sorted(idxs, key=lambda i: entry_score(items[i]), reverse=True)
        keep_idx = ranked[0]
        keep_item = items[keep_idx]
        keep_rel = keep_item.get("relative_path") or keep_item.get("path") or ""
        # 保留的那一份的所有檔案。這組底下沒有任何一個標記可以碰到它們。
        keep_files = set(xkb_index.files_for_item(keep_item, by_source=False))

        for idx in ranked[1:]:
            item = items[idx]
            rel_path = item.get("relative_path") or item.get("path") or ""
            changed_here = False

            # 標在檔案上，不是標在索引列上。索引是衍生物：標在列上的話，
            # 下一次重建就把這個決定洗掉——實測正式索引 1680 筆裡帶 excluded
            # 的是 0 筆，這支腳本過去的成果一次都沒留下來。
            #
            # 而且一列代表的是一組檔案：同一份知識可能被歸進兩個分類資料夾。
            # 只標其中一份的話，重建時去重會挑「沒有被排除」的那一份留下，
            # 排除等於沒發生。
            #
            # by_source=False 是這裡的關鍵：重複組本來就共用同一個來源，用來源
            # 去找會把「要保留的那一份」也撈進來。加上 keep_files 這道保險——
            # 保留的那一份無論如何都不能被標，否則整組一起消失。
            files = [p for p in xkb_index.files_for_item(item, by_source=False)
                     if p not in keep_files]
            if not files:
                unmarkable.append(rel_path)
            for path in files:
                if xkb_frontmatter.mark_excluded(
                        path, f"duplicate_source_url; canonical={keep_rel}",
                        dry_run=args.dry_run):
                    changed_here = True

            if changed_here:
                changed += 1
            excluded_dupes += 1
            if len(examples) < 20:
                examples.append((source_url, keep_rel, rel_path))

    print(f"duplicate groups: {len(duplicate_groups)}")
    print(f"duplicate entries excluded: {excluded_dupes}")
    print(f"changed: {changed}")
    if unmarkable:
        # 標記不上跟標記了不一樣。安靜跳過會讓這批重複永遠標不掉，而輸出
        # 看起來一切正常——那正是這支腳本原本在犯的錯。
        print(f"could not mark (檔案不存在或沒有 frontmatter): {len(unmarkable)}")
        for rel in unmarkable[:5]:
            print(f"  - {rel}")
    print("\nexamples:")
    for source_url, keep_rel, rel_path in examples:
        print(f"- source: {source_url}")
        print(f"  keep: {keep_rel}")
        print(f"  drop: {rel_path}")

    # 這支腳本不再寫索引。決定寫在卡片檔案上，索引由 build_search_index.sh
    # 從檔案重算——那是唯一寫索引的地方。
    if not args.dry_run and changed > 0:
        print(f"已標記 {changed} 張卡片。索引重建後生效：")
        print("  bash scripts/build_search_index.sh --incremental")

    return 0


import xkb_usage  # noqa: E402  — 量測誰在跑，見 scripts/xkb_usage.py

if __name__ == "__main__":
    xkb_usage.record(__file__)
    raise SystemExit(main())