""" Genesis Domain Intelligence — Single Domain Audit Uses DataForSEO Domain Analytics API (pay-as-you-go, ~$0.01-0.05 per domain) Usage: python audit_domain.py sunaivadigital.com python audit_domain.py tradiechatbots.com.au Requirements: pip install requests psycopg2-binary DataForSEO account: https://dataforseo.com """ import sys import os import json import requests from datetime import datetime # Add genesis-memory to path for Elestio config sys.path.append('E:/genesis-system/data/genesis-memory') # DataForSEO credentials — set in environment or replace below DATAFORSEO_LOGIN = os.environ.get("DATAFORSEO_LOGIN", "") DATAFORSEO_PASSWORD = os.environ.get("DATAFORSEO_PASSWORD", "") DATAFORSEO_BASE = "https://api.dataforseo.com/v3" def audit_domain_whois(domain: str) -> dict: """Fetch Whois data: registrar, dates, nameservers.""" payload = [{"target": domain}] try: response = requests.post( f"{DATAFORSEO_BASE}/domain_analytics/whois/overview/live", auth=(DATAFORSEO_LOGIN, DATAFORSEO_PASSWORD), json=payload, timeout=10 ) response.raise_for_status() data = response.json() tasks = data.get("tasks", []) if tasks and tasks[0].get("result"): return tasks[0]["result"][0] or {} except Exception as e: print(f" [WHOIS ERROR] {domain}: {e}") return {} def audit_domain_technologies(domain: str) -> dict: """Fetch technology stack: CMS, analytics, CRM, e-commerce, chat.""" payload = [{"target": domain, "limit": 100}] try: response = requests.post( f"{DATAFORSEO_BASE}/domain_analytics/technologies/domain_technologies/live", auth=(DATAFORSEO_LOGIN, DATAFORSEO_PASSWORD), json=payload, timeout=10 ) response.raise_for_status() data = response.json() tasks = data.get("tasks", []) if tasks and tasks[0].get("result"): return tasks[0]["result"][0] or {} except Exception as e: print(f" [TECH ERROR] {domain}: {e}") return {} def detect_opportunities(tech_data: dict, whois_data: dict) -> list: """ Scan tech stack for revenue gaps Genesis can fill. Returns list of opportunity dicts with severity + pitch + solution. """ opportunities = [] tech_str = json.dumps(tech_data).lower() # AI Voice Receptionist (primary Genesis product) voice_signals = ["telnyx", "twilio", "vapi", "air.ai", "bland.ai"] has_voice = any(sig in tech_str for sig in voice_signals) if not has_voice: opportunities.append({ "type": "no_ai_receptionist", "severity": "critical", "pitch": "No AI voice receptionist detected. Calls going to voicemail = missed revenue. " "10 missed calls/day x $60/call = $600/day = $16,800/month lost.", "solution": "ReceptionistAI — from $497/month. Setup fee waived for launch clients.", "revenue_impact": 16800 }) # CRM crm_signals = ["gohighlevel", "hubspot", "salesforce", "zoho", "pipedrive", "activecampaign"] has_crm = any(sig in tech_str for sig in crm_signals) if not has_crm: opportunities.append({ "type": "missing_crm", "severity": "high", "pitch": "No CRM detected. Every lead that doesn't book immediately is lost forever. " "Automated follow-up sequences typically recover 20-30% of lost leads.", "solution": "GHL CRM + automation — included in ReceptionistAI Business tier ($697/month).", "revenue_impact": 3000 }) # Analytics analytics_signals = ["google-analytics", "gtm", "hotjar", "microsoft-clarity", "matomo"] has_analytics = any(sig in tech_str for sig in analytics_signals) if not has_analytics: opportunities.append({ "type": "missing_analytics", "severity": "medium", "pitch": "No web analytics detected. No visibility on where leads come from or where they drop off.", "solution": "Analytics setup + Netlify deployment — included in Business tier.", "revenue_impact": 500 }) # Google Reviews automation reviews_signals = ["trustpilot", "birdeye", "podium", "grade.us"] has_reviews = any(sig in tech_str for sig in reviews_signals) if not has_reviews: opportunities.append({ "type": "no_reviews_automation", "severity": "medium", "pitch": "No automated review collection detected. Businesses with 50+ Google reviews " "get 3-5x more calls than those with fewer than 10.", "solution": "Google Reviews automation — included in ReceptionistAI Enterprise tier ($997/month).", "revenue_impact": 2000 }) return opportunities def calculate_audit_score(opportunities: list) -> int: """Composite health score 0-100. Starts at 100, deducted per issue.""" score = 100 for opp in opportunities: if opp["severity"] == "critical": score -= 35 elif opp["severity"] == "high": score -= 20 elif opp["severity"] == "medium": score -= 10 elif opp["severity"] == "low": score -= 5 return max(0, score) def audit_domain(domain: str) -> dict: """ Full domain intelligence audit. Returns structured result ready for PostgreSQL insert and report generation. """ print(f"Auditing {domain}...") # Strip http/https/www if present clean_domain = domain.replace("https://", "").replace("http://", "").replace("www.", "").rstrip("/") whois_data = audit_domain_whois(clean_domain) tech_data = audit_domain_technologies(clean_domain) opportunities = detect_opportunities(tech_data, whois_data) audit_score = calculate_audit_score(opportunities) result = { "domain": clean_domain, "audit_date": datetime.utcnow().isoformat(), "whois": whois_data, "technologies": tech_data, "opportunities": opportunities, "audit_score": audit_score, "total_revenue_at_risk": sum(o.get("revenue_impact", 0) for o in opportunities), "critical_issues": sum(1 for o in opportunities if o["severity"] == "critical"), "high_issues": sum(1 for o in opportunities if o["severity"] == "high"), "medium_issues": sum(1 for o in opportunities if o["severity"] == "medium"), } print(f" Score: {audit_score}/100 | Issues: {len(opportunities)} | " f"Revenue at risk: ${result['total_revenue_at_risk']:,}/month") return result def save_audit_to_db(audit_result: dict): """Save audit result to Elestio PostgreSQL.""" try: from elestio_config import PostgresConfig import psycopg2 conn = psycopg2.connect(**PostgresConfig.get_connection_params()) cur = conn.cursor() cur.execute(""" INSERT INTO domain_audits (domain, audit_date, full_tech_stack, opportunities, audit_score, raw_dataforseo_response) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (domain) DO UPDATE SET audit_date = EXCLUDED.audit_date, full_tech_stack = EXCLUDED.full_tech_stack, opportunities = EXCLUDED.opportunities, audit_score = EXCLUDED.audit_score, raw_dataforseo_response = EXCLUDED.raw_dataforseo_response """, ( audit_result["domain"], audit_result["audit_date"], json.dumps(audit_result.get("technologies", {})), json.dumps(audit_result.get("opportunities", [])), audit_result.get("audit_score", 0), json.dumps(audit_result) )) conn.commit() cur.close() conn.close() print(f" Saved to PostgreSQL.") except Exception as e: print(f" [DB WARNING] Could not save to PostgreSQL: {e}") print(f" Result saved to local JSON instead.") # Fallback: save to local JSON on E: drive output_path = f"E:/genesis-system/tools/domain_audit/results/{audit_result['domain']}.json" os.makedirs(os.path.dirname(output_path), exist_ok=True) with open(output_path, "w") as f: json.dump(audit_result, f, indent=2) if __name__ == "__main__": if not DATAFORSEO_LOGIN: print("ERROR: Set DATAFORSEO_LOGIN and DATAFORSEO_PASSWORD environment variables.") print(" Get credentials at: https://dataforseo.com") sys.exit(1) domain = sys.argv[1] if len(sys.argv) > 1 else "sunaivadigital.com" result = audit_domain(domain) save_audit_to_db(result) print("\n--- AUDIT RESULT ---") print(json.dumps(result, indent=2))