#!/usr/bin/env python3 """ SERP Telemetry Poisoning Auditor (Sept-Oct 2026 Fingerprint Detector) Created by Chris Slater | AllClear Digital (allcleardigital.com | chrisslater.ai) Detects synthetic Google scraper poisoning in Semrush, BrightEdge, and Ahrefs CSV exports. Usage: python3 serp_poison_auditor.py --demo python3 serp_poison_auditor.py path/to/rankings.csv --json report.json """ import csv import sys import json import re import argparse from datetime import datetime from collections import defaultdict KNOWN_PEAK_DATES = { "2026-09-13", "2026-09-14", "2026-09-25", "2026-09-27", "2026-09-28", "2026-09-29", "2026-09-30", "2026-10-01", "2026-10-02", "2026-10-07", "2026-10-16", "2026-10-17", "2026-10-18", "2026-10-20", "2026-10-21", "2026-10-22" } DATE_REGEX = re.compile(r"(\d{4}[-/]\d{2}[-/]\d{2}|\d{2}[-/]\d{2}[-/]\d{4})") def normalize_date(date_str): date_str = date_str.strip() match = DATE_REGEX.search(date_str) if not match: for fmt in ("%b %d, %Y", "%b %d %Y", "%B %d, %Y", "%Y%m%d"): try: return datetime.strptime(date_str, fmt).strftime("%Y-%m-%d") except ValueError: continue return None raw = match.group(1).replace("/", "-") parts = raw.split("-") return raw if len(parts[0]) == 4 else f"{parts[2]}-{parts[0]}-{parts[1]}" def parse_rank(val): if val is None: return 101 s = str(val).strip().lower() if s in ("", "-", "n/a", "none", "null", "not in top 100", ">100", "0"): return 101 s = re.sub(r"[^\d.]", "", s) try: r = int(float(s)) return 101 if r <= 0 or r > 100 else r except ValueError: return 101 def load_csv_series(filepath): series = defaultdict(dict) with open(filepath, mode="r", encoding="utf-8-sig", errors="replace") as f: reader = csv.reader(f) headers = [h.strip() for h in next(reader, [])] if not headers: raise ValueError("CSV file is empty.") date_cols = {} for idx, h in enumerate(headers): if any(s in h.lower() for s in ("_type", "_landing", "url", "volume", "diff", "cpc")): continue norm_d = normalize_date(h) if norm_d: date_cols[idx] = norm_d kw_idx = 0 for idx, h in enumerate(headers): if h.lower() in ("keyword", "query", "search term", "target keyword", "phrase"): kw_idx = idx break if len(date_cols) >= 3: for row in reader: if len(row) <= kw_idx: continue kw = row[kw_idx].strip() if not kw: continue for col_idx, d_str in date_cols.items(): if col_idx < len(row): series[kw][d_str] = parse_rank(row[col_idx]) else: date_idx, rank_idx = None, None for idx, h in enumerate(headers): hl = h.lower() if date_idx is None and any(k in hl for k in ("date", "day", "time", "week")): date_idx = idx if rank_idx is None and any(k in hl for k in ("pos", "rank", "avg")): rank_idx = idx if date_idx is None or rank_idx is None: raise ValueError(f"Could not auto-detect Date/Position columns in: {headers}") for row in reader: if max(kw_idx, date_idx, rank_idx) >= len(row): continue kw = row[kw_idx].strip() d_str = normalize_date(row[date_idx]) if kw and d_str: series[kw][d_str] = parse_rank(row[rank_idx]) return { kw: sorted(dmap.items(), key=lambda x: x[0]) for kw, dmap in series.items() if len(dmap) >= 3 } def audit_series(sorted_series): anomalies = [] date_drop_counts = defaultdict(int) date_total_counts = defaultdict(int) for kw, timeline in sorted_series.items(): n = len(timeline) for d_str, _ in timeline: date_total_counts[d_str] += 1 i = 1 while i < n - 1: prev_date, prev_rank = timeline[i - 1] curr_date, curr_rank = timeline[i] if prev_rank <= 30 and (curr_rank - prev_rank) >= 12: recovered = False for lookahead in range(1, min(5, n - i)): next_date, next_rank = timeline[i + lookahead] if abs(next_rank - prev_rank) <= 5 or (prev_rank <= 10 and next_rank <= 12): anomalies.append({ "keyword": kw, "pre_date": prev_date, "pre_rank": prev_rank, "poison_date": curr_date, "poison_rank": "Vanished (>100)" if curr_rank > 100 else curr_rank, "raw_drop": curr_rank - prev_rank, "recovery_date": next_date, "recovery_rank": next_rank, "duration_points": lookahead, "vanished_completely": curr_rank > 100, "matches_nozzle_peak": curr_date in KNOWN_PEAK_DATES }) date_drop_counts[curr_date] += 1 i += lookahead recovered = True break if recovered: continue i += 1 daily_summary = [] for d_str in sorted(date_total_counts.keys()): total = date_total_counts[d_str] drops = date_drop_counts.get(d_str, 0) pct = round((drops / total) * 100, 1) if total > 0 else 0.0 daily_summary.append({ "date": d_str, "total_tracked": total, "ghost_drops": drops, "poison_rate_pct": pct, "known_google_event": d_str in KNOWN_PEAK_DATES }) total_kws = len(sorted_series) affected_kws = len(set(a["keyword"] for a in anomalies)) max_daily_rate = max((d["poison_rate_pct"] for d in daily_summary), default=0) vanished_ratio = ( sum(1 for a in anomalies if a["vanished_completely"]) / len(anomalies) if anomalies else 0 ) contamination_score = min(100, round( (affected_kws / max(total_kws, 1)) * 45 + (max_daily_rate * 0.40) + (vanished_ratio * 15) )) return { "total_keywords_audited": total_kws, "poisoned_keywords_detected": affected_kws, "affected_keyword_pct": round((affected_kws / max(total_kws, 1)) * 100, 1), "total_ghost_bounce_events": len(anomalies), "peak_daily_poison_rate_pct": max_daily_rate, "telemetry_contamination_score": contamination_score, "daily_breakdown": [d for d in daily_summary if d["ghost_drops"] > 0], "anomalies": sorted(anomalies, key=lambda x: x["raw_drop"], reverse=True) } def generate_demo_csv(filename="demo_semrush_oct2026.csv"): dates = [ "2026-09-25", "2026-09-28", "2026-09-30", "2026-10-02", "2026-10-07", "2026-10-12", "2026-10-18", "2026-10-21", "2026-10-25" ] rows = [ ["Keyword"] + dates, ["ai compliance audit agency", 3, 3, 101, 3, 4, 3, 3, 101, 3], ["enterprise geo consulting", 5, 4, 101, 5, 5, 4, 68, 74, 4], ["headless wordpress nextjs setup", 2, 2, 54, 2, 2, 2, 2, 101, 2], ["agentic engineering services", 1, 1, 1, 1, 88, 1, 1, 1, 1], ["b2b technical seo audit", 6, 6, 101, 6, 7, 6, 101, 101, 6], ["schema grounding architecture", 4, 4, 101, 4, 4, 4, 4, 49, 4], ["custom mcp server development", 8, 7, 7, 8, 7, 7, 8, 7, 7], ["local ai directory architecture", 3, 3, 101, 3, 92, 3, 3, 101, 3], ] with open(filename, "w", newline="", encoding="utf-8") as f: csv.writer(f).writerows(rows) return filename if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("csv_file", nargs="?") parser.add_argument("--demo", action="store_true") parser.add_argument("--json") args = parser.parse_args() target = generate_demo_csv() if (args.demo or not args.csv_file) else args.csv_file report = audit_series(load_csv_series(target)) print(json.dumps(report, indent=2)) if args.json: with open(args.json, "w", encoding="utf-8") as jf: json.dump(report, jf, indent=2)