spb/vrai-prix

1// Connecteur incrémental « nouvelles ventes » — source api.qub.ca2// (le même feed que le widget « Transactions immobilières » du Journal de3// Montréal, https://www.journaldemontreal.com/argent/immobilier/transactions-immobilieres).4//5// Contrairement au pipeline batch complet (docs/PIPELINE-DONNEES.md, qui vit sur6// le laptop et reconstruit tout le .db), ce connecteur tourne SUR LE NŒUD et7// ajoute uniquement les ventes récentes directement dans data/vraiprix.db :8//9//   1. Récupère un jeton Bearer QUB (scripts/qub-token.mjs, via Scrapfly).10//   2. Liste les ~1478 secteurs de /real-estate-service/v1/locations/all.11//   3. Pour chaque secteur, interroge /v1/map (résultats triés du plus récent12//      au plus ancien, plafonnés à `nb_transactions`). Si un secteur sature le13//      plafond sans atteindre la date plancher, on subdivise sa bbox (quadtree).14//   4. Mappe chaque vente vers le schéma `transactions`, joint spatialement à15//      `units` pour id_provinc / valeur_role, puis INSERT OR IGNORE (dédup par16//      id — identique au pipeline batch, la source fournit le même id nanoid).17//18// Les `id` déjà connus sont ignorés : lancer ce script régulièrement (cron/pm2)19// garde la base à jour sans doublon. L'app lit le .db en WAL, les nouvelles20// lignes sont visibles immédiatement, sans redémarrage.21//22// Env : SCRAPFLY_KEY, QUB_EMAIL, QUB_PASSWORD (login), VRAIPRIX_DB (optionnel).23// Args : --since=YYYY-MM-DD  date plancher (défaut : max(date en base) - 45 j)24//        --region="Nom"      ne traiter que ce(s) secteur(s) (sous-chaîne)25//        --dry-run           n'écrit rien, affiche seulement le décompte26//        --concurrency=N     requêtes /map simultanées (défaut 6)2728import Database from 'better-sqlite3';29import fs from 'node:fs';30import path from 'node:path';31import { fileURLToPath } from 'node:url';32import { getQubToken } from './qub-token.mjs';3334// Charge .env.local (racine du repo) si présent — rend le script autonome sous35// cron/pm2 (SCRAPFLY_KEY, QUB_EMAIL, QUB_PASSWORD, QUB_SESSION, VRAIPRIX_DB).36(function loadEnvLocal() {37  const envPath = path.join(fileURLToPath(new URL('..', import.meta.url)), '.env.local');38  if (!fs.existsSync(envPath)) return;39  for (const line of fs.readFileSync(envPath, 'utf8').split('\n')) {40    const m = line.match(/^\s*([A-Z0-9_]+)\s*=\s*(.*)\s*$/);41    if (m && !(m[1] in process.env)) process.env[m[1]] = m[2].replace(/^["']|["']$/g, '');42  }43})();4445const API = 'https://api.qub.ca/real-estate-service/v1';46const PAGE_CAP = 500; // plafond de /map ; si atteint on subdivise47const HDRS = {48  Origin: 'https://www.journaldemontreal.com',49  Referer: 'https://www.journaldemontreal.com/',50  'User-Agent':51    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36',52};5354// ---------- args ----------55const args = Object.fromEntries(56  process.argv.slice(2).map((a) => {57    const m = a.match(/^--([^=]+)(?:=(.*))?$/);58    return m ? [m[1], m[2] ?? true] : [a, true];59  }),60);61const DRY = !!args['dry-run'];62const CONC = Math.max(1, parseInt(args.concurrency || '6', 10));63const REGION = args.region ? String(args.region).toLowerCase() : null;6465// ---------- db ----------66const DB_PATH =67  process.env.VRAIPRIX_DB || path.join(process.cwd(), 'data', 'vraiprix.db');68const db = new Database(DB_PATH);69db.pragma('journal_mode = WAL');70db.pragma('busy_timeout = 15000');7172const sinceArg = typeof args.since === 'string' ? args.since : null;73const maxDate = db.prepare('SELECT MAX(date) d FROM transactions').get().d;74function minusDays(iso, n) {75  const d = new Date(iso + 'T00:00:00Z');76  d.setUTCDate(d.getUTCDate() - n);77  return d.toISOString().slice(0, 10);78}79const SINCE = sinceArg || minusDays(maxDate || '2026-01-01', 45);8081// ---------- helpers ----------82const sleep = (ms) => new Promise((r) => setTimeout(r, ms));8384async function apiGet(url, token, tries = 4) {85  for (let i = 0; i < tries; i++) {86    let res;87    try {88      res = await fetch(url, { headers: { ...HDRS, Authorization: `Bearer ${token}` } });89    } catch (e) {90      if (i === tries - 1) throw e;91      await sleep(500 * (i + 1));92      continue;93    }94    if (res.ok) return res.json();95    if (res.status === 401 || res.status === 403)96      throw Object.assign(new Error(`auth ${res.status}`), { authError: true });97    if (res.status === 429 || res.status >= 500) {98      await sleep(800 * (i + 1));99      continue;100    }101    throw new Error(`HTTP ${res.status} sur ${url}`);102  }103  throw new Error(`échec après ${tries} essais : ${url}`);104}105106async function fetchLocations(token) {107  const list = await apiGet(`${API}/locations/all`, token);108  return list109    .map((l) => {110      // shape.coordinates est un anneau « plat » [[lng,lat],…] chez cette API111      // (parfois imbriqué [[[lng,lat],…]] en GeoJSON standard) — on gère les deux.112      let ring = l.shape?.coordinates || [];113      if (Array.isArray(ring[0]?.[0])) ring = ring[0];114      if (!ring.length || typeof ring[0]?.[0] !== 'number') return null;115      const lngs = ring.map((c) => c[0]);116      const lats = ring.map((c) => c[1]);117      return {118        name: l.name,119        sw_lat: Math.min(...lats),120        ne_lat: Math.max(...lats),121        sw_lng: Math.min(...lngs),122        ne_lng: Math.max(...lngs),123      };124    })125    .filter(Boolean);126}127128// Récupère les ventes d'une bbox, subdivise (quadtree) si le plafond est saturé129// alors que les ventes retournées sont encore >= SINCE (⇒ il en manque).130async function fetchBox(box, token, out, depth = 0) {131  const url =132    `${API}/map?ne_lat=${box.ne_lat}&ne_lng=${box.ne_lng}` +133    `&sw_lat=${box.sw_lat}&sw_lng=${box.sw_lng}&nb_transactions=${PAGE_CAP}&cid=0`;134  const rows = await apiGet(url, token);135  for (const t of rows) if (t.date >= SINCE) out.set(t.id, t);136137  const oldest = rows.length ? rows[rows.length - 1].date : null;138  const saturated = rows.length >= PAGE_CAP && oldest && oldest >= SINCE;139  if (saturated && depth < 6) {140    const mlat = (box.ne_lat + box.sw_lat) / 2;141    const mlng = (box.ne_lng + box.sw_lng) / 2;142    const quads = [143      { sw_lat: box.sw_lat, sw_lng: box.sw_lng, ne_lat: mlat, ne_lng: mlng },144      { sw_lat: box.sw_lat, sw_lng: mlng, ne_lat: mlat, ne_lng: box.ne_lng },145      { sw_lat: mlat, sw_lng: box.sw_lng, ne_lat: box.ne_lat, ne_lng: mlng },146      { sw_lat: mlat, sw_lng: mlng, ne_lat: box.ne_lat, ne_lng: box.ne_lng },147    ];148    for (const q of quads) await fetchBox(q, token, out, depth + 1);149  }150}151152// ---------- spatial join vers units (id_provinc, valeur_role) ----------153const unitsInBox = db.prepare(154  `SELECT id_provinc, lat, lng, valeur_role, type_prop FROM units155   WHERE lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?`,156);157function haversine(aLat, aLng, bLat, bLng) {158  const R = 6371000, toR = Math.PI / 180;159  const dLat = (bLat - aLat) * toR, dLng = (bLng - aLng) * toR;160  const s =161    Math.sin(dLat / 2) ** 2 +162    Math.cos(aLat * toR) * Math.cos(bLat * toR) * Math.sin(dLng / 2) ** 2;163  return 2 * R * Math.asin(Math.sqrt(s));164}165// Rayons d'appariement croissants (~65 m, ~220 m, ~550 m).166const DEGS = [0.0006, 0.002, 0.005];167function matchUnit(lat, lng, propertyType) {168  for (const d of DEGS) {169    const cands = unitsInBox.all(lat - d, lat + d, lng - d, lng + d);170    if (!cands.length) continue;171    let best = null, bestDist = Infinity;172    for (const u of cands) {173      let dist = haversine(lat, lng, u.lat, u.lng);174      if (propertyType && u.type_prop && u.type_prop !== propertyType) dist += 25; // léger biais type175      if (dist < bestDist) { bestDist = dist; best = u; }176    }177    if (best) return best;178  }179  return null;180}181182function toRow(t) {183  const coords = t.geometries?.[0]?.coordinates || [];184  const lng = coords[0], lat = coords[1];185  if (typeof lat !== 'number' || typeof lng !== 'number') return null;186  const ar = t.ar || {};187  const yb = ar.yearBuilt != null ? parseInt(ar.yearBuilt, 10) : null;188  const unit = matchUnit(lat, lng, t.propertyType);189  return {190    id: t.id,191    date: t.date,192    amount: t.amount,193    street: t.address?.street ?? null,194    city: t.address?.city ?? null,195    lat,196    lng,197    property_type: t.propertyType ?? null,198    year_built: Number.isFinite(yb) ? yb : null,199    floor_area: ar.floorArea ?? null,200    building_type: ar.buildingType ?? null,201    id_provinc: unit?.id_provinc ?? null,202    valeur_role: unit?.valeur_role ?? ar.totalArValue ?? null,203    land_area: ar.parcelArea ?? null,204  };205}206207const insert = db.prepare(208  `INSERT OR IGNORE INTO transactions209   (id,date,amount,street,city,lat,lng,property_type,year_built,floor_area,210    building_type,id_provinc,valeur_role,land_area)211   VALUES (@id,@date,@amount,@street,@city,@lat,@lng,@property_type,@year_built,212    @floor_area,@building_type,@id_provinc,@valeur_role,@land_area)`,213);214215// ---------- main ----------216async function main() {217  console.log(`[ingest-jdm] base : ${DB_PATH}`);218  console.log(`[ingest-jdm] date max en base : ${maxDate} → plancher SINCE=${SINCE}`);219  if (DRY) console.log('[ingest-jdm] DRY-RUN : aucune écriture');220221  const token = await getQubToken({222    session: process.env.QUB_SESSION || 'qub-vraiprix',223    log: (m) => console.log('[ingest-jdm]', m),224  });225226  let locations = await fetchLocations(token);227  if (REGION) locations = locations.filter((l) => l.name.toLowerCase().includes(REGION));228  console.log(`[ingest-jdm] secteurs à balayer : ${locations.length}`);229230  // Balayage concurrent par secteur.231  const collected = new Map();232  let done = 0;233  async function worker(queue) {234    for (;;) {235      const box = queue.pop();236      if (!box) return;237      try {238        await fetchBox(box, token, collected);239      } catch (e) {240        if (e.authError) throw e;241        console.warn(`[ingest-jdm] secteur "${box.name}" : ${e.message}`);242      }243      if (++done % 100 === 0)244        console.log(`[ingest-jdm]   ${done}/${locations.length} secteurs, ${collected.size} ventes ≥ ${SINCE}`);245    }246  }247  const queue = [...locations];248  await Promise.all(Array.from({ length: CONC }, () => worker(queue)));249  console.log(`[ingest-jdm] ventes candidates (≥ ${SINCE}) : ${collected.size}`);250251  // Mapping + insertion.252  const before = db.prepare('SELECT COUNT(*) n FROM transactions').get().n;253  let inserted = 0, skippedGeo = 0, matched = 0;254  const rows = [];255  for (const t of collected.values()) {256    const r = toRow(t);257    if (!r) { skippedGeo++; continue; }258    if (r.id_provinc) matched++;259    rows.push(r);260  }261262  if (!DRY) {263    const tx = db.transaction((batch) => {264      for (const r of batch) inserted += insert.run(r).changes;265    });266    tx(rows);267  }268269  const after = DRY ? before : db.prepare('SELECT COUNT(*) n FROM transactions').get().n;270  console.log('[ingest-jdm] ---------- résumé ----------');271  console.log(`  candidates mappées   : ${rows.length}`);272  console.log(`  appariées à une unité: ${matched} (${rows.length ? ((matched / rows.length) * 100).toFixed(1) : 0} %)`);273  console.log(`  sans géométrie       : ${skippedGeo}`);274  console.log(`  NOUVELLES insérées   : ${DRY ? '(dry-run)' : inserted}`);275  console.log(`  transactions totales : ${before} → ${after}`);276  const nd = db.prepare('SELECT MAX(date) d FROM transactions').get().d;277  console.log(`  date max en base     : ${nd}`);278  db.close();279}280281main().catch((e) => {282  console.error('[ingest-jdm] ERREUR', e);283  db.close();284  process.exit(1);285});286