From 3cba8c7f37f4d97c5cb8337197e1917472cefdbb Mon Sep 17 00:00:00 2001 From: Dusan Milicevic Date: Thu, 30 Jul 2026 13:25:00 -0500 Subject: [PATCH] Add corpus index text dedupe command Piece ids include the source path, so re-ingesting a post from a second export path lands it twice and append_index's id check cannot see it. dedupe-index groups pieces by normalized text (plus near-identical captures that differ only in trailing link/hashtag noise), keeps one deterministic keeper per group, and refuses to drop holdout ids. Report-only by default; --apply backs up the index before rewriting it and prints the index-voice / select / build-style-profile rebuild those counts depend on. Local run: 24 groups / 44 pieces removed, corpus 1647 -> 1603. Co-authored-by: Cursor --- README.md | 1 + src/personality_protect/cli.py | 93 ++++++++- src/personality_protect/corpus_dedupe.py | 237 +++++++++++++++++++++++ tests/test_corpus_dedupe.py | 209 ++++++++++++++++++++ 4 files changed, 537 insertions(+), 3 deletions(-) create mode 100644 src/personality_protect/corpus_dedupe.py create mode 100644 tests/test_corpus_dedupe.py diff --git a/README.md b/README.md index 7f3fdf6..71cd542 100644 --- a/README.md +++ b/README.md @@ -197,6 +197,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 | +| `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`) | diff --git a/src/personality_protect/cli.py b/src/personality_protect/cli.py index 0910dec..8484ee9 100644 --- a/src/personality_protect/cli.py +++ b/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, diff --git a/src/personality_protect/corpus_dedupe.py b/src/personality_protect/corpus_dedupe.py new file mode 100644 index 0000000..1cf1fa1 --- /dev/null +++ b/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, + ) diff --git a/tests/test_corpus_dedupe.py b/tests/test_corpus_dedupe.py new file mode 100644 index 0000000..35b8ce3 --- /dev/null +++ b/tests/test_corpus_dedupe.py @@ -0,0 +1,209 @@ +"""Contoso-safe tests for duplicate-text cleanup of the corpus index.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from personality_protect.cli import app +from personality_protect.config import init_profile +from personality_protect.corpus_dedupe import ( + DEFAULT_NEAR_RATIO, + dedupe_pieces, + duplicate_key, + find_duplicate_groups, +) +from personality_protect.models import Piece, load_index, save_index + +runner = CliRunner() + +CONTOSO_POST = ( + "Contoso shipped a pricing test this quarter. We compared renewal signals " + "against customer value, then killed the variant that only moved clicks. " + "The boring version won and the rollback plan stayed one page long.\n\n" + "Three things made it work. We wrote the failure condition before the " + "experiment started, so nobody could relabel a flat quarter as momentum. " + "We gave the support team the rollback switch instead of routing it through " + "a release meeting. And we published the losing numbers in the same note as " + "the winning ones, which is the part most teams quietly skip.\n\n" + "The uncomfortable finding: our best-converting plan was also our worst " + "renewing plan. Cheaper entry pricing pulled in accounts that never " + "onboarded, and the churn showed up two quarters later where nobody was " + "looking for it. Pricing is a retention decision wearing an acquisition " + "costume, and the dashboard that tracks signups will never tell you that." +) + + +def _piece(piece_id: str, **overrides) -> Piece: + fields = { + "id": piece_id, + "source": "linkedin_post", + "text": CONTOSO_POST, + "date": "2024-03-04", + } + fields.update(overrides) + return Piece(**fields) + + +def test_duplicate_key_ignores_case_whitespace_and_export_wrapping(): + assert duplicate_key(" Contoso Pricing\n\n") == duplicate_key("contoso pricing") + assert duplicate_key('"Contoso pricing."') == duplicate_key("Contoso pricing.") + assert duplicate_key("

Contoso pricing

