awesome-kubernetes/src/dedup.py

from __future__ import annotations

import re
import asyncio
from collections import defaultdict
from difflib import SequenceMatcher
from typing import Dict, List, Tuple
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse

from src.inventory_manager import load_inventory, save_inventory
from src.logger import log_event

TRACKING_PARAMS = {"utm_source", "utm_medium", "utm_campaign", "utm_content", "utm_term",
                   "ref", "source", "fbclid", "gclid", "mc_cid", "mc_eid", "s", "share"}


def normalize_url_deep(url: str) -> str:
    parsed = urlparse(url.strip().lower())
    scheme = "https"
    netloc = parsed.netloc.removeprefix("www.")
    path = parsed.path.rstrip("/") or "/"
    params = parse_qs(parsed.query)
    clean_params = {k: v for k, v in params.items() if k not in TRACKING_PARAMS}
    query = urlencode(clean_params, doseq=True) if clean_params else ""
    return urlunparse((scheme, netloc, path, "", query, ""))


def normalize_title(title: str) -> str:
    if not title:
        return ""
    t = title.lower().strip()
    t = re.sub(r'^[\w.-]+\.\w{2,}:\s*', '', t)
    t = re.sub(r'[^\w\s]', ' ', t)
    t = re.sub(r'\s+', ' ', t).strip()
    return t


def find_url_duplicates(inventory: Dict) -> List[Tuple[str, str]]:
    norm_map = defaultdict(list)
    for url in inventory:
        if not isinstance(inventory[url], dict):
            continue
        deep = normalize_url_deep(url)
        norm_map[deep].append(url)

    duplicates = []
    for norm, urls in norm_map.items():
        if len(urls) > 1:
            for i in range(1, len(urls)):
                duplicates.append((urls[0], urls[i]))
    return duplicates


def find_hash_duplicates(inventory: Dict) -> List[List[str]]:
    hash_map = defaultdict(list)
    for url, entry in inventory.items():
        if not isinstance(entry, dict):
            continue
        ch = entry.get("content_hash")
        if ch and ch != "N/A":
            hash_map[ch].append(url)
    return [urls for urls in hash_map.values() if len(urls) > 1]


def find_title_duplicates(inventory: Dict, threshold: float = 0.85) -> List[Tuple[str, str, float]]:
    entries = []
    for url, entry in inventory.items():
        if not isinstance(entry, dict):
            continue
        title = entry.get("title", "")
        norm = normalize_title(title)
        if len(norm) < 10:
            continue
        entries.append((url, norm, entry.get("stars") or 0))

    log_event(f"[Dedup] Building title index for {len(entries)} entries...")

    prefix_groups = defaultdict(list)
    for url, norm, stars in entries:
        words = norm.split()
        prefix = " ".join(words[:3]) if len(words) >= 3 else norm
        prefix_groups[prefix].append((url, norm, stars))

    duplicates = []
    checked = 0
    for prefix, group in prefix_groups.items():
        if len(group) < 2:
            continue
        for i in range(len(group)):
            for j in range(i + 1, len(group)):
                url1, norm1, stars1 = group[i]
                url2, norm2, stars2 = group[j]
                if stars1 >= 4 and stars2 >= 4:
                    continue
                ratio = SequenceMatcher(None, norm1, norm2).ratio()
                if ratio >= threshold:
                    duplicates.append((url1, url2, ratio))
        checked += 1
        if checked % 500 == 0:
            log_event(f"[Dedup] Checked {checked}/{len(prefix_groups)} prefix groups...")

    log_event(f"[Dedup] Title scan complete: {len(duplicates)} potential duplicates found")
    return duplicates


def _entry_score(entry: Dict) -> Tuple:
    return (
        entry.get("stars") or 0,
        1 if entry.get("ai_summary") else 0,
        1 if entry.get("hierarchy") else 0,
        len(entry.get("tags", [])),
        -len(str(entry.get("url", "")))
    )


def resolve_duplicates(inventory: Dict, duplicate_pairs: List[Tuple[str, str, float]]) -> int:
    resolved = 0
    seen = set()

    for url1, url2, score in sorted(duplicate_pairs, key=lambda x: -x[2]):
        if url1 in seen or url2 in seen:
            continue

        entry1 = inventory.get(url1, {})
        entry2 = inventory.get(url2, {})

        if not isinstance(entry1, dict) or not isinstance(entry2, dict):
            continue

        score1 = _entry_score(entry1)
        score2 = _entry_score(entry2)

        if score1 >= score2:
            winner, loser = url1, url2
        else:
            winner, loser = url2, url1

        inventory[loser]["status"] = "duplicate"
        inventory[loser]["duplicate_of"] = winner
        seen.add(loser)
        resolved += 1

    return resolved


async def run_dedup(dry_run: bool = True) -> Dict:
    log_event("STARTING DEDUPLICATION SCAN", section_break=True)
    inventory = load_inventory()

    url_dups = find_url_duplicates(inventory)
    log_event(f"[Dedup] URL duplicates: {len(url_dups)}")

    hash_groups = find_hash_duplicates(inventory)
    hash_dups = []
    for group in hash_groups:
        for i in range(1, len(group)):
            hash_dups.append((group[0], group[i], 1.0))
    log_event(f"[Dedup] Content hash duplicates: {len(hash_dups)}")

    title_dups = find_title_duplicates(inventory)

    all_dups = [(u1, u2, 1.0) for u1, u2 in url_dups] + hash_dups + title_dups
    unique_pairs = {}
    for u1, u2, s in all_dups:
        key = tuple(sorted([u1, u2]))
        if key not in unique_pairs or s > unique_pairs[key]:
            unique_pairs[key] = s
    deduped_pairs = [(k[0], k[1], v) for k, v in unique_pairs.items()]

    stats = {
        "url_duplicates": len(url_dups),
        "hash_duplicates": len(hash_dups),
        "title_duplicates": len(title_dups),
        "total_unique_pairs": len(deduped_pairs),
    }

    if dry_run:
        log_event(f"[Dedup] DRY RUN — {len(deduped_pairs)} duplicates found, no changes made")
        for u1, u2, score in sorted(deduped_pairs, key=lambda x: -x[2])[:20]:
            t1 = inventory.get(u1, {}).get("title", "?")[:60]
            t2 = inventory.get(u2, {}).get("title", "?")[:60]
            log_event(f"  [{score:.0%}] {t1} <-> {t2}")
    else:
        resolved = resolve_duplicates(inventory, deduped_pairs)
        stats["resolved"] = resolved
        save_inventory(inventory)
        log_event(f"[Dedup] Resolved {resolved} duplicates")

    log_event(f"DEDUP COMPLETE: {stats}")
    return stats


if __name__ == "__main__":
    asyncio.run(run_dedup(dry_run=True))