源修改

aliyun-sync/aliyun-aps-sync/.env.example

@@ -1,4 +1,5 @@
 ALIYUN_APS_BASE_URL=https://aps.aliyun.com
+ALIYUN_APS_SOURCE_ID=default
 ALIYUN_APS_HEADLESS=false
 ALIYUN_APS_BROWSER_MODE=launch
 ALIYUN_APS_BROWSER_CHANNEL=

aliyun-sync/aliyun-aps-sync/README.md

@@ -259,6 +259,7 @@ ALIYUN_APS_HOT_FINAL_STATUSES=已完成,已关闭,已取消,已退款完成
 `.env` 需要配置：
 
 ```env
+ALIYUN_APS_SOURCE_ID=default
 ALIYUN_APS_DB_HOST=
 ALIYUN_APS_DB_PORT=3306
 ALIYUN_APS_DB_USER=
@@ -268,6 +269,39 @@ ALIYUN_APS_DB_CHARSET=utf8mb4
 ALIYUN_APS_DB_CONNECTION_LIMIT=5
 ```
 
+### 多账号 source_id
+
+如果两个 APS 账号写入同一个数据库，每个账号必须配置不同的 `ALIYUN_APS_SOURCE_ID`：
+
+```env
+# 账号 A
+ALIYUN_APS_SOURCE_ID=aliyun_account_a
+
+# 账号 B
+ALIYUN_APS_SOURCE_ID=aliyun_account_b
+```
+
+同步写库时会把 `source_id` 写入：
+
+- `aps_customer`
+- `aps_order`
+- `aps_order_detail`
+- `aps_bill`
+- `aliyun_aps_messages`
+
+增量水位也会按 `source_id` 查询，避免两个账号互相影响。
+
+建议两个账号使用不同项目目录或不同 `data/.browser` 目录，避免本地登录态和 checkpoint 互相覆盖。
+
+生产库建议把唯一键调整为 `source_id + 业务唯一键`，例如：
+
+```sql
+-- 示例，实际约束名以生产库为准
+-- aps_order: UNIQUE(source_id, order_id)
+-- aps_order_detail: UNIQUE(source_id, order_id)
+-- aliyun_aps_messages: UNIQUE(source_id, msg_id)
+```
+
 ## 浏览器配置
 
 默认不再强制使用 Google Chrome。

aliyun-sync/aliyun-aps-sync/src/config.js

