spb/vrai-prix

1// Auteur : Simon-Pierre Boucher — contact@spboucher.ai2/**3 * Persistance des connecteurs : observations brutes (historisées, hachées),4 * prix canoniques, grilles de main-d'œuvre, indices, runs, sources.5 * N'altère pas le schéma de db.ts : `ensureConnectorSchema` ajoute seulement6 * ses propres index/colonnes de façon tolérante.7 */8import type Database from "better-sqlite3";9import { getCostDb, metaSet, sourceIdByKey } from "../db";10import { invalidateCache } from "../cache";11import type { CanonicalIndexObservation, CanonicalLabourObservation, CanonicalPriceObservation, ConnectorKey, RawDocument, RawObservation, RunLogEntry, RunStats } from "./types";1213export function ensureConnectorSchema(d: Database.Database = getCostDb()): void {14  d.exec(`15    CREATE INDEX IF NOT EXISTS idx_raw_obs_url ON cost_raw_observations(source_url, retrieved_at);16    CREATE INDEX IF NOT EXISTS idx_item_prices_source_date ON cost_item_prices(source_id, observation_date);17    CREATE INDEX IF NOT EXISTS idx_labour_trade ON labour_rates(trade_code, sector, effective_from);18    CREATE INDEX IF NOT EXISTS idx_connector_runs_conn ON connector_runs(connector, started_at);19  `);20}2122const nowIso = () => new Date().toISOString();23const dateOf = (iso: string) => iso.slice(0, 10);2425/* -------------------------------------------------------- documents bruts */2627/** Le dernier hash vu pour une URL (détection de changement). */28export function lastHashFor(url: string, d: Database.Database = getCostDb()): string | null {29  const r = d.prepare("SELECT content_hash FROM cost_raw_observations WHERE source_url=? AND content_hash IS NOT NULL ORDER BY retrieved_at DESC LIMIT 1").get(url) as { content_hash: string } | undefined;30  return r?.content_hash ?? null;31}3233export function saveRawObservation(connector: ConnectorKey, obs: RawObservation, doc: RawDocument, status: "new" | "accepted" | "rejected" | "unchanged", rejectReason: string | null, parserVersion: string, d: Database.Database = getCostDb()): number {34  const sid = sourceIdByKey(connector, d);35  const r = d.prepare(`INSERT INTO cost_raw_observations(source_id,external_id,source_url,retrieved_at,effective_date,raw_title,raw_description,raw_unit,raw_price,raw_regular_price,raw_sale_price,raw_currency,raw_location,raw_category,raw_payload_json,content_hash,parser_version,status,reject_reason)36    VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`).run(37    sid, obs.externalId, obs.sourceUrl, obs.retrievedAt, obs.effectiveDate, obs.title.slice(0, 500), obs.description?.slice(0, 2000) ?? null, obs.unit, obs.price, obs.regularPrice, obs.salePrice, obs.currency, obs.location, obs.category,38    JSON.stringify(obs.payload).slice(0, 20000), doc.contentHash, parserVersion, status, rejectReason,39  );40  return Number(r.lastInsertRowid);41}4243/** Marque simplement qu'un document a été vu inchangé (une ligne légère par run). */44export function saveUnchangedMarker(connector: ConnectorKey, doc: RawDocument, d: Database.Database = getCostDb()): void {45  const sid = sourceIdByKey(connector, d);46  d.prepare(`INSERT INTO cost_raw_observations(source_id,source_url,retrieved_at,raw_title,raw_currency,raw_payload_json,content_hash,parser_version,status) VALUES(?,?,?,?,?,?,?,?,'unchanged')`)47    .run(sid, doc.url, doc.fetchedAt, String(doc.metadata.title ?? "").slice(0, 500), "CAD", "{}", doc.contentHash, "marker");48}4950/* ---------------------------------------------------------------- prix */5152export function itemIdByCode(code: string, d: Database.Database = getCostDb()): number | null {53  const r = d.prepare("SELECT id FROM cost_items WHERE canonical_code=?").get(code) as { id: number } | undefined;54  return r?.id ?? null;55}5657/** Une observation par jour et par (article, source) : la seconde du même jour remplace la première (même URL) sinon s'ajoute. */58export function savePrice(connector: ConnectorKey, o: CanonicalPriceObservation, rawId: number | null, confidence: number, isOutlier: boolean, outlierReason: string | null, d: Database.Database = getCostDb()): "inserted" | "updated" {59  const sid = sourceIdByKey(connector, d);60  const iid = itemIdByCode(o.itemCode, d);61  if (!iid) throw new Error(`article inconnu ${o.itemCode}`);62  const existing = d.prepare("SELECT id FROM cost_item_prices WHERE cost_item_id=? AND source_id=? AND observation_date=? AND source_url=?").get(iid, sid, o.observationDate, o.sourceUrl) as { id: number } | undefined;63  if (existing) {64    d.prepare(`UPDATE cost_item_prices SET total_cost=?, material_cost=?, source_unit=?, conversion_factor=?, is_regular_price=?, is_outlier=?, outlier_reason=?, confidence_score=?, raw_observation_id=? WHERE id=?`)65      .run(o.totalCost, o.totalCost, o.sourceUnit, o.conversionFactor, o.isRegularPrice ? 1 : 0, isOutlier ? 1 : 0, outlierReason, confidence, rawId, existing.id);66    return "updated";67  }68  d.prepare(`INSERT INTO cost_item_prices(cost_item_id,source_id,location_code,observation_date,price_kind,material_cost,total_cost,currency,source_unit,conversion_factor,is_regular_price,is_outlier,outlier_reason,confidence_score,source_url,raw_observation_id,verified)69    VALUES(?,?,?,?,'observed',?,?,?,?,?,?,?,?,?,?,?,0)`).run(iid, sid, o.locationCode, o.observationDate, o.totalCost, o.totalCost, o.currency, o.sourceUnit, o.conversionFactor, o.isRegularPrice ? 1 : 0, isOutlier ? 1 : 0, outlierReason, confidence, o.sourceUrl, rawId);70  return "inserted";71}7273/** Historique récent des prix canoniques d'un article (autres sources incluses) — pour la détection d'aberrations. */74export function recentPricesFor(itemCode: string, days = 120, d: Database.Database = getCostDb()): number[] {75  return (d.prepare(`SELECT p.total_cost v FROM cost_item_prices p JOIN cost_items i ON i.id=p.cost_item_id WHERE i.canonical_code=? AND p.is_outlier=0 AND p.observation_date >= date('now', ?) ORDER BY p.observation_date DESC LIMIT 60`).all(itemCode, `-${days} days`) as { v: number }[]).map((r) => r.v);76}7778export function referencePriceFor(itemCode: string, d: Database.Database = getCostDb()): number | null {79  const r = d.prepare("SELECT reference_price v FROM cost_items WHERE canonical_code=?").get(itemCode) as { v: number | null } | undefined;80  return r?.v ?? null;81}8283/* ------------------------------------------------------------ main-d'œuvre */8485export function saveLabour(connector: ConnectorKey, o: CanonicalLabourObservation, rawId: number | null, d: Database.Database = getCostDb()): "inserted" | "updated" {86  const sid = sourceIdByKey(connector, d);87  const ex = d.prepare("SELECT id FROM labour_rates WHERE trade_code=? AND sector=? AND classification=? AND region=? AND effective_from=? AND source_id=?").get(o.tradeCode, o.sector, o.classification, o.region, o.effectiveFrom, sid) as { id: number } | undefined;88  if (ex) {89    d.prepare(`UPDATE labour_rates SET trade_name_fr=?, base_wage=?, vacation_cost=?, benefits_cost=?, employer_contributions=?, other_contributions=?, total_employer_cost=?, source_url=?, confidence_score=?, raw_observation_id=? WHERE id=?`)90      .run(o.tradeNameFr, o.baseWage, o.vacationCost, o.benefitsCost, o.employerContributions, o.otherContributions, o.totalEmployerCost, o.sourceUrl, o.confidence, rawId, ex.id);91    return "updated";92  }93  d.prepare(`INSERT INTO labour_rates(trade_code,trade_name_fr,trade_name_en,sector,classification,region,effective_from,effective_to,base_wage,vacation_cost,benefits_cost,employer_contributions,other_contributions,total_employer_cost,source_id,source_url,confidence_score,raw_observation_id)94    VALUES(?,?,?,?,?,?,?,NULL,?,?,?,?,?,?,?,?,?,?)`).run(o.tradeCode, o.tradeNameFr, null, o.sector, o.classification, o.region, o.effectiveFrom, o.baseWage, o.vacationCost, o.benefitsCost, o.employerContributions, o.otherContributions, o.totalEmployerCost, sid, o.sourceUrl, o.confidence, rawId);95  return "inserted";96}9798/** Recalcule effective_to = date de la grille suivante pour chaque série (métier, secteur, classification, région, source). */99export function closeLabourPeriods(connector: ConnectorKey, d: Database.Database = getCostDb()): void {100  const sid = sourceIdByKey(connector, d);101  d.exec(`UPDATE labour_rates SET effective_to = (102      SELECT MIN(l2.effective_from) FROM labour_rates l2103      WHERE l2.trade_code=labour_rates.trade_code AND l2.sector=labour_rates.sector AND l2.classification=labour_rates.classification AND l2.region=labour_rates.region AND l2.source_id=labour_rates.source_id AND l2.effective_from > labour_rates.effective_from)104    WHERE source_id=${sid}`);105}106107/* -------------------------------------------------------------- indices */108109export function saveIndex(o: CanonicalIndexObservation, d: Database.Database = getCostDb()): void {110  d.prepare(`INSERT INTO construction_cost_indices(source,index_code,geography,building_type,division,period,index_value,pct_change_qoq,pct_change_yoy,retrieved_at) VALUES(?,?,?,?,?,?,?,?,?,?)111    ON CONFLICT(index_code,period) DO UPDATE SET index_value=excluded.index_value, pct_change_qoq=excluded.pct_change_qoq, pct_change_yoy=excluded.pct_change_yoy, retrieved_at=excluded.retrieved_at`)112    .run("Statistique Canada 18-10-0289-01", o.indexCode, o.geography, o.buildingType, o.division, o.period, o.value, o.pctQoq, o.pctYoy, o.retrievedAt);113}114115/* ---------------------------------------------------------------- runs */116117export function startRun(connector: ConnectorKey, d: Database.Database = getCostDb()): RunStats {118  const startedAt = nowIso();119  const r = d.prepare("INSERT INTO connector_runs(connector,started_at,status) VALUES(?,?,'running')").run(connector, startedAt);120  return { connector, runId: Number(r.lastInsertRowid), startedAt, finishedAt: null, status: "running", pages: 0, observations: 0, accepted: 0, rejected: 0, unchanged: 0, errors: 0, log: [] };121}122123export function finishRun(run: RunStats, d: Database.Database = getCostDb()): RunStats {124  run.finishedAt = nowIso();125  if (run.status === "running") run.status = run.errors > 0 ? (run.accepted > 0 ? "partial" : "error") : "ok";126  d.prepare("UPDATE connector_runs SET finished_at=?, status=?, pages=?, observations=?, accepted=?, rejected=?, unchanged=?, errors=?, duration_ms=?, log_json=? WHERE id=?")127    .run(run.finishedAt, run.status, run.pages, run.observations, run.accepted, run.rejected, run.unchanged, run.errors, new Date(run.finishedAt).getTime() - new Date(run.startedAt).getTime(), JSON.stringify({ message: run.message ?? null, entries: run.log.slice(-400) }), run.runId);128  const sid = sourceIdByKey(run.connector, d);129  if (run.status === "ok" || run.status === "partial") d.prepare("UPDATE cost_sources SET last_successful_sync=?, last_error=?, updated_at=datetime('now') WHERE id=?").run(run.finishedAt, run.status === "partial" ? `${run.errors} erreur(s) — voir connector_runs #${run.runId}` : null, sid);130  else d.prepare("UPDATE cost_sources SET last_error=?, updated_at=datetime('now') WHERE id=?").run(run.message ?? `${run.status} — connector_runs #${run.runId}`, sid);131  if (run.accepted > 0) {132    metaSet("last_ingest", run.finishedAt, d);133    invalidateCache();134  }135  return run;136}137138export function pushLog(run: RunStats, e: RunLogEntry, cb?: (e: RunLogEntry) => void): void {139  run.log.push(e);140  cb?.(e);141}142143export function lastSuccessfulSync(connector: ConnectorKey, d: Database.Database = getCostDb()): string | null {144  const r = d.prepare("SELECT last_successful_sync v FROM cost_sources WHERE key=?").get(connector) as { v: string | null } | undefined;145  return r?.v ?? null;146}147148export function isSourceActive(connector: ConnectorKey, d: Database.Database = getCostDb()): boolean {149  const r = d.prepare("SELECT is_active v FROM cost_sources WHERE key=?").get(connector) as { v: number } | undefined;150  return (r?.v ?? 0) === 1;151}152153export { dateOf, nowIso };154