Spaces:
Sleeping
Sleeping
| 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) |