How one can Construct a Proactive Pre-Emptive Churn Prevention Agent with Clever Commentary and Technique Formation

import os
import time
import json
import random
from datetime import datetime, timedelta
from typing import Listing, Dict, Any
import textwrap


strive:
   import google.generativeai as genai
besides ImportError:
   !pip set up -q -U google-generativeai
   import google.generativeai as genai


from google.colab import userdata
import getpass

def setup_gemini():
   print("--- 🔐 Safety Test ---")
   strive:
       api_key = userdata.get('GEMINI_API_KEY')
   besides:
       print("Please enter your Google Gemini API Key:")
       api_key = getpass.getpass("API Key: ")
   if not api_key:
       increase ValueError("API Secret's required to run the agent.")
   genai.configure(api_key=api_key)
   return genai.GenerativeModel('gemini-2.5-flash')


class MockCustomerDB:
   def __init__(self):
       self.as we speak = datetime.now()
       self.customers = self._generate_mock_users()


   def _generate_mock_users(self) -> Listing[Dict]:
       profiles = [
           {"id": "U001", "name": "Sarah Connor", "plan": "Enterprise",
            "last_login_days_ago": 2, "top_features": ["Reports", "Admin Panel"], "total_spend": 5000},
           {"id": "U002", "title": "John Smith", "plan": "Fundamental",
            "last_login_days_ago": 25, "top_features": ["Image Editor"], "total_spend": 50},
           {"id": "U003", "title": "Emily Chen", "plan": "Professional",
            "last_login_days_ago": 16, "top_features": ["API Access", "Data Export"], "total_spend": 1200},
           {"id": "U004", "title": "Marcus Aurelius", "plan": "Enterprise",
            "last_login_days_ago": 45, "top_features": ["Team Management"], "total_spend": 8000}
       ]
       return profiles


   def fetch_at_risk_users(self, threshold_days=14) -> Listing[Dict]:
       return [u for u in self.users if u['last_login_days_ago'] >= threshold_days]

class ChurnPreventionAgent:
   def __init__(self, mannequin):
       self.mannequin = mannequin


   def analyze_and_strategize(self, person: Dict) -> Dict:
       print(f"   ... 🧠 Analyzing technique for {person['name']}...")
       immediate = f"""
       You're a Buyer Success AI Specialist.
       Analyze this person profile and decide one of the best 'Win-Again Technique'.
       USER PROFILE:
       - Identify: {person['name']}
       - Plan: {person['plan']}
       - Days Inactive: {person['last_login_days_ago']}
       - Favourite Options: {', '.be a part of(person['top_features'])}
       - Whole Spend: ${person['total_spend']}
       TASK:
       1. Decide the 'Churn Likelihood' (Medium/Excessive/Essential).
       2. Choose a particular INCENTIVE.
       3. Clarify your reasoning briefly.
       OUTPUT FORMAT:
       {{
           "risk_level": "Excessive",
           "incentive_type": "Particular Incentive",
           "reasoning": "One sentence clarification."
       }}
       """
       strive:
           response = self.mannequin.generate_content(immediate)
           clean_json = response.textual content.exchange("```json", "").exchange("```", "").strip()
           return json.masses(clean_json)
       besides Exception as e:
           return {
               "risk_level": "Unknown",
               "incentive_type": "Basic Test-in",
               "reasoning": f"Evaluation failed: {str(e)}"
           }

def draft_engagement_email(self, person: Dict, technique: Dict) -> str:
       print(f"   ... ✍️  Drafting e-mail for {person['name']} utilizing '{technique['incentive_type']}'...")
       immediate = f"""
       Write a brief, empathetic, skilled re-engagement e-mail.
       TO: {person['name']}
       CONTEXT: They have not logged in for {person['last_login_days_ago']} days.
       STRATEGY: {technique['incentive_type']}
       REASONING: {technique['reasoning']}
       USER HISTORY: They love {', '.be a part of(person['top_features'])}.
       TONE: Useful and concise.
       """
       response = self.mannequin.generate_content(immediate)
       return response.textual content

class ManagerDashboard:
   def review_draft(self, user_name, technique, draft_text):
       print("n" + "="*60)
       print(f"🚨 REVIEW REQUIRED: Re-engagement for {user_name}")
       print(f"🎯 Technique: {technique['incentive_type']}")
       print(f"📝 Threat Degree: {technique['risk_level']}")
       print("-" * 60)
       print("📨 DRAFT EMAIL:n")
       print(textwrap.indent(draft_text, '    '))
       print("-" * 60)
       print("n[Auto-Simulation] Supervisor reviewing...")
       time.sleep(1.5)
       if technique['risk_level'] == "Essential":
           print("✅ MANAGER DECISION: Authorized (Precedence Ship)")
           return True
       else:
           print("✅ MANAGER DECISION: Authorized")
           return True

def principal():
   print("Initializing Agentic System...")
   strive:
       mannequin = setup_gemini()
       db = MockCustomerDB()
       agent = ChurnPreventionAgent(mannequin)
       supervisor = ManagerDashboard()
   besides Exception as e:
       print(f"Setup failed: {e}")
       return


   print("n🔍 AGENT STATUS: Scanning Database for inactive customers (>14 days)...")
   at_risk_users = db.fetch_at_risk_users(threshold_days=14)
   print(f"Discovered {len(at_risk_users)} at-risk customers.n")


   for person in at_risk_users:
       print(f"--- Processing Case: {person['id']} ({person['name']}) ---")
       technique = agent.analyze_and_strategize(person)
       email_draft = agent.draft_engagement_email(person, technique)
       permitted = supervisor.review_draft(person['name'], technique, email_draft)
       if permitted:
           print(f"🚀 ACTION: E mail queued for sending to {person['name']}.")
       else:
           print(f"🛑 ACTION: E mail rejected.")
       print("n")
       time.sleep(1)


if __name__ == "__main__":
   principal()