Add corpus index text dedupe command

README.md

@@ -211,7 +211,7 @@ Global flags (most commands): `--profile`, `--home`, `--json`, plus branding `--
 | `init` | Create profile under `~/.personality-protect/` |
 | `download` | Prefetch quantized MLX or GGUF base |
 | `ingest` | Index LinkedIn export and/or local paths |
-| `select` | Gate corpus by length / year — required before `build-style-profile` |
+| `dedupe-index` | Report pieces repeating another piece's text; `--apply` backs up then rewrites the index |
 | `index-voice` | Build local voice retrieval index |
 | `build-style-profile` | Build cadence / length / banned-filler style card |
 | `write` | Draft a post or article (`--channel post\|article`) |

src/personality_protect/cli.py

@@ -4,6 +4,7 @@
 
 import json
 import sys
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
 
@@ -39,6 +40,7 @@
     init_profile,
     load_config,
 )
+from personality_protect.corpus_dedupe import DEFAULT_NEAR_RATIO, dedupe_pieces
 from personality_protect.demo import run_demo
 from personality_protect.download import run_download
 from personality_protect.eval_compare import (
@@ -73,7 +75,7 @@
     DEFAULT_MAX_SEQ_LENGTH,
     PROOF_MAX_STEPS,
 )
-from personality_protect.models import load_index, summarize_by_source_year
+from personality_protect.models import load_index, save_index, summarize_by_source_year
 from personality_protect.pair_gate import (
     MAX_INPUT_PROPER_PER_1K,
     MAX_STERILE_FRAG_DELTA,
@@ -192,8 +194,8 @@ def main(
         typer.echo("Run with --help for commands. Data never leaves this machine.")
         typer.echo("")
         typer.echo(
-            "Commands: init | download | ingest | index-voice | build-style-profile | "
-            "select | write | eval-write-holdout | train | filter | "
+            "Commands: init | download | ingest | dedupe-index | index-voice | "
+            "build-style-profile | select | write | eval-write-holdout | train | filter | "
             "eval | compare | scorecard | pair-gate | sterile-check | "
             "translator-eval | demo | api | logo | status"
         )
@@ -362,6 +364,91 @@ def index_voice_cmd(
     console.print(f"Voice index: {result['voice_index']}")
 
 
+@app.command("dedupe-index")
+def dedupe_index_cmd(
+    ctx: typer.Context,
+    apply: bool = typer.Option(
+        False,
+        "--apply",
+        help="Rewrite the index (default: report only).",
+    ),
+    near_ratio: float = typer.Option(
+        DEFAULT_NEAR_RATIO,
+        "--near-ratio",
+        help="Similarity for near-identical captures; 1.0 for exact text only.",
+    ),
+    holdout_id: Optional[list[str]] = typer.Option(
+        None,
+        "--holdout-id",
+        help="Id that must survive as keeper (repeatable; adds to the local holdout file).",
+    ),
+    profile: str = typer.Option(DEFAULT_PROFILE, "--profile"),
+    home: Optional[Path] = typer.Option(None, "--home"),
+    as_json: bool = typer.Option(False, "--json"),
+) -> None:
+    """Drop pieces repeating another piece's text (re-ingest under a new path).
+
+    Rebuild ``index-voice``, ``select`` and ``build-style-profile`` after
+    applying, otherwise they keep counting the dropped pieces.
+    """
+    from personality_protect.writer_sft import load_holdout_id_set
+
+    _banner_from_ctx(ctx, json_mode=as_json)
+    paths = get_paths(profile, home=home)
+    pieces = load_index(paths.index_path)
+    if not pieces:
+        console.print(f"[red]No corpus index at {paths.index_path}.[/red]")
+        raise typer.Exit(1)
+
+    holdouts = load_holdout_id_set(paths) | {
+        str(piece_id) for piece_id in (holdout_id or []) if str(piece_id).strip()
+    }
+    result = dedupe_pieces(
+        pieces,
+        holdout_ids=holdouts,
+        near_ratio=None if near_ratio >= 1.0 else near_ratio,
+    )
+
+    payload: dict[str, object] = {
+        "index": str(paths.index_path),
+        "applied": False,
+        "near_ratio": near_ratio,
+        "holdout_ids": sorted(holdouts),
+        "before": summarize_by_source_year(pieces),
+        "after": summarize_by_source_year(result.kept),
+        **result.to_report(),
+    }
+    if apply and result.groups:
+        stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
+        backup = paths.index_path.with_name(f"{paths.index_path.name}.bak-{stamp}")
+        backup.write_bytes(paths.index_path.read_bytes())
+        payload["backup"] = str(backup)
+        payload["written"] = save_index(paths.index_path, result.kept)
+        payload["applied"] = True
+
+    if as_json:
+        typer.echo(json.dumps(payload, indent=2))
+        return
+    console.print(
+        f"[bold]{len(result.groups)}[/bold] duplicate-text groups "
+        f"({payload['exact_groups']} exact, {payload['near_groups']} near); "
+        f"[bold]{payload['dropped']}[/bold] pieces removable."
+    )
+    console.print(f"by source: {payload['dropped_by_source']}")
+    if result.groups and payload["cross_source_groups"]:
+        console.print(
+            f"[yellow]{len(payload['cross_source_groups'])} groups span sources "
+            "— identical text kept under the more specific source.[/yellow]"
+        )
+    if payload["applied"]:
+        console.print(
+            f"Index now [bold]{payload['written']}[/bold] pieces. Backup: {payload['backup']}"
+        )
+        console.print("Rebuild: index-voice, select, build-style-profile.")
+    elif result.groups:
+        console.print("Report only. Pass --apply to rewrite the index.")
+
+
 @app.command("select")
 def select_cmd(
     ctx: typer.Context,

src/personality_protect/corpus_dedupe.py

@@ -0,0 +1,237 @@
+"""Collapse corpus pieces that repeat the same text under different ids.
+
+Piece ids are derived partly from the source path, so the same post ingested
+from two export paths (a fresh scrape plus the periodic full export) lands twice
+with two ids and ``append_index``'s id check cannot see it. Text is the only
+identity that survives a re-ingest, so dedupe compares normalized text.
+
+Near-duplicates matter for the same reason: two captures of one post differ by a
+trailing link or hashtag, so an exact key alone leaves the pair in the corpus and
+double-weights it in retrieval and in the style stats.
+"""
+
+from __future__ import annotations
+
+import re
+from collections import defaultdict
+from dataclasses import dataclass, field
+from difflib import SequenceMatcher
+from typing import Any, Iterable, Sequence
+
+from personality_protect.corpus_text import normalize_corpus_text
+from personality_protect.models import Piece
+
+# Same post captured twice differs only in trailing link/hashtag noise. Anything
+# looser starts merging distinct posts that share a template opening.
+DEFAULT_NEAR_RATIO = 0.99
+# Most specific/valuable source wins when identical text spans sources.
+SOURCE_PRIORITY: tuple[str, ...] = ("linkedin_article", "linkedin_post", "linkedin_comment")
+_WHITESPACE_RE = re.compile(r"\s+")
+
+
+def duplicate_key(text: str) -> str:
+    """Comparison key: cleaned corpus text, whitespace-collapsed, case-folded."""
+    return _WHITESPACE_RE.sub(" ", normalize_corpus_text(text)).casefold().strip()
+
+
+def _source_rank(source: str) -> int:
+    try:
+        return SOURCE_PRIORITY.index(source)
+    except ValueError:
+        return len(SOURCE_PRIORITY)
+
+
+def _keeper_sort_key(piece: Piece, holdout_ids: frozenset[str], key: str) -> tuple[Any, ...]:
+    return (
+        0 if piece.id in holdout_ids else 1,
+        _source_rank(piece.source),
+        0 if piece.date else 1,
+        -len(key),  # the fuller capture of a near-duplicate pair
+        -len((piece.title or "").strip()),
+        -len(piece.meta or {}),
+        piece.id,
+    )
+
+
+@dataclass
+class DuplicateGroup:
+    """One set of pieces holding the same text, with the chosen keeper first."""
+
+    keeper: Piece
+    dropped: list[Piece]
+    exact: bool
+
+    @property
+    def members(self) -> list[Piece]:
+        return [self.keeper, *self.dropped]
+
+    @property
+    def sources(self) -> list[str]:
+        return sorted({piece.source for piece in self.members})
+
+    @property
+    def cross_source(self) -> bool:
+        return len(self.sources) > 1
+
+    def to_report(self) -> dict[str, Any]:
+        return {
+            "keeper": self.keeper.id,
+            "keeper_source": self.keeper.source,
+            "keeper_words": self.keeper.word_count,
+            "dropped": [piece.id for piece in self.dropped],
+            "sources": self.sources,
+            "cross_source": self.cross_source,
+            "exact": self.exact,
+        }
+
+
+@dataclass
+class DedupeResult:
+    kept: list[Piece]
+    groups: list[DuplicateGroup] = field(default_factory=list)
+
+    @property
+    def dropped_ids(self) -> list[str]:
+        return [piece.id for group in self.groups for piece in group.dropped]
+
+    def to_report(self) -> dict[str, Any]:
+        dropped_by_source: dict[str, int] = defaultdict(int)
+        for group in self.groups:
+            for piece in group.dropped:
+                dropped_by_source[piece.source] += 1
+        return {
+            "groups": len(self.groups),
+            "exact_groups": sum(1 for group in self.groups if group.exact),
+            "near_groups": sum(1 for group in self.groups if not group.exact),
+            "dropped": len(self.dropped_ids),
+            "dropped_by_source": dict(sorted(dropped_by_source.items())),
+            "cross_source_groups": [
+                group.to_report() for group in self.groups if group.cross_source
+            ],
+            "group_reports": [group.to_report() for group in self.groups],
+        }
+
+
+def _near_duplicate_links(
+    pieces: Sequence[Piece],
+    keys: dict[str, str],
+    near_ratio: float,
+) -> list[tuple[str, str]]:
+    """Pair ids whose text is near-identical, comparing only plausible candidates.
+
+    Candidates share a source and date, which is what a re-ingested capture of
+    the same piece looks like; that keeps the quadratic comparison inside small
+    buckets instead of across the whole corpus.
+    """
+    buckets: dict[tuple[str, str], list[Piece]] = defaultdict(list)
+    for piece in pieces:
+        if keys[piece.id]:
+            buckets[(piece.source, piece.date or "")].append(piece)
+
+    links: list[tuple[str, str]] = []
+    for bucket in buckets.values():
+        if len(bucket) < 2:
+            continue
+        ordered = sorted(bucket, key=lambda piece: piece.id)
+        for index, left in enumerate(ordered):
+            for right in ordered[index + 1 :]:
+                matcher = SequenceMatcher(None, keys[left.id], keys[right.id])
+                # Cheap upper bounds first: length, then shared character counts.
+                if matcher.real_quick_ratio() < near_ratio:
+                    continue
+                if matcher.quick_ratio() < near_ratio:
+                    continue
+                if matcher.ratio() >= near_ratio:
+                    links.append((left.id, right.id))
+    return links
+
+
+def find_duplicate_groups(
+    pieces: Iterable[Piece],
+    *,
+    holdout_ids: Iterable[str] = (),
+    near_ratio: float | None = None,
+) -> list[DuplicateGroup]:
+    """Group pieces sharing text and pick a deterministic keeper for each group.
+
+    ``near_ratio`` also folds in near-identical captures of the same piece; pass
+    ``None`` to compare exact normalized text only.
+    """
+    pieces = list(pieces)
+    holdouts = frozenset(str(piece_id) for piece_id in holdout_ids)
+    keys = {piece.id: duplicate_key(piece.text) for piece in pieces}
+
+    parent: dict[str, str] = {piece.id: piece.id for piece in pieces}
+
+    def find(piece_id: str) -> str:
+        while parent[piece_id] != piece_id:
+            parent[piece_id] = parent[parent[piece_id]]
+            piece_id = parent[piece_id]
+        return piece_id
+
+    def union(left: str, right: str) -> None:
+        left_root, right_root = find(left), find(right)
+        if left_root != right_root:
+            parent[max(left_root, right_root)] = min(left_root, right_root)
+
+    exact_members: dict[str, list[str]] = defaultdict(list)
+    for piece in pieces:
+        if keys[piece.id]:
+            exact_members[keys[piece.id]].append(piece.id)
+    for members in exact_members.values():
+        for piece_id in members[1:]:
+            union(members[0], piece_id)
+
+    exact_ids = {
+        piece_id
+        for members in exact_members.values()
+        if len(members) > 1
+        for piece_id in members
+    }
+    if near_ratio is not None:
+        for left, right in _near_duplicate_links(pieces, keys, near_ratio):
+            union(left, right)
+
+    by_root: dict[str, list[Piece]] = defaultdict(list)
+    for piece in pieces:
+        if keys[piece.id]:
+            by_root[find(piece.id)].append(piece)
+
+    groups: list[DuplicateGroup] = []
+    for members in by_root.values():
+        if len(members) < 2:
+            continue
+        ordered = sorted(members, key=lambda p: _keeper_sort_key(p, holdouts, keys[p.id]))
+        groups.append(
+            DuplicateGroup(
+                keeper=ordered[0],
+                dropped=ordered[1:],
+                exact=all(piece.id in exact_ids for piece in members)
+                and len({keys[piece.id] for piece in members}) == 1,
+            )
+        )
+    groups.sort(key=lambda group: group.keeper.id)
+    return groups
+
+
+def dedupe_pieces(
+    pieces: Iterable[Piece],
+    *,
+    holdout_ids: Iterable[str] = (),
+    near_ratio: float | None = None,
+) -> DedupeResult:
+    """Drop duplicate-text pieces, keeping input order and every holdout id."""
+    pieces = list(pieces)
+    holdouts = frozenset(str(piece_id) for piece_id in holdout_ids)
+    groups = find_duplicate_groups(pieces, holdout_ids=holdouts, near_ratio=near_ratio)
+
+    dropped = {piece.id for group in groups for piece in group.dropped}
+    protected = dropped & holdouts
+    if protected:
+        raise AssertionError(
+            "refusing to drop holdout ids: " + ", ".join(sorted(protected))
+        )
+    return DedupeResult(
+        kept=[piece for piece in pieces if piece.id not in dropped],
+        groups=groups,
+    )