spb/ka6 Public
ka6 — bot FLAGSHIP du Groupe KA : explorateur structuré du web québécois. Crawling concurrent robuste, Claude Sonnet+Haiku, graphe de connaissances, dashboard temps réel.
Python 97.8%
Shell 2.2%
1# Author: Simon-Pierre Boucher2# Contact: contact@spboucher.ai3"""Persistance SQLite — graphe de connaissances du web québécois (v2).45Entités enrichies et normalisées, relations typées (avec rôle + provenance),6mentions (provenance multi-sources), déduplication par résolution d'entités,7plus missions / événements / réglages / file de crawl pour le bot long terme.8"""910from __future__ import annotations1112import hashlib13import json14import sqlite315import time16from typing import Any, Iterable, Optional1718from .normalize import canonical_location, domain, normalize_name192021def _to_int(v: Any) -> Optional[int]:22 """Parse un nombre d'abonnés : 12000, '12,5 K', '3.2M', '1 200 abonnés' -> int."""23 if v is None or v == "":24 return None25 if isinstance(v, (int, float)):26 return int(v)27 s = str(v).lower().replace(" ", " ").strip()28 import re as _re29 m = _re.search(r"([\d]+(?:[.,]\d+)?)\s*([km])?", s.replace(" ", ""))30 if not m:31 return None32 num = float(m.group(1).replace(",", "."))33 mult = {"k": 1_000, "m": 1_000_000}.get(m.group(2) or "", 1)34 try:35 return int(num * mult)36 except (ValueError, OverflowError):37 return None383940REL_TYPES = {41 "WORKS_AT", "FOUNDER_OF", "OWNS", "MEMBER_OF", "PARTNER_OF",42 "SUBSIDIARY_OF", "PARENT_OF", "AFFILIATED_WITH", "LOCATED_IN", "SUPPLIER_OF",43 # relations créateurs / influenceurs44 "REPRESENTED_BY", "COLLABORATES_WITH", "CREATES_ON", "SPONSORED_BY", "PROMOTES",45 "MANAGES", "APPEARS_WITH",46}4748SCHEMA = """49CREATE TABLE IF NOT EXISTS entities (50 id INTEGER PRIMARY KEY AUTOINCREMENT,51 type TEXT NOT NULL,52 name TEXT NOT NULL,53 description TEXT,54 url TEXT,55 location TEXT,56 email TEXT,57 phone TEXT,58 confidence REAL DEFAULT 0.5,59 raw_json TEXT,60 created_at REAL NOT NULL,61 updated_at REAL NOT NULL,62 UNIQUE(type, name, url)63);6465CREATE TABLE IF NOT EXISTS social_links (66 id INTEGER PRIMARY KEY AUTOINCREMENT,67 entity_id INTEGER NOT NULL, platform TEXT NOT NULL, url TEXT NOT NULL,68 UNIQUE(entity_id, url),69 FOREIGN KEY(entity_id) REFERENCES entities(id) ON DELETE CASCADE70);7172CREATE TABLE IF NOT EXISTS relations (73 id INTEGER PRIMARY KEY AUTOINCREMENT,74 from_entity INTEGER NOT NULL, to_entity INTEGER NOT NULL, relation_type TEXT NOT NULL,75 UNIQUE(from_entity, to_entity, relation_type)76);7778CREATE TABLE IF NOT EXISTS entity_mentions (79 id INTEGER PRIMARY KEY AUTOINCREMENT,80 entity_id INTEGER NOT NULL, source_url TEXT NOT NULL, seen_at REAL NOT NULL,81 UNIQUE(entity_id, source_url),82 FOREIGN KEY(entity_id) REFERENCES entities(id) ON DELETE CASCADE83);8485CREATE TABLE IF NOT EXISTS sources (86 id INTEGER PRIMARY KEY AUTOINCREMENT,87 url TEXT NOT NULL UNIQUE, title TEXT, content_hash TEXT, scraped_at REAL NOT NULL88);8990CREATE TABLE IF NOT EXISTS crawl_queue (91 id INTEGER PRIMARY KEY AUTOINCREMENT,92 url TEXT NOT NULL UNIQUE, status TEXT NOT NULL DEFAULT 'pending',93 depth INTEGER DEFAULT 0, added_at REAL NOT NULL94);9596CREATE TABLE IF NOT EXISTS missions (97 id INTEGER PRIMARY KEY AUTOINCREMENT,98 goal TEXT NOT NULL, seed TEXT, sector TEXT, region TEXT,99 priority INTEGER DEFAULT 5, status TEXT NOT NULL DEFAULT 'pending',100 runs_count INTEGER DEFAULT 0, last_run REAL, created_at REAL NOT NULL,101 UNIQUE(goal, region)102);103104CREATE TABLE IF NOT EXISTS events (105 id INTEGER PRIMARY KEY AUTOINCREMENT,106 ts REAL NOT NULL, kind TEXT NOT NULL, message TEXT, url TEXT, data_json TEXT107);108109CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT);110111CREATE TABLE IF NOT EXISTS archives (112 id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT, created_at REAL NOT NULL,113 entities INTEGER, relations INTEGER, sources INTEGER, payload TEXT114);115116CREATE INDEX IF NOT EXISTS idx_entities_type ON entities(type);117CREATE INDEX IF NOT EXISTS idx_queue_status ON crawl_queue(status);118CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts);119CREATE INDEX IF NOT EXISTS idx_rel_from ON relations(from_entity);120CREATE INDEX IF NOT EXISTS idx_rel_to ON relations(to_entity);121"""122123# Colonnes ajoutées par migration aux DB existantes124_ENTITY_COLS = [125 ("canonical_name", "TEXT"), ("norm_name", "TEXT"), ("domain", "TEXT"), ("sector", "TEXT"),126 ("address", "TEXT"), ("city", "TEXT"), ("region", "TEXT"), ("postal_code", "TEXT"),127 ("neq", "TEXT"), ("founded", "TEXT"), ("size", "TEXT"), ("tags", "TEXT"),128 ("source_url", "TEXT"), ("first_seen", "REAL"), ("last_seen", "REAL"),129 # champs CRÉATEURS / INFLUENCEURS (ka6)130 ("niche", "TEXT"), ("handle", "TEXT"), ("platform", "TEXT"), ("followers", "INTEGER"),131 ("languages", "TEXT"),132]133_RELATION_COLS = [("role", "TEXT"), ("source_url", "TEXT"), ("confidence", "REAL")]134135136class Store:137 def __init__(self, path: str):138 self.conn = sqlite3.connect(path, timeout=30, check_same_thread=False)139 self.conn.row_factory = sqlite3.Row140 self.conn.execute("PRAGMA foreign_keys = ON")141 self.conn.execute("PRAGMA journal_mode = WAL")142 self.conn.execute("PRAGMA busy_timeout = 30000")143 self.conn.executescript(SCHEMA)144 self._migrate()145 self.conn.execute("CREATE INDEX IF NOT EXISTS idx_entities_norm ON entities(norm_name)")146 self.conn.execute("CREATE INDEX IF NOT EXISTS idx_entities_domain ON entities(domain)")147 self.conn.commit()148149 def _migrate(self) -> None:150 ecols = {r["name"] for r in self.conn.execute("PRAGMA table_info(entities)")}151 for name, typ in _ENTITY_COLS:152 if name not in ecols:153 self.conn.execute(f"ALTER TABLE entities ADD COLUMN {name} {typ}")154 rcols = {r["name"] for r in self.conn.execute("PRAGMA table_info(relations)")}155 for name, typ in _RELATION_COLS:156 if name not in rcols:157 self.conn.execute(f"ALTER TABLE relations ADD COLUMN {name} {typ}")158 # social_links.followers (abonnés par plateforme) + crawl_queue.attempts (robustesse)159 scols = {r["name"] for r in self.conn.execute("PRAGMA table_info(social_links)")}160 if "followers" not in scols:161 self.conn.execute("ALTER TABLE social_links ADD COLUMN followers INTEGER")162 qcols = {r["name"] for r in self.conn.execute("PRAGMA table_info(crawl_queue)")}163 if "attempts" not in qcols:164 self.conn.execute("ALTER TABLE crawl_queue ADD COLUMN attempts INTEGER DEFAULT 0")165 self.conn.commit()166 self._backfill()167168 def _backfill(self) -> None:169 """Normalise les entités héritées (norm_name/domain/region) pour la dédup et l'explorer."""170 rows = self.conn.execute(171 "SELECT id, name, url, location FROM entities WHERE norm_name IS NULL"172 ).fetchall()173 for r in rows:174 nn = normalize_name(r["name"]) or domain(r["url"])175 self.conn.execute(176 "UPDATE entities SET norm_name=?, domain=?, region=COALESCE(region,?) WHERE id=?",177 (nn, domain(r["url"]) or None, canonical_location(r["location"]) or None, r["id"]),178 )179 if rows:180 self.conn.commit()181182 # -- résolution + upsert d'entités -------------------------------------183 def _resolve(self, etype: str, nn: str, dom: str, region: str) -> Optional[int]:184 if not nn:185 return None186 c = self.conn187 if dom:188 r = c.execute(189 "SELECT id FROM entities WHERE type=? AND norm_name=? AND domain=? LIMIT 1",190 (etype, nn, dom),191 ).fetchone()192 if r:193 return r["id"]194 if region:195 r = c.execute(196 "SELECT id FROM entities WHERE type=? AND norm_name=? AND region=? "197 "AND (domain IS NULL OR domain='') LIMIT 1",198 (etype, nn, region),199 ).fetchone()200 if r:201 return r["id"]202 r = c.execute(203 "SELECT id FROM entities WHERE type=? AND norm_name=? "204 "AND (domain IS NULL OR domain='') AND (region IS NULL OR region='') LIMIT 1",205 (etype, nn),206 ).fetchone()207 return r["id"] if r else None208209 def upsert_entity(self, e: dict[str, Any], source_url: Optional[str] = None) -> int:210 now = time.time()211 etype = (e.get("type") or "website").strip().lower()212 name = (e.get("name") or "").strip()213 website = (e.get("website") or e.get("url") or "").strip() or None214 dom = domain(website)215 region = canonical_location(e.get("region") or e.get("city") or e.get("location") or "")216 city = (e.get("city") or "").strip() or None217 nn = normalize_name(name) or dom218 if not name:219 name = website or "inconnu"220 conf = float(e.get("confidence", 0.5) or 0.5)221 tags = json.dumps(e.get("tags"), ensure_ascii=False) if e.get("tags") else None222 langs = e.get("languages")223 languages = ", ".join(langs) if isinstance(langs, list) else (langs or None)224 followers = _to_int(e.get("followers"))225226 eid = self._resolve(etype, nn, dom, region)227 vals = {228 "canonical_name": e.get("canonical_name") or name,229 "description": e.get("description"), "url": website, "domain": dom or None,230 "sector": e.get("sector"), "address": e.get("address"), "city": city,231 "region": region or None, "location": e.get("location") or region or city,232 "email": e.get("email"), "phone": e.get("phone"), "postal_code": e.get("postal_code"),233 "neq": e.get("neq"), "founded": e.get("founded"), "size": e.get("size"),234 "tags": tags, "confidence": conf, "source_url": source_url,235 "niche": e.get("niche"), "handle": e.get("handle"), "platform": e.get("platform"),236 "followers": followers, "languages": languages,237 }238 if eid:239 self.conn.execute(240 """UPDATE entities SET241 canonical_name=COALESCE(canonical_name,?), description=COALESCE(description,?),242 url=COALESCE(url,?), domain=COALESCE(NULLIF(domain,''),?), sector=COALESCE(sector,?),243 address=COALESCE(address,?), city=COALESCE(city,?), region=COALESCE(region,?),244 location=COALESCE(location,?), email=COALESCE(email,?), phone=COALESCE(phone,?),245 postal_code=COALESCE(postal_code,?), neq=COALESCE(neq,?), founded=COALESCE(founded,?),246 size=COALESCE(size,?), tags=COALESCE(tags,?), confidence=MAX(IFNULL(confidence,0),?),247 source_url=COALESCE(source_url,?), niche=COALESCE(niche,?), handle=COALESCE(handle,?),248 platform=COALESCE(platform,?), followers=MAX(IFNULL(followers,0),?),249 languages=COALESCE(languages,?), last_seen=?, updated_at=? WHERE id=?""",250 (251 vals["canonical_name"], vals["description"], vals["url"], vals["domain"],252 vals["sector"], vals["address"], vals["city"], vals["region"], vals["location"],253 vals["email"], vals["phone"], vals["postal_code"], vals["neq"], vals["founded"],254 vals["size"], vals["tags"], vals["confidence"], vals["source_url"],255 vals["niche"], vals["handle"], vals["platform"], followers or 0, vals["languages"],256 now, now, eid,257 ),258 )259 else:260 cur = self.conn.execute(261 """INSERT INTO entities(type, name, canonical_name, norm_name, description, url, domain,262 sector, address, city, region, location, email, phone, postal_code, neq, founded,263 size, tags, confidence, raw_json, source_url, niche, handle, platform, followers,264 languages, first_seen, last_seen, created_at, updated_at)265 VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",266 (267 etype, name, vals["canonical_name"], nn, vals["description"], vals["url"], vals["domain"],268 vals["sector"], vals["address"], vals["city"], vals["region"], vals["location"],269 vals["email"], vals["phone"], vals["postal_code"], vals["neq"], vals["founded"],270 vals["size"], vals["tags"], conf, json.dumps(e, ensure_ascii=False), source_url,271 vals["niche"], vals["handle"], vals["platform"], followers, vals["languages"],272 now, now, now, now,273 ),274 )275 eid = cur.lastrowid276277 for link in e.get("social_links", []) or []:278 self.add_social_link(eid, link.get("platform", "?"), link.get("url", ""),279 _to_int(link.get("followers")))280 if source_url:281 self.conn.execute(282 "INSERT OR IGNORE INTO entity_mentions(entity_id, source_url, seen_at) VALUES(?,?,?)",283 (eid, source_url, now),284 )285 self.conn.commit()286 return eid287288 def add_social_link(self, entity_id: int, platform: str, url: str,289 followers: Optional[int] = None) -> None:290 url = (url or "").strip()291 if not url:292 return293 self.conn.execute(294 "INSERT INTO social_links(entity_id, platform, url, followers) VALUES(?,?,?,?) "295 "ON CONFLICT(entity_id, url) DO UPDATE SET followers=COALESCE(excluded.followers, followers)",296 (entity_id, (platform or "?").strip().lower(), url, followers),297 )298299 # -- relations ---------------------------------------------------------300 def add_relation(self, from_id: int, to_id: int, rel_type: str, role: str = "",301 source_url: str = "", confidence: float = 0.6) -> None:302 rel_type = (rel_type or "").strip().upper()303 if not from_id or not to_id or from_id == to_id or rel_type not in REL_TYPES:304 return305 self.conn.execute(306 """INSERT INTO relations(from_entity, to_entity, relation_type, role, source_url, confidence)307 VALUES(?,?,?,?,?,?)308 ON CONFLICT(from_entity, to_entity, relation_type) DO UPDATE SET309 role=COALESCE(NULLIF(excluded.role,''), role),310 source_url=COALESCE(NULLIF(excluded.source_url,''), source_url),311 confidence=MAX(IFNULL(confidence,0), excluded.confidence)""",312 (from_id, to_id, rel_type, role or None, source_url or None, confidence),313 )314 self.conn.commit()315316 def entity_relations(self, entity_id: int) -> list[dict[str, Any]]:317 rows = self.conn.execute(318 """SELECT r.relation_type, r.role, r.confidence, r.from_entity, r.to_entity,319 e.id oid, e.name oname, e.type otype, e.region oregion, e.sector osector320 FROM relations r321 JOIN entities e ON e.id = CASE WHEN r.from_entity=? THEN r.to_entity ELSE r.from_entity END322 WHERE r.from_entity=? OR r.to_entity=?323 ORDER BY r.confidence DESC""",324 (entity_id, entity_id, entity_id),325 ).fetchall()326 out = []327 for r in rows:328 outgoing = r["from_entity"] == entity_id329 out.append({330 "id": r["oid"], "name": r["oname"], "type": r["otype"],331 "region": r["oregion"], "sector": r["osector"],332 "relation": r["relation_type"], "role": r["role"],333 "direction": "out" if outgoing else "in",334 "confidence": r["confidence"],335 })336 return out337338 # -- sources / file ----------------------------------------------------339 def record_source(self, url: str, title: str, content: str) -> None:340 h = hashlib.sha256((content or "").encode("utf-8")).hexdigest()341 self.conn.execute(342 "INSERT OR REPLACE INTO sources(url, title, content_hash, scraped_at) VALUES(?,?,?,?)",343 (url, title, h, time.time()),344 )345 self.conn.commit()346347 def source_seen(self, url: str) -> bool:348 return self.conn.execute("SELECT 1 FROM sources WHERE url=?", (url,)).fetchone() is not None349350 def enqueue(self, urls: Iterable[str], depth: int = 0) -> int:351 n = 0352 for u in urls:353 u = (u or "").strip()354 if not u:355 continue356 n += self.conn.execute(357 "INSERT OR IGNORE INTO crawl_queue(url, status, depth, added_at) VALUES(?,?,?,?)",358 (u, "pending", depth, time.time()),359 ).rowcount360 self.conn.commit()361 return n362363 def next_pending(self) -> Optional[sqlite3.Row]:364 return self.conn.execute(365 "SELECT * FROM crawl_queue WHERE status='pending' ORDER BY depth, id LIMIT 1"366 ).fetchone()367368 def pending_batch(self, n: int, max_attempts: int = 99) -> list[sqlite3.Row]:369 return self.conn.execute(370 "SELECT * FROM crawl_queue WHERE status='pending' AND IFNULL(attempts,0) < ? "371 "ORDER BY depth, id LIMIT ?", (max_attempts, max(1, n)),372 ).fetchall()373374 def mark(self, queue_id: int, status: str) -> None:375 self.conn.execute("UPDATE crawl_queue SET status=? WHERE id=?", (status, queue_id))376 self.conn.commit()377378 def bump_attempt(self, queue_id: int, max_attempts: int) -> None:379 """Incrémente le compteur de tentatives ; passe en 'error' au-delà du seuil (dead-letter)."""380 self.conn.execute(381 "UPDATE crawl_queue SET attempts=IFNULL(attempts,0)+1, "382 "status=CASE WHEN IFNULL(attempts,0)+1 >= ? THEN 'error' ELSE 'pending' END WHERE id=?",383 (max_attempts, queue_id),384 )385 self.conn.commit()386387 # -- settings ----------------------------------------------------------388 def get_setting(self, key: str, default: str = "") -> str:389 row = self.conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()390 return row["value"] if row else default391392 def set_setting(self, key: str, value: str) -> None:393 self.conn.execute(394 "INSERT INTO settings(key, value) VALUES(?,?) "395 "ON CONFLICT(key) DO UPDATE SET value=excluded.value",396 (key, value),397 )398 self.conn.commit()399400 # -- missions ----------------------------------------------------------401 def add_mission(self, goal: str, seed: str = "", sector: str = "", region: str = "",402 priority: int = 5) -> int:403 cur = self.conn.execute(404 """INSERT OR IGNORE INTO missions(goal, seed, sector, region, priority, status, created_at)405 VALUES(?,?,?,?,?, 'pending', ?)""",406 (goal, seed or None, sector or None, region or None, priority, time.time()),407 )408 self.conn.commit()409 return cur.lastrowid or 0410411 def next_mission(self) -> Optional[sqlite3.Row]:412 return self.conn.execute(413 "SELECT * FROM missions WHERE status='pending' ORDER BY priority, last_run IS NOT NULL, "414 "IFNULL(last_run, 0), id LIMIT 1"415 ).fetchone()416417 def due_mission(self, revisit_seconds: float) -> Optional[sqlite3.Row]:418 threshold = time.time() - revisit_seconds419 return self.conn.execute(420 "SELECT * FROM missions WHERE status IN ('pending','done') "421 "AND (last_run IS NULL OR last_run <= ?) "422 "ORDER BY priority, IFNULL(last_run, 0), id LIMIT 1",423 (threshold,),424 ).fetchone()425426 def set_mission_status(self, mission_id: int, status: str, bump_run: bool = False) -> None:427 if bump_run:428 self.conn.execute(429 "UPDATE missions SET status=?, runs_count=runs_count+1, last_run=? WHERE id=?",430 (status, time.time(), mission_id),431 )432 else:433 self.conn.execute("UPDATE missions SET status=? WHERE id=?", (status, mission_id))434 self.conn.commit()435436 def list_missions(self) -> list[dict[str, Any]]:437 return [dict(r) for r in self.conn.execute("SELECT * FROM missions ORDER BY priority, id")]438439 def delete_mission(self, mission_id: int) -> None:440 self.conn.execute("DELETE FROM missions WHERE id=?", (mission_id,))441 self.conn.commit()442443 def prioritize_mission(self, mission_id: int) -> None:444 """Passe la mission tout en haut de la file et la rend « due » immédiatement."""445 self.conn.execute(446 "UPDATE missions SET priority=0, status='pending', last_run=NULL WHERE id=?",447 (mission_id,),448 )449 self.conn.commit()450451 def next_forced(self) -> Optional[sqlite3.Row]:452 """Mission « lancer maintenant » (priorité 0) à exécuter avant tout le reste."""453 return self.conn.execute(454 "SELECT * FROM missions WHERE status='pending' AND priority <= 0 "455 "ORDER BY IFNULL(last_run, 0), id LIMIT 1"456 ).fetchone()457458 def requeue_stale_missions(self) -> None:459 self.conn.execute("UPDATE missions SET status='pending' WHERE status='running'")460 self.conn.commit()461462 def pages_since(self, since_ts: float) -> int:463 return self.conn.execute(464 "SELECT COUNT(*) n FROM sources WHERE scraped_at >= ?", (since_ts,)465 ).fetchone()["n"]466467 # -- événements --------------------------------------------------------468 def log_event(self, kind: str, message: str = "", url: str = "", data: Any = None) -> None:469 self.conn.execute(470 "INSERT INTO events(ts, kind, message, url, data_json) VALUES(?,?,?,?,?)",471 (time.time(), kind, message, url or None,472 json.dumps(data, ensure_ascii=False) if data is not None else None),473 )474 self.conn.commit()475476 def recent_events(self, limit: int = 100, after_id: int = 0) -> list[dict[str, Any]]:477 rows = self.conn.execute(478 "SELECT * FROM events WHERE id > ? ORDER BY id DESC LIMIT ?", (after_id, limit)479 ).fetchall()480 return [dict(r) for r in rows]481482 # -- lecture / agrégats ------------------------------------------------483 def get_entity(self, entity_id: int) -> Optional[dict[str, Any]]:484 e = self.conn.execute("SELECT * FROM entities WHERE id=?", (entity_id,)).fetchone()485 if not e:486 return None487 d = dict(e)488 d["social_links"] = [489 {"platform": r["platform"], "url": r["url"], "followers": r["followers"]}490 for r in self.conn.execute(491 "SELECT platform, url, followers FROM social_links WHERE entity_id=?", (entity_id,)492 )493 ]494 d["sources"] = [495 r["source_url"] for r in self.conn.execute(496 "SELECT source_url FROM entity_mentions WHERE entity_id=? ORDER BY seen_at DESC LIMIT 15",497 (entity_id,),498 )499 ]500 return d501502 def distinct_locations(self) -> list[dict[str, Any]]:503 rows = self.conn.execute(504 "SELECT COALESCE(NULLIF(region,''), location) loc, COUNT(*) n FROM entities "505 "WHERE COALESCE(NULLIF(region,''), location) IS NOT NULL "506 "AND COALESCE(NULLIF(region,''), location)<>'' GROUP BY loc ORDER BY n DESC"507 ).fetchall()508 return [{"location": r["loc"], "count": r["n"]} for r in rows]509510 def platform_counts(self) -> list[dict[str, Any]]:511 rows = self.conn.execute(512 "SELECT platform, COUNT(*) n FROM social_links GROUP BY platform ORDER BY n DESC"513 ).fetchall()514 return [{"platform": r["platform"], "count": r["n"]} for r in rows]515516 def degrees(self) -> dict[int, int]:517 deg: dict[int, int] = {}518 for r in self.conn.execute("SELECT from_entity a, to_entity b FROM relations"):519 deg[r["a"]] = deg.get(r["a"], 0) + 1520 deg[r["b"]] = deg.get(r["b"], 0) + 1521 return deg522523 def top_connected(self, limit: int = 8) -> list[dict[str, Any]]:524 deg = self.degrees()525 if not deg:526 return []527 top = sorted(deg.items(), key=lambda kv: kv[1], reverse=True)[:limit]528 out = []529 for eid, d in top:530 e = self.conn.execute(531 "SELECT id, name, type, sector, region FROM entities WHERE id=?", (eid,)532 ).fetchone()533 if e:534 out.append({**dict(e), "degree": d})535 return out536537 def graph(self, limit: int = 140, etype: str = "", region: str = "", sector: str = "",538 connected_only: bool = True) -> dict[str, Any]:539 q = "SELECT id, name, type, sector, region, location FROM entities WHERE 1=1"540 args: list[Any] = []541 if etype:542 q += " AND type=?"543 args.append(etype)544 if region:545 q += " AND COALESCE(NULLIF(region,''), location)=?"546 args.append(region)547 if sector:548 q += " AND sector LIKE ?"549 args.append(f"%{sector}%")550 rows = self.conn.execute(q, args).fetchall()551 deg = self.degrees()552 if connected_only:553 rows = [r for r in rows if deg.get(r["id"], 0) > 0]554 rows = sorted(rows, key=lambda r: deg.get(r["id"], 0), reverse=True)[:limit]555 ids = {r["id"] for r in rows}556 nodes = [{"id": r["id"], "name": r["name"], "type": r["type"],557 "degree": deg.get(r["id"], 0)} for r in rows]558 edges = []559 for r in self.conn.execute(560 "SELECT from_entity a, to_entity b, relation_type t, role FROM relations"561 ):562 if r["a"] in ids and r["b"] in ids:563 edges.append({"s": r["a"], "t": r["b"], "type": r["t"], "role": r["role"]})564 return {"nodes": nodes, "edges": edges, "truncated": len(ids)}565566 def sector_counts(self) -> list[dict[str, Any]]:567 rows = self.conn.execute(568 "SELECT sector, COUNT(*) n FROM entities WHERE sector IS NOT NULL AND sector<>'' "569 "GROUP BY sector ORDER BY n DESC"570 ).fetchall()571 return [{"sector": r["sector"], "count": r["n"]} for r in rows]572573 def stats(self) -> dict[str, Any]:574 c = self.conn575 by_type = {r["type"]: r["n"] for r in c.execute("SELECT type, COUNT(*) n FROM entities GROUP BY type")}576 return {577 "entities": c.execute("SELECT COUNT(*) n FROM entities").fetchone()["n"],578 "by_type": by_type,579 "relations": c.execute("SELECT COUNT(*) n FROM relations").fetchone()["n"],580 "social_links": c.execute("SELECT COUNT(*) n FROM social_links").fetchone()["n"],581 "sources": c.execute("SELECT COUNT(*) n FROM sources").fetchone()["n"],582 "queue_pending": c.execute(583 "SELECT COUNT(*) n FROM crawl_queue WHERE status='pending'"584 ).fetchone()["n"],585 }586587 def export(self) -> list[dict[str, Any]]:588 out = []589 for e in self.conn.execute("SELECT * FROM entities ORDER BY type, name"):590 d = dict(e)591 d["social_links"] = [592 {"platform": r["platform"], "url": r["url"], "followers": r["followers"]}593 for r in self.conn.execute(594 "SELECT platform, url, followers FROM social_links WHERE entity_id=?", (e["id"],)595 )596 ]597 out.append(d)598 return out599600 # -- archives (snapshot + reset + réutilisation) -----------------------601 def _snapshot(self) -> dict[str, Any]:602 c = self.conn603 return {604 "entities": self.export(),605 "relations": [dict(r) for r in c.execute(606 "SELECT from_entity, to_entity, relation_type, role, source_url, confidence FROM relations")],607 "missions": [dict(r) for r in c.execute(608 "SELECT goal, seed, sector, region, priority FROM missions")],609 "sources": [dict(r) for r in c.execute("SELECT url, title FROM sources")],610 }611612 def archive_and_reset(self, label: str) -> dict[str, Any]:613 """Sauvegarde tout le graphe dans l'historique puis remet l'espace de travail à VIDE."""614 st = self.stats()615 payload = json.dumps(self._snapshot(), ensure_ascii=False)616 lab = (label or "").strip() or time.strftime("Archive %Y-%m-%d %H:%M", time.localtime())617 cur = self.conn.execute(618 "INSERT INTO archives(label, created_at, entities, relations, sources, payload) VALUES(?,?,?,?,?,?)",619 (lab, time.time(), st["entities"], st["relations"], st["sources"], payload),620 )621 for t in ("relations", "social_links", "entity_mentions", "entities", "sources",622 "crawl_queue", "events", "missions"):623 self.conn.execute(f"DELETE FROM {t}")624 self.conn.execute("DELETE FROM settings WHERE key IN ('active_mission_id','abort_current')")625 self.conn.commit()626 return {"id": cur.lastrowid, "label": lab, "entities": st["entities"], "relations": st["relations"]}627628 def list_archives(self) -> list[dict[str, Any]]:629 return [dict(r) for r in self.conn.execute(630 "SELECT id, label, created_at, entities, relations, sources FROM archives ORDER BY id DESC")]631632 def get_archive(self, archive_id: int) -> Optional[dict[str, Any]]:633 r = self.conn.execute("SELECT * FROM archives WHERE id=?", (archive_id,)).fetchone()634 if not r:635 return None636 d = dict(r)637 d["data"] = json.loads(d.pop("payload") or "{}")638 return d639640 def delete_archive(self, archive_id: int) -> None:641 self.conn.execute("DELETE FROM archives WHERE id=?", (archive_id,))642 self.conn.commit()643644 def restore_archive(self, archive_id: int) -> dict[str, Any]:645 """Recharge une archive dans l'espace de travail (remap des identifiants)."""646 a = self.get_archive(archive_id)647 if not a:648 return {"ok": False}649 data = a["data"]650 idmap: dict[Any, int] = {}651 for e in data.get("entities", []):652 old = e.get("id")653 nid = self.upsert_entity(dict(e))654 if old is not None:655 idmap[old] = nid656 rel = 0657 for r in data.get("relations", []):658 f, t = idmap.get(r.get("from_entity")), idmap.get(r.get("to_entity"))659 if f and t:660 self.add_relation(f, t, r.get("relation_type", ""), r.get("role") or "",661 r.get("source_url") or "", float(r.get("confidence") or 0.6))662 rel += 1663 for m in data.get("missions", []):664 self.add_mission(m.get("goal", ""), m.get("seed") or "", m.get("sector") or "",665 m.get("region") or "", int(m.get("priority") or 5))666 for s in data.get("sources", []):667 self.record_source(s.get("url", ""), s.get("title") or "", "")668 return {"ok": True, "entities": len(idmap), "relations": rel}669670 def close(self) -> None:671 self.conn.close()672