数据源修改

2026-08-06 15:24:55 +08:00
parent b10b1c44db
commit ec15f40323
24 changed files with 2289 additions and 131 deletions
@@ -1,7 +1,12 @@
 /**
- * 拉所有启用的数据源, upsert 到 Indicator 表
- * 用法: npm run data:refresh
- * 幂等: 同 name 覆盖 data, 跨源独立, 错误隔离
+ * 拉所有启用的数据源, upsert 到 Indicator + IndicatorPoint 明细表
+ * 用法:
+ *   npm run data:refresh                        # 只拉 DB 里没有的指标
+ *   npm run data:refresh -- --force a,b,c       # 强制刷新指定指标 (忽略 DB 已有)
+ *
+ * 数据结构 (2026-08 重构):
+ *   Indicator      元数据: name/groupName/title/unit/source/frequency/hidden
+ *   IndicatorPoint 明细:   (indicatorId, period, value), period 兼容年/季/月/日
 *
 * DB 优先策略:
 *   1. 先从 DB 查已存在的 name 集合
@@ -14,6 +19,15 @@
 require('dotenv').config()
 const mysql = require('mysql2/promise')

+// 隐藏指标: 数据与获取脚本保留, 但不参与前端展示 (API 过滤)
+const HIDDEN_INDICATORS = new Set([
+  'china-gdp-usd',     // 中国 GDP (现价美元)
+  'china-gdp-growth',  // 中国 GDP 同比 (World Bank)
+  'china-gdp-yoy',     // 中国 GDP 同比 (妙想季度)
+  'china-unemploy',    // 中国失业率
+  'usa-gdp-usd',       // 美国 GDP
+])
+
 function parseUrl(url) {
  const u = new URL(url)
  return {
@@ -25,13 +39,63 @@ function parseUrl(url) {
  }
 }

-async function upsertIndicator(conn, payload) {
-  const sql =
-    'INSERT INTO `Indicator` (`name`, `groupName`, `data`, `createdAt`, `updatedAt`) ' +
-    'VALUES (?, ?, ?, NOW(3), NOW(3)) ' +
-    'ON DUPLICATE KEY UPDATE `groupName` = VALUES(`groupName`), `data` = VALUES(`data`), `updatedAt` = NOW(3)'
-  const [r] = await conn.execute(sql, [payload.name, payload.groupName, JSON.stringify(payload.data)])
-  return r.affectedRows
+function parseForceArg(argv) {
+  const i = argv.indexOf('--force')
+  if (i === -1) return new Set()
+  return new Set(String(argv[i + 1] || '').split(',').map((s) => s.trim()).filter(Boolean))
+}
+
+// 推断频率: year | quarter | month | day (优先用 fetcher 显式声明)
+function inferFrequency(data) {
+  if (data && data.frequency) return data.frequency
+  const p = String(((data && data.rows) || [])[0]?.period || '')
+  if (/^\d{4}$/.test(p)) return 'year'
+  if (/^\d{4}-Q[1-4]$/.test(p)) return 'quarter'
+  if (/^\d{4}-\d{2}$/.test(p)) return 'month'
+  if (/^\d{4}-\d{2}-\d{2}$/.test(p)) return 'day'
+  return 'month'
+}
+
+// 写 Indicator 元数据 + 全量替换明细点 (每指标一个事务)
+async function upsertIndicatorWithPoints(conn, payload) {
+  const { name, groupName, data, hidden } = payload
+  const title = data.title || name
+  const unit = data.unit || ''
+  const source = data.source || ''
+  const frequency = inferFrequency(data)
+  const rows = Array.isArray(data.rows) ? data.rows : []
+
+  await conn.beginTransaction()
+  try {
+    const [res] = await conn.execute(
+      'INSERT INTO `Indicator` (`name`, `groupName`, `title`, `unit`, `source`, `frequency`, `hidden`, `createdAt`, `updatedAt`) ' +
+      'VALUES (?, ?, ?, ?, ?, ?, ?, NOW(3), NOW(3)) ' +
+      'ON DUPLICATE KEY UPDATE `groupName` = VALUES(`groupName`), `title` = VALUES(`title`), ' +
+      '`unit` = VALUES(`unit`), `source` = VALUES(`source`), `frequency` = VALUES(`frequency`), ' +
+      '`hidden` = VALUES(`hidden`), `updatedAt` = NOW(3)',
+      [name, groupName, title, unit, source, frequency, hidden ? 1 : 0],
+    )
+    const indicatorId = res.insertId
+
+    // 明细: 先删旧点再插入新点 (全量替换, 保证与数据源一致)
+    await conn.execute('DELETE FROM `IndicatorPoint` WHERE `indicatorId` = ?', [indicatorId])
+    let points = 0
+    for (const r of rows) {
+      const period = String(r.period !== undefined ? r.period : (r.month !== undefined ? r.month : r.year))
+      const value = Number(r.value)
+      if (!period || Number.isNaN(value)) continue
+      await conn.execute(
+        'INSERT INTO `IndicatorPoint` (`indicatorId`, `period`, `value`, `createdAt`, `updatedAt`) VALUES (?, ?, ?, NOW(3), NOW(3))',
+        [indicatorId, period, value],
+      )
+      points++
+    }
+    await conn.commit()
+    return { indicatorId, points }
+  } catch (e) {
+    await conn.rollback()
+    throw e
+  }
 }

 async function fetchExistingNames(conn) {
@@ -44,48 +108,63 @@ async function main() {
    console.error('缺少 DATABASE_URL 环境变量')
    process.exit(1)
  }
+  const force = parseForceArg(process.argv.slice(2))
+  if (force.size) console.log('[Refresh] 强制刷新: ' + [...force].join(', '))
+
  const conn = await mysql.createConnection(parseUrl(process.env.DATABASE_URL))
  try {
    // 1) 查 DB 已存在的指标名
    const existing = await fetchExistingNames(conn)
-    console.log(`[Refresh] DB 已有 ${existing.size} 个指标: ${[...existing].join(', ') || '(空)'}`)
+    console.log('[Refresh] DB 已有 ' + existing.size + ' 个指标: ' + ([...existing].join(', ') || '(空)'))

-    // 2) 跑所有 fetcher, 传入 existing (DB 优先)
+    // 2) 跑所有 fetcher, 传入 existing + force (DB 优先, force 覆盖)
    const { runFetchers } = await import('./fetchers/index.js')
-    const fetcherResults = await runFetchers({ skipNames: existing })
+    const fetcherResults = await runFetchers({ skipNames: existing, forceNames: force })

-    // 3) 汇总
+    // 3) 汇总: force 名单里的指标即使 DB 已有也写入
    const allItems = []
    let skipped = 0, failed = 0
    for (const r of fetcherResults) {
      if (r.skipped) { skipped++; continue }
      if (!r.ok) { failed++; continue }
-      // fetcher 内部已经按 skipNames 过滤过, 这里再做一次防御性去重
      for (const it of r.items) {
-        if (existing.has(it.name)) continue
+        if (existing.has(it.name) && !force.has(it.name)) continue
        allItems.push(it)
      }
    }
-    console.log(`\n[Refresh] 汇总: ${allItems.length} 个新指标 (跳过 ${skipped} 个数据源, 失败 ${failed} 个, DB 已存在 ${existing.size} 个)`)
+    console.log('\n[Refresh] 汇总: ' + allItems.length + ' 个指标待写 (跳过 ' + skipped + ' 个数据源, 失败 ' + failed + ' 个, DB 已存在 ' + existing.size + ' 个)')

    if (failed > 0) {
      console.error('[Refresh] 警告: 有数据源失败, 仅写入成功的指标')
    }

+    // 5) 同步隐藏标记: 即使指标被 DB 优先跳过, 名单内的也确保 hidden=true
+    if (HIDDEN_INDICATORS.size > 0) {
+      const names = [...HIDDEN_INDICATORS]
+      const ph = names.map(() => '?').join(',')
+      const sql = "UPDATE `Indicator` SET `hidden` = 1, `updatedAt` = NOW(3) WHERE `name` IN (" + ph + ")"
+      await conn.execute(sql, names)
+      console.log('[Refresh] 同步隐藏标记: ' + names.length + ' 个指标 hidden=true (数据保留, 不展示)')
+    }
+
    if (allItems.length === 0) {
-      console.log('[Refresh] 无新指标, 不写库 (DB 优先策略生效)')
+      console.log('[Refresh] 无指标待写, 不写库 (DB 优先策略生效)')
      return
    }

    // 4) 写库
    console.log('[Refresh] 写入数据库...')
-    let ins = 0, upd = 0
+    let ins = 0, upd = 0, totalPoints = 0
    for (const it of allItems) {
-      const n = await upsertIndicator(conn, { name: it.name, groupName: it.groupName, data: it.data })
-      if (n === 1) ins++
-      else if (n === 2) upd++
+      const hidden = HIDDEN_INDICATORS.has(it.name)
+      const { points } = await upsertIndicatorWithPoints(conn, { name: it.name, groupName: it.groupName, data: it.data, hidden })
+      if (points > 0) ins++
+      else upd++
+      totalPoints += points
+      console.log('[Refresh]   ' + it.name + (hidden ? ' (hidden)' : '') + ': ' + points + ' 个明细点')
    }
-    console.log(`[Refresh] 写入完成: ${ins} 新增, ${upd} 更新\n`)
+    console.log('[Refresh] 写入完成: ' + ins + ' 个指标有数据, ' + upd + ' 个无数据, 共 ' + totalPoints + ' 个明细点\n')
+
  } finally {
    await conn.end()
  }
@@ -95,3 +174,4 @@ main().catch((e) => {
  console.error('[Refresh] 异常:', e)
  process.exit(1)
 })
+