# Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai """Orchestrateur ka4 — version plus large / agressive / capable. Nouveautés vs ka2 : - Crawling CONCURRENT : fetch (Firecrawl/Scrapfly) + extraction (Haiku) en parallèle via un pool de threads ; les écritures SQLite restent sérialisées sur le thread principal (une seule connexion) pour rester sûres. - Frontière AGRESSIVE : chaque page auto-alimente la file avec les liens découverts (nouveaux domaines priorisés) pour élargir la couverture en continu. - Contrôleur agentique sur un modèle PLUS PUISSANT (Sonnet), extraction sur Haiku. """ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any from urllib.parse import urlparse from anthropic import Anthropic from .config import Config from .extractor import Extractor from .llm_controller import Controller from .scraper import Scraper from .storage import Store _SKIP_HOSTS = ("facebook.", "instagram.", "twitter.", "x.com", "linkedin.", "youtube.", "tiktok.", "pinterest.", "google.", "maps.", "apple.", "amazon.") _SKIP_EXT = (".jpg", ".jpeg", ".png", ".pdf", ".zip", ".mp4", ".svg", ".gif", ".webp", ".css", ".js", ".xml", ".ico", ".woff", ".woff2") class Orchestrator: def __init__(self, cfg: Config): cfg.validate() self.cfg = cfg self.anthropic = Anthropic(api_key=cfg.anthropic_api_key) self.fc = Scraper(cfg) self.store = Store(cfg.db_path) # extraction sur le modèle rapide ; contrôleur sur cfg.model (plus puissant) self.extractor = Extractor(self.anthropic, cfg.extract_model) self._pages_scraped = 0 self._domain_fails: dict[str, int] = {} # coupe-circuit par domaine (robustesse) # -- fetch pur (THREAD-SAFE : réseau + LLM, aucun accès DB) ------------- def _fetch_only(self, url: str) -> dict[str, Any]: data = self.fc.scrape(url) if data.get("blocked"): return {"url": url, "blocked": True} if data.get("error"): return {"url": url, "error": data["error"]} md = data.get("markdown", "") or "" meta = data.get("metadata", {}) or {} title = meta.get("title", "") or "" graph = (self.extractor.extract_graph(url, title, md) if md.strip() else {"entities": [], "relations": []}) return {"url": url, "title": title, "backend": data.get("backend"), "markdown": md, "links": data.get("links", []) or [], "graph": graph} # -- persistance (THREAD PRINCIPAL uniquement) ------------------------- def _save_result(self, r: dict[str, Any]) -> dict[str, Any]: url = r["url"] if r.get("blocked"): self.store.log_event("info", "robots.txt interdit", url) return {"entities_found": 0, "relations_found": 0, "blocked": True} if r.get("error"): self._domain_fails[urlparse(url).netloc.lower()] = \ self._domain_fails.get(urlparse(url).netloc.lower(), 0) + 1 self.store.log_event("error", r["error"], url) return {"entities_found": 0, "relations_found": 0, "error": r["error"]} self.store.record_source(url, r.get("title", ""), r.get("markdown", "")) self.store.log_event("scrape", f"{r.get('title') or url} [{r.get('backend')}]", url, {"chars": len(r.get("markdown", ""))}) graph = r.get("graph", {}) or {} # robustesse : le LLM peut occasionnellement renvoyer un item non-dict -> on filtre entities = [e for e in (graph.get("entities") or []) if isinstance(e, dict)] relations = [x for x in (graph.get("relations") or []) if isinstance(x, dict)] idmap: dict[str, int] = {} for e in entities: eid = self.store.upsert_entity(e, source_url=url) if e.get("temp_id"): idmap[e["temp_id"]] = eid rel = 0 for rr in relations: fid, tid = idmap.get(rr.get("from")), idmap.get(rr.get("to")) if fid and tid: self.store.add_relation(fid, tid, rr.get("type", ""), rr.get("role", ""), source_url=url, confidence=float(rr.get("confidence", 0.6) or 0.6)) rel += 1 self._enqueue_frontier(url, r.get("links", [])) self.store.log_event("extract", f"{len(idmap)} entité(s), {rel} relation(s)", url, {"relations": rel}) return {"entities_found": len(idmap), "relations_found": rel, "links_on_page": (r.get("links", []) or [])[:25]} def _enqueue_frontier(self, url: str, links: list) -> None: if self.cfg.frontier_per_page <= 0 or not links: return base = urlparse(url).netloc.lower() picked = [] for l in links: u = l["url"] if isinstance(l, dict) else l if not isinstance(u, str) or not u.startswith("http"): continue h = urlparse(u).netloc.lower() if not h or any(s in h for s in _SKIP_HOSTS) or u.lower().endswith(_SKIP_EXT): continue picked.append(u) # priorise les NOUVEAUX domaines (élargit la couverture) picked.sort(key=lambda u: urlparse(u).netloc.lower() == base) self.store.enqueue(picked[: self.cfg.frontier_per_page]) # -- crawling CONCURRENT ---------------------------------------------- def _tripped(self, url: str) -> bool: """Coupe-circuit : True si le domaine a trop échoué dans ce run.""" return self._domain_fails.get(urlparse(url).netloc.lower(), 0) >= self.cfg.domain_fail_threshold def crawl_batch(self, urls: list[str]) -> dict[str, Any]: seen, todo = set(), [] for u in urls: if not u or u in seen: continue seen.add(u) if self._tripped(u): continue # domaine coupé (trop d'échecs) if not self.store.source_seen(u): todo.append(u) remaining = max(0, self.cfg.max_pages_per_run - self._pages_scraped) todo = todo[:remaining] if not todo: return {"pages": 0, "entities_found": 0, "relations_found": 0} results = [] with ThreadPoolExecutor(max_workers=max(1, self.cfg.concurrency)) as ex: futs = [ex.submit(self._fetch_only, u) for u in todo] for f in as_completed(futs): try: results.append(f.result()) except Exception as e: # noqa: BLE001 self.store.log_event("error", f"fetch: {e}", "") tot_e = tot_r = pages = 0 for r in results: out = self._save_result(r) if not (r.get("blocked") or r.get("error")): self._pages_scraped += 1 pages += 1 tot_e += out.get("entities_found", 0) tot_r += out.get("relations_found", 0) return {"pages": pages, "entities_found": tot_e, "relations_found": tot_r} # -- scrape unitaire (utilisé par l'agent) ---------------------------- def _scrape_and_extract(self, url: str) -> dict[str, Any]: if self._pages_scraped >= self.cfg.max_pages_per_run: return {"stopped": "limite de pages atteinte", "entities_found": 0} if self.store.source_seen(url): return {"skipped": "déjà scrapé", "entities_found": 0} r = self._fetch_only(url) out = self._save_result(r) if not (r.get("blocked") or r.get("error")): self._pages_scraped += 1 return {**out, "url": url, "title": r.get("title", ""), "backend": r.get("backend")} # -- crawling centré-entité (concurrent) ------------------------------ _DEEP_KEYWORDS = ("about", "a-propos", "apropos", "propos", "equipe", "team", "notre-equipe", "leadership", "direction", "gouvernance", "membres", "conseil", "contact", "coordonnees", "nous-joindre", "qui-sommes", "notre-histoire", "carrieres") def deep_dive(self, url: str, max_pages: int = 8) -> dict[str, Any]: parts = urlparse(url if "://" in url else "http://" + url) root = f"{parts.scheme or 'https'}://{parts.netloc or parts.path}" try: links = self.fc.map(root) except Exception as e: # noqa: BLE001 links = [root] self.store.log_event("error", f"map deep_dive: {e}", root) def score(u: str) -> int: lu = u.lower() return sum(2 if k in lu else 0 for k in self._DEEP_KEYWORDS) targets = [t for t in sorted(set(links), key=score, reverse=True) if score(t) > 0][: max_pages - 1] targets = [root] + targets self.store.log_event("agent", f"deep_dive {parts.netloc}: {len(targets)} page(s), concurrent", root) res = self.crawl_batch(targets) return {"domain": parts.netloc, "pages": res["pages"], "entities_found": res["entities_found"], "relations_found": res["relations_found"]} # -- mode agent (contrôleur = modèle plus puissant) ------------------- def run_agent(self, goal: str, seed: str | None, max_steps: int, should_abort=None) -> str: self.store.log_event("agent", f"Mission: {goal}", "") def t_search(inp: dict[str, Any]) -> dict[str, Any]: res = self.fc.search(inp["query"], int(inp.get("limit", 10))) self.store.log_event("search", inp["query"], "", {"results": len(res)}) return {"results": [{"url": r.get("url"), "title": r.get("title"), "desc": r.get("description")} for r in res]} def t_map(inp: dict[str, Any]) -> dict[str, Any]: links = self.fc.map(inp["url"], inp.get("search")) self.store.enqueue(links[:150]) self.store.log_event("map", inp["url"], inp["url"], {"links": len(links)}) return {"count": len(links), "sample": links[:40]} def t_scrape(inp: dict[str, Any]) -> dict[str, Any]: return self._scrape_and_extract(inp["url"]) def t_deep(inp: dict[str, Any]) -> dict[str, Any]: return self.deep_dive(inp["url"], int(inp.get("max_pages", 8))) def t_enqueue(inp: dict[str, Any]) -> dict[str, Any]: return {"enqueued": self.store.enqueue(inp.get("urls", []))} def t_drain(inp: dict[str, Any]) -> dict[str, Any]: return self.crawl_batch([r["url"] for r in self.store.pending_batch( int(inp.get("count", self.cfg.concurrency)))]) def t_plan(inp: dict[str, Any]) -> dict[str, Any]: self.store.log_event("agent", "PLAN: " + " | ".join(inp.get("plan", [])), "", {"rationale": inp.get("rationale", "")}) return {"ack": True} def t_note(inp: dict[str, Any]) -> dict[str, Any]: self.store.log_event("agent", "NOTE: " + inp.get("note", ""), "") return {"ack": True} controller = Controller( self.anthropic, self.cfg.model, {"search_web": t_search, "map_site": t_map, "scrape_page": t_scrape, "deep_dive": t_deep, "crawl_queue": t_drain, "enqueue_urls": t_enqueue, "update_plan": t_plan, "note_progress": t_note}, on_event=lambda kind, msg: self.store.log_event(kind, msg, ""), max_context_tokens=self.cfg.max_context_tokens, ) summary = controller.run(goal=goal, seed=seed, max_steps=max_steps, should_abort=should_abort) self.store.log_event("agent", "Mission conclue", "", {"summary": summary}) return summary # -- pipeline (concurrent) -------------------------------------------- def run_pipeline(self, query: str, limit: int = 10, per_site_pages: int = 3) -> dict[str, Any]: results = self.fc.search(query, limit) urls = [r.get("url") for r in results if r.get("url")] res = self.crawl_batch(urls) # approfondir les domaines découverts for u in urls[:per_site_pages]: if self._pages_scraped >= self.cfg.max_pages_per_run: break self.deep_dive(u, max_pages=per_site_pages + 2) return {"query": query, "pages_scraped": self._pages_scraped, "entities_found": res["entities_found"], "stats": self.store.stats()} # -- file d'attente (drainée en concurrence) -------------------------- def drain_queue(self, max_pages: int | None = None, should_abort=None) -> dict[str, Any]: budget = max_pages or self.cfg.max_pages_per_run processed = 0 while processed < budget and self._pages_scraped < self.cfg.max_pages_per_run: if should_abort and should_abort(): break rows = self.store.pending_batch(min(self.cfg.concurrency, budget - processed), max_attempts=self.cfg.max_attempts) if not rows: break self.crawl_batch([r["url"] for r in rows]) # robustesse : succès -> done ; échec -> retry (attempts++) puis dead-letter for r in rows: if self.store.source_seen(r["url"]): self.store.mark(r["id"], "done") else: self.store.bump_attempt(r["id"], self.cfg.max_attempts) processed += len(rows) return {"processed": processed, "stats": self.store.stats()} def close(self) -> None: self.store.close()