#!/usr/bin/env python3
"""
Tradie Website Audit Pipeline (V2)
Iterates through tradie leads, visits website, runs audit (Vision + PageSpeed), generates report.
Usage:
python3 scripts/tradie_audit_pipeline.py --limit 10 --offset 0
"""
import os
import sys
import csv
import json
import time
import requests
import re
import argparse
from pathlib import Path
from datetime import datetime
# Add scripts dir to path to import browser_agent
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from browser_agent import BrowserAgent
# Configuration
LEADS_FILE = "/mnt/e/genesis-system/data/LEADS/Manus Email enriched Tradie Leads/ZOHO_SIMPLE_IMPORT_ENRICHED.csv"
OUTPUT_DIR = Path("/mnt/e/genesis-system/data/AUDITS")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
PAGESPEED_API_KEY = os.environ.get("GOOGLE_API_KEY")
def clean_filename(s):
"""Clean string for filename."""
s = str(s).strip().replace("http://", "").replace("https://", "").replace("www.", "")
return re.sub(r'[^\w\-_.]', '_', s)
def get_pagespeed_score(url, strategy="mobile"):
"""Get PageSpeed Insights score."""
if not PAGESPEED_API_KEY:
return None
api_url = f"https://www.googleapis.com/pagespeedonline/v5/runPagespeed?url={url}&strategy={strategy}&category=PERFORMANCE&category=SEO&key={PAGESPEED_API_KEY}"
try:
response = requests.get(api_url, timeout=30)
if response.status_code == 200:
data = response.json()
lighthouse = data.get("lighthouseResult", {})
categories = lighthouse.get("categories", {})
return {
"performance": categories.get("performance", {}).get("score"),
"seo": categories.get("seo", {}).get("score"),
"loading_experience": data.get("loadingExperience", {}).get("overall_category")
}
elif response.status_code == 403:
# Just return None silently or log warning, don't crash
return None
else:
return None
except Exception:
return None
def generate_audit_prompt(business_name, industry):
return f"""
You are an expert Digital Marketing Auditor specializing in Australian Trade Businesses.
Analyze this website screenshot for '{business_name}' ({industry}).
Provide a professional, constructive audit report in Markdown format.
Focus on:
1. **First Impressions (3-second test)**: Does it look trustworthy? Is the service clear?
2. **Conversion Elements**: Phone number visibility, 'Get Quote' buttons, contact forms.
3. **Trust Signals**: Reviews, licenses, insurance, 'About Us', real photos vs stock.
4. **Mobile Experience**: (Infer from layout) Is text readable? Are buttons clickable?
5. **SEO Basics**: Headlines, content relevance, local keywords.
**Output Format:**
# Website Audit for {business_name}
## 🚦 Executive Summary
[1-2 sentences on overall health]
## ✅ What You're Doing Well
* [Point 1]
* [Point 2]
## ⚠️ High-Priority Fixes (The "Money Leaks")
* [Critical Issue 1]
* [Critical Issue 2]
* [Critical Issue 3]
## 💡 Growth Opportunities
[Strategic advice]
## 📱 Mobile & Trust Score
* **Mobile-Friendly:** [Yes/No/Needs Improvement]
* **Trust Factor:** [High/Medium/Low]
Tone: Professional, helpful, direct (no fluff). Speak to a busy tradie.
"""
def generate_html_report(lead, audit_md, pagespeed_data, screenshot_path):
"""Generate HTML report."""
ps_html = ""
if pagespeed_data:
perf = int(pagespeed_data.get('performance', 0) * 100) if pagespeed_data.get('performance') else "N/A"
seo = int(pagespeed_data.get('seo', 0) * 100) if pagespeed_data.get('seo') else "N/A"
def get_color(score):
if score == "N/A": return "gray"
if score >= 90: return "green"
if score >= 50: return "orange"
return "red"
ps_html = f"""
"""
else:
ps_html = """
Performance
N/A
Requires PageSpeed API Key
"""
# Simple Markdown to HTML conversion
html_content = audit_md
# Headers
html_content = re.sub(r'^# (.*)', r'\1
', html_content, flags=re.MULTILINE)
html_content = re.sub(r'^## (.*)', r'\1
', html_content, flags=re.MULTILINE)
html_content = re.sub(r'^### (.*)', r'\1
', html_content, flags=re.MULTILINE)
# Lists
html_content = re.sub(r'^\* (.*)', r'\1', html_content, flags=re.MULTILINE)
# Bold
html_content = re.sub(r'\*\*(.*?)\*\*', r'\1', html_content)
# Wrap lists in (heuristic: sequences of - )
# This is a hacky way to do it without a parser, but sufficient for simple reports
template = f"""
Sunaiva Audit: {lead['Company']}
"""
return template
def main():
parser = argparse.ArgumentParser(description="Tradie Website Audit Pipeline")
parser.add_argument("--limit", type=int, default=5, help="Max audits to run")
parser.add_argument("--offset", type=int, default=0, help="Skip first N leads")
args = parser.parse_args()
print(f"Starting Tradie Audit Pipeline (Limit: {args.limit}, Offset: {args.offset})")
if not os.path.exists(LEADS_FILE):
print(f"Error: Leads file not found at {LEADS_FILE}")
return
agent = BrowserAgent(headless=True)
agent.start()
count = 0
skipped = 0
try:
with open(LEADS_FILE, 'r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f)
rows = list(reader) # Read all to handle offset easily
total_rows = len(rows)
print(f"Found {total_rows} leads in CSV.")
for i, row in enumerate(rows):
if i < args.offset:
continue
if count >= args.limit and args.limit > 0:
break
website = row.get('Website')
email = row.get('Email')
if not website or not email:
continue
if not website.startswith('http'):
url = f"https://{website}"
else:
url = website
company_slug = clean_filename(row['Company'])
company_dir = OUTPUT_DIR / company_slug
if company_dir.exists() and (company_dir / "report.html").exists():
# print(f"Skipping {row['Company']} (already audited)")
skipped += 1
continue
company_dir.mkdir(exist_ok=True)
print(f"\\n[{count+1}/{args.limit}] Processing: {row['Company']} ({url})")
# PageSpeed
ps_data = get_pagespeed_score(url)
# Browser
success = agent.navigate(url)
if not success:
if "www." not in url:
url = url.replace("https://", "https://www.")
print(f" Retrying with {url}...")
success = agent.navigate(url)
if success:
screenshot_filename = "screenshot.png"
screenshot_path = agent.screenshot(str(company_dir / screenshot_filename))
prompt = generate_audit_prompt(row['Company'], row['Industry'])
analysis = agent.analyze_page(prompt, screenshot_path)
report_html = generate_html_report(row, analysis, ps_data, screenshot_path)
(company_dir / "report.html").write_text(report_html)
(company_dir / "audit.md").write_text(analysis)
(company_dir / "lead_data.json").write_text(json.dumps(row, indent=2))
print(f" ✅ Complete: {company_dir}/report.html")
count += 1
else:
print(f" ❌ Failed: {url}")
except KeyboardInterrupt:
print("Stopped by user.")
except Exception as e:
print(f"Pipeline Error: {e}")
finally:
agent.stop()
print(f"\\nPipeline finished. Processed {count} audits (Skipped {skipped} existing).")
if __name__ == "__main__":
main()
