74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""
|
|
comparator.py — Compares two Account Profiles CSVs and returns new entries.
|
|
|
|
Logic
|
|
─────
|
|
• Unique key : "Account Domain" (the website domain, e.g. dell.com)
|
|
• New entry : a domain present in the NEW csv that is absent in the OLD csv.
|
|
• If OLD csv is missing (first ever run) → every row counts as "new".
|
|
"""
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Column used to identify a unique account across exports
|
|
_UNIQUE_KEY = "Account Domain"
|
|
|
|
|
|
def _load(path: str) -> pd.DataFrame:
|
|
return pd.read_csv(path, dtype=str).fillna("-").rename(
|
|
columns=lambda c: c.strip()
|
|
)
|
|
|
|
|
|
def find_new_entries(
|
|
new_csv: str,
|
|
old_csv: str | None,
|
|
) -> tuple[pd.DataFrame, bool]:
|
|
"""
|
|
Parameters
|
|
----------
|
|
new_csv : path to the freshly downloaded CSV.
|
|
old_csv : path to the previously stored CSV, or None.
|
|
|
|
Returns
|
|
-------
|
|
(new_entries_df, is_first_run)
|
|
new_entries_df — DataFrame of rows that are brand-new.
|
|
is_first_run — True when there was no previous CSV to compare against.
|
|
"""
|
|
new_df = _load(new_csv)
|
|
|
|
if _UNIQUE_KEY not in new_df.columns:
|
|
# The downloaded CSV might have a slightly different column order;
|
|
# treat the first column as the domain key.
|
|
logger.warning(
|
|
f"Column '{_UNIQUE_KEY}' not found. "
|
|
f"Available columns: {list(new_df.columns)}"
|
|
)
|
|
first_col = new_df.columns[0]
|
|
new_df.rename(columns={first_col: _UNIQUE_KEY}, inplace=True)
|
|
logger.warning(f"Using '{first_col}' as the unique key instead.")
|
|
|
|
if old_csv is None or not Path(old_csv).exists():
|
|
logger.info("No previous CSV found — treating ALL entries as new (first run).")
|
|
return new_df, True
|
|
|
|
old_df = _load(old_csv)
|
|
old_keys = set(
|
|
old_df[_UNIQUE_KEY].str.strip().str.lower()
|
|
if _UNIQUE_KEY in old_df.columns
|
|
else old_df.iloc[:, 0].str.strip().str.lower()
|
|
)
|
|
mask = ~new_df[_UNIQUE_KEY].str.strip().str.lower().isin(old_keys)
|
|
new_entries = new_df[mask].copy().reset_index(drop=True)
|
|
|
|
logger.info(
|
|
f"Comparison complete: {len(new_entries)} new / "
|
|
f"{len(new_df)} total (previous had {len(old_df)} rows)"
|
|
)
|
|
return new_entries, False
|