import gradio as gr import sys import os import json from typing import List, Dict, Any, Optional from pathlib import Path from linkedin_data import fetch_linkedin_profile, get_cache_stats from data_enrichment import DataEnrichmentManager # --- Data Storage Configuration --- DATA_DIR = Path("/data") USER_DATA_DIR = DATA_DIR / "users" USER_DATA_DIR.mkdir(parents=True, exist_ok=True) def get_user_dir(username: str) -> Path: """Get or create user-specific data directory""" user_dir = USER_DATA_DIR / username user_dir.mkdir(parents=True, exist_ok=True) return user_dir def load_user_data(username: str, data_type: str) -> Any: """Load user-specific data from their directory""" user_dir = get_user_dir(username) data_file = user_dir / f"{data_type}.json" if data_file.exists(): with open(data_file, 'r') as f: return json.load(f) return [] def save_user_data(username: str, data_type: str, data: Any): """Save user-specific data to their directory""" user_dir = get_user_dir(username) data_file = user_dir / f"{data_type}.json" with open(data_file, 'w') as f: json.dump(data, f, indent=2) def get_enrichment_manager(username: str) -> DataEnrichmentManager: """Get or create DataEnrichmentManager for a user""" return DataEnrichmentManager(username) # --- Consent/Event Audit Panel --- def consent_audit_panel(consent_events: List[Dict[str, Any]], profile: Optional[gr.OAuthProfile]) -> str: """ UI Panel to show all consent events, opt-outs, and compliance flags per contact and batch. Accepts a list of consent event dicts with structure as returned from backend JSON. """ if profile is None: return "# 🔒 Authentication Required\n\nPlease sign in with your Hugging Face account to access this feature." username = profile.username # Load user-specific consent events user_events = load_user_data(username, "consent_events") if not user_events: user_events = consent_events # Use provided sample if no user data audit_md = f"# Consent/Event Audit\n\n**User:** {username}\n\nPer-contact and batch consent history." for event in user_events: audit_md += f"\n- Contact: {event.get('contact_id','?')} | Event: {event.get('event_type','?')} | Opt-out: {event.get('opt_out','?')}" # LinkedIn enrichment section audit_md += "\n\n---\n\n## 🔗 LinkedIn Profile Enrichment\n\n" # Get cache stats cache_stats = get_cache_stats(username) audit_md += f"**Cached LinkedIn Profiles:** {cache_stats.get('cached_profiles', 0)}\n\n" # Sample LinkedIn enrichment (demonstrating the feature) audit_md += "**To enrich a contact with LinkedIn data:**\n" audit_md += "1. Add a 'linkedin_url' field to your consent events\n" audit_md += "2. The system will automatically fetch and cache profile data\n" audit_md += "3. Cached profiles are stored in your isolated data directory\n\n" # Example with sample LinkedIn URL (if configured) sample_linkedin_url = "https://linkedin.com/in/sample-profile" audit_md += f"**Example:** To test LinkedIn enrichment, add RAPIDAPI_KEY to Space Secrets.\n\n" if user_events and len(user_events) > 0: # Check if any event has LinkedIn data for event in user_events: if 'linkedin_url' in event and event['linkedin_url']: linkedin_url = event['linkedin_url'] audit_md += f"\n### LinkedIn Profile for Contact {event.get('contact_id', '?')}\n\n" # Fetch LinkedIn profile result = fetch_linkedin_profile(linkedin_url, username=username) if result.get('success'): profile_data = result.get('data', {}) from_cache = result.get('from_cache', False) audit_md += f"**Full Name:** {profile_data.get('full_name', 'N/A')}\n\n" audit_md += f"**Headline:** {profile_data.get('headline', 'N/A')}\n\n" audit_md += f"**Location:** {profile_data.get('location', 'N/A')}\n\n" audit_md += f"**Data Source:** {'Cached (30-day TTL)' if from_cache else 'Fresh from API'}\n\n" # Show first experience if available experiences = profile_data.get('experiences', []) if experiences and len(experiences) > 0: exp = experiences[0] audit_md += f"**Current Role:** {exp.get('title', 'N/A')} at {exp.get('company', 'N/A')}\n\n" else: error_msg = result.get('message', 'Unknown error') audit_md += f"⚠️ **Error:** {error_msg}\n\n" audit_md += f"\n- Contact: {event.get('contact_id','?')} | Event: {event.get('event_type','?')} | Opt-out: {event.get('opt_out','?')}" return gr.Markdown(audit_md) # --- Dynamic Branching Composer for Cadence Editing --- def branching_composer(branches: List[Dict[str, Any]], profile: Optional[gr.OAuthProfile]) -> str: """ Visual sequence builder for editing cadences. Structure assumed to be a list of dict branches with JSON mirroring backend. """ if profile is None: return "# 🔒 Authentication Required\n\nPlease sign in with your Hugging Face account to access this feature." username = profile.username # Load user-specific branches user_branches = load_user_data(username, "branches") if not user_branches: user_branches = branches # Use provided sample if no user data builder_md = f"# Dynamic Cadence Branching\n\n**User:** {username}\n\n## Sequence Tree:" for b in user_branches: builder_md += f"\n- Step {b.get('step','?')}: {b.get('description','?')} (Next: {b.get('next','?')})" return gr.Markdown(builder_md) # --- Live Diagnostics Panel for Cadence Errors/Health --- def diagnostics_panel(events: List[Dict[str, Any]], profile: Optional[gr.OAuthProfile]) -> str: """ Incident/event log UI widget to show errors and health of cadence system. Format is a simple list of event dicts from backend state. """ if profile is None: return "# 🔒 Authentication Required\n\nPlease sign in with your Hugging Face account to access this feature." username = profile.username # Load user-specific diagnostic events user_events = load_user_data(username, "diagnostics") if not user_events: user_events = events # Use provided sample if no user data diag_md = f"# Live Diagnostics: Cadence Health\n\n**User:** {username}" for e in user_events: diag_md += f"\n- [{e.get('timestamp','?')}] {e.get('type','?')}: {e.get('message','?')} {'[ERROR]' if e.get('level')=='error' else ''}" return gr.Markdown(diag_md) # --- Audit Trail Export Widget --- def audit_trail_widget(profile: Optional[gr.OAuthProfile]) -> str: """ CSV export widget for compliance officer. Returns download button for current batch consent data. """ if profile is None: return "# 🔒 Authentication Required\n\nPlease sign in with your Hugging Face account to access this feature." username = profile.username export_md = f"# Audit Trail Export\n\n**User:** {username}\n\nDownload compliance reports in CSV format.\n\n*Export functionality will be implemented based on your specific requirements.*" return gr.Markdown(export_md) # Sample Data (for demo purposes when no user data exists) consent_events_sample = [ {"contact_id": "c123", "event_type": "opt_in", "opt_out": False, "timestamp": "2025-01-10"}, {"contact_id": "c456", "event_type": "opt_out", "opt_out": True, "timestamp": "2025-01-11"}, {"contact_id": "c789", "event_type": "consent_updated", "opt_out": False, "timestamp": "2025-01-12"}, ] branches_sample = [ {"step": 1, "description": "Initial outreach email", "next": 2}, {"step": 2, "description": "Follow-up call", "next": 3}, {"step": 3, "description": "Send proposal", "next": 4}, {"step": 4, "description": "Close or nurture", "next": None}, ] diagnostics_sample = [ {"timestamp": "2025-01-15 10:30", "type": "rate_limit", "message": "LinkedIn API approaching limit (85%)", "level": "warning"}, {"timestamp": "2025-01-15 11:00", "type": "success", "message": "Batch processed successfully", "level": "info"}, {"timestamp": "2025-01-15 11:30", "type": "error", "message": "Failed to send email to c999", "level": "error"}, ] audit_history_sample = [ {"batch_id": "b001", "contacts": 150, "opt_outs": 5, "timestamp": "2025-01-10"}, {"batch_id": "b002", "contacts": 200, "opt_outs": 8, "timestamp": "2025-01-12"}, ] # --- Main Application --- with gr.Blocks(title="StrategyStack UI") as demo: gr.Markdown("# 🚀 StrategyStack Recovery, Enrichment & Outreach Platform") # OAuth Login Button gr.LoginButton() gr.Markdown(""" ## Welcome to StrategyStack **Customer-Ready Multi-Tenant Platform** - ✅ **Secure Authentication**: Sign in with your Hugging Face account - ✅ **Data Isolation**: Your data is stored separately and securely - ✅ **Privacy First**: We don't store your data centrally - you own it - ✅ **Compliance Ready**: Full audit trails and consent management Please sign in above to access your personalized dashboard. """) # Create tabbed interface with gr.Tabs(): with gr.Tab("Consent/Event Audit"): consent_output = gr.Markdown() consent_refresh = gr.Button("Refresh Data") user_profile_1 = gr.State() with gr.Tab("Cadence Branching Composer"): branching_output = gr.Markdown() branching_refresh = gr.Button("Refresh Data") user_profile_2 = gr.State() with gr.Tab("Live Diagnostics"): diagnostics_output = gr.Markdown() diagnostics_refresh = gr.Button("Refresh Data") user_profile_3 = gr.State() with gr.Tab("Audit Trail Export"): audit_output = gr.Markdown() audit_refresh = gr.Button("Refresh Data") user_profile_4 = gr.State() with gr.Tab("Data Enrichment"): enrichment_output = gr.Markdown() enrichment_refresh = gr.Button("Refresh Data") user_profile_5 = gr.State() # Load initial data def load_consent_data(profile: Optional[gr.OAuthProfile] = None): return consent_audit_panel(consent_events_sample, profile) def load_branching_data(profile: Optional[gr.OAuthProfile] = None): return branching_composer(branches_sample, profile) def load_diagnostics_data(profile: Optional[gr.OAuthProfile] = None): return diagnostics_panel(diagnostics_sample, profile) def load_audit_data(profile: Optional[gr.OAuthProfile] = None): return audit_trail_widget(profile) def load_enrichment_data(profile: Optional[gr.OAuthProfile] = None): if profile is None: return "# 👤 Authentication Required\n\nPlease sign in with your Hugging Face account to access data enrichment features." username = profile.username manager = get_enrichment_manager(username) stats = manager.get_enrichment_stats() enrichment_md = f"# 📊 Data Enrichment Dashboard\n\n**User:** {username}\n\n" enrichment_md += "## Enrichment Statistics\n\n" enrichment_md += f"- **Total API Calls:** {stats.get('total_api_calls', 0)}\n" enrichment_md += f"- **Cache Hits:** {stats.get('cache_hits', 0)}\n" enrichment_md += f"- **Cache Misses:** {stats.get('cache_misses', 0)}\n" enrichment_md += f"- **Total Enrichments:** {stats.get('total_enrichments', 0)}\n" enrichment_md += f"- **Successful Enrichments:** {stats.get('successful_enrichments', 0)}\n" enrichment_md += f"- **Failed Enrichments:** {stats.get('failed_enrichments', 0)}\n\n" # Calculate success rate if stats.get('total_enrichments', 0) > 0: success_rate = (stats.get('successful_enrichments', 0) / stats['total_enrichments']) * 100 enrichment_md += f"- **Success Rate:** {success_rate:.1f}%\n\n" enrichment_md += "## Available APIs\n\n" enrichment_md += "- ✅ LinkedIn Profile Enrichment (4 providers with fallback)\n" enrichment_md += "- ✅ Company Profile Enrichment\n" enrichment_md += "- ✅ Apollo.io Person Enrichment\n" enrichment_md += "- ✅ Twitter User Data\n" enrichment_md += "- ✅ Facebook Data\n" enrichment_md += "- ✅ Google Search\n" enrichment_md += "- ✅ Mediafy Enrichment\n" enrichment_md += "- ✅ Local Business Data (LetScrape)\n" enrichment_md += "- ✅ Website Contacts Scraper (LetScrape)\n\n" enrichment_md += "## Features\n\n" enrichment_md += "- 🔒 **Per-User Data Isolation**: Your enrichment data is stored separately\n" enrichment_md += "- ⚡ **Intelligent Caching**: 30-day TTL to minimize API costs\n" enrichment_md += "- 🔄 **Automatic Fallback**: Multiple providers ensure high availability\n" enrichment_md += "- 📈 **Real-time Statistics**: Track API usage and success rates\n\n" enrichment_md += "## Usage\n\n" enrichment_md += "To use data enrichment in your workflows:\n" enrichment_md += "1. Add contact data with fields like `linkedin_url`, `email`, `domain`\n" enrichment_md += "2. The system automatically enriches contacts using all available APIs\n" enrichment_md += "3. Enriched data is cached for 30 days to optimize performance\n" enrichment_md += "4. View statistics and monitor API usage from this dashboard\n\n" return gr.Markdown(enrichment_md) # Set up event handlers demo.load(load_consent_data, inputs=None, outputs=consent_output) demo.load(load_branching_data, inputs=None, outputs=branching_output) demo.load(load_diagnostics_data, inputs=None, outputs=diagnostics_output) demo.load(load_audit_data, inputs=None, outputs=audit_output) demo.load(load_enrichment_data, inputs=None, outputs=enrichment_output) consent_refresh.click(load_consent_data, inputs=None, outputs=consent_output) branching_refresh.click(load_branching_data, inputs=None, outputs=branching_output) diagnostics_refresh.click(load_diagnostics_data, inputs=None, outputs=diagnostics_output) audit_refresh.click(load_audit_data, inputs=None, outputs=audit_output) enrichment_refresh.click(load_enrichment_data, inputs=None, outputs=enrichment_output) if __name__ == "__main__": demo.launch()