""" Genesis Clone Pipeline — QA Checker Module ========================================== Quality assurance checks for cloned websites: 1. Google PageSpeed Insights API (mobile + desktop scores) 2. Basic HTML validation (structure checks) 3. Before/after report generation Google PageSpeed API is free, no auth required for basic use. """ import os import sys import json import time import logging from pathlib import Path from typing import Optional from datetime import datetime import requests logger = logging.getLogger("clone_pipeline.qa_checker") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- PAGESPEED_API_URL = "https://www.googleapis.com/pagespeedonline/v5/runPagespeed" PAGESPEED_API_KEY = os.environ.get("GOOGLE_PAGESPEED_API_KEY", "") # Try genesis secrets loader try: sys.path.insert(0, "/mnt/e/genesis-system/core") from secrets_loader import _get_env if not PAGESPEED_API_KEY: PAGESPEED_API_KEY = _get_env("GOOGLE_PAGESPEED_API_KEY") except Exception: pass TARGET_MOBILE_SCORE = 85 TARGET_DESKTOP_SCORE = 90 # --------------------------------------------------------------------------- # PageSpeed check # --------------------------------------------------------------------------- def check_pagespeed(url: str, strategy: str = "mobile") -> dict: """ Check PageSpeed score for a URL using Google PageSpeed Insights API. Args: url: URL to check strategy: 'mobile' or 'desktop' Returns dict with keys: - score: int (0-100, -1 if check failed) - performance: int - accessibility: int - seo: int - best_practices: int - fcp: float (First Contentful Paint in seconds) - lcp: float (Largest Contentful Paint in seconds) - cls: float (Cumulative Layout Shift) - tbt: float (Total Blocking Time in ms) - error: str or None """ result = { "score": -1, "performance": -1, "accessibility": -1, "seo": -1, "best_practices": -1, "fcp": None, "lcp": None, "cls": None, "tbt": None, "error": None, "url": url, "strategy": strategy, } if not PAGESPEED_API_KEY: result["error"] = "Google PageSpeed API key not found. Skipping PageSpeed check." logger.warning(f"[QA] {result['error']}") return result params = { "url": url, "strategy": strategy, "category": ["performance", "accessibility", "seo", "best-practices"], "key": PAGESPEED_API_KEY, # Always include key if present } logger.info(f"[QA] Checking PageSpeed ({strategy}): {url}") try: resp = requests.get(PAGESPEED_API_URL, params=params, timeout=60) resp.raise_for_status() data = resp.json() categories = data.get("lighthouseResult", {}).get("categories", {}) audits = data.get("lighthouseResult", {}).get("audits", {}) # Extract category scores (0-100) result["performance"] = int( (categories.get("performance", {}).get("score") or 0) * 100 ) result["accessibility"] = int( (categories.get("accessibility", {}).get("score") or 0) * 100 ) result["seo"] = int( (categories.get("seo", {}).get("score") or 0) * 100 ) result["best_practices"] = int( (categories.get("best-practices", {}).get("score") or 0) * 100 ) result["score"] = result["performance"] # Primary score is performance # Web vitals fcp = audits.get("first-contentful-paint", {}).get("numericValue") lcp = audits.get("largest-contentful-paint", {}).get("numericValue") cls = audits.get("cumulative-layout-shift", {}).get("numericValue") tbt = audits.get("total-blocking-time", {}).get("numericValue") result["fcp"] = round(fcp / 1000, 2) if fcp else None # ms -> seconds result["lcp"] = round(lcp / 1000, 2) if lcp else None result["cls"] = round(cls, 3) if cls else None result["tbt"] = round(tbt, 0) if tbt else None # ms logger.info( f"[QA] PageSpeed ({strategy}): Performance={result['performance']}, " f"Accessibility={result['accessibility']}, SEO={result['seo']}" ) except requests.exceptions.Timeout: result["error"] = f"PageSpeed API timeout after 60s" logger.warning(f"[QA] {result['error']}") except requests.exceptions.HTTPError as e: if e.response.status_code in [403, 429]: result["error"] = f"PageSpeed API key issue or rate limit: HTTP {e.response.status_code}. Please ensure GOOGLE_PAGESPEED_API_KEY is valid and API is enabled." logger.warning(f"[QA] {result['error']}") else: result["error"] = f"PageSpeed API HTTP error: {e.response.status_code}" logger.warning(f"[QA] {result['error']}") except Exception as e: result["error"] = f"PageSpeed check failed: {e}" logger.warning(f"[QA] {result['error']}") return result def check_pagespeed_both(url: str) -> dict: """ Run PageSpeed checks for both mobile and desktop. Returns dict with 'mobile' and 'desktop' keys. """ logger.info(f"[QA] Running full PageSpeed audit: {url}") mobile = check_pagespeed(url, strategy="mobile") time.sleep(2) # Avoid rate limits desktop = check_pagespeed(url, strategy="desktop") return { "mobile": mobile, "desktop": desktop, "url": url, "passed": ( mobile["score"] >= TARGET_MOBILE_SCORE or desktop["score"] >= TARGET_DESKTOP_SCORE ), } # --------------------------------------------------------------------------- # HTML validation checks # --------------------------------------------------------------------------- def validate_html(html: str, business_name: str = "", phone: str = "") -> dict: """ Basic structural validation of the generated HTML. Returns dict with: - valid: bool - checks: list of (check_name, passed, message) tuples - score: int (0-100) """ checks = [] def check(name: str, condition: bool, pass_msg: str, fail_msg: str): checks.append({ "name": name, "passed": condition, "message": pass_msg if condition else fail_msg, }) html_lower = html.lower() # Structure checks check("DOCTYPE", "" in html_lower, "DOCTYPE present", "Missing ") check("HTML tag", " tag") check("Head section", "" in html_lower, "Head section present", "Missing section") check("Body section", "" in html_lower, "Body section present", "Missing section") check("Title tag", "" in html_lower, "Title tag present", "Missing <title> tag") check("Meta viewport", 'name="viewport"' in html_lower, "Viewport meta present (mobile-ready)", "Missing viewport meta tag") check("H1 heading", "<h1" in html_lower, "H1 heading present", "Missing H1 heading") check("Tailwind CSS", "tailwindcss" in html_lower or "tailwind" in html_lower, "Tailwind CSS included", "Tailwind CSS not found") # SEO checks check("Meta description", 'name="description"' in html_lower, "Meta description present", "Missing meta description") # Content checks if business_name: check("Business name", business_name.lower() in html_lower, f"Business name '{business_name}' found", f"Business name '{business_name}' not found in HTML") if phone: # Check both formatted and unformatted phone clean_phone = "".join(c for c in phone if c.isdigit()) check("Phone number", phone in html or clean_phone in html, f"Phone {phone} found in HTML", f"Phone {phone} not found in HTML") check("Click-to-call", "tel:" in html_lower, "Click-to-call (tel: link) present", "Missing click-to-call link") # Widget check check("Chatbot widget", "genesis-chat" in html_lower or "genesis_chat" in html_lower or "genesisinit" in html_lower.replace(" ", ""), "Chatbot widget snippet found", "Chatbot widget not injected") # Schema check check("Schema markup", "schema.org" in html_lower, "Schema.org markup present", "Missing schema.org structured data") passed_count = sum(1 for c in checks if c["passed"]) score = int((passed_count / len(checks)) * 100) if checks else 0 valid = score >= 70 # 70%+ is considered valid return { "valid": valid, "checks": checks, "score": score, "passed": passed_count, "total": len(checks), } # --------------------------------------------------------------------------- # Report generation # --------------------------------------------------------------------------- def generate_report( original_url: str, new_url: Optional[str], business_name: str, client_slug: str, pagespeed_scores: Optional[dict] = None, html_validation: Optional[dict] = None, extraction_method: str = "unknown", elapsed_seconds: float = 0, output_dir: Optional[Path] = None, ) -> str: """ Generate a markdown before/after comparison report. Returns the markdown report string, and optionally saves to output_dir. """ timestamp = datetime.now().strftime("%Y-%m-%d %H:%M") # PageSpeed summaries ps_text = "" if pagespeed_scores: mobile = pagespeed_scores.get("mobile", {}) desktop = pagespeed_scores.get("desktop", {}) mobile_score = mobile.get("score", -1) desktop_score = desktop.get("score", -1) mobile_icon = "PASS" if mobile_score >= TARGET_MOBILE_SCORE else ("WARN" if mobile_score >= 50 else "FAIL") desktop_icon = "PASS" if desktop_score >= TARGET_DESKTOP_SCORE else ("WARN" if desktop_score >= 50 else "FAIL") ps_text = f""" ## PageSpeed Scores | Metric | Mobile | Desktop | |--------|--------|---------| | Performance | {mobile_score}/100 [{mobile_icon}] | {desktop_score}/100 [{desktop_icon}] | | Accessibility | {mobile.get('accessibility', '-')}/100 | {desktop.get('accessibility', '-')}/100 | | SEO | {mobile.get('seo', '-')}/100 | {desktop.get('seo', '-')}/100 | | Best Practices | {mobile.get('best_practices', '-')}/100 | {desktop.get('best_practices', '-')}/100 | | FCP | {mobile.get('fcp', '-')}s | {desktop.get('fcp', '-')}s | | LCP | {mobile.get('lcp', '-')}s | {desktop.get('lcp', '-')}s | | CLS | {mobile.get('cls', '-')} | {desktop.get('cls', '-')} | | TBT | {mobile.get('tbt', '-')}ms | {desktop.get('tbt', '-')}ms | Target: Mobile >= {TARGET_MOBILE_SCORE}, Desktop >= {TARGET_DESKTOP_SCORE} Result: {'PASSED' if pagespeed_scores.get('passed') else 'NEEDS IMPROVEMENT'} """ # HTML validation summary val_text = "" if html_validation: passed = html_validation.get("passed", 0) total = html_validation.get("total", 0) score = html_validation.get("score", 0) val_text = f""" ## HTML Validation Score: {score}/100 ({passed}/{total} checks passed) | Check | Status | |-------|--------| """ for c in html_validation.get("checks", []): status = "PASS" if c["passed"] else "FAIL" val_text += f"| {c['name']} | [{status}] {c['message']} |\n" report = f"""# Genesis Website Clone Report Generated: {timestamp} ## Client: {business_name} - **Slug**: {client_slug} - **Original URL**: {original_url} - **New URL**: {new_url or 'Saved locally (no Netlify token)'} - **Extraction Method**: {extraction_method} - **Processing Time**: {elapsed_seconds:.1f} seconds {ps_text} {val_text} ## Delivery Checklist - [ ] Review all business information is accurate - [ ] Phone number is correct and click-to-call works - [ ] Services listed are complete and accurate - [ ] Brand colours are appropriate - [ ] Chatbot widget responds correctly - [ ] Mobile display tested on real device - [ ] Point client's domain to: {new_url or 'local file'} ## Upsell Opportunities - Talking Voice Widget: $197-$597/mo (Telnyx NaturalHD voice AI) - ReceptionistAI: Handles all inbound calls 24/7 from $497/mo - Monthly SEO package: Ongoing content and ranking improvement --- *Report generated by Genesis Clone Pipeline | {timestamp}* """ if output_dir: output_dir.mkdir(parents=True, exist_ok=True) report_path = output_dir / "clone_report.md" report_path.write_text(report, encoding="utf-8") logger.info(f"[QA] Report saved: {report_path}") return report # --------------------------------------------------------------------------- # CLI test # --------------------------------------------------------------------------- if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") url = sys.argv[1] if len(sys.argv) > 1 else "https://example.com" print(f"\nChecking PageSpeed for: {url}\n") scores = check_pagespeed_both(url) print(json.dumps(scores, indent=2)) report = generate_report( original_url=url, new_url="https://test.netlify.app", business_name="Test Business", client_slug="test-business", pagespeed_scores=scores, elapsed_seconds=42.5, ) print("\n" + report)