") == duplicate_key("Contoso pricing") + assert duplicate_key(" ") == "" + + +def test_exact_duplicate_from_second_export_path_is_dropped(): + result = dedupe_pieces( + [ + _piece("aaaa", path="exports/first.csv"), + _piece("bbbb", path="exports/second.csv"), + _piece("cccc", text="Contoso platform migrations need boring rollback plans."), + ] + ) + + assert result.dropped_ids == ["bbbb"] + assert [piece.id for piece in result.kept] == ["aaaa", "cccc"] + assert result.to_report()["exact_groups"] == 1 + assert result.to_report()["dropped_by_source"] == {"linkedin_post": 1} + + +def test_keeper_prefers_specific_source_then_date_then_metadata(): + groups = find_duplicate_groups( + [ + _piece("aaaa", source="linkedin_comment"), + _piece("bbbb", source="linkedin_article"), + _piece("cccc", source="linkedin_post"), + ] + ) + assert [groups[0].keeper.id, *sorted(piece.id for piece in groups[0].dropped)] == [ + "bbbb", + "aaaa", + "cccc", + ] + assert groups[0].cross_source is True + assert groups[0].sources == ["linkedin_article", "linkedin_comment", "linkedin_post"] + + dated = find_duplicate_groups([_piece("aaaa", date=None), _piece("bbbb")]) + assert dated[0].keeper.id == "bbbb" + + titled = find_duplicate_groups( + [_piece("aaaa"), _piece("bbbb", title="Contoso pricing test")] + ) + assert titled[0].keeper.id == "bbbb" + + richer_meta = find_duplicate_groups( + [_piece("bbbb"), _piece("aaaa", meta={"file": "shares.csv"})] + ) + assert richer_meta[0].keeper.id == "aaaa" + + +def test_identical_pieces_collapse_to_the_lowest_id(): + groups = find_duplicate_groups([_piece("cccc"), _piece("aaaa"), _piece("bbbb")]) + assert groups[0].keeper.id == "aaaa" + assert [piece.id for piece in groups[0].dropped] == ["bbbb", "cccc"] + + +def test_near_identical_recapture_needs_the_ratio_and_keeps_fuller_text(): + scraped = _piece("aaaa", text=CONTOSO_POST, path="scrape/post.txt") + exported = _piece( + "bbbb", + text=CONTOSO_POST + " #contoso #pricing", + path="exports/shares.csv", + ) + + exact_only = dedupe_pieces([scraped, exported], near_ratio=None) + assert exact_only.dropped_ids == [] + + near = dedupe_pieces([scraped, exported], near_ratio=DEFAULT_NEAR_RATIO) + assert near.dropped_ids == ["aaaa"] + assert near.groups[0].keeper.id == "bbbb" + assert near.groups[0].exact is False + + +def test_distinct_pieces_sharing_an_opening_are_not_near_duplicates(): + tail = CONTOSO_POST.rsplit("\n\n", 1)[0] + result = dedupe_pieces( + [ + _piece("aaaa", text=tail + "\n\nThe renewal cohort barely moved this month."), + _piece("bbbb", text=tail + "\n\nSupport tickets dropped by a third instead."), + ], + near_ratio=DEFAULT_NEAR_RATIO, + ) + assert result.dropped_ids == [] + + +def test_holdout_id_always_becomes_the_keeper(): + result = dedupe_pieces( + [ + _piece("aaaa", source="linkedin_article", title="Contoso pricing"), + _piece("zzzz", source="linkedin_comment", date=None), + ], + holdout_ids={"zzzz"}, + ) + + assert result.groups[0].keeper.id == "zzzz" + assert result.dropped_ids == ["aaaa"] + assert "zzzz" in {piece.id for piece in result.kept} + + +def test_dedupe_refuses_when_a_holdout_would_be_dropped(): + groups = find_duplicate_groups([_piece("aaaa"), _piece("zzzz")]) + assert {piece.id for piece in groups[0].dropped} == {"zzzz"} + + with pytest.raises(AssertionError, match="zzzz"): + dedupe_pieces([_piece("aaaa"), _piece("zzzz")], holdout_ids={"aaaa", "zzzz"}) + + +def test_empty_text_pieces_never_group_together(): + result = dedupe_pieces([_piece("aaaa", text=""), _piece("bbbb", text=" ")]) + assert result.dropped_ids == [] + + +def test_cli_dedupe_index_reports_then_applies_with_backup(tmp_path: Path): + paths, _, _ = init_profile("contoso", home=tmp_path) + paths.root.joinpath("dogfood_holdout_ids.json").write_text( + json.dumps({"holdout_ids": ["zzzz"]}), encoding="utf-8" + ) + save_index( + paths.index_path, + [ + _piece("aaaa", source="linkedin_article"), + _piece("zzzz", source="linkedin_comment"), + _piece("cccc", text="Contoso platform migrations need boring rollback plans."), + ], + ) + + base_args = ["--logo", "off", "dedupe-index", "--profile", "contoso", "--home", str(tmp_path)] + report = runner.invoke(app, [*base_args, "--json"]) + assert report.exit_code == 0, report.output + payload = json.loads(report.output) + assert payload["applied"] is False + assert payload["dropped"] == 1 + assert payload["group_reports"][0]["keeper"] == "zzzz" + assert len(load_index(paths.index_path)) == 3 + + applied = runner.invoke(app, [*base_args, "--apply", "--json"]) + assert applied.exit_code == 0, applied.output + payload = json.loads(applied.output) + assert payload["applied"] is True + assert payload["written"] == 2 + assert payload["after"]["pieces"] == 2 + assert {piece.id for piece in load_index(paths.index_path)} == {"zzzz", "cccc"} + assert Path(payload["backup"]).is_file() + assert len(load_index(Path(payload["backup"]))) == 3 + + again = runner.invoke(app, [*base_args, "--apply", "--json"]) + assert json.loads(again.output)["dropped"] == 0 + + +def test_cli_dedupe_index_fails_without_an_index(tmp_path: Path): + init_profile("contoso", home=tmp_path) + result = runner.invoke( + app, + ["--logo", "off", "dedupe-index", "--profile", "contoso", "--home", str(tmp_path)], + ) + assert result.exit_code == 1 + assert "No corpus index" in result.output