"""Orchestrator: runs configured modules per site; scheduler keeps it running.""" from __future__ import annotations import datetime import json from pathlib import Path import yaml from . import audit, geo, content, report, fingerprint, citations from .db import get_db def load_config(path: str = "config.yaml") -> dict: return yaml.safe_load(Path(path).read_text(encoding="utf-8")) def run_site_audit(site: dict, cfg: dict, log=print) -> dict: mods = cfg["modules"] crawler = audit.SiteCrawler(cfg) root = site["url"].rstrip("/") db = get_db() prev_started = None prev = db.q("SELECT started_at FROM crawls WHERE site=? ORDER BY id DESC LIMIT 1", (site["name"],)) if prev: prev_started = prev[0][0] crawl_id = db.new_crawl(site["name"]) log(f"→ Crawling {root} (max {cfg['crawl']['max_pages']} pages)...") pages = crawler.discover(root) log(f" found {len(pages)} pages") # ---- change intelligence: fingerprint every page ---- for u, resp in pages: db.save_fingerprint(crawl_id, u, fingerprint.extract_fingerprint(u, resp.text, resp.status_code)) changes = fingerprint.diff_and_record(db, site["name"], crawl_id, prev_started) change_summary = fingerprint.summarize_changes(changes) if change_summary.get("headline"): log(f"⚠️ {change_summary['headline']}") r: dict = { "site": site, "generated_at": datetime.datetime.now().isoformat(timespec="seconds"), "pages_crawled": len(pages), } # ---- llms.txt presence (used by geo audit) ---- llms_resp = crawler.fetch(root + "/llms.txt") llms_exists = llms_resp is not None and llms_resp.status_code == 200 # ---- technical ---- if mods.get("technical_seo_audit"): log(f" Checking technical SEO on all {len(pages)} page(s) " f"(titles, meta tags, canonical, schema…)") page_audits = [audit.audit_page(u, resp) for u, resp in pages] r["technical"] = { "pages": [{"url": p.url, "score": p.score, "issues": p.issues, "warnings": p.warnings, "data": p.data} for p in page_audits], "avg_score": round(sum(p.score for p in page_audits) / max(len(page_audits), 1), 1), } n_issues = sum(len(p.issues) for p in page_audits) log(f" found {n_issues} technical issue(s) across the site") if mods.get("sitemap_check"): log(" Checking sitemap.xml…") r["sitemap"] = audit.check_sitemap(crawler, root) if mods.get("robots_check"): log(" Checking robots.txt…") r["robots"] = audit.check_robots(crawler, root) # ---- GEO / AI readiness ---- if mods.get("geo_ai_audit"): log(" Checking AI readiness — this is free, built-in, and separate " "from the optional paid AI citation tracker below") geo_pages = [geo.audit_ai_readiness(u, resp.text, llms_exists, is_homepage=(u.rstrip("/") == root.rstrip("/"))) for u, resp in pages] r["geo"] = { "pages": geo_pages, "avg_score": round(sum(p["score"] for p in geo_pages) / max(len(geo_pages), 1), 1), } log(f" AI readiness score: {r['geo']['avg_score']}/100") # ---- content ---- if mods.get("content_analysis"): log(" Checking content quality and keyword coverage…") c_pages = [content.analyze_content(u, resp.text, site.get("target_keywords", [])) for u, resp in pages] cov = [] for p in c_pages: hit = sum(1 for k in p["keywords"] if k["verdict"] != "missing") cov.append(100 * hit / max(len(p["keywords"]), 1)) r["content"] = {"pages": c_pages, "avg_score": round(sum(cov) / max(len(cov), 1), 1)} # ---- lead gen ---- if mods.get("lead_gen_audit"): log(" Checking lead-generation signals (contact forms, click-to-call, reviews…)") l_pages = [content.audit_lead_gen(u, resp.text) for u, resp in pages] r["lead_gen"] = {"pages": l_pages, "avg_score": round(sum(p["score"] for p in l_pages) / max(len(l_pages), 1), 1)} # ---- generators ---- gen_dir = Path(cfg["reports"]["output_dir"]) / "generated" gen_dir.mkdir(parents=True, exist_ok=True) slug = site["name"].lower().replace(" ", "-") if mods.get("llms_txt_generator") and r.get("technical"): page_meta = [{"url": p["url"], "title": p["data"].get("title", ""), "meta_description": p["data"].get("meta_description", "")} for p in r["technical"]["pages"]] (gen_dir / f"{slug}-llms.txt").write_text( geo.generate_llms_txt(site, page_meta), encoding="utf-8") r["generated_llms_txt"] = str(gen_dir / f"{slug}-llms.txt") if mods.get("structured_data_generator"): (gen_dir / f"{slug}-organization-schema.json").write_text( geo.generate_org_schema(site), encoding="utf-8") r["generated_org_schema"] = str(gen_dir / f"{slug}-organization-schema.json") # ---- scoring + diff ---- for k in ("technical", "geo", "content", "lead_gen"): r.setdefault(k, {"avg_score": 0, "pages": []}) r["overall_score"] = report.overall_score(r, cfg["scoring"]) # ---- broken links, for the dead_links auto-fixer ---- broken = [p["url"].rstrip("/") for p in r["technical"]["pages"] if any(i.startswith("HTTP 4") or i.startswith("HTTP 5") for i in p.get("issues", []))] broken_path = Path(cfg["reports"]["output_dir"]) / "broken_links.json" broken_path.parent.mkdir(parents=True, exist_ok=True) broken_path.write_text(json.dumps(broken), encoding="utf-8") crawled = [p["url"].rstrip("/") for p in r["technical"]["pages"]] crawled_path = Path(cfg["reports"]["output_dir"]) / "crawled_urls.json" crawled_path.write_text(json.dumps(crawled), encoding="utf-8") db.finish_crawl(crawl_id, r["overall_score"], len(pages)) r["changes"] = {"summary": change_summary, "items": [c for c in changes if c["severity"] in ("critical", "warning")][:25]} # ---- AI citation tracking (optional, needs API keys) ---- cit = cfg.get("citation_tracking", {}) if cit.get("enabled") and cit.get("run_with_audit"): log("→ Running AI citation checks…") citations.run_citation_checks(db, site, cit, log) r["citations"] = db.citation_summary(site["name"]) return r def run_all(config_path: str = "config.yaml") -> None: cfg = load_config(config_path) for site in cfg["sites"]: r = run_site_audit(site, cfg) files = report.save_reports(r, cfg["reports"]["output_dir"], cfg["reports"]["formats"], cfg["reports"].get("keep_history", 30)) if cfg["scheduler"].get("report_diff"): prev = report.load_previous(cfg["reports"]["output_dir"], site["name"]) if prev: r["diff"] = report.diff_reports(prev, r) if (cfg["scheduler"].get("alert_on_regression") and r["diff"]["overall"]["delta"] < -5): print(f"⚠️ REGRESSION: {site['name']} overall score dropped " f"{r['diff']['overall']['delta']} points!") print(f"✓ {site['name']}: overall {r['overall_score']}/100 " f"(tech {r['technical']['avg_score']}, " f"AI {r['geo']['avg_score']}, " f"content {r['content']['avg_score']}, " f"leads {r['lead_gen']['avg_score']})") for f in files: print(f" report: {f}") def run_scheduler(config_path: str = "config.yaml") -> None: """Continuous mode — full audits + quick drift checks on your cadence.""" from apscheduler.schedulers.blocking import BlockingScheduler cfg = load_config(config_path) sched = BlockingScheduler() sched.add_job(run_all, "interval", hours=cfg["scheduler"]["audit_interval_hours"], args=[config_path], next_run_time=datetime.datetime.now()) print(f"Scheduler started — full audit every " f"{cfg['scheduler']['audit_interval_hours']}h. Ctrl+C to stop.") try: sched.start() except (KeyboardInterrupt, SystemExit): print("Scheduler stopped.") def main(): import argparse p = argparse.ArgumentParser(prog="seo-ai-optimizer", description="White-hat SEO + AI-search (GEO) optimizer") p.add_argument("command", choices=["audit", "schedule", "generate"], help="audit = run once; schedule = run continuously; " "generate = only produce llms.txt & schema files") p.add_argument("--config", default="config.yaml") args = p.parse_args() if args.command == "audit": run_all(args.config) elif args.command == "schedule": run_scheduler(args.config) elif args.command == "generate": cfg = load_config(args.config) for site in cfg["sites"]: gen_dir = Path(cfg["reports"]["output_dir"]) / "generated" gen_dir.mkdir(parents=True, exist_ok=True) slug = site["name"].lower().replace(" ", "-") (gen_dir / f"{slug}-organization-schema.json").write_text( geo.generate_org_schema(site), encoding="utf-8") print(f"wrote {gen_dir}/{slug}-organization-schema.json") if __name__ == "__main__": main()