import streamlit as st import sqlite3 import pandas as pd import asyncio import os import threading import time from dotenv import load_dotenv load_dotenv() from pipeline import run_pipeline, PIPELINE_TRACKER st.set_page_config(page_title="Ops Command Center", layout="wide") DATABASE_PATH = 'leads.db' GLOBAL_MATRIX = { "India (IN - West Hubs)": [ "Pune-Chakan-Pimpri (Automotive & Heavy Eng)", "Ahmedabad-Sanand (GIDC Manufacturing Belt)", "Surat-Hazira (Heavy Industrial & Textiles)", "Vadodara-Ankleshwar (Chemical & Pharma Clusters)", "Nagpur-Butibori (Logistics & MIDC Manufacturing)", "Nashik-Aurangabad (Engineering & Tooling Hubs)", "Mumbai-Thane-Vapi (Industrial Processing & Tech)" ], "India (IN - South Hubs)": [ "Bangalore-Peenya (KIADB Tech & Advanced Precision)", "Chennai-Sriperumbudur (Automotive & Electronics Assembly)", "Coimbatore (Industrial Machinery & Textile Engineering)", "Hosur (Heavy Manufacturing & Component Fabricators)", "Hyderabad-Pashamylaram (Pharma & Automation Clusters)", "Visakhapatnam-Sri City (Seaport Logistics & Special Tech Zones)" ], "India (IN - North & Central Hubs)": [ "Gurugram-Manesar (Automotive & Core Software Integration)", "Noida-Greater Noida (Electronics Packaging & Smart Infrastructure)", "Faridabad-Ghaziabad (Light Engineering & Component Casting)", "Ludhiana (Heavy Machining & Textile Manufacturing Hubs)", "Indore-Pithampur (Automotive & Pharma Corridors)", "Pantnagar-Haridwar (Industrial Manufacturing Estates)" ], "India (IN - East Hubs)": [ "Jamshedpur-Bokaro (Steel, Mining Equipment & Heavy Metals)", "Kolkata-Durgapur-Asansol (Core Industrial, Logistics & Metal Alloys)" ], "Mexico (MX)": ["Monterrey", "Querétaro", "Ciudad Juárez", "Guadalajara", "Puebla"], "Poland (PL)": ["Wrocław", "Poznań", "Katowice", "Kraków", "Warsaw"], "Vietnam (VN)": ["Ho Chi Minh City", "Hanoi", "Haiphong", "Bình Dương"], "Saudi Arabia (SA)": ["Riyadh", "Jeddah", "Dammam", "NEOM (Tabuk Region)"], "United Arab Emirates (AE)": ["Dubai", "Abu Dhabi", "Sharjah"], "Australia (AU)": ["Perth", "Brisbane", "Melbourne", "Sydney"], "Malaysia (MY)": ["Penang", "Kuala Lumpur", "Johor Bahru"], "Brazil (BR)": ["São Paulo", "Campinas", "Belo Horizonte", "Curitiba"], "Morocco (MA)": ["Casablanca", "Tangier", "Kenitra"] } def inject_custom_css(): st.markdown( """ """, unsafe_allow_html=True ) def init_db(): conn = sqlite3.connect(DATABASE_PATH) c = conn.cursor() c.execute(''' CREATE TABLE IF NOT EXISTS leads ( id INTEGER PRIMARY KEY AUTOINCREMENT, company TEXT, country TEXT, city TEXT, email TEXT, status TEXT, pitch TEXT, industry_tier TEXT, market_priority TEXT ) ''') # Database Schema Self-Healing Patch c.execute("PRAGMA table_info(leads)") columns = [info[1] for info in c.fetchall()] required_columns = { "email": "TEXT", "country": "TEXT", "industry_tier": "TEXT", "market_priority": "TEXT", "website_url": "TEXT", "phone_number": "TEXT", "social_links": "TEXT" } for col, dtype in required_columns.items(): if col not in columns: c.execute(f"ALTER TABLE leads ADD COLUMN {col} {dtype};") conn.commit() conn.close() def get_pending_leads(country: str, cities: list): """Ghost Leak Fix: Filter strictly by the selected country and city view state.""" if not cities: return pd.DataFrame() conn = sqlite3.connect(DATABASE_PATH) placeholders = ','.join(['?'] * len(cities)) query = f"SELECT * FROM leads WHERE status='PENDING_REVIEW' AND country=? AND city IN ({placeholders})" params = [country] + cities df = pd.read_sql_query(query, conn, params=params) conn.close() return df def discard_lead(lead_id): conn = sqlite3.connect(DATABASE_PATH) c = conn.cursor() c.execute("UPDATE leads SET status = 'DISCARDED' WHERE id = ?", (lead_id,)) conn.commit() conn.close() def main(): inject_custom_css() init_db() st.title("🌐 Live Operations Engine: Emerging Markets") st.markdown("**(Powered by DuckDuckGo X-Ray Engine & Deep Domain Crawling)**") if not os.environ.get("GEMINI_API_KEY"): st.warning("⚠️ GEMINI_API_KEY not found in environment.") st.header("🎛️ Live Pipeline Control Panel") with st.expander("Web Scraper Configuration Engine", expanded=True): col1, col2 = st.columns(2) with col1: market_priority = st.selectbox("Market Priority", ["International P1", "National P2"]) # Dynamic Country Filtering Rule if market_priority == "International P1": country_options = [k for k in GLOBAL_MATRIX.keys() if "India" not in k] else: country_options = [k for k in GLOBAL_MATRIX.keys() if "India" in k] selected_country = st.selectbox("Target Emerging Country", country_options) with col2: available_cities = GLOBAL_MATRIX[selected_country] select_all = st.checkbox("Select All Cities", value=True) if select_all: selected_cities = available_cities st.multiselect("Target Hubs", available_cities, default=available_cities, disabled=True) else: selected_cities = st.multiselect("Target Hubs", available_cities, default=[]) if st.button("🚀 Execute DuckDuckGo X-Ray Search", type="primary", disabled=PIPELINE_TRACKER["is_running"]): if selected_cities: conn = sqlite3.connect(DATABASE_PATH) c = conn.cursor() # Workspace Lead Isolation c.execute("DELETE FROM leads WHERE status = 'PENDING_REVIEW';") conn.commit() conn.close() def bg_run(priority, country, cities): asyncio.run(run_pipeline(priority, country, cities)) t = threading.Thread(target=bg_run, args=(market_priority, selected_country, selected_cities)) try: from streamlit.runtime.scriptrunner import add_script_run_ctx add_script_run_ctx(t) except ImportError: pass t.start() st.rerun() else: st.warning("Please select at least one target hub.") if PIPELINE_TRACKER["is_running"]: with st.spinner("Pipeline is running in background. You can interact with the dashboard..."): time.sleep(1) st.rerun() elif PIPELINE_TRACKER["result"] is not None: count = PIPELINE_TRACKER["result"] if count > 0: st.success(f"Pipeline complete! Digested {count} REAL verified enterprise leads.") else: st.warning("No leads found for these hubs. This can happen if domains are completely locked or contacts are hidden.") PIPELINE_TRACKER["result"] = None st.divider() st.header("📋 Operational Review Queue (Live Data)") # Filter only based on what's active in the UI to fix Ghost Leaks leads_df = get_pending_leads(selected_country, selected_cities) if leads_df.empty: st.info("Queue is empty for this region. Execute the Live DDG X-Ray Search above.") else: total_leads = len(leads_df) st.markdown(f"**Total Pending Verification in View:** {total_leads}") items_per_page = 10 total_pages = max(1, (total_leads - 1) // items_per_page + 1) if total_pages > 1: page = st.number_input("Pagination (Page)", min_value=1, max_value=total_pages, value=1, step=1) start_idx = (page - 1) * items_per_page end_idx = start_idx + items_per_page leads_to_display = leads_df.iloc[start_idx:end_idx] else: leads_to_display = leads_df for index, row in leads_to_display.iterrows(): st.markdown('