#!/usr/bin/env python3 """ Ghost Agency - AI Ads Manager (Google Ads + Meta Ads) Automated local ads management: keyword research, ad copy, bidding, A/B test, reporting """ import os import json import sqlite3 import hashlib import time import random import requests from pathlib import Path from datetime import datetime, timedelta from typing import Dict, List, Any, Optional from dataclasses import dataclass, asdict DB_FILE = Path("/home/ubuntu/GhostAgency") / "revenue_bots.db" ADS_DB = Path("/home/ubuntu/GhostAgency") / "ai_ads_manager.db" # ============================================================ # AI ADS MANAGER ENGINE # ============================================================ class AIAdsManager: def __init__(self): self.db_path = str(ADS_DB) self.openrouter_api_key = os.getenv("OPENROUTER_API_KEY", "") self.google_ads_client_id = os.getenv("GOOGLE_ADS_CLIENT_ID", "") self.google_ads_client_secret = os.getenv("GOOGLE_ADS_CLIENT_SECRET", "") self.google_ads_refresh_token = os.getenv("GOOGLE_ADS_REFRESH_TOKEN", "") self.google_ads_developer_token = os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN", "") self.meta_app_id = os.getenv("META_APP_ID", "") self.meta_app_secret = os.getenv("META_APP_SECRET", "") self.meta_access_token = os.getenv("META_ACCESS_TOKEN", "") self.init_db() def init_db(self): conn = sqlite3.connect(self.db_path) c = conn.cursor() # Ad Accounts c.execute("""CREATE TABLE IF NOT EXISTS ad_accounts ( id TEXT PRIMARY KEY, client_id TEXT NOT NULL, business_name TEXT, platform TEXT, -- google, meta account_id TEXT, access_token TEXT, refresh_token TEXT, currency TEXT DEFAULT 'EUR', timezone TEXT DEFAULT 'Europe/Rome', status TEXT DEFAULT 'active', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Campaigns c.execute("""CREATE TABLE IF NOT EXISTS campaigns ( id TEXT PRIMARY KEY, account_id TEXT NOT NULL, client_id TEXT NOT NULL, platform TEXT, campaign_name TEXT, campaign_type TEXT, -- search, display, shopping, video, social objective TEXT, -- conversions, traffic, awareness, leads status TEXT DEFAULT 'paused', -- active, paused, removed daily_budget REAL, bid_strategy TEXT, -- maximize_conversions, target_cpa, target_roas, manual_cpc target_cpa REAL, target_roas REAL, start_date DATE, end_date DATE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (account_id) REFERENCES ad_accounts(id) )""") # Ad Groups c.execute("""CREATE TABLE IF NOT EXISTS ad_groups ( id TEXT PRIMARY KEY, campaign_id TEXT NOT NULL, client_id TEXT NOT NULL, ad_group_name TEXT, status TEXT DEFAULT 'paused', bid_amount REAL, target_cpa REAL, target_roas REAL, keywords TEXT, -- JSON array negative_keywords TEXT, -- JSON array audience_targeting TEXT, -- JSON created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (campaign_id) REFERENCES campaigns(id) )""") # Ads/Creatives c.execute("""CREATE TABLE IF NOT EXISTS ads_creatives ( id TEXT PRIMARY KEY, ad_group_id TEXT NOT NULL, client_id TEXT NOT NULL, platform TEXT, ad_type TEXT, -- text, responsive_search, image, video, carousel, collection headline_1 TEXT, headline_2 TEXT, headline_3 TEXT, description_1 TEXT, description_2 TEXT, description_3 TEXT, display_url TEXT, final_url TEXT, image_urls TEXT, -- JSON array video_url TEXT, call_to_action TEXT, status TEXT DEFAULT 'draft', -- draft, pending_review, approved, disapproved, paused quality_score REAL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (ad_group_id) REFERENCES ad_groups(id) )""") # Keywords c.execute("""CREATE TABLE IF NOT EXISTS keywords ( id TEXT PRIMARY KEY, ad_group_id TEXT NOT NULL, client_id TEXT NOT NULL, keyword TEXT, match_type TEXT, -- exact, phrase, broad, broad_match_modifier bid_amount REAL, quality_score REAL, ctr REAL, cpc REAL, impressions INTEGER, clicks INTEGER, conversions REAL, cost REAL, status TEXT DEFAULT 'active', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (ad_group_id) REFERENCES ad_groups(id) )""") # Performance Metrics (daily) c.execute("""CREATE TABLE IF NOT EXISTS performance_metrics ( id TEXT PRIMARY KEY, entity_id TEXT, -- campaign, ad_group, ad, keyword entity_type TEXT, -- campaign, ad_group, ad, keyword client_id TEXT NOT NULL, date DATE, impressions INTEGER DEFAULT 0, clicks INTEGER DEFAULT 0, cost REAL DEFAULT 0, conversions REAL DEFAULT 0, conversion_value REAL DEFAULT 0, ctr REAL DEFAULT 0, cpc REAL DEFAULT 0, cpa REAL DEFAULT 0, roas REAL DEFAULT 0, quality_score REAL, impression_share REAL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # A/B Tests c.execute("""CREATE TABLE IF NOT EXISTS ab_tests ( id TEXT PRIMARY KEY, client_id TEXT NOT NULL, test_name TEXT, test_type TEXT, -- ad_copy, headline, description, image, cta, landing_page, audience control_id TEXT, variant_ids TEXT, -- JSON array metric TEXT, -- ctr, cpa, roas, conversion_rate status TEXT DEFAULT 'running', -- running, completed, stopped confidence_level REAL DEFAULT 0.95, min_sample_size INTEGER DEFAULT 100, start_date DATE, end_date DATE, winner_id TEXT, results TEXT, -- JSON created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Optimization Rules c.execute("""CREATE TABLE IF NOT EXISTS optimization_rules ( id TEXT PRIMARY KEY, client_id TEXT NOT NULL, rule_name TEXT, trigger_type TEXT, -- schedule, threshold, performance trigger_config TEXT, -- JSON action_type TEXT, -- pause, enable, adjust_bid, adjust_budget, add_negative, create_ad action_config TEXT, -- JSON status TEXT DEFAULT 'active', last_run TIMESTAMP, run_count INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Negative Keywords c.execute("""CREATE TABLE IF NOT EXISTS negative_keywords ( id TEXT PRIMARY KEY, campaign_id TEXT, ad_group_id TEXT, client_id TEXT NOT NULL, keyword TEXT, match_type TEXT, -- exact, phrase, broad level TEXT, -- campaign, ad_group source TEXT, -- search_terms, manual, ai_suggested added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Search Terms Report c.execute("""CREATE TABLE IF NOT EXISTS search_terms ( id TEXT PRIMARY KEY, campaign_id TEXT, ad_group_id TEXT, client_id TEXT NOT NULL, search_term TEXT, match_type TEXT, impressions INTEGER, clicks INTEGER, cost REAL, conversions REAL, ctr REAL, cpc REAL, cpa REAL, is_negative INTEGER DEFAULT 0, date DATE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Budgets c.execute("""CREATE TABLE IF NOT EXISTS budget_pacing ( id TEXT PRIMARY KEY, campaign_id TEXT NOT NULL, client_id TEXT NOT NULL, daily_budget REAL, monthly_budget REAL, spend_to_date REAL DEFAULT 0, projected_spend REAL DEFAULT 0, pacing_status TEXT DEFAULT 'on_track', -- on_track, under_pacing, over_pacing date DATE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Client Packages c.execute("""CREATE TABLE IF NOT EXISTS ads_packages ( id TEXT PRIMARY KEY, client_id TEXT NOT NULL, package_name TEXT, platform TEXT, -- google, meta, both management_fee_pct REAL, -- 15-20% setup_fee REAL, min_monthly_spend REAL, max_monthly_spend REAL, included_services TEXT, -- JSON array status TEXT DEFAULT 'active', start_date DATE, end_date DATE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") conn.commit() conn.close() def register_client(self, client_data: Dict) -> str: """Register a client for ads management.""" client_id = hashlib.md5(f"{client_data.get('business_name', '')}{time.time()}".encode()).hexdigest()[:12] conn = sqlite3.connect(self.db_path) c = conn.cursor() # Create package package_id = hashlib.md5(f"package_{client_id}{time.time()}".encode()).hexdigest()[:12] c.execute(""" INSERT INTO ads_packages (id, client_id, package_name, platform, management_fee_pct, setup_fee, min_monthly_spend, max_monthly_spend, included_services, status, start_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', date('now')) """, ( package_id, client_id, client_data.get("package_name", "Gestione Ads Completa"), client_data.get("platform", "both"), client_data.get("management_fee_pct", 0.15), client_data.get("setup_fee", 200), client_data.get("min_monthly_spend", 300), client_data.get("max_monthly_spend", 5000), json.dumps(client_data.get("included_services", [ "keyword_research", "ad_copy_creation", "bid_management", "ab_testing", "negative_keywords", "weekly_reports", "monthly_optimization" ])) )) conn.commit() conn.close() return client_id def create_google_search_campaign(self, client_data: Dict, campaign_config: Dict) -> Dict: """Create a Google Search campaign structure.""" client_id = client_data.get("client_id", "") business_name = client_data.get("business_name", "") category = client_data.get("category", "") location = client_data.get("location", "Parma") # Generate keywords based on category keywords = self._generate_keywords(category, location, business_name) # Generate ad copy using AI ad_copies = self._generate_ad_copies(business_name, category, location, keywords[:10]) campaign_id = hashlib.md5(f"camp_{client_id}{time.time()}".encode()).hexdigest()[:12] ad_group_id = hashlib.md5(f"ag_{campaign_id}{time.time()}".encode()).hexdigest()[:12] campaign = { "id": campaign_id, "client_id": client_id, "platform": "google", "campaign_name": f"{business_name} - Search - {location}", "campaign_type": "search", "objective": "conversions", "status": "draft", "daily_budget": campaign_config.get("daily_budget", 20), "bid_strategy": "maximize_conversions", "target_cpa": campaign_config.get("target_cpa", 30), "start_date": datetime.now().date().isoformat(), "ad_groups": [{ "id": ad_group_id, "ad_group_name": f"{category} - {location}", "status": "draft", "keywords": keywords, "ads": ad_copies, "negative_keywords": self._get_negative_keywords(category) }] } # Save to DB self._save_campaign(campaign) return campaign def create_meta_campaign(self, client_data: Dict, campaign_config: Dict) -> Dict: """Create a Meta (Facebook/Instagram) campaign structure.""" client_id = client_data.get("client_id", "") business_name = client_data.get("business_name", "") category = client_data.get("category", "") location = client_data.get("location", "Parma") # Generate ad creatives creatives = self._generate_meta_creatives(business_name, category, location) # Generate audiences audiences = self._generate_meta_audiences(category, location) campaign_id = hashlib.md5(f"meta_camp_{client_id}{time.time()}".encode()).hexdigest()[:12] ad_set_id = hashlib.md5(f"adset_{campaign_id}{time.time()}".encode()).hexdigest()[:12] campaign = { "id": campaign_id, "client_id": client_id, "platform": "meta", "campaign_name": f"{business_name} - Social - {location}", "campaign_type": "social", "objective": "conversions", "status": "draft", "daily_budget": campaign_config.get("daily_budget", 15), "bid_strategy": "lowest_cost_without_cap", "ad_sets": [{ "id": ad_set_id, "ad_set_name": f"{category} - {location} - Broad", "status": "draft", "targeting": audiences, "daily_budget": campaign_config.get("daily_budget", 15), "ads": creatives }] } return campaign def _generate_keywords(self, category: str, location: str, business_name: str) -> List[Dict]: """Generate keyword list for a category.""" base_keywords = { "ristorante": [ "ristorante {location}", "ristorante italiano {location}", "cena {location}", "pranzo {location}", "ristorante centro {location}", "miglior ristorante {location}", "ristorante pesce {location}", "ristorante carne {location}", "ristorante romantico {location}", "ristorante famiglia {location}", "ristorante consegna {location}", "takeaway {location}" ], "pizzeria": [ "pizzeria {location}", "pizza napoletana {location}", "pizza da asporto {location}", "pizzeria centro {location}", "miglior pizza {location}", "pizza a domicilio {location}", "pizzeria aperta cena {location}", "pizzeria aperta pranzo {location}", "pizza senza glutine {location}" ], "gelateria": [ "gelateria {location}", "gelato artigianale {location}", "gelato {location}", "miglior gelato {location}", "gelateria centro {location}", "gelato consegna {location}" ], "caffetteria": [ "caffetteria {location}", "caffè {location}", "cappuccino {location}", "colazione {location}", "café {location}", "specialty coffee {location}" ], "bar": [ "bar {location}", "aperitivo {location}", "cocktail {location}", "locale notturno {location}", "happy hour {location}", "birra {location}" ], "parrucchiere": [ "parrucchiere {location}", "taglio capelli {location}", "colore capelli {location}", "parrucchiere donna {location}", "parrucchiere uomo {location}", "acconciature {location}" ], "centro_estetico": [ "centro estetico {location}", "estetista {location}", "trattamento viso {location}", "massaggio {location}", "epilazione {location}", "manicure {location}", "pedicure {location}" ], "autofficina": [ "autofficina {location}", "meccanico {location}", "tagliando {location}", "revisione {location}", "gomme {location}", "riparazione auto {location}" ], "dentista": [ "dentista {location}", "sbiancamento denti {location}", "ortodonzia {location}", "implantologia {location}", "igiene orale {location}", "emergenza dentista {location}" ] } templates = base_keywords.get(category.lower(), [ "{category} {location}", "{business_name} {location}", "{category} centro {location}" ]) keywords = [] for template in templates: kw = template.format( location=location, business_name=business_name, category=category ) keywords.append({ "keyword": kw, "match_type": "phrase", "bid_amount": None }) return keywords[:30] def _generate_ad_copies(self, business_name: str, category: str, location: str, keywords: List[str]) -> List[Dict]: """Generate ad copy variations using AI/templates.""" # Templates by category templates = { "ristorante": [ { "headlines": ["{name} - Ristorante a {location}", "Cucina Italiana Autentica", "Prenota Il Tuo Tavolo"], "descriptions": ["Ingredienti freschi, piatti tradizionali. Menu stagionale, cantina selezionata.", "Ambiente accogliente, servizio impeccabile. Aperto pranzo e cena. Parcheggio vicino."], "cta": "Prenota Ora" }, { "headlines": ["Miglior Ristorante {location}", "{name} - Tradizione & Qualità", "Menu Stagionale"], "descriptions": ["Scopri i nostri piatti signature. Pasta fatta in casa, secondi di carne e pesce.", "Prenotazione online facile e veloce. Offerte speciali per gruppi e eventi."], "cta": "Vedi Menu" } ], "pizzeria": [ { "headlines": ["Pizza Napoletana a {location}", "{name} - Verace Pizza", "Impasto 24h Lievitazione"], "descriptions": ["Pomodoro San Marzano DOP, Mozzarella di Bufala, forno a legna 450°C.", "Da asporto e consegna. Ordina online, pronto in 20 min."], "cta": "Ordina Ora" }, { "headlines": ["Miglior Pizza {location}", "{name} - Tradizione Napoletana", "Ingredienti DOP"], "descriptions": ["Margherita, Diavola, Capricciosa e speciali della casa. Impasto leggero e digeribile.", "Aperto tutti i giorni. Parcheggio riservato clienti."], "cta": "Scopri Menu" } ], "gelateria": [ { "headlines": ["Gelato Artigianale {location}", "{name} - Fatto Ogni Giorno", "Ingredienti Naturali"], "descriptions": ["Senza coloranti né aromi artificiali. Latte fresco, frutta di stagione, nocciole IGP.", "Coni, coppette, vaschette da asporto. Nuovi gusti ogni settimana."], "cta": "Vieni a Provarlo" } ], "default": [ { "headlines": ["{name} - {category} a {location}", "Qualità e Professionalità", "Prenota Ora"], "descriptions": ["Servizio eccellente, prezzi onesti. Anni di esperienza al tuo servizio.", "Contattaci per info e prenotazioni. Aperti tutti i giorni."], "cta": "Contattaci" } ] } cat_templates = templates.get(category.lower(), templates["default"]) ads = [] for i, tmpl in enumerate(cat_templates): ad = { "headline_1": tmpl["headlines"][0].format(name=business_name, location=location, category=category), "headline_2": tmpl["headlines"][1].format(name=business_name, location=location, category=category) if len(tmpl["headlines"]) > 1 else "", "headline_3": tmpl["headlines"][2].format(name=business_name, location=location, category=category) if len(tmpl["headlines"]) > 2 else "", "description_1": tmpl["descriptions"][0].format(name=business_name, location=location, category=category), "description_2": tmpl["descriptions"][1].format(name=business_name, location=location, category=category) if len(tmpl["descriptions"]) > 1 else "", "display_url": f"{business_name.lower().replace(' ', '')}.it", "final_url": f"https://{business_name.lower().replace(' ', '')}.it", "call_to_action": tmpl["cta"], "ad_type": "responsive_search" } ads.append(ad) return ads def _generate_meta_creatives(self, business_name: str, category: str, location: str) -> List[Dict]: """Generate Meta ad creatives.""" creatives = [] # Image ad creatives.append({ "ad_type": "image", "primary_text": f"������� Cerchi il miglior {category.lower()} a {location}?\n\nDa {business_name} trovi qualità, tradizione e passione in ogni piatto.\n\n��� Ingredienti freschi\n��� Menu stagionale\n��� Ambiente accogliente\n\n���� {location} centro\n���� Aperti tutti i giorni", "headline": f"{business_name} - {category.title()} a {location}", "description": "Prenota ora il tuo tavolo", "call_to_action": "BOOK_TRAVEL", "image_prompt": f"Professional photo of {category} interior, welcoming atmosphere, {location} style" }) # Video ad creatives.append({ "ad_type": "video", "primary_text": f"���� Scopri {business_name} in 30 secondi!\n\nDalla cucina al tuo tavolo: ingredienti freschi, passione, tradizione.\n\n���� Guarda il video e prenota", "headline": f"{business_name} - Esperienza {category.title()}", "description": "Guarda il video", "call_to_action": "WATCH_MORE", "video_prompt": f"Short video tour of {category} {business_name}, food preparation, happy customers, {location} atmosphere" }) # Carousel ad creatives.append({ "ad_type": "carousel", "primary_text": f"���� I nostri bestseller da {business_name}:\n\nSwipe per vedere i piatti più amati! ������", "headline": "I Più Amati", "description": "Scopri i preferiti dei clienti", "call_to_action": "SHOP_NOW", "cards": [ {"headline": "Piatto Signature", "description": "Il nostro cavallo di battaglia", "image_prompt": f"Best selling dish at {business_name}"}, {"headline": "Specialità della Casa", "description": "Tradizione e innovazione", "image_prompt": f"House specialty at {business_name}"}, {"headline": "Menu Stagionale", "description": "Ingredienti di stagione", "image_prompt": f"Seasonal menu at {business_name}"}, {"headline": "Prenota Ora", "description": "Tavolo riservato per te", "image_prompt": f"Reservation booking at {business_name}"} ] }) return creatives def _generate_meta_audiences(self, category: str, location: str) -> List[Dict]: """Generate Meta audience targeting.""" audiences = [] # Broad audience audiences.append({ "name": f"Broad - {location} 25-55", "type": "broad", "location": location, "age_min": 25, "age_max": 55, "interests": [] }) # Interest-based interest_map = { "ristorante": ["Cucina italiana", "Ristoranti", "Cibo e bevande", "Cena", "Gastronomia", "Vino"], "pizzeria": ["Pizza", "Cucina napoletana", "Cibo da asporto", "Ristoranti italiani"], "gelateria": ["Gelato", "Dolci", "Dessert", "Gelateria artigianale"], "caffetteria": ["Caffè", "Cappuccino", "Colazione", "Specialty coffee", "Caffetteria"], "bar": ["Cocktail", "Aperitivo", "Vita notturna", "Bar", "Happy hour"], "parrucchiere": ["Parrucchiere", "Taglio capelli", "Colore capelli", "Acconciature", "Bellezza"], "centro_estetico": ["Estetica", "Trattamenti viso", "Massaggi", "Epilazione", "Benessere", "Spa"], "autofficina": ["Auto", "Manutenzione auto", "Riparazione auto", "Meccanico", "Gomme"], "dentista": ["Dentista", "Igiene orale", "Sbiancamento denti", "Ortodonzia", "Salute dentale"] } interests = interest_map.get(category.lower(), ["Piccole imprese", "Servizi locali"]) audiences.append({ "name": f"Interest - {category} - {location}", "type": "interest", "location": location, "age_min": 25, "age_max": 55, "interests": interests }) # Lookalike (would need pixel data) audiences.append({ "name": f"Lookalike - Clienti - {location}", "type": "lookalike", "source": "pixel_purchasers", "location": location, "percentage": 1 }) # Retargeting audiences.append({ "name": f"Retargeting - Website Visitors 30d - {location}", "type": "custom", "source": "website_visitors_30d", "location": location }) return audiences def _get_negative_keywords(self, category: str) -> List[str]: """Get negative keywords for category.""" base_negatives = [ "gratis", "gratuito", "free", "economico", "low cost", "scontato", "offerta", "lavoro", "assunzioni", "cerco lavoro", "stage", "tirocinio", "apprendistato", "corso", "formazione", "scuola", "impara", "tutorial", "come fare", "fai da te", "ricetta", "ingredienti", "cucinare", "preparare", "fatto in casa", "usato", "seconda mano", "vendita privata", "subito.it", "bakeca", "porno", "adulti", "xxx", "sex", "erotico", "incontri", "dating", "truffa", "falso", "fake", "clone", "replica", "imitazione" ] category_negatives = { "ristorante": ["ricette", "cucina casalinga", "chef a domicilio", "catering matrimoni"], "pizzeria": ["pizza surgelata", "pizza fatta in casa", "forno pizza casa"], "gelateria": ["gelato confezionato", "gelato supermercato", "ricetta gelato casa"], "parrucchiere": ["taglio capelli casa", "tinta capelli casa", "parrucchiere fai da te"], "centro_estetico": ["creme viso supermercato", "fai da te viso", "epilazione casa"], "autofficina": ["riparazione auto fai da te", "meccanico fai da te", "pezzi usati"], "dentista": ["rimedi naturali denti", "sbiancamento casa", "dentista fai da te"] } return base_negatives + category_negatives.get(category.lower(), []) def _save_campaign(self, campaign: Dict): """Save campaign structure to DB.""" conn = sqlite3.connect(self.db_path) c = conn.cursor() # Save campaign c.execute(""" INSERT OR REPLACE INTO campaigns (id, account_id, client_id, platform, campaign_name, campaign_type, objective, status, daily_budget, bid_strategy, target_cpa, start_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( campaign["id"], f"acc_{campaign['client_id']}", campaign["client_id"], campaign["platform"], campaign["campaign_name"], campaign["campaign_type"], campaign["objective"], campaign["status"], campaign["daily_budget"], campaign["bid_strategy"], campaign.get("target_cpa"), campaign["start_date"] )) # Save ad groups for ag in campaign.get("ad_groups", []): c.execute(""" INSERT OR REPLACE INTO ad_groups (id, campaign_id, client_id, ad_group_name, status, keywords, negative_keywords) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( ag["id"], campaign["id"], campaign["client_id"], ag["ad_group_name"], ag["status"], json.dumps(ag["keywords"]), json.dumps(ag.get("negative_keywords", [])) )) # Save ads for ad in ag.get("ads", []): ad_id = hashlib.md5(f"ad_{ag['id']}{time.time()}".encode()).hexdigest()[:12] c.execute(""" INSERT OR REPLACE INTO ads_creatives (id, ad_group_id, client_id, platform, ad_type, headline_1, headline_2, headline_3, description_1, description_2, display_url, final_url, call_to_action, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'draft') """, ( ad_id, ag["id"], campaign["client_id"], campaign["platform"], ad.get("ad_type", "responsive_search"), ad.get("headline_1", ""), ad.get("headline_2", ""), ad.get("headline_3", ""), ad.get("description_1", ""), ad.get("description_2", ""), ad.get("display_url", ""), ad.get("final_url", ""), ad.get("call_to_action", "") )) conn.commit() conn.close() def create_optimization_rules(self, client_id: str) -> List[Dict]: """Create default optimization rules for a client.""" rules = [ { "rule_name": "Pause High CPA Keywords", "trigger_type": "threshold", "trigger_config": json.dumps({ "metric": "cpa", "condition": ">", "value": 50, "lookback_days": 14, "min_clicks": 20 }), "action_type": "pause", "action_config": json.dumps({"entity": "keyword"}), "status": "active" }, { "rule_name": "Increase Bid High ROAS Keywords", "trigger_type": "threshold", "trigger_config": json.dumps({ "metric": "roas", "condition": ">", "value": 4, "lookback_days": 14, "min_conversions": 5 }), "action_type": "adjust_bid", "action_config": json.dumps({"adjustment_pct": 15, "max_bid": 10}), "status": "active" }, { "rule_name": "Add Negative Keywords from Search Terms", "trigger_type": "schedule", "trigger_config": json.dumps({ "frequency": "daily", "time": "06:00" }), "action_type": "add_negative", "action_config": json.dumps({ "min_impressions": 10, "max_ctr": 0.005, "min_cost": 10, "max_conversions": 0 }), "status": "active" }, { "rule_name": "Budget Pacing Check", "trigger_type": "schedule", "trigger_config": json.dumps({ "frequency": "daily", "time": "09:00" }), "action_type": "adjust_budget", "action_config": json.dumps({ "target_pacing": 1.0, "adjustment_limit_pct": 20 }), "status": "active" }, { "rule_name": "Pause Underperforming Ads", "trigger_type": "threshold", "trigger_config": json.dumps({ "metric": "ctr", "condition": "<", "value": 0.01, "lookback_days": 30, "min_impressions": 1000 }), "action_type": "pause", "action_config": json.dumps({"entity": "ad"}), "status": "active" } ] conn = sqlite3.connect(self.db_path) c = conn.cursor() for rule in rules: rule_id = hashlib.md5(f"rule_{client_id}{rule['rule_name']}{time.time()}".encode()).hexdigest()[:12] c.execute(""" INSERT INTO optimization_rules (id, client_id, rule_name, trigger_type, trigger_config, action_type, action_config, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( rule_id, client_id, rule["rule_name"], rule["trigger_type"], rule["trigger_config"], rule["action_type"], rule["action_config"], rule["status"] )) conn.commit() conn.close() return rules # ============================================================ # EXECUTION # ============================================================ def run_ai_ads_manager(): """Run AI Ads Manager for all active clients.""" print(f"\n{'='*60}") print(f" AI ADS MANAGER - GOOGLE & META") print(f"{'='*60}\n") manager = AIAdsManager() # Get clients from leads broker conn = sqlite3.connect("/home/ubuntu/GhostAgency/lead_broker.db") c = conn.cursor() c.execute(""" SELECT id, nome, categoria, citta, indirizzo 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 ads campaigns for {len(leads)} leads...") for lead in leads: lead_id, nome, categoria, citta, indirizzo = lead client_data = { "client_id": lead_id, "business_name": nome, "category": categoria, "location": citta, "address": indirizzo, "package_name": "Gestione Ads Completa", "platform": "both", "management_fee_pct": 0.15, "setup_fee": 200, "min_monthly_spend": 300, "max_monthly_spend": 2000 } # Register client manager.register_client(client_data) # Create Google Search campaign google_campaign = manager.create_google_search_campaign(client_data, { "daily_budget": 15, "target_cpa": 25 }) # Create Meta campaign meta_campaign = manager.create_meta_campaign(client_data, { "daily_budget": 12 }) # Create optimization rules manager.create_optimization_rules(lead_id) print(f" [+] {nome} ({categoria}) - Google: {len(google_campaign['ad_groups'][0]['keywords'])} kw, {len(google_campaign['ad_groups'][0]['ads'])} ads | Meta: {len(meta_campaign['ad_sets'][0]['ads'])} creatives, {len(meta_campaign['ad_sets'][0]['targeting'])} audiences") print(f"\n[+] AI Ads Manager setup complete") print(f"[+] Campaigns ready for launch (status: draft)") print(f"[+] Optimization rules active") print(f"[+] Next: Connect Google Ads API + Meta Marketing API for live deployment") if __name__ == "__main__": run_ai_ads_manager()