Move source mapping to scrape database

2026-08-05 21:40:58 +02:00
parent 17082f6d9e
commit a72bdab9ab
6 changed files with 221 additions and 14 deletions
@@ -0,0 +1,165 @@
+import fs from "node:fs/promises";
+import mariadb from "mariadb";
+
+async function readEnv() {
+  return Object.fromEntries(
+    (await fs.readFile(".env", "utf8"))
+      .split(/\r?\n/)
+      .map((line) => line.trim())
+      .filter((line) => line && !line.startsWith("#") && line.includes("="))
+      .map((line) => {
+        const index = line.indexOf("=");
+        return [line.slice(0, index).trim(), line.slice(index + 1).trim()];
+      }),
+  );
+}
+
+async function queryRows(connection, sql, params = []) {
+  const rows = await connection.query(sql, params);
+  return Array.isArray(rows) ? rows.filter((row) => row && typeof row === "object" && !("meta" in row)) : [];
+}
+
+async function tableExists(connection, schema, table) {
+  const rows = await queryRows(
+    connection,
+    "SELECT 1 FROM information_schema.tables WHERE table_schema = ? AND table_name = ? LIMIT 1",
+    [schema, table],
+  );
+  return rows.length > 0;
+}
+
+async function columnExists(connection, schema, table, column) {
+  const rows = await queryRows(
+    connection,
+    "SELECT 1 FROM information_schema.columns WHERE table_schema = ? AND table_name = ? AND column_name = ? LIMIT 1",
+    [schema, table, column],
+  );
+  return rows.length > 0;
+}
+
+async function addColumnIfMissing(connection, schema, table, column, definition) {
+  if (await columnExists(connection, schema, table, column)) return;
+  await connection.query(`ALTER TABLE \`${schema}\`.\`${table}\` ADD COLUMN \`${column}\` ${definition}`);
+}
+
+const env = await readEnv();
+const catalogDb = env.DB_NAME || "9bplus";
+const scrapeDb = env.SCRAPE_DB_NAME || "catalog_scrape";
+const connection = await mariadb.createConnection({
+  host: env.DB_HOST || "127.0.0.1",
+  port: Number(env.DB_PORT || 3306),
+  user: env.DB_USER || "root",
+  password: env.DB_PASSWORD || "",
+  multipleStatements: false,
+  bigIntAsNumber: true,
+});
+
+try {
+  await connection.query(`CREATE DATABASE IF NOT EXISTS \`${scrapeDb}\` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`);
+  const schemaSql = await fs.readFile("sql/local-scrape-db.sql", "utf8");
+  for (const statement of schemaSql.split(/;\s*(?:\r?\n|$)/).map((item) => item.trim()).filter(Boolean)) {
+    await connection.query(statement);
+  }
+
+  await addColumnIfMissing(connection, scrapeDb, "scrape_sources", "legacy_catalog_source_id", "BIGINT UNSIGNED NULL");
+  await addColumnIfMissing(connection, scrapeDb, "scrape_sources", "fallback_source_type", "VARCHAR(64) NOT NULL DEFAULT 'none'");
+  await addColumnIfMissing(connection, scrapeDb, "scrape_sources", "access_mode", "VARCHAR(64) NOT NULL DEFAULT 'browser'");
+  await addColumnIfMissing(connection, scrapeDb, "scrape_sources", "browser_engine", "VARCHAR(32) NOT NULL DEFAULT 'default'");
+  await addColumnIfMissing(connection, scrapeDb, "scrape_sources", "browser_headless", "TINYINT(1) NOT NULL DEFAULT 0");
+  await addColumnIfMissing(connection, scrapeDb, "scrape_sources", "wait_after_load_ms", "INT UNSIGNED NOT NULL DEFAULT 3000");
+  await addColumnIfMissing(connection, scrapeDb, "scrape_sources", "note", "TEXT NULL");
+
+  if (!(await tableExists(connection, catalogDb, "ps_product_catalog_source"))) {
+    throw new Error(`Source table ${catalogDb}.ps_product_catalog_source does not exist.`);
+  }
+  if (!(await tableExists(connection, catalogDb, "ps_product_catalog_manufacturer_source"))) {
+    throw new Error(`Mapping table ${catalogDb}.ps_product_catalog_manufacturer_source does not exist.`);
+  }
+
+  await connection.query(`
+    INSERT INTO \`${scrapeDb}\`.scrape_sources (
+      legacy_catalog_source_id,
+      source_key,
+      source_name,
+      source_type,
+      base_url,
+      enabled,
+      priority,
+      search_primary_key,
+      search_fallback_keys,
+      fallback_source_type,
+      search_url_template,
+      fallback_url_template,
+      search_context_keys,
+      access_mode,
+      browser_engine,
+      browser_headless,
+      wait_after_load_ms,
+      note
+    )
+    SELECT
+      id_product_catalog_source,
+      source_key,
+      source_name,
+      source_type,
+      base_url,
+      enabled,
+      priority,
+      search_primary_key,
+      search_fallback_keys,
+      COALESCE(fallback_source_type, 'none'),
+      search_url_template,
+      fallback_url_template,
+      search_context_keys,
+      COALESCE(access_mode, 'browser'),
+      COALESCE(browser_engine, 'default'),
+      COALESCE(browser_headless, 0),
+      COALESCE(wait_after_load_ms, 3000),
+      note
+    FROM \`${catalogDb}\`.ps_product_catalog_source
+    ON DUPLICATE KEY UPDATE
+      source_key = VALUES(source_key),
+      source_name = VALUES(source_name),
+      source_type = VALUES(source_type),
+      base_url = VALUES(base_url),
+      enabled = VALUES(enabled),
+      priority = VALUES(priority),
+      search_primary_key = VALUES(search_primary_key),
+      search_fallback_keys = VALUES(search_fallback_keys),
+      fallback_source_type = VALUES(fallback_source_type),
+      search_url_template = VALUES(search_url_template),
+      fallback_url_template = VALUES(fallback_url_template),
+      search_context_keys = VALUES(search_context_keys),
+      access_mode = VALUES(access_mode),
+      browser_engine = VALUES(browser_engine),
+      browser_headless = VALUES(browser_headless),
+      wait_after_load_ms = VALUES(wait_after_load_ms),
+      note = VALUES(note)
+  `);
+
+  await connection.query(`
+    INSERT INTO \`${scrapeDb}\`.scrape_manufacturer_sources (
+      manufacturer_id,
+      source_id,
+      enabled,
+      priority
+    )
+    SELECT
+      ms.id_manufacturer,
+      ss.id,
+      ms.enabled,
+      ms.priority
+    FROM \`${catalogDb}\`.ps_product_catalog_manufacturer_source ms
+    JOIN \`${scrapeDb}\`.scrape_sources ss
+      ON ss.legacy_catalog_source_id = ms.id_product_catalog_source
+    ON DUPLICATE KEY UPDATE
+      enabled = VALUES(enabled),
+      priority = VALUES(priority)
+  `);
+
+  const sources = await queryRows(connection, `SELECT COUNT(*) AS total FROM \`${scrapeDb}\`.scrape_sources`);
+  const mappings = await queryRows(connection, `SELECT COUNT(*) AS total FROM \`${scrapeDb}\`.scrape_manufacturer_sources`);
+  console.log(`Scrape mapping migration ready: ${sources[0]?.total || 0} sources, ${mappings[0]?.total || 0} manufacturer mappings.`);
+} finally {
+  await connection.end();
+}