app.py

# =============================================================================
#  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()