Spaces:
Runtime error
Runtime error
| # ============================================================================= | |
| # Fraud Log Publisher Resolver — Hugging Face Spaces (Gradio) App | |
| # ============================================================================= | |
| # Drop your Excel file → get fraud_resolved_*.xlsx + publisher_files_*.zip | |
| # ============================================================================= | |
| import gradio as gr | |
| import pandas as pd | |
| import re, os, zipfile, time, math, tempfile, shutil | |
| import xlsxwriter | |
| # ── Colour constants ────────────────────────────────────────────────────────── | |
| C_HDR="1F4E79"; C_HF="FFFFFF" | |
| C_PRI="E2EFDA"; C_FB="FCE4D6"; C_UN="FFE0E0" | |
| C_SUM="2E75B6"; C_ZEB="EBF3FB"; C_TOT="D6E4F0"; C_REMOVED="F2F2F2" | |
| # ── Helpers ─────────────────────────────────────────────────────────────────── | |
| def parse_cols(v): | |
| return [c.strip().strip('"').strip("'") for c in str(v).split(',') if c.strip()] | |
| def extract_pub_id(val): | |
| if val is None or str(val).strip() in ('', 'nan', 'None'): | |
| return None | |
| try: | |
| return int(float(str(val).strip())) | |
| except Exception: | |
| m = re.match(r'^(\d+)', str(val).strip()) | |
| return int(m.group(1)) if m else None | |
| def safe_val(v): | |
| if v is None: | |
| return '' | |
| if isinstance(v, float) and math.isnan(v): | |
| return '' | |
| return v | |
| def resolve(df, pid_col, fl_cols, pub_map, ms_map): | |
| names = pd.Series('', index=df.index) | |
| src = pd.Series('unresolved', index=df.index) | |
| mask = pd.Series(True, index=df.index) | |
| if pid_col in df.columns: | |
| pri = df[pid_col].apply(extract_pub_id).map(pub_map) | |
| ok = pri.notna() | |
| names[ok] = pri[ok]; src[ok] = 'primary'; mask &= ~ok | |
| for col in fl_cols: | |
| if col in df.columns and mask.any(): | |
| fb = df.loc[mask, col].astype(str).str.strip().str.lower().map(ms_map) | |
| ok2 = fb.notna() | |
| names.loc[ok2[ok2].index] = fb[ok2] | |
| src.loc[ok2[ok2].index] = 'fallback' | |
| mask.loc[ok2[ok2].index] = False | |
| if mask.any(): | |
| raw = df.get(pid_col, pd.Series('', index=df.index)) | |
| names[mask] = 'UNRESOLVED (id=' + raw[mask].astype(str) + ')' | |
| return names, src | |
| def sort_and_dedup(df, sort_cols, dedup_cols, tab_name, log): | |
| original_len = len(df) | |
| working = df.copy() | |
| valid_sort = [c for c in sort_cols if c in working.columns] | |
| valid_dedup = [c for c in dedup_cols if c in working.columns] | |
| for c in sort_cols: | |
| if c not in working.columns: | |
| log.append(f" ⚠️ {tab_name}: sort col not found: '{c}'") | |
| for c in dedup_cols: | |
| if c not in working.columns: | |
| log.append(f" ⚠️ {tab_name}: dedup col not found: '{c}'") | |
| if valid_sort: | |
| sort_keys = [] | |
| for col in valid_sort: | |
| try: | |
| working[f'__s_{col}'] = pd.to_datetime(working[col], errors='coerce') | |
| sort_keys.append(f'__s_{col}') | |
| except Exception: | |
| sort_keys.append(col) | |
| working = working.sort_values(sort_keys, ascending=True, na_position='last') | |
| working = working.drop(columns=[k for k in sort_keys if k.startswith('__s_')], errors='ignore') | |
| log.append(f" ✓ {tab_name}: sorted ascending by {valid_sort}") | |
| removed_df = pd.DataFrame(columns=df.columns) | |
| if valid_dedup: | |
| dup_mask = working.duplicated(subset=valid_dedup, keep='first') | |
| removed_df = working[dup_mask].copy() | |
| working = working[~dup_mask].copy() | |
| log.append(f" ✓ {tab_name}: deduped on {valid_dedup} → " | |
| f"kept {len(working):,}, removed {len(removed_df):,}") | |
| return (working.reset_index(drop=True), removed_df.reset_index(drop=True), | |
| {'original': original_len, 'kept': len(working), 'removed': len(removed_df), | |
| 'sort_cols': valid_sort, 'dedup_cols': valid_dedup}) | |
| def build_summary(grp): | |
| gb = (grp.groupby(['publishername', 'Media Source', '_tab'], sort=True) | |
| .size().reset_index(name='Rows')) | |
| pivot = gb.pivot_table(index=['publishername', 'Media Source'], | |
| columns='_tab', values='Rows', aggfunc='sum', fill_value=0).reset_index() | |
| pivot.columns.name = None | |
| for c in ['pat', 'rtb']: | |
| if c not in pivot.columns: pivot[c] = 0 | |
| pivot = pivot[['publishername', 'Media Source', 'pat', 'rtb']] | |
| pivot.rename(columns={'pat': 'PAT Rows', 'rtb': 'RTB Rows'}, inplace=True) | |
| pivot['Total Rows'] = pivot['PAT Rows'] + pivot['RTB Rows'] | |
| totals = pd.DataFrame([{'publishername': '── TOTAL ──', 'Media Source': '', | |
| 'PAT Rows': pivot['PAT Rows'].sum(), | |
| 'RTB Rows': pivot['RTB Rows'].sum(), | |
| 'Total Rows': pivot['Total Rows'].sum()}]) | |
| return pd.concat([pivot, totals], ignore_index=True) | |
| # ── xlsxwriter writers ──────────────────────────────────────────────────────── | |
| def xw_write_sheet(wb, name, df, src_list=None, src_fmts=None, hdr_fmt=None, def_fmt=None): | |
| ws = wb.add_worksheet(name[:31]) | |
| ws.freeze_panes(1, 0); ws.set_column(0, len(df.columns)-1, 18) | |
| for ci, cn in enumerate(df.columns): ws.write(0, ci, cn, hdr_fmt) | |
| pub_ci = list(df.columns).index('publishername') if 'publishername' in df.columns else None | |
| for ri, row_vals in enumerate(df.itertuples(index=False), 1): | |
| for ci, val in enumerate(row_vals): | |
| fmt = (src_fmts.get(src_list[ri-1], src_fmts['unresolved']) | |
| if (src_list and pub_ci == ci) else def_fmt) | |
| ws.write(ri, ci, safe_val(val), fmt) | |
| def xw_write_tab_with_dedup(wb, sheet_name, kept_df, removed_df, stats, | |
| src_list, src_fmts, hdr_fmt, def_fmt, rmvd_fmt): | |
| ws = wb.add_worksheet(sheet_name[:31]) | |
| ws.freeze_panes(2, 0); ws.set_column(0, len(kept_df.columns)-1, 18) | |
| nc = len(kept_df.columns) | |
| stats_fmt = wb.add_format({'bold': True, 'italic': True, 'font_name': 'Arial', 'font_size': 9, | |
| 'bg_color': '#D9E1F2', 'font_color': '#1F4E79'}) | |
| zeb = wb.add_format({'font_name': 'Arial', 'font_size': 9, 'bg_color': '#' + C_ZEB}) | |
| banner = (f"Sort: {stats['sort_cols']} | Dedup on: {stats['dedup_cols']} | " | |
| f"Original: {stats['original']:,} ✓ Kept: {stats['kept']:,} " | |
| f"✗ Removed: {stats['removed']:,}") | |
| ws.merge_range(0, 0, 0, min(nc-1, 15), banner, stats_fmt) | |
| for ci, cn in enumerate(kept_df.columns): ws.write(1, ci, cn, hdr_fmt) | |
| pub_ci = list(kept_df.columns).index('publishername') if 'publishername' in kept_df.columns else None | |
| row = 2 | |
| for ri, row_vals in enumerate(kept_df.itertuples(index=False)): | |
| for ci, val in enumerate(row_vals): | |
| if pub_ci == ci and src_list: | |
| fmt = src_fmts.get(src_list[ri], src_fmts['unresolved']) | |
| else: | |
| fmt = zeb if ri % 2 == 0 else def_fmt | |
| ws.write(row, ci, safe_val(val), fmt) | |
| row += 1 | |
| if len(removed_df) > 0: | |
| sep_fmt = wb.add_format({'bold': True, 'font_name': 'Arial', 'font_size': 10, | |
| 'bg_color': '#' + C_UN, 'font_color': '#78281F'}) | |
| ws.merge_range(row, 0, row, min(nc-1, 15), | |
| f"▼ REMOVED DUPLICATES ({stats['removed']:,} rows" | |
| f" — duplicate {stats['dedup_cols']})", sep_fmt) | |
| row += 1 | |
| for row_vals in removed_df.itertuples(index=False): | |
| for ci, val in enumerate(row_vals): ws.write(row, ci, safe_val(val), rmvd_fmt) | |
| row += 1 | |
| def xw_write_summary(wb, df, hdr_fmt, def_fmt): | |
| ws = wb.add_worksheet('Summary') | |
| ws.freeze_panes(1, 0); ws.set_column(0, 0, 24); ws.set_column(1, 1, 30); ws.set_column(2, 4, 14) | |
| for ci, cn in enumerate(df.columns): ws.write(0, ci, cn, hdr_fmt) | |
| zeb = wb.add_format({'font_name': 'Arial', 'font_size': 9, 'bg_color': '#' + C_ZEB}) | |
| tot = wb.add_format({'font_name': 'Arial', 'font_size': 9, 'bg_color': '#' + C_TOT, 'bold': True}) | |
| for ri, rv in enumerate(df.itertuples(index=False), 1): | |
| rf = tot if str(rv[0]).startswith('──') else (zeb if ri % 2 == 0 else def_fmt) | |
| for ci, val in enumerate(rv): ws.write(ri, ci, safe_val(val), rf) | |
| def xw_legend(wb, fmts): | |
| ws = wb.add_worksheet('Legend') | |
| ws.set_column(0, 0, 26); ws.set_column(1, 1, 54) | |
| for r, (a, b) in enumerate([ | |
| ('Colour', 'Meaning'), | |
| ('Green — Primary', 'Sub Param 4 → publisher tab lookup succeeded'), | |
| ('Orange — Fallback', 'Media Source → media source tab (fallback used)'), | |
| ('Red — Unresolved','Publisher could not be resolved — review manually') | |
| ]): | |
| ws.write(r, 0, a, fmts[r]); ws.write(r, 1, b, fmts[r]) | |
| # ── Core processing function ────────────────────────────────────────────────── | |
| def process_file(input_file): | |
| if input_file is None: | |
| return None, None, "❌ No file uploaded." | |
| log = [] | |
| t0 = time.time() | |
| def elapsed(): return f"[{time.time()-t0:.0f}s]" | |
| # Create a temp working directory | |
| work_dir = tempfile.mkdtemp() | |
| INPUT_FILE = input_file.name | |
| INPUT_STEM = os.path.splitext(os.path.basename(INPUT_FILE))[0] | |
| OUTPUT_MAIN = os.path.join(work_dir, f"fraud_resolved_{INPUT_STEM}.xlsx") | |
| OUTPUT_ZIP = os.path.join(work_dir, f"publisher_files_{INPUT_STEM}.zip") | |
| TMP_DIR = os.path.join(work_dir, "publisher_exports") | |
| os.makedirs(TMP_DIR, exist_ok=True) | |
| try: | |
| # ── Load ────────────────────────────────────────────────────────────── | |
| log.append(f"{elapsed()} 📊 Loading workbook…") | |
| try: | |
| all_sheets = pd.read_excel(INPUT_FILE, sheet_name=None, dtype=str, engine='calamine') | |
| except Exception: | |
| all_sheets = pd.read_excel(INPUT_FILE, sheet_name=None, dtype=str) | |
| missing = [t for t in ['input', 'pat', 'rtb', 'publisher', 'media source'] | |
| if t not in all_sheets] | |
| if missing: | |
| log.append(f"⚠️ Missing tabs: {missing}") | |
| config = {} | |
| for _, r in all_sheets.get('input', pd.DataFrame()).iterrows(): | |
| n = str(r.get('Name', '')).strip() | |
| v = str(r.get('Value', '')).strip().strip('"').strip("'") | |
| if n and n != 'nan': config[n] = v | |
| pid_col = config.get('column to fetch publisher id from fraud logs- both rtb and pat', 'Sub Param 4') | |
| fl_cols = parse_cols(config.get('fallback column in fraud logs to look for publisher name. If multiple then comma separated', 'Media Source')) | |
| ms_cols = parse_cols(config.get('fallback column in media source TAB which should be used in case , publisher name cannot be resolved in fraud logs.. If multiple then comma separated', 'Media Source')) | |
| sort_cols = parse_cols(config.get('Sort by column for fraud logs, if multiple then comma separated', '')) | |
| dedup_cols= parse_cols(config.get('Remove Duplicate basis this column name from fraud logs , if multiple then comma separated', '')) | |
| log.append(f" sort_cols={sort_cols} | dedup_cols={dedup_cols} | pid_col={pid_col}") | |
| pub_map = {} | |
| for _, r in all_sheets.get('publisher', pd.DataFrame()).iterrows(): | |
| pid = extract_pub_id(r.get('id')); pn = str(r.get('name', '')).strip() | |
| if pid and pn and pn != 'nan': pub_map[pid] = pn | |
| ms_map = {} | |
| for _, r in all_sheets.get('media source', pd.DataFrame()).iterrows(): | |
| for col in ms_cols: | |
| k = str(r.get(col, '')).strip().lower() | |
| pv = str(r.get('Publisher', '')).strip() | |
| if k and k != 'nan' and pv and pv != 'nan': ms_map[k] = pv | |
| log.append(f" publisher_map={len(pub_map)} | ms_map={len(ms_map)}") | |
| if not ms_map: | |
| log.append(" ℹ️ media source tab is empty — fallback will not resolve any rows") | |
| # ── Sort → Dedup → Resolve ──────────────────────────────────────────── | |
| log.append(f"\n{elapsed()} 🔄 Sort → Dedup → Resolve…") | |
| resolved = {}; srcs = {}; removed = {}; dedup_stats = {} | |
| for tab in ['pat', 'rtb']: | |
| if tab not in all_sheets: | |
| log.append(f" ⚠️ '{tab}' tab missing"); continue | |
| df = all_sheets[tab].copy() | |
| log.append(f"\n [{tab.upper()}] {len(df):,} rows") | |
| df_clean, df_removed, stats = sort_and_dedup(df, sort_cols, dedup_cols, tab.upper(), log) | |
| removed[tab] = df_removed; dedup_stats[tab] = stats | |
| df_clean['publishername'], s = resolve(df_clean, pid_col, fl_cols, pub_map, ms_map) | |
| resolved[tab] = df_clean; srcs[tab] = s | |
| log.append(f" ✓ publisher: 🟢primary={(s=='primary').sum():,} " | |
| f"🟠fallback={(s=='fallback').sum():,} " | |
| f"🔴unresolved={(s=='unresolved').sum():,}") | |
| frames = [] | |
| for tab in ['pat', 'rtb']: | |
| if tab in resolved: | |
| d = resolved[tab].copy(); d.insert(0, '_tab', tab) | |
| d['_src'] = srcs[tab].values; frames.append(d) | |
| pub_groups = {} | |
| if frames: | |
| comb = pd.concat(frames, ignore_index=True) | |
| for pn, grp in comb.groupby('publishername', sort=True): | |
| pub_groups[pn] = grp | |
| log.append(f"\n Publishers ({len(pub_groups)}):") | |
| for p, g in sorted(pub_groups.items()): | |
| log.append(f" • {p:<32} {len(g):>6,} " | |
| f"(PAT={(g['_tab']=='pat').sum():,} RTB={(g['_tab']=='rtb').sum():,})") | |
| # ── Main Excel ──────────────────────────────────────────────────────── | |
| log.append(f"\n{elapsed()} 📝 Writing main Excel…") | |
| wb_m = xlsxwriter.Workbook(OUTPUT_MAIN, {'constant_memory': True, 'strings_to_urls': False}) | |
| hdr_m = wb_m.add_format({'bold': True, 'font_name': 'Arial', 'font_size': 10, | |
| 'bg_color': '#'+C_HDR, 'font_color': '#'+C_HF, | |
| 'align': 'center', 'valign': 'vcenter', 'text_wrap': True}) | |
| def_m = wb_m.add_format({'font_name': 'Arial', 'font_size': 9}) | |
| srf_m = {k: wb_m.add_format({'font_name': 'Arial', 'font_size': 9, 'bg_color': '#'+bg}) | |
| for k, bg in [('primary', C_PRI), ('fallback', C_FB), ('unresolved', C_UN)]} | |
| rmvd_fmt = wb_m.add_format({'font_name': 'Arial', 'font_size': 9, | |
| 'bg_color': '#'+C_REMOVED, 'font_color': '#888888', 'italic': True}) | |
| leg_fmts = [wb_m.add_format({'bold': True, 'font_name': 'Arial', 'font_size': 10, | |
| 'bg_color': '#'+bg, 'font_color': '#'+fg}) | |
| for bg, fg in [(C_HDR, C_HF), (C_PRI, '000000'), (C_FB, '000000'), (C_UN, '000000')]] | |
| for tab in ['input', 'media source', 'publisher']: | |
| df = all_sheets.get(tab) | |
| if df is not None: | |
| xw_write_sheet(wb_m, tab, df, hdr_fmt=hdr_m, def_fmt=def_m) | |
| log.append(f" ✓ '{tab}'") | |
| for tab in ['pat', 'rtb']: | |
| if tab in resolved: | |
| xw_write_tab_with_dedup(wb_m, tab, | |
| kept_df=resolved[tab], removed_df=removed[tab], stats=dedup_stats[tab], | |
| src_list=srcs[tab].tolist(), src_fmts=srf_m, | |
| hdr_fmt=hdr_m, def_fmt=def_m, rmvd_fmt=rmvd_fmt) | |
| s = dedup_stats[tab] | |
| log.append(f" ✓ '{tab}' (kept={s['kept']:,} + removed={s['removed']:,})") | |
| xw_legend(wb_m, leg_fmts) | |
| wb_m.close() | |
| log.append(f" ✅ fraud_resolved_{INPUT_STEM}.xlsx " | |
| f"({os.path.getsize(OUTPUT_MAIN)/1024/1024:.1f} MB)") | |
| # ── Publisher ZIP ───────────────────────────────────────────────────── | |
| log.append(f"\n{elapsed()} 📦 Building publisher ZIP…") | |
| pub_files = [] | |
| for pn, grp in sorted(pub_groups.items()): | |
| safe = re.sub(r'[\\/*?:\[\]]', '_', pn) | |
| fname = f"publisher_{safe}_{INPUT_STEM}.xlsx" | |
| fpath = os.path.join(TMP_DIR, fname) | |
| pat_s = grp[grp['_tab']=='pat'].drop(columns=['_tab','_src']).reset_index(drop=True) | |
| rtb_s = grp[grp['_tab']=='rtb'].drop(columns=['_tab','_src']).reset_index(drop=True) | |
| pat_src = grp.loc[grp['_tab']=='pat', '_src'].tolist() | |
| rtb_src = grp.loc[grp['_tab']=='rtb', '_src'].tolist() | |
| summary = build_summary(grp) | |
| wb_p = xlsxwriter.Workbook(fpath, {'constant_memory': True, 'strings_to_urls': False}) | |
| hdr_s = wb_p.add_format({'bold': True, 'font_name': 'Arial', 'font_size': 10, | |
| 'bg_color': '#'+C_SUM, 'font_color': '#'+C_HF, | |
| 'align': 'center', 'valign': 'vcenter', 'text_wrap': True}) | |
| hdr_d = wb_p.add_format({'bold': True, 'font_name': 'Arial', 'font_size': 10, | |
| 'bg_color': '#'+C_HDR, 'font_color': '#'+C_HF, | |
| 'align': 'center', 'valign': 'vcenter', 'text_wrap': True}) | |
| def_p = wb_p.add_format({'font_name': 'Arial', 'font_size': 9}) | |
| srf_p = {k: wb_p.add_format({'font_name': 'Arial', 'font_size': 9, 'bg_color': '#'+bg}) | |
| for k, bg in [('primary', C_PRI), ('fallback', C_FB), ('unresolved', C_UN)]} | |
| def move_pub_last(df, src): | |
| if 'publishername' not in df.columns: return df, src | |
| cols = [c for c in df.columns if c != 'publishername'] + ['publishername'] | |
| return df[cols], src | |
| pat_s, pat_src = move_pub_last(pat_s, pat_src) | |
| rtb_s, rtb_src = move_pub_last(rtb_s, rtb_src) | |
| xw_write_summary(wb_p, summary, hdr_s, def_p) | |
| if len(pat_s): xw_write_sheet(wb_p, 'pat', pat_s, src_list=pat_src, src_fmts=srf_p, hdr_fmt=hdr_d, def_fmt=def_p) | |
| if len(rtb_s): xw_write_sheet(wb_p, 'rtb', rtb_s, src_list=rtb_src, src_fmts=srf_p, hdr_fmt=hdr_d, def_fmt=def_p) | |
| wb_p.close() | |
| pub_files.append(fpath) | |
| log.append(f" • {fname} ({os.path.getsize(fpath)//1024:,} KB)") | |
| with zipfile.ZipFile(OUTPUT_ZIP, 'w', zipfile.ZIP_DEFLATED) as zf: | |
| for fp in pub_files: zf.write(fp, os.path.basename(fp)) | |
| log.append(f" ✅ publisher_files_{INPUT_STEM}.zip " | |
| f"({os.path.getsize(OUTPUT_ZIP)/1024/1024:.1f} MB)") | |
| log.append(f"\n{elapsed()} 🎉 Done!") | |
| return OUTPUT_MAIN, OUTPUT_ZIP, "\n".join(log) | |
| except Exception as e: | |
| import traceback | |
| log.append(f"\n❌ ERROR: {e}") | |
| log.append(traceback.format_exc()) | |
| return None, None, "\n".join(log) | |
| # ── Gradio UI ───────────────────────────────────────────────────────────────── | |
| with gr.Blocks(title="Fraud Log Publisher Resolver", theme=gr.themes.Soft()) as demo: | |
| gr.Markdown(""" | |
| # 🔍 Fraud Log Publisher Resolver | |
| Upload your Excel file to sort, deduplicate, and resolve publisher names across **PAT** and **RTB** fraud log tabs. | |
| ### Expected Excel tabs | |
| | Tab | Purpose | | |
| |-----|---------| | |
| | `input` | Configuration (Name / Value pairs) | | |
| | `pat` | PAT fraud log rows | | |
| | `rtb` | RTB fraud log rows | | |
| | `publisher` | Publisher ID → Name mapping | | |
| | `media source` | Media source fallback mapping | | |
| ### Colour key in output | |
| 🟢 **Green** — resolved via Sub Param 4 → publisher tab | |
| 🟠 **Orange** — resolved via Media Source fallback | |
| 🔴 **Red** — unresolved | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| file_input = gr.File( | |
| label="📂 Upload Excel File (.xlsx)", | |
| file_types=[".xlsx", ".xls"], | |
| type="filepath" | |
| ) | |
| run_btn = gr.Button("▶️ Process", variant="primary", size="lg") | |
| with gr.Column(scale=1): | |
| out_main = gr.File(label="📄 Download: fraud_resolved_*.xlsx") | |
| out_zip = gr.File(label="📦 Download: publisher_files_*.zip") | |
| log_box = gr.Textbox( | |
| label="📋 Processing Log", | |
| lines=20, | |
| max_lines=40, | |
| interactive=False, | |
| placeholder="Processing log will appear here after you click ▶️ Process…" | |
| ) | |
| # Wire up | |
| class FileWrapper: | |
| def __init__(self, path): self.name = path | |
| def run(filepath): | |
| if filepath is None: | |
| return None, None, "❌ Please upload a file first." | |
| return process_file(FileWrapper(filepath)) | |
| run_btn.click(fn=run, inputs=[file_input], outputs=[out_main, out_zip, log_box]) | |
| gr.Markdown(""" | |
| --- | |
| *Built with [Gradio](https://gradio.app) · Deploy on [Hugging Face Spaces](https://huggingface.co/spaces)* | |
| """) | |
| if __name__ == "__main__": | |
| demo.launch() |