# Author: Simon-Pierre Boucher
# Contact: contact@spboucher.ai
"""API + dashboard de suivi de ka4 (FastAPI).
Interface alignée sur le système de design du Groupe KA (néo-brutalisme éditorial :
papier chaud, encre, accent lime, bordures nettes, ombres dures décalées).
Temps réel via SSE, réglages personnalisables (backend, budget, session, accent).
Le bot de crawl tourne dans un processus séparé partageant la même base SQLite (WAL).
"""
from __future__ import annotations
import asyncio
import json
import time
from typing import Optional
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse
from src.config import config
from src.storage import Store
app = FastAPI(title="ka4 — cartographe du web québécois", version="2.0.0")
SESSION_SECONDS = 15 * 60 # bloc de session (bouton « Continuer »)
# Réglages personnalisables : nom public -> (clé en base, défaut effectif, type)
CONFIG_KEYS = {
"daily_budget": ("cfg_daily_budget", 800, int),
"cycle_delay": ("cfg_cycle_delay", 20.0, float),
"queue_batch": ("cfg_queue_batch", 8, int),
"session_minutes": ("cfg_session_minutes", 15.0, float),
"max_pages": ("cfg_max_pages", config.max_pages_per_run, int),
"backend": ("cfg_backend", config.scraper_backend, str),
"respect_robots": ("cfg_respect_robots", "1" if config.respect_robots else "0", str),
"delay": ("cfg_delay", config.request_delay, float),
}
def store() -> Store:
return Store(config.db_path)
def _domain(url: Optional[str]) -> str:
"""Domaine racine simplifié d'une URL (pour relier entités du même site)."""
if not url:
return ""
u = url.strip().lower()
for pref in ("https://", "http://"):
if u.startswith(pref):
u = u[len(pref):]
u = u.split("/")[0]
if u.startswith("www."):
u = u[4:]
return u
def _session(s: Store) -> dict:
try:
run_until = float(s.get_setting("run_until", "0") or 0)
except ValueError:
run_until = 0.0
remaining = max(0, int(run_until - time.time()))
return {"run_until": run_until, "session_remaining": remaining, "session_active": remaining > 0}
@app.get("/health")
def health():
return {"status": "ok", "service": "ka4"}
@app.get("/api/stats")
def api_stats():
s = store()
try:
data = s.stats()
data["paused"] = s.get_setting("paused", "0") == "1"
data["active_mission_id"] = s.get_setting("active_mission_id", "")
data.update(_session(s))
return data
finally:
s.close()
@app.get("/api/events")
def api_events(after_id: int = 0, limit: int = 80):
s = store()
try:
return {"events": s.recent_events(limit=limit, after_id=after_id)}
finally:
s.close()
def _filter_entities(s: Store, type=None, q=None, region=None, sector=None):
rows = s.export()
if type:
rows = [r for r in rows if r["type"] == type]
if region:
rl = region.lower()
rows = [r for r in rows if rl in ((r.get("region") or r.get("location") or "").lower())]
if sector:
sl = sector.lower()
rows = [r for r in rows if sl in ((r.get("sector") or "").lower())]
if q:
ql = q.lower()
rows = [r for r in rows if ql in (r["name"] or "").lower()
or ql in (r.get("description") or "").lower()
or ql in (r.get("sector") or "").lower()]
deg = s.degrees()
for r in rows:
r["domain"] = r.get("domain") or _domain(r.get("url"))
r["relations_count"] = deg.get(r["id"], 0)
r["social_count"] = len(r.get("social_links") or [])
return rows
@app.get("/api/entities")
def api_entities(type: Optional[str] = None, q: Optional[str] = None, region: Optional[str] = None,
sector: Optional[str] = None, sort: str = "name", offset: int = 0, limit: int = 60):
s = store()
try:
rows = _filter_entities(s, type, q, region, sector)
if sort == "degree":
rows.sort(key=lambda r: (r["relations_count"], r.get("social_count", 0)), reverse=True)
elif sort == "recent":
rows.sort(key=lambda r: (r.get("last_seen") or r.get("updated_at") or 0), reverse=True)
else:
rows.sort(key=lambda r: (r.get("name") or "").lower())
total = len(rows)
return {"count": total, "entities": rows[offset:offset + limit],
"offset": offset, "limit": limit}
finally:
s.close()
@app.get("/api/export.json")
def export_json(type: Optional[str] = None, q: Optional[str] = None,
region: Optional[str] = None, sector: Optional[str] = None):
s = store()
try:
rows = _filter_entities(s, type, q, region, sector)
finally:
s.close()
import json as _json
return StreamingResponse(
iter([_json.dumps(rows, ensure_ascii=False, indent=2)]),
media_type="application/json",
headers={"Content-Disposition": "attachment; filename=ka4-entites.json"},
)
@app.get("/api/export.csv")
def export_csv(type: Optional[str] = None, q: Optional[str] = None,
region: Optional[str] = None, sector: Optional[str] = None):
s = store()
try:
rows = _filter_entities(s, type, q, region, sector)
finally:
s.close()
import csv as _csv
import io as _io
cols = ["id", "type", "name", "sector", "region", "city", "url", "email", "phone",
"founded", "size", "neq", "relations_count", "social_count"]
buf = _io.StringIO()
w = _csv.writer(buf)
w.writerow(cols)
for r in rows:
w.writerow([r.get(c, "") for c in cols])
return StreamingResponse(
iter([buf.getvalue()]), media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=ka4-entites.csv"},
)
@app.get("/api/explore/summary")
def explore_summary():
s = store()
try:
st = s.stats()
return {
"total": st["entities"],
"relations": st["relations"],
"by_type": st["by_type"],
"regions": s.distinct_locations()[:40],
"sectors": s.sector_counts()[:40],
"platforms": s.platform_counts()[:16],
"top_connected": s.top_connected(10),
}
finally:
s.close()
@app.get("/api/graph")
def api_graph(type: Optional[str] = None, region: Optional[str] = None,
sector: Optional[str] = None, limit: int = 140):
s = store()
try:
return s.graph(limit=limit, etype=type or "", region=region or "",
sector=sector or "", connected_only=True)
finally:
s.close()
@app.get("/api/entity/{entity_id}")
def api_entity(entity_id: int):
s = store()
try:
e = s.get_entity(entity_id)
if not e:
return JSONResponse({"error": "introuvable"}, status_code=404)
e["domain"] = e.get("domain") or _domain(e.get("url"))
# Relations RÉELLES extraites (typées, avec rôle) — le cœur du graphe
e["relations"] = s.entity_relations(entity_id)
# Contexte secondaire : autres entités de la même région
region = (e.get("region") or e.get("location") or "").strip()
same_region = []
if region:
linked = {r["id"] for r in e["relations"]}
for r in s.export():
if r["id"] == entity_id or r["id"] in linked:
continue
if (r.get("region") or r.get("location") or "").strip() == region:
same_region.append({"id": r["id"], "name": r["name"], "type": r["type"]})
e["same_region"] = same_region[:20]
finally:
s.close()
return e
@app.get("/api/missions")
def api_missions():
s = store()
try:
return {"missions": s.list_missions()}
finally:
s.close()
@app.post("/api/missions")
def add_mission(payload: dict):
s = store()
try:
mid = s.add_mission(
goal=payload.get("goal", "").strip(),
seed=payload.get("seed", ""),
sector=payload.get("sector", ""),
region=payload.get("region", ""),
priority=int(payload.get("priority", 5)),
)
return {"id": mid, "ok": bool(mid)}
finally:
s.close()
@app.post("/api/missions/{mission_id}/{op}")
def mission_op(mission_id: int, op: str):
s = store()
try:
if op == "delete":
s.delete_mission(mission_id)
s.log_event("info", f"Mission #{mission_id} supprimée.")
elif op == "prioritize":
# « Lancer maintenant » : priorité max + interrompt l'activité en cours
s.prioritize_mission(mission_id)
s.set_setting("abort_current", "1")
s.set_setting("paused", "0")
s.log_event("info", f"Mission #{mission_id} priorisée (lancer maintenant).")
else:
return JSONResponse({"error": "op inconnue"}, status_code=400)
return {"ok": True}
finally:
s.close()
@app.post("/api/archive")
def archive_now(payload: dict):
s = store()
try:
# 1) stopper l'activité en cours (pause + interruption) pour éviter la course
s.set_setting("paused", "1")
s.set_setting("abort_current", "1")
# 2) attendre que le bot confirme qu'il est au repos (≤20s) avant de vider
for _ in range(40):
if s.get_setting("bot_busy", "0") != "1":
break
time.sleep(0.5)
# 3) snapshot + reset
res = s.archive_and_reset(payload.get("label", ""))
s.log_event("info", f"Archive « {res['label']} » créée ({res['entities']} entités). Espace de travail vidé.")
return res
finally:
s.close()
@app.get("/api/archives")
def api_archives():
s = store()
try:
return {"archives": s.list_archives()}
finally:
s.close()
@app.post("/api/archive/{archive_id}/{op}")
def archive_op(archive_id: int, op: str):
s = store()
try:
if op == "restore":
return s.restore_archive(archive_id)
if op == "delete":
s.delete_archive(archive_id)
return {"ok": True}
return JSONResponse({"error": "op inconnue"}, status_code=400)
finally:
s.close()
@app.get("/api/archive/{archive_id}/export.json")
def archive_export(archive_id: int):
s = store()
try:
a = s.get_archive(archive_id)
finally:
s.close()
if not a:
return JSONResponse({"error": "introuvable"}, status_code=404)
import json as _json
return StreamingResponse(
iter([_json.dumps(a["data"], ensure_ascii=False, indent=2)]),
media_type="application/json",
headers={"Content-Disposition": f"attachment; filename=ka-archive-{archive_id}.json"},
)
@app.post("/api/control/{action}")
def control(action: str):
s = store()
try:
if action == "pause":
s.set_setting("paused", "1")
elif action == "resume":
s.set_setting("paused", "0")
elif action == "skip":
# arrête l'agent / le drainage en cours et passe à la suite
s.set_setting("abort_current", "1")
s.log_event("info", "Interruption demandée (arrêter l'activité en cours).")
elif action == "extend":
try:
cur = float(s.get_setting("run_until", "0") or 0)
except ValueError:
cur = 0.0
s.set_setting("run_until", str(max(cur, time.time()) + SESSION_SECONDS))
s.set_setting("paused", "0")
s.log_event("info", "Session prolongée de 15 min (bouton Continuer).")
elif action == "stop":
s.set_setting("run_until", str(time.time()))
else:
return JSONResponse({"error": "action inconnue"}, status_code=400)
out = {"paused": s.get_setting("paused", "0") == "1"}
out.update(_session(s))
return out
finally:
s.close()
@app.get("/api/config")
def get_config():
s = store()
try:
out = {}
for name, (key, default, cast) in CONFIG_KEYS.items():
raw = s.get_setting(key, "")
try:
out[name] = cast(raw) if raw != "" else default
except ValueError:
out[name] = default
return out
finally:
s.close()
@app.post("/api/config")
def set_config(payload: dict):
s = store()
try:
applied = {}
for name, value in payload.items():
if name not in CONFIG_KEYS:
continue
key, _default, cast = CONFIG_KEYS[name]
try:
val = cast(value)
except (ValueError, TypeError):
continue
s.set_setting(key, str(val))
applied[name] = val
if applied:
s.log_event("info", "Réglages: " + ", ".join(f"{k}={v}" for k, v in applied.items()))
return {"applied": applied}
finally:
s.close()
@app.get("/api/stream")
async def stream(request: Request):
"""Flux SSE : pousse stats + nouveaux événements (remplace le polling lourd)."""
async def gen():
last_id = 0
while True:
if await request.is_disconnected():
break
s = store()
try:
stats = s.stats()
stats["paused"] = s.get_setting("paused", "0") == "1"
stats.update(_session(s))
events = s.recent_events(limit=40, after_id=last_id)
if events:
last_id = max(e["id"] for e in events)
finally:
s.close()
payload = {"stats": stats, "events": list(reversed(events))}
yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
await asyncio.sleep(2.5)
return StreamingResponse(
gen(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"},
)
@app.get("/ka4.css")
def ka4_css():
# CSS partagé (extrait du dashboard) — source unique pour les deux pages.
css = HTML.split("", 1)[0]
return StreamingResponse(iter([css]), media_type="text/css",
headers={"Cache-Control": "max-age=300"})
@app.get("/", response_class=HTMLResponse)
def index():
return HTML
@app.get("/explore", response_class=HTMLResponse)
def explore():
return EXPLORE_HTML
# =========================================================================
# Dashboard mono-fichier — design system Groupe KA (néo-brutalisme éditorial)
# =========================================================================
HTML = r"""
ka4 · explorateur structuré du web — Groupe KA
"""
# =========================================================================
# Page « Répertoire structuré » v3 — 3 vues (cartes / tableau / graphe global)
# + facettes (type, secteur, région), tri, insights, export CSV/JSON.
# =========================================================================
EXPLORE_HTML = r"""
ka4 · Répertoire structuré — graphe du web québécois
Répertoire structuré
Répertoire structuré
Entreprise
Organisation
Personne
Site web
"""