186 lines
7.1 KiB
JavaScript
186 lines
7.1 KiB
JavaScript
/**
|
|
* 拉所有启用的数据源, 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 集合
|
|
* 2. 把集合传给每个 fetcher, fetcher 跳过已存在的
|
|
* 3. 只对 DB 没有的指标调远程拉取, 节省 token (尤其妙想 MCP 这种 LLM 接口)
|
|
*
|
|
* 加新数据源: 写 scripts/fetchers/<name>.fetcher.js 导出 META + fetchAll(opts),
|
|
* 然后在 scripts/fetchers/index.js 的 FETCHERS 注册
|
|
*/
|
|
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 {
|
|
host: u.hostname,
|
|
port: Number(u.port) || 3306,
|
|
user: decodeURIComponent(u.username),
|
|
password: decodeURIComponent(u.password),
|
|
database: u.pathname.replace(/^\//, ''),
|
|
}
|
|
}
|
|
|
|
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) {
|
|
const [rows] = await conn.execute('SELECT `name` FROM `Indicator`')
|
|
return new Set(rows.map((r) => r.name))
|
|
}
|
|
|
|
async function main() {
|
|
if (!process.env.DATABASE_URL) {
|
|
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(', ') || '(空)'))
|
|
|
|
// 2) 跑所有 fetcher, 传入 existing + force (DB 优先, force 覆盖)
|
|
const { runFetchers } = await import('./fetchers/index.js')
|
|
const fetcherResults = await runFetchers({ skipNames: existing, forceNames: force })
|
|
|
|
// 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 }
|
|
for (const it of r.items) {
|
|
if (existing.has(it.name) && !force.has(it.name)) continue
|
|
allItems.push(it)
|
|
}
|
|
}
|
|
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 (数据保留, 不展示)')
|
|
}
|
|
|
|
// 6) 跨源对比报警: 全量跑一次 (即便无新指标也跑, 防止上次漏检)
|
|
try {
|
|
const m = await import('../server/alert-engine.js')
|
|
const a = await m.checkAfterRefresh(conn)
|
|
console.log('[Refresh] 报警检测: 新增 ' + a.inserted + ' 条 (扫描 ' + a.scanned + ')')
|
|
} catch (e) { console.error('[Refresh] 报警检测异常:', e.message) }
|
|
|
|
if (allItems.length === 0) {
|
|
console.log('[Refresh] 无指标待写, 不写库 (DB 优先策略生效)')
|
|
return
|
|
}
|
|
|
|
// 4) 写库
|
|
console.log('[Refresh] 写入数据库...')
|
|
let ins = 0, upd = 0, totalPoints = 0
|
|
for (const it of allItems) {
|
|
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 + ' 个明细点')
|
|
}
|
|
|
|
|
|
} finally {
|
|
await conn.end()
|
|
}
|
|
}
|
|
|
|
main().catch((e) => {
|
|
console.error('[Refresh] 异常:', e)
|
|
process.exit(1)
|
|
})
|
|
|
|
|