@@ -36,6 +36,7 @@ const ensureDir = (dirPath) => {
 
 export const config = {
   rootDir,
+  sourceId: process.env.ALIYUN_APS_SOURCE_ID || 'default',
   baseUrl: process.env.ALIYUN_APS_BASE_URL || 'https://aps.aliyun.com',
   headless: toBool(process.env.ALIYUN_APS_HEADLESS, false),
   browserMode: (process.env.ALIYUN_APS_BROWSER_MODE || 'launch').trim().toLowerCase(),

aliyun-sync/aliyun-aps-sync/src/db.js

@@ -7,6 +7,7 @@ let customerMapCache = null;
 const MESSAGE_TABLE_DDL = `
 CREATE TABLE IF NOT EXISTS aliyun_aps_messages (
   id bigint NOT NULL AUTO_INCREMENT,
+  source_id varchar(64) NOT NULL DEFAULT 'default' COMMENT '数据来源账号标识',
   msg_id varchar(128) NULL DEFAULT NULL COMMENT '消息原始ID',
   title text NULL COMMENT '消息标题',
   content text NULL COMMENT '消息内容',
@@ -28,11 +29,13 @@ CREATE TABLE IF NOT EXISTS aliyun_aps_messages (
   extra_data json NULL COMMENT '其他字段(原始JSON)',
   crawl_time datetime NULL DEFAULT CURRENT_TIMESTAMP COMMENT '爬取时间',
   PRIMARY KEY (id),
-  UNIQUE KEY uk_msg_id (msg_id)
+  UNIQUE KEY uk_source_msg_id (source_id, msg_id),
+  KEY idx_source_message_time (source_id, gmt_modified, gmt_created)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='阿里云APS站内消息'
 `;
 
 let customerLifecycleEnsured = false;
+const ensuredSourceTables = new Set();
 
 function hasDbConfig() {
   return Boolean(config.db.host && config.db.user && config.db.database);
@@ -135,7 +138,8 @@ async function getCustomerMap() {
   if (customerMapCache) {
     return customerMapCache;
   }
-  const [rows] = await getPool().query('SELECT account_id, login_name FROM aps_customer');
+  await ensureSourceColumn('aps_customer');
+  const [rows] = await getPool().execute('SELECT account_id, login_name FROM aps_customer WHERE source_id = ?', [config.sourceId]);
   const map = new Map();
   for (const row of rows) {
     const loginName = safeString(row.login_name);
@@ -158,8 +162,8 @@ function resolveCustomerAccountId(customerMap, customerAccount) {
   return customerMap.get(normalized) || customerMap.get(normalized.replace(/\s+/g, '')) || null;
 }
 
-async function queryLatestValue(sql) {
-  const [rows] = await getPool().query(sql);
+async function queryLatestValue(sql, params = []) {
+  const [rows] = await getPool().execute(sql, params);
   const row = Array.isArray(rows) ? rows[0] : null;
   if (!row) {
     return null;
@@ -175,15 +179,18 @@ async function queryLatestValue(sql) {
 }
 
 export async function getLatestOrderTimeFromDb() {
-  return queryLatestValue('SELECT MAX(order_time) AS latest_time FROM aps_order');
+  await ensureSourceColumn('aps_order');
+  return queryLatestValue('SELECT MAX(order_time) AS latest_time FROM aps_order WHERE source_id = ?', [config.sourceId]);
 }
 
 export async function getLatestBillConsumptionTimeFromDb() {
-  return queryLatestValue('SELECT MAX(consumption_time) AS latest_time FROM aps_bill');
+  await ensureSourceColumn('aps_bill');
+  return queryLatestValue('SELECT MAX(consumption_time) AS latest_time FROM aps_bill WHERE source_id = ?', [config.sourceId]);
 }
 
 export async function getLatestMessageTimeFromDb() {
-  return queryLatestValue("SELECT MAX(COALESCE(NULLIF(gmt_modified, ''), NULLIF(gmt_created, ''))) AS latest_time FROM aliyun_aps_messages");
+  await ensureMessagesTable();
+  return queryLatestValue("SELECT MAX(COALESCE(NULLIF(gmt_modified, ''), NULLIF(gmt_created, ''))) AS latest_time FROM aliyun_aps_messages WHERE source_id = ?", [config.sourceId]);
 }
 
 export async function closeDbPool() {
@@ -197,12 +204,14 @@ export async function closeDbPool() {
 
 export async function ensureMessagesTable() {
   await getPool().query(MESSAGE_TABLE_DDL);
+  await ensureSourceColumn('aliyun_aps_messages');
 }
 
 export async function ensureCustomerLifecycleColumns() {
   if (customerLifecycleEnsured) {
     return;
   }
+  await ensureSourceColumn('aps_customer');
   await ensureColumnExists('aps_customer', 'active', "ALTER TABLE aps_customer ADD COLUMN active TINYINT(1) NOT NULL DEFAULT 1 COMMENT '是否有效 1=有效 0=释放'");
   await ensureColumnExists('aps_customer', 'status', "ALTER TABLE aps_customer ADD COLUMN status VARCHAR(32) DEFAULT 'active' COMMENT '客户状态'");
   await ensureColumnExists('aps_customer', 'released_at', "ALTER TABLE aps_customer ADD COLUMN released_at DATETIME NULL COMMENT '释放时间'");
@@ -210,6 +219,14 @@ export async function ensureCustomerLifecycleColumns() {
   customerLifecycleEnsured = true;
 }
 
+async function ensureSourceColumn(tableName) {
+  if (ensuredSourceTables.has(tableName)) {
+    return;
+  }
+  await ensureColumnExists(tableName, 'source_id', `ALTER TABLE ${tableName} ADD COLUMN source_id VARCHAR(64) NOT NULL DEFAULT 'default' COMMENT '数据来源账号标识'`);
+  ensuredSourceTables.add(tableName);
+}
+
 async function ensureColumnExists(tableName, columnName, alterSql) {
   const [rows] = await getPool().execute(
     `SELECT COUNT(*) AS cnt
@@ -233,13 +250,13 @@ export async function upsertCustomers(records) {
   await ensureCustomerLifecycleColumns();
   const sql = `
     INSERT INTO aps_customer (
-      account_id, login_name, real_name, report_source, report_type, trade_mode,
+      source_id, account_id, login_name, real_name, report_source, report_type, trade_mode,
       real_name_status, relation_date, follow_staff, payment_notice_status,
       invite_register_type, is_new_customer, performance_start_point_reached,
       customer_category, remark, no_consumption_months,
       planned_release_time, planned_release_reason,
       active, status, released_at, release_reason
-    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, 'active', NULL, NULL)
+    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, 'active', NULL, NULL)
     ON DUPLICATE KEY UPDATE
       login_name=VALUES(login_name),
       real_name=VALUES(real_name),
@@ -271,6 +288,7 @@ export async function upsertCustomers(records) {
       continue;
     }
     await getPool().execute(sql, [
+      config.sourceId,
       accountId,
       loginName,
       safeString(record.realName),
@@ -309,7 +327,7 @@ export async function upsertCustomerDetails(records) {
       department = ?,
       payment_notice_status = COALESCE(?, payment_notice_status),
       updated_at = CURRENT_TIMESTAMP
-    WHERE account_id = ?
+    WHERE source_id = ? AND account_id = ?
   `;
   for (const record of records) {
     const accountId = safeString(record.accountId);
@@ -324,6 +342,7 @@ export async function upsertCustomerDetails(records) {
       safeString(record.phone),
       safeString(record.department),
       safeString(record.paymentNoticeStatus),
+      config.sourceId,
       accountId,
     ]);
   }
@@ -336,8 +355,8 @@ async function findCustomerAccountIdByName(customerName) {
     return null;
   }
   const [rows] = await getPool().execute(
-    'SELECT account_id FROM aps_customer WHERE customer_name = ? OR real_name = ? OR login_name = ? LIMIT 1',
-    [normalized, normalized, normalized],
+    'SELECT account_id FROM aps_customer WHERE source_id = ? AND (customer_name = ? OR real_name = ? OR login_name = ?) LIMIT 1',
+    [config.sourceId, normalized, normalized, normalized],
   );
   return Array.isArray(rows) && rows.length > 0 ? safeString(rows[0].account_id) : null;
 }
@@ -349,8 +368,8 @@ async function markCustomerReleased(customerName, reason, releasedAt) {
   }
   await ensureCustomerLifecycleColumns();
   await getPool().execute(
-    'UPDATE aps_customer SET active = 0, status = ?, released_at = ?, release_reason = ? WHERE account_id = ?',
-    ['released', releasedAt || new Date().toISOString().slice(0, 19).replace('T', ' '), safeString(reason), accountId],
+    'UPDATE aps_customer SET active = 0, status = ?, released_at = ?, release_reason = ? WHERE source_id = ? AND account_id = ?',
+    ['released', releasedAt || new Date().toISOString().slice(0, 19).replace('T', ' '), safeString(reason), config.sourceId, accountId],
   );
   return true;
 }
@@ -362,8 +381,8 @@ async function markCustomerActive(customerName) {
   }
   await ensureCustomerLifecycleColumns();
   await getPool().execute(
-    'UPDATE aps_customer SET active = 1, status = ?, released_at = NULL, release_reason = NULL WHERE account_id = ?',
-    ['active', accountId],
+    'UPDATE aps_customer SET active = 1, status = ?, released_at = NULL, release_reason = NULL WHERE source_id = ? AND account_id = ?',
+    ['active', config.sourceId, accountId],
   );
   return true;
 }
@@ -389,13 +408,14 @@ export async function upsertOrders(records) {
   if (!records?.length) {
     return { inserted: 0 };
   }
+  await ensureSourceColumn('aps_order');
   const customerMap = await getCustomerMap();
   const sql = `
     INSERT INTO aps_order (
-      order_id, customer_account_id, customer_login_name,
+      source_id, order_id, customer_account_id, customer_login_name,
       customer_category, order_type, original_price_cny, paid_amount_cny,
       status, order_time, order_month
-    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON DUPLICATE KEY UPDATE
       customer_account_id=VALUES(customer_account_id),
       customer_login_name=VALUES(customer_login_name),
@@ -415,6 +435,7 @@ export async function upsertOrders(records) {
     const customerLoginName = safeString(record.customerAccount) || '';
     const accountId = resolveCustomerAccountId(customerMap, customerLoginName);
     await getPool().execute(sql, [
+      config.sourceId,
       orderId,
       accountId,
       customerLoginName,
@@ -434,14 +455,15 @@ export async function upsertOrderDetails(records) {
   if (!records?.length) {
     return { inserted: 0 };
   }
+  await ensureSourceColumn('aps_order_detail');
   const sql = `
     INSERT INTO aps_order_detail (
-      order_id, order_type, status, trade_type, customer_category,
+      source_id, order_id, order_type, status, trade_type, customer_category,
       dealer_name, dealer_uid, customer_type, opportunity_id,
       payment_time, order_time, product_name, product_code,
       original_price_cny, paid_amount_cny, discount,
       payable_amount_cny, coupon_amount_cny
-    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON DUPLICATE KEY UPDATE
       order_type=VALUES(order_type),
       status=VALUES(status),
@@ -467,6 +489,7 @@ export async function upsertOrderDetails(records) {
       continue;
     }
     await getPool().execute(sql, [
+      config.sourceId,
       orderId,
       safeString(record.orderType),
       safeString(record.status),
@@ -494,10 +517,12 @@ export async function upsertBills(records) {
   if (!records?.length) {
     return { inserted: 0 };
   }
+  await ensureSourceColumn('aps_bill');
   const customerMap = await getCustomerMap();
   const selectSql = `
     SELECT id FROM aps_bill
-    WHERE billing_month = ?
+    WHERE source_id = ?
+      AND billing_month = ?
       AND commission_month = ?
       AND customer_login_name = ?
       AND consumption_time = ?
@@ -507,7 +532,7 @@ export async function upsertBills(records) {
   `;
   const insertSql = `
     INSERT INTO aps_bill (
-      billing_month, customer_account_id, customer_login_name,
+      source_id, billing_month, customer_account_id, customer_login_name,
       bill_type, consumption_time,
       customer_category,
       product_name, product_category,
@@ -517,7 +542,7 @@ export async function upsertBills(records) {
       rebated,
       invite_register_type,
       service_start_time, service_end_time
-    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
   `;
   const updateSql = `
     UPDATE aps_bill SET
@@ -546,6 +571,7 @@ export async function upsertBills(records) {
     const consumptionTime = safeString(record.consumeDate);
     const originalPrice = safeNumber(record.originalPriceCny);
     const [rows] = await getPool().execute(selectSql, [
+      config.sourceId,
       billingMonth,
       commissionMonth,
       customerLoginName,
@@ -572,6 +598,7 @@ export async function upsertBills(records) {
     }
 
     await getPool().execute(insertSql, [
+      config.sourceId,
       billingMonth,
       accountId,
       customerLoginName,
@@ -601,11 +628,11 @@ export async function upsertMessages(records) {
   await ensureMessagesTable();
   const sql = `
     INSERT INTO aliyun_aps_messages (
-      msg_id, title, content, msg_type, from_app, biz_code, msg_channel,
+      source_id, msg_id, title, content, msg_type, from_app, biz_code, msg_channel,
       category_id, category_name, lv1_category_id, lv2_category_id, lv3_category_id,
       message_classification, customer_name, order_no, status,
       gmt_created, gmt_modified, extra_data
-    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON DUPLICATE KEY UPDATE
       title=VALUES(title),
       content=VALUES(content),
@@ -633,6 +660,7 @@ export async function upsertMessages(records) {
       continue;
     }
     await getPool().execute(sql, [
+      config.sourceId,
       msgId,
       safeString(record.title),
       safeString(record.content),