#!/usr/bin/env python3 """ Ghost Agency - AI GMB Optimizer Automated Google My Business optimization: posts, photos, Q&A, hours, attributes, insights """ import os import json import sqlite3 import hashlib import time import random from pathlib import Path from datetime import datetime, timedelta from typing import Dict, List, Any, Optional GMB_DB = Path("/home/ubuntu/GhostAgency") / "ai_gmb_optimizer.db" # ============================================================ # AI GMB OPTIMIZER ENGINE # ============================================================ class AIGMBOptimizer: def __init__(self): self.db_path = str(GMB_DB) self.openrouter_api_key = os.getenv("OPENROUTER_API_KEY", "") self.google_places_api_key = os.getenv("GOOGLE_MAPS_API_KEY", "") self.init_db() def init_db(self): conn = sqlite3.connect(self.db_path) c = conn.cursor() # GMB Locations c.execute("""CREATE TABLE IF NOT EXISTS gmb_locations ( id TEXT PRIMARY KEY, client_id TEXT NOT NULL, business_name TEXT, place_id TEXT, location_name TEXT, address TEXT, phone TEXT, website TEXT, primary_category TEXT, additional_categories TEXT, hours TEXT, special_hours TEXT, attributes TEXT, description TEXT, verified INTEGER DEFAULT 0, verification_method TEXT, access_token TEXT, refresh_token TEXT, last_synced TIMESTAMP, status TEXT DEFAULT 'pending', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # GMB Posts c.execute("""CREATE TABLE IF NOT EXISTS gmb_posts ( id TEXT PRIMARY KEY, location_id TEXT NOT NULL, client_id TEXT NOT NULL, post_type TEXT, -- standard, event, offer, product, covid19 title TEXT, body TEXT, call_to_action TEXT, cta_url TEXT, media_urls TEXT, start_date DATE, end_date DATE, publish_at TIMESTAMP, published_at TIMESTAMP, status TEXT DEFAULT 'draft', language TEXT DEFAULT 'it', views INTEGER DEFAULT 0, clicks INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # GMB Photos c.execute("""CREATE TABLE IF NOT EXISTS gmb_photos ( id TEXT PRIMARY KEY, location_id TEXT NOT NULL, client_id TEXT NOT NULL, photo_type TEXT, -- cover, profile, interior, exterior, food, menu, team, logo, other url TEXT, caption TEXT, is_primary INTEGER DEFAULT 0, uploaded_at TIMESTAMP, views INTEGER DEFAULT 0, source TEXT, -- owner, customer, street_view created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # GMB Q&A c.execute("""CREATE TABLE IF NOT EXISTS gmb_qa ( id TEXT PRIMARY KEY, location_id TEXT NOT NULL, client_id TEXT NOT NULL, question TEXT, answer TEXT, asked_by TEXT, asked_at TIMESTAMP, answered_by TEXT, answered_at TIMESTAMP, is_owner_answer INTEGER DEFAULT 0, upvotes INTEGER DEFAULT 0, status TEXT DEFAULT 'unanswered', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # GMB Reviews (sync) c.execute("""CREATE TABLE IF NOT EXISTS gmb_reviews ( id TEXT PRIMARY KEY, location_id TEXT NOT NULL, client_id TEXT NOT NULL, external_review_id TEXT, author_name TEXT, author_photo_url TEXT, rating INTEGER, text TEXT, language TEXT, published_at TIMESTAMP, reply_text TEXT, replied_at TIMESTAMP, replied_by TEXT, is_replied INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # GMB Insights c.execute("""CREATE TABLE IF NOT EXISTS gmb_insights ( id TEXT PRIMARY KEY, location_id TEXT NOT NULL, client_id TEXT NOT NULL, date DATE, metric_name TEXT, metric_value INTEGER, period TEXT, -- day, week, month, quarter created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Optimization Tasks c.execute("""CREATE TABLE IF NOT EXISTS gmb_optimization_tasks ( id TEXT PRIMARY KEY, location_id TEXT NOT NULL, client_id TEXT NOT NULL, task_type TEXT, -- post, photo, qa, hours, attributes, description, category priority INTEGER, title TEXT, description TEXT, current_state TEXT, recommended_action TEXT, estimated_impact TEXT, status TEXT DEFAULT 'pending', assigned_to TEXT, completed_at TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") # Competitor Tracking c.execute("""CREATE TABLE IF NOT EXISTS gmb_competitors ( id TEXT PRIMARY KEY, location_id TEXT NOT NULL, client_id TEXT NOT NULL, competitor_place_id TEXT, competitor_name TEXT, distance_meters INTEGER, rating REAL, review_count INTEGER, categories TEXT, last_analyzed TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""") conn.commit() conn.close() self._init_post_templates() def _init_post_templates(self): conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("SELECT COUNT(*) FROM gmb_posts WHERE id LIKE 'template_%'") if c.fetchone()[0] > 0: conn.close() return templates = [ { "id": "template_offer_weekend", "post_type": "offer", "title": "Offerta Weekend: {offerta_speciale}", "body": "Solo questo weekend da {business_name}: {dettaglio_offerta}!\n\nPrenota ora: {cta_url}\n\nValido: {data_inizio} - {data_fine}", "cta": "PRENOTA", "category": "ristorante" }, { "id": "template_new_dish", "post_type": "standard", "title": "Nuovo Piatto: {nome_piatto}", "body": "Da {business_name} arriva il nuovo {nome_piatto}: {descrizione}.\n\nVieni a provarlo!\n\n{cta_url}", "cta": "SCOPRI_DI_PIU", "category": "ristorante" }, { "id": "template_event_live_music", "post_type": "event", "title": "Serata Live Music: {nome_artista}", "body": "Questa {giorno} sera musica dal vivo da {business_name} con {nome_artista}!\n\nInizio: {ora_inizio}\nPrenota il tuo tavolo: {cta_url}", "cta": "PRENOTA", "category": "bar" }, { "id": "template_seasonal_menu", "post_type": "standard", "title": "Menu Stagionale: {stagione}", "body": "Il nuovo menu {stagione} e arrivato da {business_name}!\n\nIngredienti di stagione, piatti nuovi, stessi sapori autentici.\n\nVedi il menu: {cta_url}", "cta": "VISITA_SITO_WEB", "category": "ristorante" }, { "id": "template_happy_hour", "post_type": "offer", "title": "Happy Hour: {sconto}% su tutti i cocktail", "body": "Ogni giorno dalle {ora_inizio} alle {ora_fine}: {sconto}% su tutti i cocktail e aperitivi!\n\nTi aspettiamo da {business_name}.\n\n{cta_url}", "cta": "OTTIENI_OFFERTA", "category": "bar" }, { "id": "template_gusto_settimana", "post_type": "standard", "title": "Gusto della Settimana: {gusto}", "body": "Questa settimana da {business_name}: {gusto} - {descrizione}.\n\nFatto come una volta, ingredienti veri.\n\nVieni ad assaggiarlo: {cta_url}", "cta": "VISITA_SITO_WEB", "category": "gelateria" }, { "id": "template_new_coffee", "post_type": "standard", "title": "Nuova Miscela: {nome_miscela}", "body": "Da {business_name} arriva la nuova miscela {nome_miscela}: {note_gusto}.\n\n100% Arabica, tostata artigianalmente.\n\nProvala oggi: {cta_url}", "cta": "VISITA_SITO_WEB", "category": "caffetteria" }, { "id": "template_appointment_reminder", "post_type": "standard", "title": "Prenota il tuo {servizio}", "body": "Hai bisogno di {servizio}? Da {business_name} trovi professionalita e qualita.\n\nPrenota ora: {cta_url}\n\nTelefono: {telefono}", "cta": "PRENOTA", "category": "parrucchiere" } ] for tmpl in templates: c.execute(""" INSERT INTO gmb_posts (id, location_id, client_id, post_type, title, body, call_to_action, cta_url, status, created_at) VALUES (?, 'template', 'template', ?, ?, ?, ?, '', 'template', CURRENT_TIMESTAMP) """, ( tmpl["id"], tmpl["post_type"], tmpl["title"], tmpl["body"], tmpl["cta"] )) conn.commit() conn.close() def register_location(self, client_data: Dict) -> str: location_id = hashlib.md5(f"gmb_{client_data.get('client_id', '')}{time.time()}".encode()).hexdigest()[:12] conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" INSERT INTO gmb_locations (id, client_id, business_name, address, phone, website, primary_category, additional_categories, hours, description, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending') """, ( location_id, client_data.get("client_id", ""), client_data.get("business_name", ""), client_data.get("address", ""), client_data.get("phone", ""), client_data.get("website", ""), client_data.get("category", ""), "", "", f"{client_data.get('business_name', '')} a {client_data.get('location', 'Parma')}. Qualita, tradizione e passione." )) conn.commit() conn.close() return location_id def generate_posts_for_location(self, location_data: Dict, weeks: int = 4) -> List[Dict]: location_id = location_data.get("location_id", "") client_id = location_data.get("client_id", "") business_name = location_data.get("business_name", "") category = location_data.get("category", "").lower() location = location_data.get("location", "Parma") posts = [] # Get templates for this category conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute("SELECT * FROM gmb_posts WHERE client_id='template' AND (category=? OR category='')", (category,)) templates = c.fetchall() conn.close() if not templates: # Use default templates templates = [ ("template", "standard", "Novita da {business_name}", "Scopri le novita da {business_name}! {cta_url}", "VISITA_SITO_WEB", "") ] # Generate posts for next N weeks start_date = datetime.now().date() for week in range(weeks): week_start = start_date + timedelta(weeks=week) # 2-3 posts per week posts_per_week = random.randint(2, 3) for i in range(posts_per_week): day_offset = random.randint(0, 6) post_date = week_start + timedelta(days=day_offset) template = random.choice(templates) post_type = template[1] # Fill template variables title = template[2].format( business_name=business_name, location=location, offerta_speciale="2x1 su antipasti", dettaglio_offerta="Prendi 2 antipasti, paghi 1", nome_piatto="Risotto ai funghi porcini", descrizione="Cremoso, profumato, autentico", nome_artista="Marco Rossi", giorno="venerdi", ora_inizio="20:00", stagione="autunnale", sconto="20", ora_fine="22:00", gusto="Pistacchio di Bronte", nome_miscela="Ethiopia Sidamo", note_gusto="note floreali e agrumate", servizio="taglio e piega", telefono="+39 378 088 3157" ) body = template[3].format( business_name=business_name, location=location, offerta_speciale="2x1 su antipasti", dettaglio_offerta="Prendi 2 antipasti, paghi 1", nome_piatto="Risotto ai funghi porcini", descrizione_piatto="Cremoso, profumato, autentico", nome_artista="Marco Rossi", giorno="venerdi", ora_inizio="20:00", stagione="autunnale", sconto="20", ora_fine="22:00", gusto="Pistacchio di Bronte", descrizione_gusto="Crema di pistacchio DOP, granella croccante", nome_miscela="Ethiopia Sidamo", note_gusto="note floreali e agrumate", servizio="taglio e piega", telefono="+39 378 088 3157", cta_url=f"https://{business_name.lower().replace(' ', '')}.it" ) post_id = hashlib.md5(f"post_{location_id}{post_date}{i}{time.time()}".encode()).hexdigest()[:12] post = { "id": post_id, "location_id": location_id, "client_id": client_id, "post_type": post_type, "title": title, "body": body, "call_to_action": template[4], "cta_url": f"https://{business_name.lower().replace(' ', '')}.it", "publish_at": post_date.isoformat(), "status": "scheduled", "language": "it" } posts.append(post) # Save to DB conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" INSERT INTO gmb_posts (id, location_id, client_id, post_type, title, body, call_to_action, cta_url, publish_at, status, language) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'scheduled', 'it') """, ( post_id, location_id, client_id, post_type, title, body, template[4], f"https://{business_name.lower().replace(' ', '')}.it", post_date.isoformat() )) conn.commit() conn.close() return posts def generate_qa_for_location(self, location_data: Dict) -> List[Dict]: location_id = location_data.get("location_id", "") client_id = location_data.get("client_id", "") business_name = location_data.get("business_name", "") category = location_data.get("category", "").lower() address = location_data.get("address", "") phone = location_data.get("phone", "") qa_pairs = [ { "question": f"Quali sono gli orari di apertura di {business_name}?", "answer": f"{business_name} e aperto: Lun-Ven 12:00-15:00 e 19:00-23:00, Sab 19:00-23:30, Dom 12:00-15:00 e 19:00-22:00. Chiuso il lunedi a pranzo." }, { "question": f"Dove si trova {business_name}?", "answer": f"Ci trovi in {address}, {location_data.get('location', 'Parma')}. Telefono: {phone}" }, { "question": f"Accettate prenotazioni?", "answer": "Si, accettiamo prenotazioni telefoniche al {phone} e online sul nostro sito web. Consigliamo di prenotare soprattutto per il weekend." }, { "question": f"Avete opzioni vegetariane/vegane/senza glutine?", "answer": "Certamente! Il nostro menu include diverse opzioni vegetariane, vegane e senza glutine. Il nostro staff e a disposizione per consigliarvi al meglio." }, { "question": f"C'e parcheggio vicino a {business_name}?", "answer": "Si, c'e parcheggio pubblico a 50 metri e parcheggio privato convenzionato (chiedi allo staff)." }, { "question": f"Fate consegne a domicilio o takeaway?", "answer": "Si, offriamo sia takeaway (ritiro in loco) che consegna a domicilio tramite i nostri partner. Ordina online o chiamaci al {phone}." }, { "question": f"Accettate pagamenti con carta?", "answer": "Si, accettiamo contanti, carte di credito/debito, bancomat, Satispay e pagamenti contactless." }, { "question": f"E' adatto per bambini/famiglie?", "answer": "Assolutamente si! Abbiamo menu bambini, seggioloni e spazio per passeggini. Le famiglie sono benvenute." } ] saved = [] for qa in qa_pairs: qa_id = hashlib.md5(f"qa_{location_id}{qa['question']}{time.time()}".encode()).hexdigest()[:12] conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" INSERT INTO gmb_qa (id, location_id, client_id, question, answer, asked_by, asked_at, answered_by, answered_at, is_owner_answer, status) VALUES (?, ?, ?, ?, ?, 'AI Assistant', CURRENT_TIMESTAMP, 'Owner', CURRENT_TIMESTAMP, 1, 'answered') """, ( qa_id, location_id, client_id, qa["question"], qa["answer"] )) conn.commit() conn.close() saved.append({"id": qa_id, **qa}) return saved def create_optimization_tasks(self, location_data: Dict) -> List[Dict]: location_id = location_data.get("location_id", "") client_id = location_data.get("client_id", "") business_name = location_data.get("business_name", "") tasks = [ { "task_type": "photo", "priority": 9, "title": "Aggiungere foto copertina e profilo", "description": "Manca foto copertina (1080x608) e foto profilo (250x250). Queste sono le prime immagini che vedono i clienti.", "current_state": "Nessuna foto copertina/profilo", "recommended_action": "Carica foto professionali: esterno insegna, interno sala, piatti signature, team", "estimated_impact": "high" }, { "task_type": "photo", "priority": 8, "title": "Caricare 20+ foto: interni, esterni, cibo, team", "description": "Le schede con 20+ foto ricevono 42% piu richieste indicazioni e 35% piu click al sito.", "current_state": "Poche/nessuna foto", "recommended_action": "Programma shooting: 5 esterni, 5 interni, 5 piatti, 3 team, 2 dettagli", "estimated_impact": "high" }, { "task_type": "hours", "priority": 9, "title": "Verificare e aggiornare orari (incl. festivita)", "description": "Orari errati = clienti persi e recensioni negative. Aggiorna per Natale, Capodanno, Pasqua, ferie estive.", "current_state": "Orari base impostati, festivita mancanti", "recommended_action": "Imposta orari speciali per tutte le festivita 2024-2025", "estimated_impact": "high" }, { "task_type": "attributes", "priority": 8, "title": "Impostare attributi: WiFi, parcheggio, accessibilita, cani ammessi, ecc.", "description": "Gli attributi aiutano i clienti a decidere velocemente e migliorano il ranking locale.", "current_state": "Attributi non impostati", "recommended_action": "Seleziona tutti gli attributi applicabili: WiFi gratis, parcheggio, accessibile sedie a rotelle, cani ammessi all'esterno, adatto bambini, menu bambini, takeaway, consegna", "estimated_impact": "medium" }, { "task_type": "description", "priority": 7, "title": "Ottimizzare descrizione attivita (750 caratteri)", "description": "Descrizione ricca di keyword locali (categoria + citta + quartiere) migliora SEO locale.", "current_state": "Descrizione generica/breve", "recommended_action": "Riscrivi includendo: categoria, specialita, citta, quartiere, anni di attivita, filosofia, parole chiave naturali", "estimated_impact": "medium" }, { "task_type": "category", "priority": 8, "title": "Verificare categoria primaria e aggiungere categorie secondarie", "description": "Categoria primaria corretta = ranking corretto. Aggiungi 3-5 categorie secondarie rilevanti.", "current_state": "Solo categoria primaria impostata", "recommended_action": "Imposta primaria corretta + secondarie: es. Ristorante italiano (primaria) + Ristorante pesce + Ristorante carne + Pizzeria + Ristorante per famiglie", "estimated_impact": "high" }, { "task_type": "qa", "priority": 7, "title": "Popolare Q&A con 8-10 domande frequenti", "description": "Le risposte del proprietario alle FAQ appaiono in evidenza e riducono chiamate telefoniche.", "current_state": "Q&A vuoto", "recommended_action": "Rispondi a: orari, indirizzo, prenotazioni, menu speciali, parcheggio, pagamenti, bambini, consegna, accessibilita, eventi", "estimated_impact": "medium" }, { "task_type": "post", "priority": 6, "title": "Programmare post settimanali (2-3/settimana)", "description": "Post regolari = engagement + ranking. Alterna: offerte, novita, eventi, dietro le quinte, recensioni clienti.", "current_state": "Nessun post programmato", "recommended_action": "Usa template: lunedi offerta, mercoledi novita, venerdi evento/weekend", "estimated_impact": "medium" }, { "task_type": "review", "priority": 9, "title": "Rispondere a TUTTE le recensioni (positive e negative)", "description": "Risposte del proprietario = fiducia + ranking. Rispondi entro 24-48h. Ringrazia per positive, risolvi per negative.", "current_state": "Recensioni senza risposta", "recommended_action": "Attiva alert email per nuove recensioni. Template risposte pronte per 1-5 stelle", "estimated_impact": "high" } ] saved = [] for task in tasks: task_id = hashlib.md5(f"task_{location_id}{task['title']}{time.time()}".encode()).hexdigest()[:12] conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" INSERT INTO gmb_optimization_tasks (id, location_id, client_id, task_type, priority, title, description, current_state, recommended_action, estimated_impact, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending') """, ( task_id, location_id, client_id, task["task_type"], task["priority"], task["title"], task["description"], task["current_state"], task["recommended_action"], task["estimated_impact"] )) conn.commit() conn.close() saved.append({"id": task_id, **task}) return saved def analyze_competitors(self, location_data: Dict, radius_meters: int = 2000) -> List[Dict]: location_id = location_data.get("location_id", "") client_id = location_data.get("client_id", "") category = location_data.get("category", "") location = location_data.get("location", "Parma") # Simulated competitors competitors = [ {"name": f"Concorrente {i} {category.title()}", "distance": random.randint(100, radius_meters), "rating": round(random.uniform(3.5, 4.8), 1), "reviews": random.randint(50, 500), "categories": [category.title(), f"{category.title()} tradizionale", "Ristorante italiano"]} for i in range(1, 6) ] saved = [] for comp in competitors: comp_id = hashlib.md5(f"comp_{location_id}{comp['name']}{time.time()}".encode()).hexdigest()[:12] conn = sqlite3.connect(self.db_path) c = conn.cursor() c.execute(""" INSERT INTO gmb_competitors (id, location_id, client_id, competitor_name, distance_meters, rating, review_count, categories, last_analyzed) VALUES (?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) """, ( comp_id, location_id, client_id, comp["name"], comp["distance"], comp["rating"], comp["reviews"], json.dumps(comp["categories"]) )) conn.commit() conn.close() saved.append({"id": comp_id, **comp}) return saved def run_ai_gmb_optimizer(): print(f"\n{'='*60}") print(f" AI GMB OPTIMIZER") print(f"{'='*60}\n") optimizer = AIGMBOptimizer() conn = sqlite3.connect("/home/ubuntu/GhostAgency/lead_broker.db") c = conn.cursor() c.execute(""" SELECT id, nome, categoria, citta, indirizzo, 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"[*] Optimizing GMB for {len(leads)} leads...") total_posts = 0 total_qa = 0 total_tasks = 0 total_competitors = 0 for lead in leads: lead_id, nome, categoria, citta, indirizzo, telefono = lead location_data = { "client_id": lead_id, "business_name": nome, "category": categoria, "location": citta, "address": indirizzo, "phone": telefono } location_id = optimizer.register_location(location_data) location_data["location_id"] = location_id # Generate posts posts = optimizer.generate_posts_for_location(location_data, weeks=4) total_posts += len(posts) # Generate Q&A qa = optimizer.generate_qa_for_location(location_data) total_qa += len(qa) # Create optimization tasks tasks = optimizer.create_optimization_tasks(location_data) total_tasks += len(tasks) # Analyze competitors competitors = optimizer.analyze_competitors(location_data) total_competitors += len(competitors) print(f" [+] {nome} ({categoria}) - Posts: {len(posts)} | Q&A: {len(qa)} | Tasks: {len(tasks)} | Competitors: {len(competitors)}") print(f"\n[+] AI GMB Optimizer setup complete") print(f"[+] Total posts scheduled: {total_posts}") print(f"[+] Total Q&A created: {total_qa}") print(f"[+] Total optimization tasks: {total_tasks}") print(f"[+] Total competitors tracked: {total_competitors}") print(f"[+] Next: Connect Google My Business API for live deployment") if __name__ == "__main__": run_ai_gmb_optimizer()