spb/vrai-prix

1// Auteur : Simon-Pierre Boucher — contact@spboucher.ai2/**3 * Orchestrateur d'un connecteur : discover → fetch (hash) → extract → normalize4 * → validate → persistance, avec journal structuré et statistiques de run.5 * Utilisé par scripts/cost-sync.ts (planificateur) et l'API admin.6 */7import type Database from "better-sqlite3";8import { getCostDb } from "../db";9import { CONNECTORS, type RunnableKey } from "./registry";10import { closeLabourPeriods, ensureConnectorSchema, finishRun, isSourceActive, lastSuccessfulSync, pushLog, saveIndex, saveLabour, savePrice, saveRawObservation, saveUnchangedMarker, startRun } from "./store";11import type { CanonicalObservation, RawDocument, RunLogEntry, RunOptions, RunStats } from "./types";12import { APCHQ_PARSER_VERSION } from "./apchq";13import { STATCAN_PARSER_VERSION } from "./statcan";14import { CCQ_PARSER_VERSION } from "./ccq";15import { RETAIL_PARSER_VERSION } from "./retail";1617const PARSER_VERSION: Record<RunnableKey, string> = { apchq: APCHQ_PARSER_VERSION, ccq: CCQ_PARSER_VERSION, statcan: STATCAN_PARSER_VERSION, canac: RETAIL_PARSER_VERSION, bmr: RETAIL_PARSER_VERSION, patrickmorin: RETAIL_PARSER_VERSION, rona: RETAIL_PARSER_VERSION, homedepot: RETAIL_PARSER_VERSION };1819/** Le connecteur est-il dû (fréquence nominale écoulée) ? */20export function isDue(key: RunnableKey, d: Database.Database = getCostDb()): boolean {21  const last = lastSuccessfulSync(key, d);22  if (!last) return true;23  return Date.now() - new Date(last).getTime() >= CONNECTORS[key].config.refreshDays * 86400000 - 3600000;24}2526export async function runConnector(key: RunnableKey, opts: RunOptions = {}): Promise<RunStats> {27  const d = opts.db ?? getCostDb();28  ensureConnectorSchema(d);29  const c = CONNECTORS[key];30  const run = startRun(key, d);31  const log = (e: RunLogEntry) => pushLog(run, e, opts.log);32  const ts = () => new Date().toISOString();33  if (!isSourceActive(key, d) && !opts.force) {34    run.status = "skipped";35    run.message = "source inactive (cost_sources.is_active = 0)";36    log({ connector: key, url: null, status: "skipped", duration: 0, observations: 0, validationErrors: 0, timestamp: ts(), message: run.message });37    return finishRun(run, d);38  }39  let docs: Awaited<ReturnType<typeof c.discover>> = [];40  try {41    docs = await c.discover({ ...opts, db: d, log });42    log({ connector: key, url: null, status: "ok", duration: 0, observations: docs.length, validationErrors: 0, timestamp: ts(), message: `${docs.length} document(s) découvert(s)` });43  } catch (e) {44    run.errors++;45    run.status = "error";46    run.message = `découverte échouée : ${(e as Error).message}`;47    log({ connector: key, url: null, status: "error", duration: 0, observations: 0, validationErrors: 0, timestamp: ts(), message: run.message });48    return finishRun(run, d);49  }50  const max = opts.maxPages ?? c.config.maxPages;51  const queue = docs.slice(0, max);52  const workers = Math.max(1, c.config.concurrency);53  let idx = 0;54  const worker = async () => {55    for (;;) {56      const i = idx++;57      if (i >= queue.length) return;58      const doc = queue[i];59      const t0 = Date.now();60      let raw: RawDocument | null = null;61      try {62        raw = await c.fetch(doc, { ...opts, db: d, log });63        run.pages++;64        if (raw.statusCode != null && raw.statusCode >= 400) {65          run.errors++;66          log({ connector: key, url: doc.url, status: "error", duration: Date.now() - t0, observations: 0, validationErrors: 0, timestamp: ts(), message: `HTTP ${raw.statusCode}` });67          continue;68        }69        if (raw.unchanged) {70          run.unchanged++;71          saveUnchangedMarker(key, raw, d);72          log({ connector: key, url: doc.url, status: "unchanged", duration: Date.now() - t0, observations: 0, validationErrors: 0, timestamp: ts() });73          continue;74        }75        const rawObs = await c.extract(raw, { ...opts, db: d, log });76        if (!rawObs.length) {77          const unavailable = key === "ccq" && !(raw.metadata.hasRateTable as boolean | undefined);78          if (unavailable) { run.status = "unavailable"; run.message = String(raw.metadata.incident ?? "aucune grille de taux détectable"); }79          else run.errors++;80          saveRawObservation(key, { externalId: null, sourceUrl: doc.url, retrievedAt: raw.fetchedAt, effectiveDate: null, title: String(raw.metadata.title ?? doc.title ?? ""), description: null, unit: null, price: null, regularPrice: null, salePrice: null, currency: "CAD", location: null, category: null, payload: { statusCode: raw.statusCode, unavailable } }, raw, "rejected", unavailable ? "source indisponible" : "aucune observation extraite (structure inattendue ou page sans prix)", PARSER_VERSION[key], d);81          log({ connector: key, url: doc.url, status: unavailable ? "unavailable" : "rejected", duration: Date.now() - t0, observations: 0, validationErrors: 1, timestamp: ts(), message: unavailable ? run.message : "aucune observation extraite" });82          continue;83        }84        const canon = await c.normalize(rawObs, raw, { ...opts, db: d, log });85        const { accepted, rejected } = c.validate(canon, { ...opts, db: d, log });86        run.observations += canon.length;87        const tx = d.transaction(() => {88          for (const o of accepted) persist(key, o, raw!, PARSER_VERSION[key], d);89          for (const r of rejected) saveRawObservation(key, rawOf(r.obs, raw!, doc.url), raw!, "rejected", r.reason, PARSER_VERSION[key], d);90        });91        tx();92        run.accepted += accepted.length;93        run.rejected += rejected.length;94        log({ connector: key, url: doc.url, status: accepted.length ? "ok" : "rejected", duration: Date.now() - t0, observations: accepted.length, validationErrors: rejected.length, timestamp: ts(), message: rejected.length ? rejected.slice(0, 3).map((r) => r.reason).join(" ; ") : undefined });95      } catch (e) {96        run.errors++;97        log({ connector: key, url: doc.url, status: "error", duration: Date.now() - t0, observations: 0, validationErrors: 0, timestamp: ts(), message: (e as Error).message });98      }99    }100  };101  await Promise.all(Array.from({ length: workers }, worker));102  if (key === "apchq" && run.accepted) closeLabourPeriods("apchq", d);103  if (run.status === "unavailable" && run.accepted === 0) return finishRun(run, d);104  return finishRun(run, d);105}106107function rawOf(o: CanonicalObservation, raw: RawDocument, url: string) {108  if (o.kind === "index") return { externalId: o.indexCode, sourceUrl: url, retrievedAt: raw.fetchedAt, effectiveDate: o.period, title: o.indexCode, description: null, unit: "index", price: o.value, regularPrice: null, salePrice: null, currency: "CAD", location: o.geography, category: "index", payload: {} };109  return o.raw;110}111112function persist(key: RunnableKey, o: CanonicalObservation, raw: RawDocument, parser: string, d: Database.Database): void {113  if (o.kind === "index") { saveIndex(o, d); return; }114  if (o.kind === "labour") {115    const rawId = saveRawObservation(key, o.raw, raw, "accepted", null, parser, d);116    saveLabour(key, o, rawId, d);117    return;118  }119  const rawId = saveRawObservation(key, o.raw, raw, "accepted", null, parser, d);120  const conf = Math.round(60 + (o.isRegularPrice ? 15 : 0) + (o.conversionFactor === 1 ? 10 : 5));121  savePrice(key, o, rawId, conf, false, null, d);122}123