spb/uqo-eval

1// UQO Éval — Métré : pipeline d'ingestion des sources (discover → download → parse → normalize → classify → validate → dedupe → publish). SERVEUR seulement.2/**3 * Chaque document téléchargé est conservé tel quel (data/metre-files/_sources/<clé>/<hash>.<ext>)4 * avec URL, éditeur, date de récupération, licence, hash, version d'analyseur et5 * statut QA. Idempotence : même (source, hash) → document ignoré ; même6 * (source, identifiant externe) → observation ignorée (INSERT OR IGNORE).7 */8import { createHash, randomUUID } from "crypto";9import fs from "fs";10import path from "path";11import type Database from "better-sqlite3";12import { audit, metreFilesDir, nowIso } from "../db";13import { HttpError, type Ctx } from "../auth";14import { PERMITS_DATASET_URL, PERMITS_PARSER_VERSION, parsePermitStatsCsv, summarizePermitDetailCsv } from "./permits";15import { SCHEDULE_PARSER_VERSION, parseScheduleCsv } from "./schedule-import";16import { SEAO_DATASET_URL, SEAO_PARSER_VERSION, listWeeklyResources, parseSeaoPackage } from "./seao";1718const UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.1 Safari/605.1.15 UQO-Eval-Metre/1.0 (+https://www.uqo-eval.app)";19const FETCH_TIMEOUT_MS = 120_000;20const MAX_DOWNLOAD = 200 * 1024 * 1024;2122async function download(url: string): Promise<{ buf: Buffer; status: number; contentType: string | null }> {23  const ac = new AbortController();24  const t = setTimeout(() => ac.abort(), FETCH_TIMEOUT_MS);25  try {26    const r = await fetch(url, { headers: { "User-Agent": UA, Accept: "application/json, text/csv, */*" }, signal: ac.signal, redirect: "follow" });27    const ab = await r.arrayBuffer();28    if (ab.byteLength > MAX_DOWNLOAD) throw new Error("téléchargement trop volumineux");29    return { buf: Buffer.from(ab), status: r.status, contentType: r.headers.get("content-type") };30  } finally { clearTimeout(t); }31}3233function rawDir(sourceKey: string): string { const d = path.join(metreFilesDir(), "_sources", sourceKey); fs.mkdirSync(d, { recursive: true }); return d; }3435interface DocMeta { url: string | null; documentId: string | null; title: string; publisher: string; publicationDate: string | null; effectiveDate: string | null; license: string; parserVersion: string; ext: string }3637/** Enregistre un document brut ; renvoie null s'il existe déjà (même hash). */38function storeDocument(d: Database.Database, sourceKey: string, buf: Buffer, meta: DocMeta): { id: string; hash: string; rawPath: string } | null {39  const hash = createHash("sha256").update(buf).digest("hex");40  const exists = d.prepare("SELECT id FROM metre_source_documents WHERE source_key=? AND content_hash=?").get(sourceKey, hash) as { id: string } | undefined;41  if (exists) return null;42  const rawPath = path.join(rawDir(sourceKey), `${hash}.${meta.ext}`);43  if (!fs.existsSync(rawPath)) fs.writeFileSync(rawPath, buf);44  const id = randomUUID();45  d.prepare("INSERT INTO metre_source_documents(id,source_key,url,document_id,title,publisher,publication_date,effective_date,retrieved_at,license,content_hash,raw_path,bytes,parser_version,qa_status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,'new')")46    .run(id, sourceKey, meta.url, meta.documentId, meta.title, meta.publisher, meta.publicationDate, meta.effectiveDate, nowIso(), meta.license, hash, rawPath, buf.length, meta.parserVersion);47  return { id, hash, rawPath };48}4950interface ObsInput { externalId: string; resourceCode?: string | null; description: string; originalUnit: string | null; originalAmount: number | null; normalizedUnit?: string | null; normalizedAmount?: number | null; includesMaterial?: boolean | null; includesLabour?: boolean | null; includesEquipment?: boolean | null; taxBasis?: string | null; geography?: string | null; regionCode?: string | null; buyer?: string | null; classificationScheme?: string | null; classificationCode?: string | null; classificationLabel?: string | null; dateBasis?: string | null; effectiveFrom?: string | null; effectiveTo?: string | null; extractionMethod: string; lineage: unknown; uncertaintyNote?: string | null; sourceUrl?: string | null }5152function insertObservations(d: Database.Database, sourceKey: string, documentId: string, evidenceClass: string, license: string, obs: ObsInput[]): { inserted: number; skipped: number } {53  const ins = d.prepare(`INSERT OR IGNORE INTO metre_price_observations(id,source_key,document_id,evidence_class,external_id,resource_code,original_description,original_unit,original_amount,normalized_unit,normalized_amount,currency,includes_material,includes_labour,includes_equipment,tax_basis,geography,region_code,buyer,classification_scheme,classification_code,classification_label,date_basis,effective_from,effective_to,license,extraction_method,lineage_json,verification_status,uncertainty_note,source_url)54    VALUES(?,?,?,?,?,?,?,?,?,?,?,'CAD',?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'unverified',?,?)`);55  let inserted = 0, skipped = 0;56  const b = (v: boolean | null | undefined) => (v == null ? null : v ? 1 : 0);57  const tx = d.transaction(() => {58    for (const o of obs) {59      const r = ins.run(randomUUID(), sourceKey, documentId, evidenceClass, o.externalId, o.resourceCode ?? null, o.description, o.originalUnit, o.originalAmount, o.normalizedUnit ?? o.originalUnit, o.normalizedAmount ?? o.originalAmount, b(o.includesMaterial), b(o.includesLabour), b(o.includesEquipment), o.taxBasis ?? "unknown", o.geography ?? null, o.regionCode ?? null, o.buyer ?? null, o.classificationScheme ?? null, o.classificationCode ?? null, o.classificationLabel ?? null, o.dateBasis ?? null, o.effectiveFrom ?? null, o.effectiveTo ?? null, license, o.extractionMethod, JSON.stringify(o.lineage).slice(0, 2000), o.uncertaintyNote ?? null, o.sourceUrl ?? null);60      if (r.changes) inserted++; else skipped++;61    }62  });63  tx();64  return { inserted, skipped };65}6667function flag(d: Database.Database, entityType: string, entityId: string, severity: "info" | "warning" | "error", code: string, message: string): void {68  d.prepare("INSERT INTO metre_quality_flags(entity_type,entity_id,severity,code,message) VALUES(?,?,?,?,?)").run(entityType, entityId, severity, code, message.slice(0, 500));69}7071function startRun(d: Database.Database, sourceKey: string, by: string): number {72  return Number(d.prepare("INSERT INTO metre_ingest_runs(source_key,started_at,triggered_by) VALUES(?,?,?)").run(sourceKey, nowIso(), by).lastInsertRowid);73}74function endRun(d: Database.Database, id: number, status: string, stats: { documents: number; records: number; skipped: number; errors: number }, log: unknown): void {75  d.prepare("UPDATE metre_ingest_runs SET finished_at=?, status=?, documents=?, records=?, skipped=?, errors=?, log_json=? WHERE id=?").run(nowIso(), status, stats.documents, stats.records, stats.skipped, stats.errors, JSON.stringify(log).slice(0, 20000), id);76  const src = (d.prepare("SELECT source_key FROM metre_ingest_runs WHERE id=?").get(id) as { source_key: string }).source_key;77  d.prepare("UPDATE metre_sources SET last_run_at=?, last_status=?, last_error=? WHERE key=?").run(nowIso(), status, status === "error" ? JSON.stringify(log).slice(0, 500) : null, src);78}7980/* ------------------------------------------------------------------ SEAO */8182export async function ingestSeao(ctx: Ctx, opts: { weeks?: number } = {}): Promise<{ runId: number; documents: number; records: number; skipped: number; log: unknown[] }> {83  const d = ctx.db;84  const runId = startRun(d, "seao", ctx.user.id);85  const log: unknown[] = [];86  const stats = { documents: 0, records: 0, skipped: 0, errors: 0 };87  try {88    const pk = await download(SEAO_DATASET_URL);89    if (pk.status !== 200) throw new Error(`CKAN package_show HTTP ${pk.status}`);90    const resources = listWeeklyResources(JSON.parse(pk.buf.toString("utf8"))).slice(0, Math.min(12, Math.max(1, opts.weeks ?? 2)));91    for (const res of resources) {92      const known = d.prepare("SELECT id FROM metre_source_documents WHERE source_key='seao' AND url=?").get(res.url) as { id: string } | undefined;93      if (known) { stats.skipped++; log.push({ resource: res.name, status: "déjà ingéré (URL)" }); continue; }94      const dl = await download(res.url);95      if (dl.status !== 200) { stats.errors++; log.push({ resource: res.name, status: `HTTP ${dl.status}` }); continue; }96      let json: unknown;97      try { json = JSON.parse(dl.buf.toString("utf8")); } catch { stats.errors++; log.push({ resource: res.name, status: "JSON invalide" }); continue; }98      const parsed = parseSeaoPackage(json);99      const doc = storeDocument(d, "seao", dl.buf, { url: res.url, documentId: res.name, title: `SEAO hebdo ${res.from} → ${res.to}`, publisher: "Secrétariat du Conseil du trésor", publicationDate: parsed.publishedDate?.slice(0, 10) ?? null, effectiveDate: res.to, license: "CC-BY 4.0", parserVersion: SEAO_PARSER_VERSION, ext: "json" });100      if (!doc) { stats.skipped++; log.push({ resource: res.name, status: "déjà ingéré (hash)" }); continue; }101      stats.documents++;102      const r = insertObservations(d, "seao", doc.id, "awarded_contract_total", "CC-BY 4.0", parsed.observations.map((o) => ({103        externalId: o.externalId, description: `${o.title}${o.itemDescription ? ` — ${o.itemDescription}` : ""}`.slice(0, 400), originalUnit: "contrat", originalAmount: o.amount, normalizedUnit: "contrat", normalizedAmount: o.amount,104        includesMaterial: null, includesLabour: null, includesEquipment: null, taxBasis: "unknown", geography: o.supplierLocality ? `${o.supplierLocality}, ${o.supplierRegion ?? ""}` : o.supplierRegion, regionCode: null, buyer: o.buyer,105        classificationScheme: o.unspscCode ? "UNSPSC" : null, classificationCode: o.unspscCode, classificationLabel: o.unspscLabel, dateBasis: o.awardDate, effectiveFrom: o.awardDate, extractionMethod: "ocds-json",106        lineage: { ocid: o.ocid, method: o.procurementMethod, tenderers: o.tenderers, municipal: o.buyerMunicipal, supplier: o.supplier }, uncertaintyNote: "Montant total d'un contrat octroyé : NE PAS interpréter comme un prix unitaire ni comme un coût résidentiel installé.", sourceUrl: o.noticeUrl,107      })));108      d.prepare("UPDATE metre_source_documents SET qa_status='published', records=?, qa_notes=? WHERE id=?").run(r.inserted, JSON.stringify({ releases: parsed.releases, works: parsed.observations.length, skipped: parsed.skipped }), doc.id);109      stats.records += r.inserted; stats.skipped += r.skipped;110      if (parsed.observations.length === 0) flag(d, "source_document", doc.id, "warning", "no_records", `Aucune release « works » avec octroi dans ${res.name}`);111      log.push({ resource: res.name, releases: parsed.releases, inserted: r.inserted, duplicates: r.skipped, skipped: parsed.skipped });112    }113    endRun(d, runId, stats.errors && !stats.documents ? "error" : stats.errors ? "partial" : "ok", stats, log);114  } catch (e) {115    stats.errors++; log.push({ error: (e as Error).message });116    endRun(d, runId, "error", stats, log);117  }118  audit(d, { userId: ctx.user.id, role: ctx.user.role, action: "ingest.seao", details: stats, ip: ctx.ip });119  return { runId, ...stats, log };120}121122/* --------------------------------------------------------- permis Montréal */123124export async function ingestPermits(ctx: Ctx, opts: { url?: string; detail?: boolean } = {}): Promise<{ runId: number; documents: number; records: number; skipped: number; log: unknown[] }> {125  const d = ctx.db;126  const runId = startRun(d, "mtl_permits", ctx.user.id);127  const log: unknown[] = [];128  const stats = { documents: 0, records: 0, skipped: 0, errors: 0 };129  try {130    let statsUrl = opts.url ?? null, detailUrl: string | null = null;131    if (!statsUrl) {132      const pk = await download(PERMITS_DATASET_URL);133      if (pk.status !== 200) throw new Error(`CKAN package_show HTTP ${pk.status}`);134      const res = ((JSON.parse(pk.buf.toString("utf8")) as { result?: { resources?: { url?: string; format?: string }[] } }).result?.resources ?? []).filter((r) => (r.format ?? "").toUpperCase() === "CSV" && r.url);135      statsUrl = res.find((r) => /statistiques/i.test(r.url!))?.url ?? null;136      detailUrl = res.find((r) => /permis-construction\.csv$/i.test(r.url!))?.url ?? null;137      if (!statsUrl) throw new Error("ressource « statistiques-permis-de-construction.csv » introuvable dans le jeu de données");138    }139    // 1. agrégats avec valeur déclarée des travaux140    const dl = await download(statsUrl);141    if (dl.status !== 200) throw new Error(`téléchargement HTTP ${dl.status} (${statsUrl}) — importer le CSV manuellement`);142    const doc = storeDocument(d, "mtl_permits", dl.buf, { url: statsUrl, documentId: path.basename(statsUrl), title: "Statistiques des permis de construction (valeur déclarée des travaux par année, arrondissement et type) — Ville de Montréal", publisher: "Ville de Montréal", publicationDate: null, effectiveDate: null, license: "CC-BY 4.0", parserVersion: PERMITS_PARSER_VERSION, ext: "csv" });143    if (!doc) { stats.skipped++; log.push({ resource: "statistiques", status: "déjà ingéré (hash)" }); }144    else {145      stats.documents++;146      const parsed = parsePermitStatsCsv(dl.buf.toString("utf8"));147      const r = insertObservations(d, "mtl_permits", doc.id, "permit_declared_value", "CC-BY 4.0", parsed.observations.map((o) => ({148        externalId: o.externalId, description: `${o.typeLabel} — ${o.borough} — ${o.year} : ${o.permits} permis, valeur déclarée totale des travaux`.slice(0, 400), originalUnit: "total annuel", originalAmount: o.declaredWorksValue, normalizedUnit: "permis (moyenne)", normalizedAmount: o.avgPerPermit, taxBasis: "unknown",149        geography: `Montréal — ${o.borough}`, regionCode: "QC-MTL", classificationScheme: "mtl_permit_type", classificationCode: o.typeCode, classificationLabel: o.typeLabel, dateBasis: `${o.year}-12-31`, effectiveFrom: `${o.year}-01-01`, effectiveTo: `${o.year}-12-31`,150        extractionMethod: "csv", lineage: { permits: o.permits, permitFees: o.permitFees, avgPerPermit: o.avgPerPermit }, uncertaintyNote: "Valeur des travaux DÉCLARÉE par les requérants, agrégée par année/arrondissement/type (sans valeur légale selon la Ville) — ni coût réel ni prix unitaire.", sourceUrl: statsUrl,151      })));152      d.prepare("UPDATE metre_source_documents SET qa_status='published', records=?, qa_notes=? WHERE id=?").run(r.inserted, JSON.stringify({ rows: parsed.rows, skipped: parsed.skipped, columns: parsed.columns }), doc.id);153      stats.records += r.inserted; stats.skipped += r.skipped;154      log.push({ resource: "statistiques", rows: parsed.rows, inserted: r.inserted, duplicates: r.skipped });155    }156    // 2. détail (sans valeur) : résumé statistique seulement, document conservé157    if (detailUrl && opts.detail !== false) {158      const dd = await download(detailUrl);159      if (dd.status !== 200) { stats.errors++; log.push({ resource: "detail", status: `HTTP ${dd.status}` }); }160      else {161        const doc2 = storeDocument(d, "mtl_permits", dd.buf, { url: detailUrl, documentId: path.basename(detailUrl), title: "Permis de construction, transformation et démolition — détail (sans valeur des travaux)", publisher: "Ville de Montréal", publicationDate: null, effectiveDate: null, license: "CC-BY 4.0", parserVersion: PERMITS_PARSER_VERSION, ext: "csv" });162        if (!doc2) { stats.skipped++; log.push({ resource: "detail", status: "déjà ingéré (hash)" }); }163        else {164          stats.documents++;165          const sum = summarizePermitDetailCsv(dd.buf.toString("utf8"), 2023);166          d.prepare("UPDATE metre_source_documents SET qa_status='published', records=0, qa_notes=? WHERE id=?").run(JSON.stringify(sum), doc2.id);167          if (!sum.hasValueColumn) flag(d, "source_document", doc2.id, "info", "no_value_column", "Le détail des permis ne publie aucune valeur de travaux : résumé statistique seulement (aucune observation).");168          log.push({ resource: "detail", rows_since_2023: sum.rows, years: sum.years, hasValueColumn: sum.hasValueColumn });169        }170      }171    }172    endRun(d, runId, stats.errors ? "partial" : "ok", stats, log);173  } catch (e) {174    stats.errors++; log.push({ error: (e as Error).message });175    endRun(d, runId, "error", stats, log);176  }177  audit(d, { userId: ctx.user.id, role: ctx.user.role, action: "ingest.mtl_permits", details: stats, ip: ctx.ip });178  return { runId, ...stats, log };179}180181/* ------------------------------------------------------- import manuel CSV */182183export function importScheduleCsv(ctx: Ctx, p: { sourceKey: string; csv: string; title: string; url: string | null; publicationDate: string | null; effectiveDate: string | null; license: string; evidenceClass: "published_schedule" | "awarded_unit_bid" | "supplier_quote" | "contractor_quote"; geography: string | null; regionCode: string | null }): { documentId: string | null; inserted: number; skipped: number; parseSkipped: { line: number; reason: string }[] } {184  const d = ctx.db;185  const src = d.prepare("SELECT key, publisher FROM metre_sources WHERE key=?").get(p.sourceKey) as { key: string; publisher: string } | undefined;186  if (!src) throw new HttpError(400, "Source inconnue.");187  if (!p.license.trim()) throw new HttpError(400, "Licence / droits de réutilisation requis pour tout import.");188  if (p.csv.length > 20 * 1024 * 1024) throw new HttpError(413, "CSV trop volumineux.");189  let parsed;190  try { parsed = parseScheduleCsv(p.csv); } catch (e) { throw new HttpError(400, (e as Error).message); }191  const runId = startRun(d, p.sourceKey, ctx.user.id);192  const doc = storeDocument(d, p.sourceKey, Buffer.from(p.csv, "utf8"), { url: p.url, documentId: null, title: p.title.slice(0, 200), publisher: src.publisher, publicationDate: p.publicationDate, effectiveDate: p.effectiveDate, license: p.license.slice(0, 200), parserVersion: SCHEDULE_PARSER_VERSION, ext: "csv" });193  if (!doc) { endRun(d, runId, "ok", { documents: 0, records: 0, skipped: 1, errors: 0 }, [{ status: "déjà importé (hash)" }]); return { documentId: null, inserted: 0, skipped: 1, parseSkipped: parsed.skipped }; }194  const r = insertObservations(d, p.sourceKey, doc.id, p.evidenceClass, p.license, parsed.rows.map((o) => ({195    externalId: `${doc.hash.slice(0, 12)}:${o.externalId}`, resourceCode: o.code, description: o.description, originalUnit: o.unit, originalAmount: o.amount, includesMaterial: o.includesMaterial, includesLabour: o.includesLabour, includesEquipment: o.includesEquipment, taxBasis: o.taxBasis,196    geography: o.region ?? p.geography, regionCode: p.regionCode, classificationLabel: o.category, dateBasis: o.effectiveDate ?? p.effectiveDate, effectiveFrom: o.effectiveDate ?? p.effectiveDate, extractionMethod: "manual-csv", lineage: { line: o.externalId, note: o.note, importedBy: ctx.user.id },197    uncertaintyNote: p.evidenceClass === "published_schedule" ? "Barème publié (infrastructure/municipal) — ne pas assimiler à un prix résidentiel installé sans vérification." : null, sourceUrl: p.url,198  })));199  d.prepare("UPDATE metre_source_documents SET qa_status='parsed', records=?, qa_notes=? WHERE id=?").run(r.inserted, JSON.stringify({ header: parsed.header, skipped: parsed.skipped.slice(0, 50) }), doc.id);200  for (const s of parsed.skipped.slice(0, 20)) flag(d, "source_document", doc.id, "info", "row_skipped", `ligne ${s.line} : ${s.reason}`);201  endRun(d, runId, "ok", { documents: 1, records: r.inserted, skipped: r.skipped + parsed.skipped.length, errors: 0 }, [{ inserted: r.inserted, duplicates: r.skipped, parseSkipped: parsed.skipped.length }]);202  audit(d, { userId: ctx.user.id, role: ctx.user.role, action: "ingest.manual", entityType: "source_document", entityId: doc.id, details: { source: p.sourceKey, inserted: r.inserted }, ip: ctx.ip });203  return { documentId: doc.id, inserted: r.inserted, skipped: r.skipped, parseSkipped: parsed.skipped };204}205206/* ------------------------------------------------------------------ QA */207208export function setDocumentQa(ctx: Ctx, documentId: string, status: "published" | "rejected", notes: string | null): void {209  ctx.db.prepare("UPDATE metre_source_documents SET qa_status=?, qa_notes=COALESCE(?, qa_notes) WHERE id=?").run(status, notes, documentId);210  ctx.db.prepare("UPDATE metre_price_observations SET verification_status=? WHERE document_id=?").run(status === "published" ? "verified" : "rejected", documentId);211  audit(ctx.db, { userId: ctx.user.id, role: ctx.user.role, action: `source.qa.${status}`, entityType: "source_document", entityId: documentId, ip: ctx.ip });212}213214export function sourcesOverview(d: Database.Database) {215  const sources = d.prepare(`SELECT s.*, (SELECT COUNT(*) FROM metre_source_documents x WHERE x.source_key=s.key) documents, (SELECT COUNT(*) FROM metre_price_observations o WHERE o.source_key=s.key) observations, (SELECT MAX(retrieved_at) FROM metre_source_documents x WHERE x.source_key=s.key) lastRetrieved FROM metre_sources s ORDER BY s.key`).all();216  const documents = d.prepare("SELECT id, source_key sourceKey, title, url, publication_date publicationDate, effective_date effectiveDate, retrieved_at retrievedAt, license, content_hash hash, bytes, parser_version parserVersion, qa_status qaStatus, records FROM metre_source_documents ORDER BY retrieved_at DESC LIMIT 100").all();217  const runs = d.prepare("SELECT id, source_key sourceKey, started_at startedAt, finished_at finishedAt, status, documents, records, skipped, errors FROM metre_ingest_runs ORDER BY started_at DESC LIMIT 40").all();218  const flags = d.prepare("SELECT id, entity_type entityType, entity_id entityId, severity, code, message, status, created_at createdAt FROM metre_quality_flags WHERE status='open' ORDER BY created_at DESC LIMIT 100").all();219  const byClass = d.prepare("SELECT evidence_class evidenceClass, COUNT(*) n, MIN(date_basis) first, MAX(date_basis) last FROM metre_price_observations GROUP BY evidence_class").all();220  return { sources, documents, runs, flags, byClass };221}222223export function searchObservations(d: Database.Database, p: { q?: string; evidenceClass?: string; sourceKey?: string; limit?: number }) {224  const where: string[] = []; const args: unknown[] = [];225  if (p.q) { where.push("(original_description LIKE ? OR classification_label LIKE ? OR buyer LIKE ?)"); const like = `%${p.q.slice(0, 80)}%`; args.push(like, like, like); }226  if (p.evidenceClass) { where.push("evidence_class=?"); args.push(p.evidenceClass); }227  if (p.sourceKey) { where.push("source_key=?"); args.push(p.sourceKey); }228  const sql = `SELECT id, source_key sourceKey, evidence_class evidenceClass, external_id externalId, original_description description, original_unit unit, original_amount amount, normalized_unit normalizedUnit, normalized_amount normalizedAmount, geography, buyer, classification_scheme scheme, classification_code code, classification_label label, date_basis dateBasis, license, extraction_method extractionMethod, verification_status verificationStatus, uncertainty_note uncertaintyNote, source_url sourceUrl FROM metre_price_observations ${where.length ? "WHERE " + where.join(" AND ") : ""} ORDER BY date_basis DESC LIMIT ?`;229  return d.prepare(sql).all(...args, Math.min(500, Math.max(1, p.limit ?? 50)));230}231