262 lines
8.4 KiB
Python
262 lines
8.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Deduplicate ODS snapshots by (business PK, content_hash).
|
|
Keep the latest row by fetched_at (tie-breaker: ctid desc).
|
|
|
|
Usage:
|
|
PYTHONPATH=. python -m scripts.repair.dedupe_ods_snapshots
|
|
PYTHONPATH=. python -m scripts.repair.dedupe_ods_snapshots --schema billiards_ods
|
|
PYTHONPATH=. python -m scripts.repair.dedupe_ods_snapshots --tables member_profiles,orders
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Iterable, Sequence
|
|
|
|
import psycopg2
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from config.settings import AppConfig
|
|
from database.connection import DatabaseConnection
|
|
|
|
|
|
def _reconfigure_stdout_utf8() -> None:
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _quote_ident(name: str) -> str:
|
|
return '"' + str(name).replace('"', '""') + '"'
|
|
|
|
|
|
def _fetch_tables(conn, schema: str) -> list[str]:
|
|
sql = """
|
|
SELECT table_name
|
|
FROM information_schema.tables
|
|
WHERE table_schema = %s AND table_type = 'BASE TABLE'
|
|
ORDER BY table_name
|
|
"""
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (schema,))
|
|
return [r[0] for r in cur.fetchall()]
|
|
|
|
|
|
def _fetch_columns(conn, schema: str, table: str) -> list[str]:
|
|
sql = """
|
|
SELECT column_name
|
|
FROM information_schema.columns
|
|
WHERE table_schema = %s AND table_name = %s
|
|
ORDER BY ordinal_position
|
|
"""
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (schema, table))
|
|
return [r[0] for r in cur.fetchall()]
|
|
|
|
|
|
def _fetch_pk_columns(conn, schema: str, table: str) -> list[str]:
|
|
sql = """
|
|
SELECT kcu.column_name
|
|
FROM information_schema.table_constraints tc
|
|
JOIN information_schema.key_column_usage kcu
|
|
ON tc.constraint_name = kcu.constraint_name
|
|
AND tc.table_schema = kcu.table_schema
|
|
WHERE tc.constraint_type = 'PRIMARY KEY'
|
|
AND tc.table_schema = %s
|
|
AND tc.table_name = %s
|
|
ORDER BY kcu.ordinal_position
|
|
"""
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (schema, table))
|
|
cols = [r[0] for r in cur.fetchall()]
|
|
return [c for c in cols if c.lower() != "content_hash"]
|
|
|
|
|
|
def _build_report_path(out_arg: str | None) -> Path:
|
|
if out_arg:
|
|
return Path(out_arg)
|
|
reports_dir = PROJECT_ROOT / "reports"
|
|
reports_dir.mkdir(parents=True, exist_ok=True)
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
return reports_dir / f"ods_snapshot_dedupe_{ts}.json"
|
|
|
|
|
|
def _print_progress(
|
|
table_label: str,
|
|
deleted: int,
|
|
total: int,
|
|
errors: int,
|
|
) -> None:
|
|
if total:
|
|
msg = f"[{table_label}] deleted {deleted}/{total} errors={errors}"
|
|
else:
|
|
msg = f"[{table_label}] deleted {deleted} errors={errors}"
|
|
print(msg, flush=True)
|
|
|
|
|
|
def _count_duplicates(conn, schema: str, table: str, key_cols: Sequence[str]) -> int:
|
|
keys_sql = ", ".join(_quote_ident(c) for c in [*key_cols, "content_hash"])
|
|
table_sql = f"{_quote_ident(schema)}.{_quote_ident(table)}"
|
|
sql = f"""
|
|
SELECT COUNT(*) FROM (
|
|
SELECT 1
|
|
FROM (
|
|
SELECT ROW_NUMBER() OVER (
|
|
PARTITION BY {keys_sql}
|
|
ORDER BY fetched_at DESC NULLS LAST, ctid DESC
|
|
) AS rn
|
|
FROM {table_sql}
|
|
) t
|
|
WHERE rn > 1
|
|
) s
|
|
"""
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql)
|
|
row = cur.fetchone()
|
|
return int(row[0] if row else 0)
|
|
|
|
|
|
def _delete_duplicate_batch(
|
|
conn,
|
|
schema: str,
|
|
table: str,
|
|
key_cols: Sequence[str],
|
|
batch_size: int,
|
|
) -> int:
|
|
keys_sql = ", ".join(_quote_ident(c) for c in [*key_cols, "content_hash"])
|
|
table_sql = f"{_quote_ident(schema)}.{_quote_ident(table)}"
|
|
sql = f"""
|
|
WITH dupes AS (
|
|
SELECT ctid
|
|
FROM (
|
|
SELECT ctid,
|
|
ROW_NUMBER() OVER (
|
|
PARTITION BY {keys_sql}
|
|
ORDER BY fetched_at DESC NULLS LAST, ctid DESC
|
|
) AS rn
|
|
FROM {table_sql}
|
|
) s
|
|
WHERE rn > 1
|
|
LIMIT %s
|
|
)
|
|
DELETE FROM {table_sql} t
|
|
USING dupes d
|
|
WHERE t.ctid = d.ctid
|
|
RETURNING 1
|
|
"""
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (int(batch_size),))
|
|
rows = cur.fetchall()
|
|
return len(rows or [])
|
|
|
|
|
|
def main() -> int:
|
|
_reconfigure_stdout_utf8()
|
|
ap = argparse.ArgumentParser(description="Deduplicate ODS snapshot rows by PK+content_hash")
|
|
ap.add_argument("--schema", default="billiards_ods", help="ODS schema name")
|
|
ap.add_argument("--tables", default="", help="comma-separated table names (optional)")
|
|
ap.add_argument("--batch-size", type=int, default=1000, help="delete batch size")
|
|
ap.add_argument("--progress-every", type=int, default=100, help="print progress every N deletions")
|
|
ap.add_argument("--out", default="", help="output report JSON path")
|
|
ap.add_argument("--dry-run", action="store_true", help="only compute duplicate counts")
|
|
args = ap.parse_args()
|
|
|
|
cfg = AppConfig.load({})
|
|
db = DatabaseConnection(dsn=cfg["db"]["dsn"], session=cfg["db"].get("session"))
|
|
try:
|
|
db.conn.rollback()
|
|
except Exception:
|
|
pass
|
|
db.conn.autocommit = True
|
|
|
|
tables = _fetch_tables(db.conn, args.schema)
|
|
if args.tables.strip():
|
|
whitelist = {t.strip() for t in args.tables.split(",") if t.strip()}
|
|
tables = [t for t in tables if t in whitelist]
|
|
|
|
report = {
|
|
"schema": args.schema,
|
|
"tables": [],
|
|
"summary": {
|
|
"total_tables": len(tables),
|
|
"checked_tables": 0,
|
|
"total_duplicates": 0,
|
|
"deleted_rows": 0,
|
|
"error_rows": 0,
|
|
"skipped_tables": 0,
|
|
},
|
|
}
|
|
|
|
for table in tables:
|
|
table_label = f"{args.schema}.{table}"
|
|
cols = _fetch_columns(db.conn, args.schema, table)
|
|
cols_lower = {c.lower() for c in cols}
|
|
if "content_hash" not in cols_lower or "fetched_at" not in cols_lower:
|
|
print(f"[{table_label}] skip: missing content_hash/fetched_at", flush=True)
|
|
report["summary"]["skipped_tables"] += 1
|
|
continue
|
|
|
|
key_cols = _fetch_pk_columns(db.conn, args.schema, table)
|
|
if not key_cols:
|
|
print(f"[{table_label}] skip: missing primary key", flush=True)
|
|
report["summary"]["skipped_tables"] += 1
|
|
continue
|
|
|
|
total_dupes = _count_duplicates(db.conn, args.schema, table, key_cols)
|
|
print(f"[{table_label}] duplicates={total_dupes}", flush=True)
|
|
deleted = 0
|
|
errors = 0
|
|
|
|
if not args.dry_run and total_dupes:
|
|
while True:
|
|
try:
|
|
batch_deleted = _delete_duplicate_batch(
|
|
db.conn,
|
|
args.schema,
|
|
table,
|
|
key_cols,
|
|
args.batch_size,
|
|
)
|
|
except psycopg2.Error:
|
|
errors += 1
|
|
break
|
|
if batch_deleted <= 0:
|
|
break
|
|
deleted += batch_deleted
|
|
if args.progress_every and deleted % int(args.progress_every) == 0:
|
|
_print_progress(table_label, deleted, total_dupes, errors)
|
|
|
|
if deleted and (not args.progress_every or deleted % int(args.progress_every) != 0):
|
|
_print_progress(table_label, deleted, total_dupes, errors)
|
|
|
|
report["tables"].append(
|
|
{
|
|
"table": table_label,
|
|
"duplicate_rows": total_dupes,
|
|
"deleted_rows": deleted,
|
|
"error_rows": errors,
|
|
}
|
|
)
|
|
report["summary"]["checked_tables"] += 1
|
|
report["summary"]["total_duplicates"] += total_dupes
|
|
report["summary"]["deleted_rows"] += deleted
|
|
report["summary"]["error_rows"] += errors
|
|
|
|
out_path = _build_report_path(args.out)
|
|
out_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"[REPORT] {out_path}", flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|