#!/usr/bin/env python3 """ Ghost Agency - Referral Program Automation Automated referral tracking, rewards, viral loops, and analytics """ import os import json import sqlite3 import hashlib import time import random import string from pathlib import Path from datetime import datetime, timedelta from typing import Dict, List, Any, Optional REFERRAL_DB = Path("/home/ubuntu/GhostAgency") / "referral_program.db" # ============================================================ # REFERRAL PROGRAM ENGINE # ============================================================ class ReferralProgram: def __init__(self): self.db_path = str(REFERRAL_DB) self.init_db() def init_db(self): conn = sqlite3.connect(self.db_path) c = conn.cursor() # Referral Programs (per client business) c.execute("""CREATE TABLE IF NOT EXISTS referral_programs ( id TEXT PRIMARY KEY, client_id TEXT NOT NULL, business_name TEXT, program_name TEXT, program_type TEXT, -- standard, tiered, milestone, contest reward_structure TEXT, -- JSON referrer_reward TEXT, -- JSON referee_reward TEXT, -- JSON conditions TEXT, -- JSON: min_spend, max_referrals, validity_days tracking_method TEXT, -- code, link, qr, email, phone status TEXT DEFAULT 'active', start_date DATE, end_date DATE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Referrers (customers who refer) c.execute("""CREATE TABLE IF NOT EXISTS referrers ( id TEXT PRIMARY KEY, program_id TEXT NOT NULL, client_id TEXT NOT NULL, customer_id TEXT, -- internal customer ID first_name TEXT, last_name TEXT, email TEXT, phone TEXT, referral_code TEXT UNIQUE, referral_link TEXT, qr_code_url TEXT, total_referrals INTEGER DEFAULT 0, successful_referrals INTEGER DEFAULT 0, pending_referrals INTEGER DEFAULT 0, total_rewards_earned REAL DEFAULT 0, total_rewards_redeemed REAL DEFAULT 0, tier TEXT DEFAULT 'bronze', -- bronze, silver, gold, platinum joined_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_referral_at TIMESTAMP, status TEXT DEFAULT 'active', FOREIGN KEY (program_id) REFERENCES referral_programs(id) )""") # Referrals (the actual referral events) c.execute("""CREATE TABLE IF NOT EXISTS referrals ( id TEXT PRIMARY KEY, program_id TEXT NOT NULL, referrer_id TEXT NOT NULL, referee_email TEXT, referee_phone TEXT, referee_first_name TEXT, referee_last_name TEXT, referral_code_used TEXT, status TEXT DEFAULT 'pending', -- pending, signed_up, first_visit, first_purchase, qualified, rewarded, expired source TEXT, -- email, whatsapp, sms, social, qr, direct_link, word_of_mouth signed_up_at TIMESTAMP, first_visit_at TIMESTAMP, first_purchase_at TIMESTAMP, qualified_at TIMESTAMP, rewarded_at TIMESTAMP, referrer_reward_value REAL DEFAULT 0, referee_reward_value REAL DEFAULT 0, referrer_reward_type TEXT, referee_reward_type TEXT, purchase_amount REAL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (referrer_id) REFERENCES referrers(id) )""") # Rewards c.execute("""CREATE TABLE IF NOT EXISTS rewards ( id TEXT PRIMARY KEY, referral_id TEXT NOT NULL, program_id TEXT NOT NULL, recipient_type TEXT, -- referrer, referee recipient_id TEXT NOT NULL, reward_type TEXT, -- discount_percent, discount_fixed, free_item, cashback, points, gift_card reward_value TEXT, -- JSON status TEXT DEFAULT 'pending', -- pending, issued, redeemed, expired, cancelled code TEXT UNIQUE, issued_at TIMESTAMP, redeemed_at TIMESTAMP, expires_at TIMESTAMP, conditions TEXT, -- JSON: min_spend, valid_days, excluded_items, max_uses created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (referral_id) REFERENCES referrals(id) )""") # Referral Campaigns (marketing pushes) c.execute("""CREATE TABLE IF NOT EXISTS referral_campaigns ( id TEXT PRIMARY KEY, program_id TEXT NOT NULL, campaign_name TEXT, campaign_type TEXT, -- launch, booster, seasonal, reactivation channel TEXT, -- email, whatsapp, sms, push, social, in_app template_id TEXT, target_audience TEXT, -- all, active_referrers, inactive_referrers, new_customers incentive_boost TEXT, -- JSON: extra reward for limited time start_date DATE, end_date DATE, status TEXT DEFAULT 'draft', sent_count INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (program_id) REFERENCES referral_programs(id) )""") # Campaign Sends c.execute("""CREATE TABLE IF NOT EXISTS referral_campaign_sends ( id TEXT PRIMARY KEY, campaign_id TEXT NOT NULL, referrer_id TEXT, channel TEXT, sent_at TIMESTAMP, delivered_at TIMESTAMP, opened_at TIMESTAMP, clicked_at TIMESTAMP, shared_at TIMESTAMP, status TEXT DEFAULT 'sent', -- sent, delivered, opened, clicked, shared, bounced created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (campaign_id) REFERENCES referral_campaigns(id) )""") # Analytics c.execute("""CREATE TABLE IF NOT EXISTS referral_analytics ( id TEXT PRIMARY KEY, program_id TEXT NOT NULL, date DATE, total_referrers INTEGER DEFAULT 0, active_referrers INTEGER DEFAULT 0, new_referrers INTEGER DEFAULT 0, total_referrals INTEGER DEFAULT 0, qualified_referrals INTEGER DEFAULT 0, conversion_rate REAL DEFAULT 0, referrer_rewards_issued INTEGER DEFAULT 0, referee_rewards_issued INTEGER DEFAULT 0, referrer_rewards_redeemed INTEGER DEFAULT 0, referee_rewards_redeemed INTEGER DEFAULT 0, total_revenue_from_referrals REAL DEFAULT 0, total_reward_cost REAL DEFAULT 0, roi REAL DEFAULT 0, viral_coefficient REAL DEFAULT 0, -- referrals per referrer created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Fraud Detection c.execute("""CREATE TABLE IF NOT EXISTS referral_fraud_flags ( id TEXT PRIMARY KEY, referral_id TEXT NOT NULL, flag_type TEXT, -- self_referral, duplicate_account, fake_purchase, code_sharing, bot_activity severity TEXT, -- low, medium, high, critical details TEXT, status TEXT DEFAULT 'flagged', -- flagged, investigating, confirmed, dismissed, resolved detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, resolved_at TIMESTAMP, FOREIGN KEY (referral_id) REFERENCES referrals(id) )""") conn.commit() conn.close() def create_program(self, client_data: Dict, program_config: Dict) -> str: """Create a referral program for a client.""" program_id = hashlib.md5(f"prog_{client_data.get('client_id', '')}{time.time()}".encode()).hexdigest()[:12] conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" INSERT INTO referral_programs (id, client_id, business_name, program_name, program_type, reward_structure, referrer_reward, referee_reward, conditions, tracking_method, status, start_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', date('now')) """, ( program_id, client_data.get("client_id", ""), client_data.get("business_name", ""), program_config.get("program_name", "Programma Referral"), program_config.get("program_type", "standard"), json.dumps(program_config.get("reward_structure", {})), json.dumps(program_config.get("referrer_reward", {"type": "discount_percent", "value": 20, "max": 30})), json.dumps(program_config.get("referee_reward", {"type": "discount_percent", "value": 15, "max": 25})), json.dumps(program_config.get("conditions", {"min_spend": 20, "max_referrals_per_month": 10, "validity_days": 90})), program_config.get("tracking_method", "code") )) conn.commit() conn.close() return program_id def register_referrer(self, program_id: str, referrer_data: Dict) -> str: """Register a customer as a referrer.""" referrer_id = hashlib.md5(f"ref_{program_id}{referrer_data.get('email', '')}{time.time()}".encode()).hexdigest()[:12] # Generate unique referral code referral_code = self._generate_referral_code(referrer_data.get("first_name", ""), referrer_data.get("last_name", "")) conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" INSERT INTO referrers (id, program_id, client_id, customer_id, first_name, last_name, email, phone, referral_code, referral_link, qr_code_url, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active') """, ( referrer_id, program_id, referrer_data.get("client_id", ""), referrer_data.get("customer_id", ""), referrer_data.get("first_name", ""), referrer_data.get("last_name", ""), referrer_data.get("email", ""), referrer_data.get("phone", ""), referral_code, f"https://svoraj.me/ref/{referral_code}", f"https://api.qrserver.com/v1/create-qr-code/?data=https://svoraj.me/ref/{referral_code}&size=200x200" )) conn.commit() conn.close() return referrer_id def _generate_referral_code(self, first_name: str, last_name: str) -> str: """Generate a unique referral code.""" base = (first_name[:3] + last_name[:3]).upper() suffix = ''.join(random.choices(string.digits, k=4)) code = f"{base}{suffix}" # Check uniqueness conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("SELECT COUNT(*) FROM referrers WHERE referral_code = ?", (code,)) if c.fetchone()[0] > 0: conn.close() return self._generate_referral_code(first_name, last_name) conn.close() return code def create_referral(self, program_id: str, referrer_id: str, referee_data: Dict) -> str: """Create a new referral when a referee signs up.""" referral_id = hashlib.md5(f"ref_{referrer_id}{referee_data.get('email', '')}{time.time()}".encode()).hexdigest()[:12] # Get referrer's code conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("SELECT referral_code FROM referrers WHERE id = ?", (referrer_id,)) referrer_code = c.fetchone()[0] c.execute(""" INSERT INTO referrals (id, program_id, referrer_id, referee_email, referee_phone, referee_first_name, referee_last_name, referral_code_used, status, source, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, CURRENT_TIMESTAMP) """, ( referral_id, program_id, referrer_id, referee_data.get("email", ""), referee_data.get("phone", ""), referee_data.get("first_name", ""), referee_data.get("last_name", ""), referrer_code, referee_data.get("source", "direct_link") )) # Update referrer stats c.execute(""" UPDATE referrers SET total_referrals = total_referrals + 1, pending_referrals = pending_referrals + 1, last_referral_at = CURRENT_TIMESTAMP WHERE id = ? """, (referrer_id,)) conn.commit() conn.close() return referral_id def qualify_referral(self, referral_id: str, purchase_amount: float = 0) -> Dict: """Mark a referral as qualified (first purchase made).""" conn = sqlite3.connect(self.db_path) c = conn.cursor() # Get referral and program details c.execute(""" SELECT r.id, r.program_id, r.referrer_id, r.referee_email, r.referee_phone, r.referee_first_name, r.referee_last_name, r.referral_code_used, r.status, r.source, p.referrer_reward, p.referee_reward, p.conditions FROM referrals r JOIN referral_programs p ON r.program_id = p.id WHERE r.id = ? """, (referral_id,)) row = c.fetchone() if not row: conn.close() return {"error": "Referral not found"} referral = { "id": row[0], "program_id": row[1], "referrer_id": row[2], "referee_email": row[3], "referee_phone": row[4], "referee_first_name": row[5], "referee_last_name": row[6], "referral_code_used": row[7], "status": row[8], "source": row[9], "referrer_reward": json.loads(row[10]), "referee_reward": json.loads(row[11]), "conditions": json.loads(row[12]) } # Check conditions min_spend = referral["conditions"].get("min_spend", 0) if purchase_amount < min_spend: conn.close() return {"error": f"Purchase amount {purchase_amount} below minimum {min_spend}"} # Update referral status c.execute(""" UPDATE referrals SET status = 'qualified', first_purchase_at = CURRENT_TIMESTAMP, qualified_at = CURRENT_TIMESTAMP, purchase_amount = ? WHERE id = ? """, (purchase_amount, referral_id)) # Update referrer stats c.execute(""" UPDATE referrers SET successful_referrals = successful_referrals + 1, pending_referrals = pending_referrals - 1 WHERE id = ? """, (referral["referrer_id"],)) # Issue rewards using existing connection referrer_reward_id = self._issue_reward(conn, referral_id, referral["program_id"], referral["referrer_id"], "referrer", referral["referrer_reward"], purchase_amount) referee_reward_id = self._issue_reward(conn, referral_id, referral["program_id"], referral["referee_email"], "referee", referral["referee_reward"], purchase_amount) conn.commit() conn.close() return { "referral_id": referral_id, "status": "qualified", "referrer_reward_id": referrer_reward_id, "referee_reward_id": referee_reward_id } def _issue_reward(self, conn, referral_id: str, program_id: str, recipient_id: str, recipient_type: str, reward_config: Dict, purchase_amount: float) -> str: """Issue a reward to referrer or referee using existing connection.""" reward_id = hashlib.md5(f"reward_{referral_id}{recipient_type}{time.time()}".encode()).hexdigest()[:12] code = f"REF{''.join(random.choices(string.ascii_uppercase + string.digits, k=8))}" reward_type = reward_config.get("type", "discount_percent") value = reward_config.get("value", 20) max_val = reward_config.get("max", 50) # Calculate actual value if reward_type == "discount_percent": actual_value = min(purchase_amount * value / 100, max_val) elif reward_type == "discount_fixed": actual_value = min(value, max_val) else: actual_value = value expires_at = (datetime.now() + timedelta(days=90)).isoformat() c = conn.cursor() c.execute(""" INSERT INTO rewards (id, referral_id, program_id, recipient_type, recipient_id, reward_type, reward_value, status, code, issued_at, expires_at, conditions) VALUES (?, ?, ?, ?, ?, ?, ?, 'issued', ?, CURRENT_TIMESTAMP, ?, ?) """, ( reward_id, referral_id, program_id, recipient_type, recipient_id, reward_type, json.dumps(reward_config), code, expires_at, json.dumps({"min_spend": reward_config.get("min_spend", 20), "valid_days": 90}) )) return reward_id def redeem_reward(self, reward_code: str, purchase_amount: float = 0) -> Dict: """Redeem a reward code.""" conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("SELECT * FROM rewards WHERE code = ?", (reward_code,)) reward = c.fetchone() if not reward: conn.close() return {"error": "Invalid reward code"} reward_id, referral_id, program_id, recipient_type, recipient_id, reward_type, reward_value, status, code, issued_at, expires_at, conditions = reward if status != "issued": conn.close() return {"error": f"Reward already {status}"} if datetime.now() > datetime.fromisoformat(expires_at): conn.close() return {"error": "Reward expired"} conditions = json.loads(conditions) min_spend = conditions.get("min_spend", 0) if purchase_amount < min_spend: conn.close() return {"error": f"Minimum spend {min_spend} not met"} # Calculate discount reward_config = json.loads(reward_value) if reward_type == "discount_percent": discount = min(purchase_amount * reward_config.get("value", 0) / 100, reward_config.get("max", 50)) elif reward_type == "discount_fixed": discount = min(reward_config.get("value", 0), reward_config.get("max", 50)) else: discount = 0 # Update reward c.execute(""" UPDATE rewards SET status = 'redeemed', redeemed_at = CURRENT_TIMESTAMP WHERE id = ? """, (reward_id,)) conn.commit() conn.close() return { "reward_id": reward_id, "discount": round(discount, 2), "reward_type": reward_type, "message": f"Reward redeemed! {discount:.2f}€ discount applied." } def get_program_stats(self, program_id: str) -> Dict: """Get program analytics.""" conn = sqlite3.connect(self.db_path) c = conn.cursor() # Total referrers c.execute("SELECT COUNT(*) FROM referrers WHERE program_id = ?", (program_id,)) total_referrers = c.fetchone()[0] # Active referrers (made at least 1 referral in last 30 days) c.execute(""" SELECT COUNT(*) FROM referrers WHERE program_id = ? AND last_referral_at > datetime('now', '-30 days') """, (program_id,)) active_referrers = c.fetchone()[0] # Total referrals c.execute("SELECT COUNT(*) FROM referrals WHERE program_id = ?", (program_id,)) total_referrals = c.fetchone()[0] # Qualified referrals c.execute("SELECT COUNT(*) FROM referrals WHERE program_id = ? AND status = 'qualified'", (program_id,)) qualified_referrals = c.fetchone()[0] # Rewards issued c.execute(""" SELECT COUNT(*) FROM rewards r JOIN referrals ref ON r.referral_id = ref.id WHERE ref.program_id = ? """, (program_id,)) rewards_issued = c.fetchone()[0] # Rewards redeemed c.execute(""" SELECT COUNT(*) FROM rewards r JOIN referrals ref ON r.referral_id = ref.id WHERE ref.program_id = ? AND r.status = 'redeemed' """, (program_id,)) rewards_redeemed = c.fetchone()[0] # Revenue from referrals c.execute(""" SELECT SUM(purchase_amount) FROM referrals WHERE program_id = ? AND status = 'qualified' """, (program_id,)) revenue = c.fetchone()[0] or 0 # Viral coefficient viral_coeff = round(qualified_referrals / total_referrers, 2) if total_referrers > 0 else 0 conn.close() return { "total_referrers": total_referrers, "active_referrers": active_referrers, "total_referrals": total_referrals, "qualified_referrals": qualified_referrals, "conversion_rate": round(qualified_referrals / total_referrals * 100, 1) if total_referrals > 0 else 0, "rewards_issued": rewards_issued, "rewards_redeemed": rewards_redeemed, "revenue_from_referrals": revenue, "viral_coefficient": viral_coeff } def detect_fraud(self, referral_id: str) -> List[Dict]: """Detect potential fraud in referrals.""" conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("SELECT * FROM referrals WHERE id = ?", (referral_id,)) referral = c.fetchone() if not referral: conn.close() return [] flags = [] # Check self-referral (same email/phone as referrer) c.execute("SELECT email, phone FROM referrers WHERE id = ?", (referral[2],)) referrer_info = c.fetchone() if referrer_info: referrer_email, referrer_phone = referrer_info if referral[3] == referrer_email: flags.append({"type": "self_referral", "severity": "high", "detail": "Referee email matches referrer email"}) if referral[4] == referrer_phone: flags.append({"type": "self_referral", "severity": "high", "detail": "Referee phone matches referrer phone"}) # Check for duplicate accounts (same email/phone used multiple times) c.execute(""" SELECT COUNT(*) FROM referrals WHERE referee_email = ? AND id != ? """, (referral[3], referral[0])) if c.fetchone()[0] > 0: flags.append({"type": "duplicate_account", "severity": "medium", "detail": "Referee email used in multiple referrals"}) # Check for rapid referrals from same referrer (bot-like) c.execute(""" SELECT COUNT(*) FROM referrals WHERE referrer_id = ? AND created_at > datetime('now', '-1 hour') """, (referral[2],)) if c.fetchone()[0] > 5: flags.append({"type": "bot_activity", "severity": "high", "detail": "More than 5 referrals in 1 hour"}) # Save flags for flag in flags: flag_id = hashlib.md5(f"flag_{referral_id}{flag['type']}{time.time()}".encode()).hexdigest()[:12] c.execute(""" INSERT INTO referral_fraud_flags (id, referral_id, flag_type, severity, details, status) VALUES (?, ?, ?, ?, ?, 'flagged') """, (flag_id, referral_id, flag["type"], flag["severity"], flag["detail"])) conn.commit() conn.close() return flags def run_referral_program(): print(f"\n{'='*60}") print(f" REFERRAL PROGRAM AUTOMATION") print(f"{'='*60}\n") program = ReferralProgram() conn = sqlite3.connect("/home/ubuntu/GhostAgency/lead_broker.db") c = conn.cursor() c.execute(""" SELECT id, nome, categoria, citta, email, telefono FROM leads_inventory WHERE venduto = 0 AND email IS NOT NULL AND email != '' LIMIT 10 """) leads = c.fetchall() conn.close() if not leads: print("[!] No leads with email found") return print(f"[*] Setting up referral programs for {len(leads)} leads...") total_referrers = 0 total_referrals = 0 for lead in leads: lead_id, nome, categoria, citta, email, telefono = lead client_data = { "client_id": lead_id, "business_name": nome, "category": categoria, "location": citta, "email": email, "phone": telefono } # Create referral program program_id = program.create_program(client_data, { "program_name": f"Referral {nome}", "program_type": "standard", "referrer_reward": {"type": "discount_percent", "value": 20, "max": 30}, "referee_reward": {"type": "discount_percent", "value": 15, "max": 25}, "conditions": {"min_spend": 25, "max_referrals_per_month": 10, "validity_days": 90}, "tracking_method": "code" }) # Register sample referrers (existing customers) referrers_count = random.randint(20, 100) for i in range(referrers_count): referrer_data = { "client_id": lead_id, "customer_id": f"cust_{i}", "first_name": f"Cliente{i}", "last_name": f"Referrer{i}", "email": f"referrer{i}_{lead_id}@example.com", "phone": f"+39 3{random.randint(20, 99)} {random.randint(1000000, 9999999)}" } program.register_referrer(program_id, referrer_data) total_referrers += referrers_count # Simulate some referrals conn = sqlite3.connect("/home/ubuntu/GhostAgency/referral_program.db") c = conn.cursor() c.execute("SELECT id FROM referrers WHERE program_id = ?", (program_id,)) referrer_ids = [row[0] for row in c.fetchall()] conn.close() # Each referrer makes 0-3 referrals for referrer_id in referrer_ids: num_refs = random.randint(0, 3) for _ in range(num_refs): referee_data = { "email": f"referee_{random.randint(1, 10000)}@example.com", "phone": f"+39 3{random.randint(20, 99)} {random.randint(1000000, 9999999)}", "first_name": f"Amico{random.randint(1, 100)}", "last_name": f"Nuovo{random.randint(1, 100)}", "source": random.choice(["whatsapp", "email", "social", "direct_link", "qr"]) } referral_id = program.create_referral(program_id, referrer_id, referee_data) total_referrals += 1 # Qualify some (60% conversion) if random.random() < 0.6: program.qualify_referral(referral_id, random.uniform(30, 150)) # Check stats stats = program.get_program_stats(program_id) print(f" [+] {nome} ({categoria}) - Referrers: {stats['total_referrers']} | Referrals: {stats['total_referrals']} | Qualified: {stats['qualified_referrals']} | Conversion: {stats['conversion_rate']}% | Viral coeff: {stats['viral_coefficient']}") print(f"\n[+] Referral Program setup complete") print(f"[+] Total referrers registered: {total_referrers}") print(f"[+] Total referrals created: {total_referrals}") print(f"[+] Reward types: discount_percent, discount_fixed, free_item, cashback, points, gift_card") print(f"[+] Fraud detection: self-referral, duplicate_account, bot_activity, code_sharing") print(f"[+] Next: Connect WhatsApp/Email/SMS APIs for viral sharing + POS integration for auto-qualification") if __name__ == "__main__": run_referral_program()