spb/trouve-ka Public

1# Trouve-KA — boucle de maintenance2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45"""Scheduler : tâches périodiques légères.67- Relance les items in_progress abandonnés (worker mort) — dégradation gracieuse §13.8- Recalcule l'autorité de domaine à partir du graphe de liens (inlinks pondérés9  par le score Québec des domaines source) — version simple, PageRank-like plus tard.10"""1112import asyncio13import signal1415from trouveka.config import get_settings16from trouveka.database import Database17from trouveka.logging import get_logger1819log = get_logger("scheduler")2021STALE_RESET_INTERVAL = 60          # secondes22AUTHORITY_INTERVAL = 15 * 60       # secondes232425class SchedulerLoop:26    def __init__(self) -> None:27        self.s = get_settings()28        self.db = Database(self.s.database_url, pool_min=1, pool_max=3)29        self.stop_event = asyncio.Event()3031    async def recompute_authority(self) -> None:32        """Autorité ∈ [0,1] : log-saturation des inlinks pondérés par le Québec-score des sources."""33        await self.db.pool.execute(34            """35            WITH weighted AS (36                SELECT dl.to_domain_id AS id,37                       sum(least(dl.link_count, 50) * greatest(d.quebec_score, 0.1)) AS w,38                       count(DISTINCT dl.from_domain_id) AS in_domains39                FROM domain_links dl JOIN domains d ON d.id = dl.from_domain_id40                GROUP BY dl.to_domain_id41            )42            UPDATE domains SET43                authority_score = least(1.0, ln(1 + w.w) / ln(1 + 5000)),44                inlink_domains = w.in_domains45            FROM weighted w WHERE domains.id = w.id46            """47        )48        log.info("autorité de domaine recalculée")4950    async def start(self) -> None:51        await self.db.connect()52        loop = asyncio.get_running_loop()53        for sig in (signal.SIGINT, signal.SIGTERM):54            loop.add_signal_handler(sig, self.stop_event.set)55        log.info("scheduler démarré")5657        elapsed_authority = AUTHORITY_INTERVAL  # premier calcul immédiat58        while not self.stop_event.is_set():59            try:60                reset = await self.db.reset_stale_items(older_than_minutes=30)61                if reset:62                    log.info("items abandonnés relancés", extra={"ctx": {"count": reset}})63                if elapsed_authority >= AUTHORITY_INTERVAL:64                    await self.recompute_authority()65                    elapsed_authority = 066            except Exception:67                log.exception("erreur scheduler (on continue)")68            try:69                await asyncio.wait_for(self.stop_event.wait(), timeout=STALE_RESET_INTERVAL)70            except TimeoutError:71                pass72            elapsed_authority += STALE_RESET_INTERVAL73        await self.db.close()747576def main() -> None:77    asyncio.run(SchedulerLoop().start())787980if __name__ == "__main__":81    main()82