import os import time import requests import json import pandas as pd import math import re import gradio as gr import plotly.express as px from plotly.subplots import make_subplots import plotly.graph_objects as go from datetime import date, timedelta from huggingface_hub import HfApi # ===================================================== # CONFIG / SECRETS # ===================================================== API_TOKEN = os.getenv("LEADFEEDER_API_TOKEN") APP_PASSWORD = os.getenv("APP_PASSWORD") HF_TOKEN = os.getenv("HF_TOKEN") SPACE_ID = os.getenv("SPACE_ID") if not API_TOKEN or not APP_PASSWORD: print("⚠️ WARNING: Secrets missing. App will launch but API calls will fail.") API_TOKEN = "PLACEHOLDER" APP_PASSWORD = "password" ACCOUNT_ID = "255333" BASE_URL = "https://api.leadfeeder.com" PAGE_SIZE = 100 HEADERS = { "Authorization": f"Token token={API_TOKEN}", "Accept": "application/json" } CAMPAIGN_CONFIG_FILE = "campaign_rules.json" # ===================================================== # 1. CAMPAIGN MANAGER & LOGIC # ===================================================== def load_campaign_rules(): if os.path.exists(CAMPAIGN_CONFIG_FILE): try: with open(CAMPAIGN_CONFIG_FILE, "r") as f: rules = json.load(f) for campaign, config in rules.items(): if isinstance(config, list): rules[campaign] = { "include": config, "exclude": [] } return rules except Exception as e: print(f"Error loading {CAMPAIGN_CONFIG_FILE}: {e}") return {} return {} def categorize_quality(score): if pd.isna(score): return "Unknown" try: s = int(score) if 8 <= s <= 10: return "High Quality (8-10)" if 5 <= s <= 7: return "Mid Quality (5-7)" if 1 <= s <= 4: return "Low Quality (1-4)" return "Low Quality (0)" except: return "Unknown" def get_campaign_match(text, rules): if not text or not isinstance(text, str): return None, None # Clean text: replace all URL symbols with spaces and pad edges clean_text = " " + re.sub(r'[\-_/.,?=&+#]', ' ', text.lower()) + " " for campaign_name, config in rules.items(): includes = config.get("include", []) excludes = config.get("exclude", []) # 1st: Check Exclusions (Exact Match) has_exclusion = False for ex in excludes: if not ex.strip(): continue kw = " " + re.sub(r'[\-_/.,?=&+#]', ' ', ex.lower().strip()) + " " if kw in clean_text: has_exclusion = True break if has_exclusion: continue # 2nd: Check Inclusions (Exact Match) for inc in includes: if not inc.strip(): continue kw = " " + re.sub(r'[\-_/.,?=&+#]', ' ', inc.lower().strip()) + " " if kw in clean_text: return campaign_name, inc.strip() return None, None def apply_business_logic(df): if df is None or df.empty: return df # 1. Quality Group if "lead_quality_score" in df.columns: df["Quality_Group"] = df["lead_quality_score"].apply(categorize_quality) # 2. Campaign Logic (Strict - URLs only) rules = load_campaign_rules() def resolve_campaign(row): # Priority 1: Landing Page if row.get("landing_page_path"): camp, word = get_campaign_match(row["landing_page_path"], rules) if camp: return pd.Series([camp, f"Landing Page matched: '{word}'"]) # Priority 2: Exit Page if row.get("exit_page_path"): camp, word = get_campaign_match(row["exit_page_path"], rules) if camp: return pd.Series([camp, f"Exit Page matched: '{word}'"]) return pd.Series(["Uncategorized", "No Match"]) df[["Campaign", "Match_Reason"]] = df.apply(resolve_campaign, axis=1) return df # ===================================================== # 2. PRESETS (UPDATED WITH COUNTRY/CITY/INDUSTRY) # ===================================================== DASHBOARD_PRESETS = { # -- SPECIAL VIEWS -- "Key Campaigns Bifurcation (MS Tech, Fintech, Adtech)": ("SPECIAL_KEY_BIFURCATION", None, None), "All Campaigns Performance": ("SPECIAL_ALL_PERFORMANCE", None, None), # -- CAMPAIGN & QUALITY -- "Visits by Campaign": ("Campaign", "total_visits", "sum"), "Leads by Campaign": ("Campaign", "company_name", "count"), "Leads by Quality Group": ("Quality_Group", "company_name", "count"), # -- GEOGRAPHY (NEW) -- "Visits by Country": ("country", "total_visits", "sum"), "Leads by Country": ("country", "company_name", "count"), "Visits by City": ("city", "total_visits", "sum"), "Leads by City": ("city", "company_name", "count"), # -- INDUSTRY (NEW) -- "Visits by Industry": ("primary_industry", "total_visits", "sum"), "Leads by Industry": ("primary_industry", "company_name", "count"), # -- OTHERS -- "Top Accounts by Visits": ("company_name", "total_visits", "sum"), } TREND_PRESETS = { # -- CAMPAIGN & QUALITY -- "Visits Trend by Campaign": ("total_visits", "sum", "Campaign"), "Leads Trend by Campaign": ("company_name", "count", "Campaign"), "Visits Trend by Quality": ("total_visits", "sum", "Quality_Group"), # -- GEOGRAPHY (NEW) -- "Visits Trend by Country": ("total_visits", "sum", "country"), "Leads Trend by Country": ("company_name", "count", "country"), "Visits Trend by City": ("total_visits", "sum", "city"), "Leads Trend by City": ("company_name", "count", "city"), # -- INDUSTRY (NEW) -- "Visits Trend by Industry": ("total_visits", "sum", "primary_industry"), "Leads Trend by Industry": ("company_name", "count", "primary_industry"), # -- GENERAL -- "Total Visits Trend": ("total_visits", "sum", None), "Active Accounts Trend": ("company_name", "count", None), } # ===================================================== # 3. API HANDLING # ===================================================== def make_request(url, params=None): retries = 3 while retries > 0: r = requests.get(url, headers=HEADERS, params=params, timeout=45) if r.status_code == 429: time.sleep(61) retries -= 1 continue r.raise_for_status() return r.json() raise Exception("Max retries exceeded") def fetch_basic_leads(start_date, end_date): page = 1 rows = [] print(f"🚀 Fetching full company list for {start_date} to {end_date}...") while True: try: js = make_request( f"{BASE_URL}/accounts/{ACCOUNT_ID}/leads", params={"start_date": start_date, "end_date": end_date, "page[number]": page, "page[size]": PAGE_SIZE, "include": "location"} ) data = js.get("data", []) if not data: break included = js.get("included", []) loc_map = {str(i["id"]): i["attributes"] for i in included if i["type"] == "locations"} for lead in data: a = lead["attributes"] loc_id = lead.get("relationships", {}).get("location", {}).get("data", {}).get("id") loc = loc_map.get(str(loc_id), {}) rows.append({ "lead_id": lead.get("id"), "company_name": a.get("name"), "website_url": a.get("website_url"), "phone": a.get("phone"), "business_id": a.get("business_id"), "primary_industry": a.get("industry"), "all_industries": ", ".join([i.get("name") for i in a.get("industries", [])]) if a.get("industries") else None, "first_visit_date": a.get("first_visit_date"), "last_visit_date": a.get("last_visit_date"), "total_visits": a.get("visits"), "lead_quality_score": a.get("quality"), "revenue": a.get("revenue"), "employee_count": a.get("employee_count"), "employees_min": a.get("employees_range", {}).get("min") if a.get("employees_range") else None, "employees_max": a.get("employees_range", {}).get("max") if a.get("employees_range") else None, "assignee": a.get("assignee"), "emailed_to": a.get("emailed_to"), "crm_lead_id": a.get("crm_lead_id"), "crm_organization_id": a.get("crm_organization_id"), "tags": ", ".join(a.get("tags", [])) if a.get("tags") else None, "linkedin_url": a.get("linkedin_url"), "twitter_handle": a.get("twitter_handle"), "facebook_url": a.get("facebook_url"), "country": loc.get("country"), "region": loc.get("region"), "city": loc.get("city"), "leadfeeder_url": a.get("view_in_leadfeeder"), "landing_page_path": None, "exit_page_path": None }) print(f"✅ Page {page} loaded. Rows: {len(rows)}") page += 1 except Exception as e: print(f"Error on page {page}: {e}") break df = pd.DataFrame(rows) if not df.empty: df["last_visit_date"] = pd.to_datetime(df["last_visit_date"], errors="coerce") return df def enrich_leads_with_visits(df, start_date, end_date, max_rows=None, progress=gr.Progress()): if df.empty: return df target_df = df.head(max_rows) if max_rows else df total = len(target_df) print(f"🕵️ Deep enriching {total} rows ({start_date} to {end_date})...") for index, row in target_df.iterrows(): if row.get("landing_page_path"): continue lead_id = row["lead_id"] try: visit_data = make_request( f"{BASE_URL}/accounts/{ACCOUNT_ID}/leads/{lead_id}/visits", params={"start_date": start_date, "end_date": end_date, "page[size]": 1, "include": "page_views"} ) visits = visit_data.get("data", []) included = visit_data.get("included", []) landing = None exit_p = None if visits: visit = visits[0] v_attrs = visit.get("attributes", {}) landing = v_attrs.get("landing_page_path") or v_attrs.get("landing_page_url") visit_route = v_attrs.get("visit_route", []) if visit_route and isinstance(visit_route, list): last_step = visit_route[-1] if not exit_p: exit_p = last_step.get("page_path") or last_step.get("page_url") if not landing: first_step = visit_route[0] landing = first_step.get("page_path") or first_step.get("page_url") if not landing or not exit_p: pv_map = {p["id"]: p["attributes"] for p in included if p["type"] == "page_views"} pv_ids = [r["id"] for r in visit.get("relationships", {}).get("page_views", {}).get("data", [])] if pv_ids: if not landing: first_pv = pv_map.get(pv_ids[0]) if first_pv: landing = first_pv.get("url") or first_pv.get("path") if not exit_p: last_pv = pv_map.get(pv_ids[-1]) if last_pv: exit_p = last_pv.get("url") or last_pv.get("path") df.at[index, "landing_page_path"] = landing df.at[index, "exit_page_path"] = exit_p except Exception as e: print(f"Failed to enrich lead {lead_id}: {e}") if max_rows and index % 5 == 0: progress(index / total, desc="Enriching...") return df # ===================================================== # 4. WRAPPERS # ===================================================== def load_preview(start, end): df = fetch_basic_leads(start, end) if df.empty: return df, pd.DataFrame(), pd.DataFrame(), "⚠️ No data found." df_preview = df.copy() df_preview = enrich_leads_with_visits(df_preview, start, end, max_rows=50) df_preview = apply_business_logic(df_preview) # Return Raw df for export, Enriched df for dashboard, and Table preview return df, df_preview, df_preview.head(50), f"✅ Loaded {len(df):,} companies. Preview top 50." def download_full_excel(df, start, end): if df is None or df.empty: return None, None print("⏳ Starting full enrichment for Excel export...") enriched_df = enrich_leads_with_visits(df.copy(), start, end) final_df = apply_business_logic(enriched_df) path = "/tmp/leadfeeder_campaign_data.xlsx" final_df.to_excel(path, index=False) # Updating the enriched state so the dashboard can use the full data return path, final_df def inspect_raw_json(lead_id, start_date, end_date): if not lead_id: return "Please enter a Lead ID" try: url = f"{BASE_URL}/accounts/{ACCOUNT_ID}/leads/{lead_id}/visits" params = {"start_date": start_date, "end_date": end_date, "page[size]": 1, "include": "page_views"} r = requests.get(url, headers=HEADERS, params=params) return json.dumps(r.json(), indent=2) except Exception as e: return str(e) # ===================================================== # 5. CHART ENGINES # ===================================================== def build_kpis(df): if df is None or df.empty: return 0, 0, 0, 0, 0, 0, 0 if "Quality_Group" not in df.columns: df = apply_business_logic(df) return ( len(df), df["total_visits"].gt(0).sum(), int(df["total_visits"].sum()), round(df["lead_quality_score"].mean(), 2), round(df["crm_organization_id"].notna().mean() * 100, 1), round(df["linkedin_url"].notna().mean() * 100, 1), df[df["Quality_Group"] == "High Quality (8-10)"].shape[0], ) def build_dashboard(df, preset, top_n): if df is None or df.empty: return px.bar(title="No Data") if "Campaign" not in df.columns: df = apply_business_logic(df) if preset == "Key Campaigns Bifurcation (MS Tech, Fintech, Adtech)": target_lower = ["ms tech", "fintech", "adtech"] df_chart = df.copy() # Ensure exact match regardless of trailing spaces or casing df_chart["Camp_Lower"] = df_chart["Campaign"].astype(str).str.strip().str.lower() filtered = df_chart[df_chart["Camp_Lower"].isin(target_lower)].copy() if filtered.empty: return px.bar(title="No Data for Key Campaigns") agg = filtered.groupby("Campaign").agg( Leads_Count=("company_name", "count"), Total_Visits=("total_visits", "sum") ).reset_index() fig = go.Figure() fig.add_trace(go.Bar(x=agg["Campaign"], y=agg["Leads_Count"], name="No. of Leads", marker_color="#00C49F")) fig.add_trace(go.Bar(x=agg["Campaign"], y=agg["Total_Visits"], name="Total Visits", marker_color="#FFBB28")) fig.update_layout(title="Key Campaigns: Leads vs Visits", barmode='group') return fig if preset == "All Campaigns Performance": agg = df.groupby("Campaign").agg( Leads_Count=("company_name", "count"), Total_Visits=("total_visits", "sum") ).reset_index().sort_values("Leads_Count", ascending=False) fig = make_subplots(specs=[[{"secondary_y": True}]]) fig.add_trace(go.Bar(x=agg["Campaign"], y=agg["Leads_Count"], name="Leads", marker_color="indigo"), secondary_y=False) fig.add_trace(go.Scatter(x=agg["Campaign"], y=agg["Total_Visits"], name="Visits", mode="lines+markers", line=dict(color="orange", width=3)), secondary_y=True) fig.update_layout(title_text="All Campaigns Performance") return fig dim, metric, agg = DASHBOARD_PRESETS[preset] if agg == "count": grouped = df.groupby(dim, dropna=False).size().reset_index(name="value") else: grouped = df.groupby(dim, dropna=False)[metric].agg(agg).reset_index(name="value") return px.bar(grouped.sort_values("value", ascending=False).head(top_n), x=dim, y="value", title=preset, color=dim) def build_trend(df, preset, grain, filter_values): if df is None or df.empty: return px.line(title="No Data") metric, agg, segment = TREND_PRESETS[preset] df_t = df.dropna(subset=["last_visit_date"]).copy() if segment and segment not in df_t.columns: df_t = apply_business_logic(df_t) if segment and filter_values: df_t = df_t[df_t[segment].isin(filter_values)] if grain == "Weekly": df_t["period"] = df_t["last_visit_date"].dt.to_period("W").astype(str) else: df_t["period"] = df_t["last_visit_date"].dt.date val = metric if agg == "count": df_t["_v"] = 1 val = "_v" if segment: ts = df_t.groupby(["period", segment])[val].agg(agg).reset_index() return px.line(ts, x="period", y=val, color=segment, title=preset, markers=True) ts = df_t.groupby("period")[val].agg(agg).reset_index() return px.line(ts, x="period", y=val, markers=True, title=preset) def get_trend_filter_options(df, preset): if df is None or df.empty: return gr.update(choices=[], value=None, visible=False) metric, agg, segment = TREND_PRESETS[preset] if not segment: return gr.update(choices=[], value=None, visible=False) if segment not in df.columns: df = apply_business_logic(df) options = sorted(df[segment].astype(str).unique().tolist()) if segment == "Campaign": defaults = options[:3] else: defaults = options[:5] if not defaults: defaults = options[:5] return gr.update(choices=options, value=defaults, visible=True, label=f"Filter {segment}") # ===================================================== # 6. UI LAYOUT # ===================================================== with gr.Blocks(title="Leadfeeder Campaign Pro") as demo: gr.Markdown("## 🚀 Leadfeeder Analytics & Campaign Manager") with gr.Row(): pwd = gr.Textbox(type="password", label="App Password") gr.Button("Auth").click(lambda p: gr.Info("Success") if p==APP_PASSWORD else gr.Error("Invalid"), pwd, None) # SPLIT STATE: One for raw data, one for enriched dashboard data df_raw_state = gr.State() df_enriched_state = gr.State() status = gr.Markdown() with gr.Tabs(): # --- TAB 1: DATA --- with gr.Tab("📋 Data & Report"): with gr.Row(): start = gr.Textbox(label="Start Date", value=(date.today()-timedelta(days=30)).isoformat()) end = gr.Textbox(label="End Date", value=date.today().isoformat()) with gr.Row(): btn_load = gr.Button("1. Load Data (Preview)", variant="primary") btn_dl = gr.Button("2. Enrich & Download Full Excel") file_dl = gr.File(label="Download Excel") table = gr.Dataframe(label="Preview (Top 50 Enriched)", interactive=True) btn_load.click(load_preview, [start, end], [df_raw_state, df_enriched_state, table, status]) btn_dl.click(download_full_excel, [df_raw_state, start, end], [file_dl, df_enriched_state]) # --- TAB 2: DASHBOARD --- with gr.Tab("📊 Dashboard"): kpis = [gr.Number(label=l) for l in ["Companies", "Active", "Visits", "Avg Quality", "CRM %", "LinkedIn %", "High Quality (8-10)"]] gr.Button("Refresh KPIs").click(build_kpis, df_enriched_state, kpis) gr.Markdown("### 📊 Charts") with gr.Row(): preset = gr.Dropdown(choices=list(DASHBOARD_PRESETS.keys()), label="Chart View", value="Key Campaigns Bifurcation (MS Tech, Fintech, Adtech)") top_n = gr.Slider(5, 50, value=10, label="Top N Items") chart = gr.Plot() gr.Button("Build View").click(build_dashboard, [df_enriched_state, preset, top_n], chart) # --- TAB 3: TRENDS --- with gr.Tab("📈 Trends"): with gr.Row(): trend_view = gr.Dropdown(choices=list(TREND_PRESETS.keys()), label="Select Trend", value="Visits Trend by Campaign") grain = gr.Radio(["Daily", "Weekly", "Monthly"], value="Daily", label="Granularity") filter_dropdown = gr.Dropdown(multiselect=True, visible=False, label="Filter Segments") plot = gr.Plot() trend_view.change(get_trend_filter_options, [df_enriched_state, trend_view], filter_dropdown) gr.Button("Show Trend", variant="primary").click(build_trend, [df_enriched_state, trend_view, grain, filter_dropdown], plot) # --- TAB 4: SETTINGS --- with gr.Tab("⚙️ Campaign Settings"): gr.Markdown("### Manage Campaign Groups") init_rules = load_campaign_rules() camp_choices = list(init_rules.keys()) + ["+ Create New Campaign"] if init_rules else ["+ Create New Campaign"] default_inc = "" default_exc = "" if init_rules and camp_choices[0] in init_rules: default_inc = ", ".join(init_rules[camp_choices[0]].get("include", [])) default_exc = ", ".join(init_rules[camp_choices[0]].get("exclude", [])) with gr.Row(): camp_dropdown = gr.Dropdown(choices=camp_choices, label="Select Campaign to Edit", value=camp_choices[0]) new_camp_name = gr.Textbox(label="New Campaign Name", visible=(not init_rules)) with gr.Row(): inc_kw_input = gr.Textbox(label="Include Keywords (comma separated)", lines=3, value=default_inc) exc_kw_input = gr.Textbox(label="Exclude Keywords (comma separated)", lines=3, value=default_exc) def update_ui_on_select(selected_camp): rules = load_campaign_rules() if selected_camp == "+ Create New Campaign": return gr.update(visible=True, value=""), gr.update(value=""), gr.update(value="") else: inc_kws = rules.get(selected_camp, {}).get("include", []) exc_kws = rules.get(selected_camp, {}).get("exclude", []) return gr.update(visible=False), gr.update(value=", ".join(inc_kws)), gr.update(value=", ".join(exc_kws)) camp_dropdown.change(update_ui_on_select, inputs=[camp_dropdown], outputs=[new_camp_name, inc_kw_input, exc_kw_input]) save_config_btn = gr.Button("💾 Save Configuration to Hugging Face", variant="primary") config_status = gr.Markdown() def save_easy_config(selected_camp, new_name, inc_string, exc_string): rules = load_campaign_rules() clean_inc = [k.strip().lower() for k in inc_string.split(",") if k.strip()] clean_exc = [k.strip().lower() for k in exc_string.split(",") if k.strip()] target_camp = new_name.strip() if selected_camp == "+ Create New Campaign" else selected_camp if not target_camp: return "❌ Error: Campaign name cannot be empty.", gr.update() rules[target_camp] = {"include": clean_inc, "exclude": clean_exc} with open(CAMPAIGN_CONFIG_FILE, "w") as f: json.dump(rules, f, indent=4) status_message = f"✅ Saved locally! Updated keywords for '{target_camp}'." if HF_TOKEN and SPACE_ID: try: api = HfApi(token=HF_TOKEN) api.upload_file( path_or_fileobj=CAMPAIGN_CONFIG_FILE, path_in_repo=CAMPAIGN_CONFIG_FILE, repo_id=SPACE_ID, repo_type="space" ) status_message = f"✅ Saved securely to Hugging Face Cloud! Updated keywords for '{target_camp}'." except Exception as e: status_message = f"⚠️ Saved locally, but failed to push to Hugging Face (Check HF_TOKEN). Error: {e}" elif not HF_TOKEN: status_message = f"⚠️ Saved locally. To make this permanent on Hugging Face, add an HF_TOKEN secret in your Space settings." updated_choices = list(rules.keys()) + ["+ Create New Campaign"] return status_message, gr.update(choices=updated_choices, value=target_camp) save_config_btn.click(save_easy_config, inputs=[camp_dropdown, new_camp_name, inc_kw_input, exc_kw_input], outputs=[config_status, camp_dropdown]) # --- TAB 5: DEBUGGER --- with gr.Tab("🛠️ Debugger"): dbg_id = gr.Textbox(label="Lead ID") dbg_btn = gr.Button("Inspect Raw JSON") dbg_out = gr.Code(language="json") dbg_btn.click(inspect_raw_json, [dbg_id, start, end], dbg_out) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False)