Files
Toolbox/src/server/global_news.js
T

1063 lines
44 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
const fs = require('fs')
const path = require('path')
const axios = require('axios')
const Database = require('better-sqlite3')
const dataDir = path.join(process.cwd(), 'data')
if (!fs.existsSync(dataDir)) fs.mkdirSync(dataDir, { recursive: true })
const dbPath = path.join(dataDir, 'global_news.db')
const db = new Database(dbPath)
db.pragma('journal_mode = WAL')
db.exec(`
CREATE TABLE IF NOT EXISTS global_news (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT NOT NULL,
country TEXT NOT NULL DEFAULT '',
continent TEXT NOT NULL DEFAULT '',
source_name TEXT NOT NULL DEFAULT '',
category TEXT NOT NULL DEFAULT 'general',
title TEXT NOT NULL DEFAULT '',
summary TEXT NOT NULL DEFAULT '',
title_cn TEXT NOT NULL DEFAULT '',
summary_cn TEXT NOT NULL DEFAULT '',
link TEXT NOT NULL DEFAULT '',
published_at TEXT NOT NULL DEFAULT '',
fetch_method TEXT NOT NULL DEFAULT 'google',
fetched_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
lang TEXT NOT NULL DEFAULT '',
UNIQUE(date, source_name, link)
);
CREATE INDEX IF NOT EXISTS idx_global_news_date ON global_news(date);
CREATE INDEX IF NOT EXISTS idx_global_news_continent ON global_news(continent, date);
CREATE INDEX IF NOT EXISTS idx_global_news_category ON global_news(category, date);
`)
try {
db.prepare('SELECT title_cn FROM global_news LIMIT 1').get()
} catch (e) {
if (String(e).includes('no such column')) {
db.exec(`ALTER TABLE global_news ADD COLUMN title_cn TEXT NOT NULL DEFAULT ''`)
db.exec(`ALTER TABLE global_news ADD COLUMN summary_cn TEXT NOT NULL DEFAULT ''`)
}
}
try {
db.prepare('SELECT ai_status FROM global_news LIMIT 1').get()
} catch (e) {
if (String(e).includes('no such column')) {
db.exec(`ALTER TABLE global_news ADD COLUMN ai_status TEXT NOT NULL DEFAULT ''`)
db.exec(`ALTER TABLE global_news ADD COLUMN ai_domestic TEXT NOT NULL DEFAULT ''`)
db.exec(`ALTER TABLE global_news ADD COLUMN ai_overseas TEXT NOT NULL DEFAULT ''`)
db.exec(`ALTER TABLE global_news ADD COLUMN ai_defensive TEXT NOT NULL DEFAULT ''`)
}
}
try {
db.prepare('SELECT lang FROM global_news LIMIT 1').get()
} catch (e) {
if (String(e).includes('no such column')) {
db.exec(`ALTER TABLE global_news ADD COLUMN lang TEXT NOT NULL DEFAULT ''`)
}
}
const listByDateStmt = db.prepare(`SELECT * FROM global_news WHERE date = ? ORDER BY CASE continent WHEN '北美' THEN 1 WHEN '欧洲' THEN 2 WHEN '亚太' THEN 3 WHEN '拉美' THEN 4 ELSE 5 END, source_name, id ASC`)
const dateRangeStmt = db.prepare(`SELECT DISTINCT date FROM global_news ORDER BY date DESC LIMIT 60`)
const CONTINENT_ORDER = { '北美': 1, '欧洲': 2, '亚太': 3, '拉美': 4, '其他': 5 }
const newsLogs = []
const MAX_LOGS = 2000
const fmtLogTime = () => {
const d = new Date()
return `${d.getFullYear()}-${String(d.getMonth()+1).padStart(2,'0')}-${String(d.getDate()).padStart(2,'0')} ${String(d.getHours()).padStart(2,'0')}:${String(d.getMinutes()).padStart(2,'0')}:${String(d.getSeconds()).padStart(2,'0')}.${String(d.getMilliseconds()).padStart(3,'0')}`
}
const fmtBatchTime = () => fmtLogTime()
const REMOTE_PROBE_CONCURRENCY = 3
const chunkArray = (list, size) => {
const out = []
const arr = Array.isArray(list) ? list : []
const step = Math.max(1, Number(size || 1) || 1)
for (let i = 0; i < arr.length; i += step) out.push(arr.slice(i, i + step))
return out
}
const newsLog = (level, msg, detail) => {
const log = { time: fmtLogTime(), level, msg, detail: detail || '' }
newsLogs.push(log)
if (newsLogs.length > MAX_LOGS) newsLogs.splice(0, newsLogs.length - MAX_LOGS)
const prefix = level === 'ERROR' ? '❌' : level === 'WARN' ? '⚠️' : 'ℹ️'
console.log(`[global_news] ${prefix} ${log.time} ${msg}${detail ? ' | ' + detail : ''}`)
}
const configPath = path.join(process.cwd(), 'config', 'global_news.json')
const getConfig = () => {
try { return JSON.parse(fs.readFileSync(configPath, 'utf-8')) } catch { return { enabled: false, sources: [] } }
}
const getMode = cfg => {
const mode = String(((cfg || {}).mode) || '').trim().toLowerCase()
return mode === 'remote_ephemeral' ? 'remote_ephemeral' : 'local_persist'
}
const isRemoteEphemeralMode = cfg => getMode(cfg) === 'remote_ephemeral'
const getCurrentPort = () => Number(process.env.PORT || '8976') || 8976
const getAutomationPort = cfg => Number((cfg || {}).automation_port || 8977) || 8977
const isAutomationInstance = cfg => getCurrentPort() === getAutomationPort(cfg)
const getRemoteProbeConfig = cfg => {
const base = String(((cfg || {}).remote_base_url) || '').trim().replace(/\/+$/, '')
const endpointRaw = String(((cfg || {}).remote_endpoint) || '/probe/gnews').trim()
const endpoint = endpointRaw
? (endpointRaw.startsWith('/') ? endpointRaw : `/${endpointRaw}`)
: '/probe/gnews'
const token = String(((cfg || {}).remote_token) || '').trim()
const timeoutRaw = Number((cfg || {}).remote_timeout_ms)
const fallbackTimeout = Number((cfg || {}).timeout_ms)
const timeout = Number.isFinite(timeoutRaw) && timeoutRaw > 0
? timeoutRaw
: (Number.isFinite(fallbackTimeout) && fallbackTimeout > 0 ? fallbackTimeout : 15000)
return { base, endpoint, token, timeout }
}
const getBeijingNow = () => formatBeijingParts(new Date()) || {
date: `${new Date().getFullYear()}-${String(new Date().getMonth() + 1).padStart(2, '0')}-${String(new Date().getDate()).padStart(2, '0')}`,
dateTime: fmtLogTime().slice(0, 19)
}
const getBeijingHourMinute = () => {
const now = getBeijingNow()
const hm = String(now.dateTime || '').slice(11, 16).split(':')
return {
date: now.date,
hour: Number(hm[0] || 0),
minute: Number(hm[1] || 0)
}
}
const proxyGlobalNewsRequest = async (req, routePath) => {
const cfg = getConfig()
const currentPort = getCurrentPort()
const automationPort = getAutomationPort(cfg)
if (currentPort === automationPort) return null
const proxied = await axios({
url: `http://127.0.0.1:${automationPort}${routePath}`,
method: req.method || 'GET',
headers: {
'Content-Type': 'application/json',
Cookie: req.headers.cookie || ''
},
params: req.query || {},
data: req.body || {},
timeout: 240000,
validateStatus: () => true
})
return {
status: proxied.status || 200,
body: Object.assign({}, proxied.data || {}, {
proxied_to_automation: true,
current_port: currentPort,
automation_port: automationPort
})
}
}
const normalizeRemoteProbeLink = raw => {
const link = String(raw || '').trim()
if (!link) return ''
if (/^https?:\/\//i.test(link)) return link
if (/^[A-Za-z0-9_-]{20,}$/.test(link)) return `https://news.google.com/rss/articles/${link}`
return link
}
const getRemoteProbeLang = source => {
const explicit = String((source && source.lang) || '').trim()
if (explicit) return explicit
const hl = String((source && source.hl) || '').trim()
if (hl) return String(hl.split(/[-_]/)[0] || '').toLowerCase() || 'en'
return 'en'
}
const { HttpsProxyAgent } = require('https-proxy-agent')
const { HttpProxyAgent } = require('http-proxy-agent')
const createHttpClient = () => {
const cfg = getConfig()
const opts = { timeout: cfg.timeout_ms || 15000, headers: { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } }
const proxyUrl = cfg.proxy || process.env.HTTP_PROXY || process.env.HTTPS_PROXY || ''
if (proxyUrl) {
try {
opts.httpsAgent = new HttpsProxyAgent(proxyUrl)
opts.httpAgent = new HttpProxyAgent(proxyUrl)
} catch {}
}
return axios.create(opts)
}
const getAiConfig = () => {
const aiCfgPath = path.join(process.cwd(), 'config', 'markets_security_calendar_ai.json')
try {
const c = JSON.parse(fs.readFileSync(aiCfgPath, 'utf-8'))
return {
enabled: c.enabled !== false,
apiKey: c.api_key || '',
apiBase: c.api_base || 'https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions',
model: c.model || 'qwen-plus-latest'
}
} catch {
return { enabled: false, apiKey: '', apiBase: '', model: '' }
}
}
const stripHtml = s => {
let t = String(s || '')
t = t.replace(/&amp;/g, '&').replace(/&lt;/g, '<').replace(/&gt;/g, '>').replace(/&quot;/g, '"').replace(/&#39;/g, "'").replace(/&rsquo;/g, "'").replace(/&lsquo;/g, "'").replace(/&rdquo;/g, '"').replace(/&ldquo;/g, '"').replace(/&mdash;/g, '—').replace(/&ndash;/g, '–')
t = t.replace(/<[^>]*>/g, '')
t = t.replace(/&amp;/g, '&').replace(/&lt;/g, '<').replace(/&gt;/g, '>')
return t.trim()
}
const cleanSummary = desc => {
const s = stripHtml(desc)
if (!s || s.length < 5 || /^https?:\/\//i.test(s)) return ''
return s.slice(0, 200)
}
const formatBeijingParts = date => {
try {
const formatter = new Intl.DateTimeFormat('sv-SE', {
timeZone: 'Asia/Shanghai',
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
hour12: false
})
const parts = formatter.formatToParts(date)
const map = {}
for (const part of parts) map[part.type] = part.value
return {
date: `${map.year}-${map.month}-${map.day}`,
dateTime: `${map.year}-${map.month}-${map.day} ${map.hour}:${map.minute}:${map.second}`
}
} catch {
return null
}
}
const normalizePublishedAtForStorage = (publishedAt, fallbackDate) => {
const raw = String(publishedAt || '').trim()
if (!raw) return { date: fallbackDate || '', publishedAt: '' }
const parsed = new Date(raw)
if (!Number.isFinite(parsed.getTime())) return { date: fallbackDate || '', publishedAt: raw }
const bj = formatBeijingParts(parsed)
if (!bj) return { date: fallbackDate || '', publishedAt: raw }
return { date: bj.date || fallbackDate || '', publishedAt: bj.dateTime || raw }
}
const addDaysToDateStr = (dateStr, offset) => {
const s = String(dateStr || '').trim()
if (!/^\d{4}-\d{2}-\d{2}$/.test(s)) return ''
const [year, month, day] = s.split('-').map(x => parseInt(x, 10))
const d = new Date(Date.UTC(year, month - 1, day))
if (!Number.isFinite(d.getTime())) return ''
d.setUTCDate(d.getUTCDate() + Number(offset || 0))
return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, '0')}-${String(d.getUTCDate()).padStart(2, '0')}`
}
const isDateNearTarget = (targetDate, itemDate) => {
const target = String(targetDate || '').trim()
const item = String(itemDate || '').trim()
if (!/^\d{4}-\d{2}-\d{2}$/.test(target) || !/^\d{4}-\d{2}-\d{2}$/.test(item)) return true
if (item === target) return true
if (item === addDaysToDateStr(target, -1)) return true
if (item === addDaysToDateStr(target, 1)) return true
return false
}
const continentSortValue = c => CONTINENT_ORDER[c] || 99
const cleanTitle = raw => {
let t = stripHtml(raw)
t = t.replace(/^[\s\-–—|•·]+/, '').replace(/[\s\-–—|•·]+$/, '')
const idx = t.lastIndexOf(' - ')
if (idx > t.length * 0.5) t = t.slice(0, idx).trim()
return t.slice(0, 200)
}
const isNoiseTitle = t => {
const low = t.toLowerCase()
const noise = ['advertisement', 'sponsored', 'subscribe', 'cookie', 'privacy policy', 'terms of service', 'video:', 'watch:', 'live:', 'breaking news:', 'newsletter', 'sign up']
return noise.some(k => low.includes(k)) || t.length < 10
}
const isEntertainmentOrSport = t => {
const low = t.toLowerCase()
const sports = ['nfl', 'nba', 'mlb', 'nhl', 'soccer', 'football', 'tennis', 'golf', 'olympics', 'champions league', 'premier league', 'world cup', 'formula 1', 'boxing', 'ufc', 'wrestling']
const entertainment = ['celebrity', 'movie', 'film', 'tv show', 'netflix', 'hbo', 'album', 'concert', 'grammy', 'oscar', 'emmy', 'red carpet', 'fashion week', 'met gala']
return sports.some(k => low.includes(k)) || entertainment.some(k => low.includes(k))
}
const parseRssItems = xml => {
const items = []
const itemRegex = /<item>([\s\S]*?)<\/item>/gi
const entryRegex = /<entry>([\s\S]*?)<\/entry>/gi
const feedRegex = itemRegex.test(xml) ? itemRegex : entryRegex
const resetRegex = new RegExp(feedRegex.source, feedRegex.flags)
let match = resetRegex.exec(xml)
while (match) {
const block = match[1]
const titleMatch = block.match(/<title>[^!]([\s\S]*?)<\/title>/i) || block.match(/<title><!\[CDATA\[([\s\S]*?)\]\]><\/title>/i) || block.match(/<title>([\s\S]*?)<\/title>/i)
const linkMatch = block.match(/<link[^>]*href="([^"]+)"/i) || block.match(/<link>([\s\S]*?)<\/link>/i)
const descMatch = block.match(/<description><!\[CDATA\[([\s\S]*?)\]\]><\/description>/i) || block.match(/<description>([\s\S]*?)<\/description>/i) || block.match(/<content[^>]*>([\s\S]*?)<\/content>/i) || block.match(/<summary[^>]*>([\s\S]*?)<\/summary>/i)
const pubMatch = block.match(/<pubDate>([\s\S]*?)<\/pubDate>/i) || block.match(/<published>([\s\S]*?)<\/published>/i) || block.match(/<updated>([\s\S]*?)<\/updated>/i) || block.match(/<dc:date>([\s\S]*?)<\/dc:date>/i)
const title = titleMatch ? (titleMatch[1] || titleMatch[2] || '') : ''
const link = linkMatch ? (linkMatch[1] || linkMatch[2] || '').trim() : ''
const desc = descMatch ? stripHtml(descMatch[1] || descMatch[2] || '') : ''
const pubDate = pubMatch ? pubMatch[1].trim() : ''
if (title && link) items.push({ title, link, desc, pubDate })
match = resetRegex.exec(xml)
}
return items
}
const fetchGoogleNews = async source => {
const queryEncoded = encodeURIComponent(source.query)
const url = `https://news.google.com/rss/search?q=${queryEncoded}&hl=${source.hl}&gl=${source.gl}&ceid=${source.ceid}`
const client = createHttpClient()
const resp = await client.get(url, { timeout: getConfig().timeout_ms || 15000, validateStatus: () => true })
if (resp.status !== 200) throw new Error(`google_http_${resp.status}`)
const items = parseRssItems(String(resp.data || ''))
return items
}
const fetchDirectRss = async source => {
const rssUrl = String(source.fallback_rss || '').trim()
if (!rssUrl) throw new Error('no_fallback_rss')
const client = createHttpClient()
const resp = await client.get(rssUrl, { timeout: getConfig().timeout_ms || 15000, validateStatus: () => true, headers: { 'User-Agent': 'Feedly/1.0 (RSS Reader)', 'Accept': 'application/rss+xml, application/atom+xml, application/xml, text/xml' } })
if (resp.status !== 200) throw new Error(`rss_http_${resp.status}`)
const items = parseRssItems(String(resp.data || ''))
if (!items.length) throw new Error('rss_empty')
return items
}
const fetchRemoteProbe = async (source, cfg) => {
const remoteCfg = getRemoteProbeConfig(cfg)
if (!remoteCfg.base) throw new Error('remote_probe_base_missing')
if (!remoteCfg.token) throw new Error('remote_probe_token_missing')
const query = String((source && (source.remote_query || source.query || source.name)) || '').trim()
if (!query) throw new Error('remote_probe_query_missing')
const resp = await axios.get(`${remoteCfg.base}${remoteCfg.endpoint}`, {
timeout: remoteCfg.timeout,
validateStatus: () => true,
params: {
token: remoteCfg.token,
q: query,
lang: getRemoteProbeLang(source)
},
headers: {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
})
if (resp.status !== 200) throw new Error(`remote_probe_http_${resp.status}`)
const payload = resp.data
if (Array.isArray(payload)) return payload
if (payload && Array.isArray(payload.items)) return payload.items
if (payload && Array.isArray(payload.rows)) return payload.rows
throw new Error('remote_probe_invalid_payload')
}
const fetchAllSources = async (targetDate) => {
const cfg = getConfig()
if (!cfg.enabled) throw new Error('global_news_disabled')
const sources = Array.isArray(cfg.sources) ? cfg.sources : []
const defaultMax = cfg.max_headlines_per_source || 3
const financeMax = cfg.finance_max_headlines || 5
const fallbackEnabled = cfg.direct_rss_fallback !== false
const mode = getMode(cfg)
const allResults = []
const stats = mode === 'remote_ephemeral'
? { remote_ok: 0, remote_fail: 0, total_sources: sources.length }
: { google_ok: 0, google_fail: 0, rss_fallback_used: 0, rss_fallback_fail: 0, total_sources: sources.length }
const buildHeadlines = (source, items, method) => {
const category = source.category || 'general'
const maxHeadlines = category === 'finance' ? financeMax : defaultMax
const headlines = []
const lang = getRemoteProbeLang(source)
for (const item of items) {
const title = cleanTitle(item.title)
if (!title || isNoiseTitle(title)) continue
if (category !== 'finance' && isEntertainmentOrSport(title)) continue
if (headlines.length >= maxHeadlines) break
const timeInfo = normalizePublishedAtForStorage(item.pubDate || item.published_at || '', targetDate || '')
if (mode === 'remote_ephemeral' && targetDate && !isDateNearTarget(targetDate, timeInfo.date || targetDate)) continue
headlines.push({
date: timeInfo.date || targetDate || '',
country: source.country,
continent: source.continent,
source_name: String(item.source || item.source_name || source.name || ''),
category,
title,
summary: cleanSummary(item.desc || item.summary || item.description || item.snippet) || title,
link: mode === 'remote_ephemeral' ? normalizeRemoteProbeLink(item.link) : item.link,
published_at: timeInfo.publishedAt || item.pubDate || item.published_at || '',
fetch_method: method,
lang
})
}
return headlines
}
if (mode === 'remote_ephemeral') {
for (const batch of chunkArray(sources, REMOTE_PROBE_CONCURRENCY)) {
const batchResults = await Promise.all(batch.map(async source => {
try {
const items = await fetchRemoteProbe(source, cfg)
return { ok: true, source, items }
} catch (e) {
return { ok: false, source, error: e }
}
}))
for (const result of batchResults) {
if (!result.ok) {
stats.remote_fail++
console.error(`[global_news] Remote probe failed for ${result.source.name}:`, result.error && result.error.message ? result.error.message : result.error)
continue
}
stats.remote_ok++
allResults.push(...buildHeadlines(result.source, result.items, 'remote_probe'))
}
}
allResults.sort((a, b) => {
const ca = continentSortValue(a.continent) - continentSortValue(b.continent)
if (ca !== 0) return ca
return String(a.source_name).localeCompare(String(b.source_name))
})
return { items: allResults, stats, mode }
}
for (const source of sources) {
const category = source.category || 'general'
const maxHeadlines = category === 'finance' ? financeMax : defaultMax
let items = []
let method = 'google'
try {
items = await fetchGoogleNews(source)
stats.google_ok++
} catch (e) {
stats.google_fail++
console.error(`[global_news] Google News failed for ${source.name}:`, e.message)
if (fallbackEnabled && String(source.fallback_rss || '').trim()) {
try {
items = await fetchDirectRss(source)
method = 'rss_fallback'
stats.rss_fallback_used++
} catch (e2) {
stats.rss_fallback_fail++
console.error(`[global_news] RSS fallback failed for ${source.name}:`, e2.message)
continue
}
} else {
continue
}
}
allResults.push(...buildHeadlines(source, items, method))
}
allResults.sort((a, b) => {
const ca = continentSortValue(a.continent) - continentSortValue(b.continent)
if (ca !== 0) return ca
return String(a.source_name).localeCompare(String(b.source_name))
})
return { items: allResults, stats, mode }
}
const saveToDb = (dateStr, items, fetchMethod, fetchedAt) => {
const insert = db.prepare(`INSERT OR REPLACE INTO global_news (date, country, continent, source_name, category, title, summary, link, published_at, fetch_method, fetched_at, lang) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
const deleteOld = db.prepare(`DELETE FROM global_news WHERE date = ?`)
const batchTime = String(fetchedAt || fmtBatchTime())
const tx = db.transaction(() => {
deleteOld.run(dateStr)
for (const item of items) {
insert.run(dateStr, item.country, item.continent, item.source_name, item.category || 'general', item.title, item.summary, item.link, item.published_at, item.fetch_method || fetchMethod || 'google', batchTime, item.lang || '')
}
})
tx()
return items.length
}
const upsertItemsByOwnDate = (items, fetchedAt) => {
const insert = db.prepare(`INSERT OR REPLACE INTO global_news (date, country, continent, source_name, category, title, summary, link, published_at, fetch_method, fetched_at, lang) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
const batchTime = String(fetchedAt || fmtBatchTime())
const tx = db.transaction(list => {
for (const item of (Array.isArray(list) ? list : [])) {
const itemDate = String(item && item.date || '').trim()
if (!itemDate) continue
insert.run(itemDate, item.country || '', item.continent || '', item.source_name || '', item.category || 'general', item.title || '', item.summary || '', item.link || '', item.published_at || '', item.fetch_method || 'remote_probe', batchTime, item.lang || '')
}
})
tx(items)
return Array.isArray(items) ? items.length : 0
}
const runFetchForDate = async date => {
const cfg = getConfig()
const mode = getMode(cfg)
newsLog('INFO', mode === 'remote_ephemeral' ? '远程抓取开始' : '手动刷新开始', `date=${date}`)
if (mode !== 'remote_ephemeral') {
const deleted = db.prepare(`DELETE FROM global_news WHERE date = ?`).run(date).changes
if (deleted > 0) newsLog('INFO', '清理旧数据', `deleted ${deleted} rows for ${date}`)
}
const result = await fetchAllSources(date)
const fetchBatchAt = fmtBatchTime()
const saved = mode === 'remote_ephemeral'
? upsertItemsByOwnDate(result.items || [], fetchBatchAt)
: saveToDb(date, result.items || [], undefined, fetchBatchAt)
const impactedDates = Array.from(new Set((result.items || []).map(item => String(item.date || '').trim()).filter(Boolean)))
const rows = listByDateStmt.all(date)
newsLog('INFO', mode === 'remote_ephemeral' ? '远程抓取完成' : '手动刷新完成', `date=${date}, saved=${saved}`)
return {
ok: true,
mode,
ephemeral: false,
date,
saved,
count: rows.length,
rows,
fetched_at: fetchBatchAt,
total_sources: Array.isArray(cfg.sources) ? cfg.sources.length : 0,
stats: result.stats || {},
finance_sources: (cfg.sources || []).filter(s => (s.category || 'general') === 'finance').length,
translated: 0,
translate_errors: [],
analyzed: 0,
analyze_errors: [],
impacted_dates: impactedDates,
processing_dates: [date],
next_actions: ['translate', 'analyze']
}
}
const runTranslateForDate = async date => {
newsLog('INFO', 'AI 翻译开始', `date=${date}`)
const result = await translateNewsByDate(date)
newsLog('INFO', 'AI 翻译完成', `date=${date}, translated=${result.translated}, errors=${result.errors.length}`)
if (result.errors.length > 0) result.errors.slice(0, 5).forEach(e => newsLog('WARN', '翻译失败', e))
return { ok: true, date, translated: result.translated, errors: result.errors }
}
const runAnalyzeForDate = async date => {
newsLog('INFO', 'AI 研判开始', `date=${date}`)
const result = await analyzeNewsByDate(date)
newsLog('INFO', 'AI 研判完成', `date=${date}, analyzed=${result.analyzed}, errors=${result.errors.length}`)
if (result.errors.length > 0) result.errors.slice(0, 5).forEach(e => newsLog('WARN', '研判失败', e))
return { ok: true, date, analyzed: result.analyzed, errors: result.errors }
}
const todayLocal = () => {
return getBeijingNow().date
}
const yesterdayLocal = () => {
return addDaysToDateStr(todayLocal(), -1)
}
let scheduledTimer = null
const schedulerState = {
running: false,
stage: '',
lastTriggerKey: ''
}
const bindRoutes = app => {
app.get('/api/global_news/list', (req, res) => {
try {
const date = String(req.query.date || yesterdayLocal()).trim()
const cfg = getConfig()
const rows = db.prepare(`SELECT * FROM global_news WHERE date = ? ORDER BY CASE continent WHEN '北美' THEN 1 WHEN '欧洲' THEN 2 WHEN '亚太' THEN 3 WHEN '拉美' THEN 4 ELSE 5 END, source_name, id ASC`).all(date)
const stats = { google: rows.filter(r => r.fetch_method === 'google').length, rss_fallback: rows.filter(r => r.fetch_method === 'rss_fallback').length }
if (isRemoteEphemeralMode(cfg)) stats.remote_probe = rows.filter(r => r.fetch_method === 'remote_probe').length
res.json({ ok: true, mode: getMode(cfg), ephemeral: false, date, count: rows.length, stats, rows })
} catch (e) {
res.status(500).json({ ok: false, error: String(e.message || e) })
}
})
app.get('/api/global_news/dates', (req, res) => {
try {
const rows = dateRangeStmt.all()
res.json({ ok: true, dates: rows.map(r => r.date) })
} catch (e) {
res.status(500).json({ ok: false, error: String(e.message || e) })
}
})
app.post('/api/global_news/fetch', async (req, res) => {
try {
const forwarded = await proxyGlobalNewsRequest(req, '/api/global_news/fetch')
if (forwarded) return res.status(forwarded.status).json(forwarded.body)
const cfg = getConfig()
if (!cfg.enabled) return res.status(400).json({ ok: false, error: 'global_news_disabled' })
const date = String((req.body && req.body.date) || yesterdayLocal()).trim()
res.json(await runFetchForDate(date))
} catch (e) {
const msg = String(e.message || e)
newsLog('ERROR', '抓取失败', msg)
if (msg.includes('global_news_disabled')) return res.status(400).json({ ok: false, error: msg })
if (msg.includes('ECONNREFUSED') || msg.includes('ETIMEDOUT') || msg.includes('ENOTFOUND') || msg.includes('ECONNABORTED')) {
return res.status(502).json({ ok: false, error: `所有数据源均不可达,请检查服务器代理:${msg}。如未开代理,Google News 和海外 RSS 均无法访问。`, need_proxy: true })
}
res.status(500).json({ ok: false, error: msg })
}
})
app.get('/api/global_news/health', (req, res) => {
try {
const cfg = getConfig()
if (!isAutomationInstance(cfg)) {
proxyGlobalNewsRequest(req, '/api/global_news/health')
.then(forwarded => {
if (forwarded) return res.status(forwarded.status).json(forwarded.body)
return res.status(500).json({ ok: false, error: 'global_news_health_proxy_failed' })
})
.catch(e => res.status(500).json({ ok: false, error: String(e.message || e) }))
return
}
const dates = dateRangeStmt.all()
const mode = getMode(cfg)
const count = db.prepare(`SELECT COUNT(*) as cnt FROM global_news WHERE date = ?`).get(yesterdayLocal())
const todayRows = db.prepare(`SELECT fetch_method, COUNT(*) as cnt FROM global_news WHERE date = ? GROUP BY fetch_method`).all(yesterdayLocal())
const methodStats = {}
for (const r of todayRows) methodStats[r.fetch_method] = r.cnt
res.json({
ok: true,
enabled: !!cfg.enabled,
mode,
sources: (cfg.sources || []).length,
finance_sources: (cfg.sources || []).filter(s => (s.category || 'general') === 'finance').length,
proxy: cfg.proxy || 'none',
remote_base_url: String(cfg.remote_base_url || '').trim() || 'none',
automation_port: getAutomationPort(cfg),
current_port: getCurrentPort(),
is_automation_instance: isAutomationInstance(cfg),
fallback_enabled: cfg.direct_rss_fallback !== false,
latest_dates: dates.slice(0, 5).map(d => d.date),
today_count: (count && count.cnt) || 0,
today_by_method: methodStats
})
} catch (e) {
res.status(500).json({ ok: false, error: String(e.message || e) })
}
})
app.get('/api/global_news/logs', (req, res) => {
try {
const cfg = getConfig()
if (!isAutomationInstance(cfg)) {
proxyGlobalNewsRequest(req, '/api/global_news/logs')
.then(forwarded => {
if (forwarded) return res.status(forwarded.status).json(forwarded.body)
return res.status(500).json({ ok: false, error: 'global_news_logs_proxy_failed' })
})
.catch(e => res.status(500).json({ ok: false, error: String(e.message || e) }))
return
}
res.json({ ok: true, logs: newsLogs.slice() })
} catch (e) {
res.status(500).json({ ok: false, error: String(e.message || e) })
}
})
app.post('/api/global_news/clear_logs', (req, res) => {
try {
const cfg = getConfig()
if (!isAutomationInstance(cfg)) {
proxyGlobalNewsRequest(req, '/api/global_news/clear_logs')
.then(forwarded => {
if (forwarded) return res.status(forwarded.status).json(forwarded.body)
return res.status(500).json({ ok: false, error: 'global_news_clear_logs_proxy_failed' })
})
.catch(e => res.status(500).json({ ok: false, error: String(e.message || e) }))
return
}
newsLogs.length = 0
res.json({ ok: true })
} catch (e) {
res.status(500).json({ ok: false, error: String(e.message || e) })
}
})
app.post('/api/global_news/translate', async (req, res) => {
try {
const forwarded = await proxyGlobalNewsRequest(req, '/api/global_news/translate')
if (forwarded) return res.status(forwarded.status).json(forwarded.body)
const date = String(req.body.date || yesterdayLocal()).trim()
res.json(await runTranslateForDate(date))
} catch (e) {
newsLog('ERROR', '翻译异常', e.message)
res.status(500).json({ ok: false, error: String(e.message || e) })
}
})
app.post('/api/global_news/analyze', async (req, res) => {
try {
const forwarded = await proxyGlobalNewsRequest(req, '/api/global_news/analyze')
if (forwarded) return res.status(forwarded.status).json(forwarded.body)
const date = String(req.body.date || yesterdayLocal()).trim()
res.json(await runAnalyzeForDate(date))
} catch (e) {
newsLog('ERROR', '研判异常', e.message)
res.status(500).json({ ok: false, error: String(e.message || e) })
}
})
}
const isJapaneseText = (text) => {
// 检测日语特征:平假名 \u3040-\u309F 或 片假名 \u30A0-\u30FF
const japaneseRegex = /[\u3040-\u309F\u30A0-\u30FF]/
return japaneseRegex.test(text)
}
const isChineseText = (text) => {
// 如果包含日语特征(平假名/片假名),则不是中文
if (isJapaneseText(text)) return false
// 检测文本是否包含中文(CJK 统一汉字 Unicode 范围)
const chineseRegex = /[\u3400-\u4DBF\u4E00-\u9FFF\uFA0E-\uFA0F\uFA11\uFA13-\uFA14\uFA1F\uFA21\uFA23-\uFA24\uFA27-\uFA29]/
return chineseRegex.test(text)
}
const isChineseLang = (lang) => {
const l = String(lang || '').toLowerCase()
return l.startsWith('zh') || l === 'cn'
}
const translateNewsByDate = async (dateStr, options = {}) => {
const ai = getAiConfig()
if (!ai.enabled || !ai.apiKey) return { translated: 0, errors: ['AI翻译未启用或无API Key'] }
const fetchedAt = String(options && options.fetchedAt || '').trim()
const rows = fetchedAt
? db.prepare(`SELECT id, title, summary, lang FROM global_news WHERE date = ? AND title_cn = '' AND fetched_at = ? ORDER BY id ASC`).all(dateStr, fetchedAt)
: db.prepare(`SELECT id, title, summary, lang FROM global_news WHERE date = ? AND title_cn = '' ORDER BY id ASC`).all(dateStr)
if (!rows.length) return { translated: 0, errors: [] }
const updateStmt = db.prepare(`UPDATE global_news SET title_cn = ?, summary_cn = ? WHERE id = ?`)
let translated = 0
const errors = []
for (let i = 0; i < rows.length; i++) {
const row = rows[i]
// 如果是中文,直接使用原文,不翻译
if (isChineseLang(row.lang) || isChineseText(row.title)) {
const s = row.summary ? cleanSummary(row.summary) : ''
updateStmt.run(row.title, s || row.title, row.id)
translated++
continue
}
const s = row.summary ? cleanSummary(row.summary) : ''
const titleForAI = row.title
const summaryForAI = s || row.title
try {
const resp = await axios.post(ai.apiBase, {
model: ai.model,
temperature: 0.1,
messages: [
{ role: 'system', content: 'Translate this news title and summary to Simplified Chinese (简体中文). Return ONLY a JSON object like {"title_cn":"中文标题","summary_cn":"中文简述"}. Keep financial terms accurate. No markdown, no explanations.' },
{ role: 'user', content: `title: ${titleForAI}\nsummary: ${summaryForAI}` }
]
}, {
timeout: 60000,
headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${ai.apiKey}` }
})
const raw = resp.data && resp.data.choices && resp.data.choices[0] && resp.data.choices[0].message
? String(resp.data.choices[0].message.content || '').trim()
: ''
if (!raw) { errors.push(`row_${row.id}_empty`); continue }
let translation = {}
try {
const jsonMatch = raw.match(/\{[\s\S]*\}/)
if (jsonMatch) {
translation = JSON.parse(jsonMatch[0])
} else {
translation = JSON.parse(raw)
}
} catch (e) {
errors.push(`row_${row.id}_parse: ${String(e.message || e).slice(0, 60)}`)
newsLog('WARN', '翻译解析失败', `row ${row.id}: raw=${raw.slice(0, 200)}`)
continue
}
if (!translation || typeof translation !== 'object') {
errors.push(`row_${row.id}_not_object`)
continue
}
const titleCn = String(translation.title_cn || '').slice(0, 300)
const summaryCn = String(translation.summary_cn || '').slice(0, 500)
if (titleCn || summaryCn) {
updateStmt.run(titleCn, summaryCn, row.id)
translated++
} else {
errors.push(`row_${row.id}_empty_result`)
}
} catch (e) {
const msg = String(e.message || e).slice(0, 100)
errors.push(`row_${row.id}_err: ${msg}`)
newsLog('ERROR', '翻译请求失败', `row ${row.id}: ${msg}`)
}
}
return { translated, errors }
}
const ANALYZE_SYSTEM = [
'你是资深A股/ETF/基金投资顾问,坐标上海。',
'',
'以下有若干条新闻。你必须逐条独立分析,不能合并。对每一条新闻,从三个方向逐一研判,每个方向50-100字:',
'1. 国内市场(A股个股、宽基指数、行业ETF、基金定投)',
'2. 海外市场(QDII、跨境ETF、港股通、海外指数)',
'3. 防御性资产(黄金、债券、红利低波、货币基金)',
'',
'相关性判断标准(放宽原则):',
'- 涉及以下任一领域即视为相关:宏观经济/货币政策/财政政策/监管政策/产业政策、',
' 科技供应链/半导体/芯片/AI、大宗商品/能源/矿产、地产/基建、',
' 地缘政治/贸易摩擦/制裁、金融监管/证券法规/交易所规则',
'- 纯社会娱乐/体育/花边/媒体人事变动才判为无关',
'',
'输出严格JSON数组,长度必须等于输入新闻条数,每条对应对应:',
'[{"status":"related或unrelated","domestic":"...","overseas":"...","defensive":"..."}]',
'',
'status="unrelated"时三类都填"【无关】"。不荐股不臆测。侧重中长期配置、低位分批建仓、行情前置信号。'
].join('\n')
const analyzeNewsByDate = async (dateStr, options = {}) => {
const ai = getAiConfig()
if (!ai.enabled || !ai.apiKey) return { analyzed: 0, errors: ['AI分析未启用或无API Key'] }
const fetchedAt = String(options && options.fetchedAt || '').trim()
const rows = fetchedAt
? db.prepare(`SELECT id, title, title_cn, summary, summary_cn FROM global_news WHERE date = ? AND ai_status = '' AND fetched_at = ? ORDER BY id ASC`).all(dateStr, fetchedAt)
: db.prepare(`SELECT id, title, title_cn, summary, summary_cn FROM global_news WHERE date = ? AND ai_status = '' ORDER BY id ASC`).all(dateStr)
if (!rows.length) return { analyzed: 0, errors: [] }
const updateStmt = db.prepare(`UPDATE global_news SET ai_status = ?, ai_domestic = ?, ai_overseas = ?, ai_defensive = ? WHERE id = ?`)
let analyzed = 0
const errors = []
const batchSize = 8
const batches = chunkArray(rows, batchSize)
for (let i = 0; i < batches.length; i++) {
const batch = batches[i]
const payloadText = batch.map((row, idx) => {
const cnTitle = row.title_cn || row.title
const cnSummary = row.summary_cn || cleanSummary(row.summary) || row.title
return `${idx + 1}. 标题:${cnTitle}\n简述:${cnSummary}`
}).join('\n\n')
try {
const resp = await axios.post(ai.apiBase, {
model: ai.model,
temperature: 0.2,
messages: [
{ role: 'system', content: ANALYZE_SYSTEM },
{ role: 'user', content: `请严格按顺序逐条分析以下 ${batch.length} 条新闻,并只返回 JSON 数组:\n\n${payloadText}` }
]
}, {
timeout: 60000,
headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${ai.apiKey}` }
})
const raw = resp.data && resp.data.choices && resp.data.choices[0] && resp.data.choices[0].message
? String(resp.data.choices[0].message.content || '').trim()
: ''
if (!raw) {
for (const row of batch) errors.push(`row_${row.id}_empty`)
continue
}
let results = null
try {
const arrMatch = raw.match(/\[[\s\S]*\]/)
results = arrMatch ? JSON.parse(arrMatch[0]) : JSON.parse(raw)
} catch (e) {
newsLog('WARN', '研判解析失败', `batch ${i + 1}/${batches.length}: raw=${raw.slice(0, 300)}`)
}
if (!Array.isArray(results) || results.length !== batch.length) {
for (const row of batch) {
const cnTitle = row.title_cn || row.title
const cnSummary = row.summary_cn || cleanSummary(row.summary) || row.title
try {
const respOne = await axios.post(ai.apiBase, {
model: ai.model,
temperature: 0.2,
messages: [
{ role: 'system', content: ANALYZE_SYSTEM },
{ role: 'user', content: `分析以下这条新闻:\n\n${cnTitle}\n简述:${cnSummary}` }
]
}, {
timeout: 60000,
headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${ai.apiKey}` }
})
const rawOne = respOne.data && respOne.data.choices && respOne.data.choices[0] && respOne.data.choices[0].message
? String(respOne.data.choices[0].message.content || '').trim()
: ''
if (!rawOne) { errors.push(`row_${row.id}_empty`); continue }
let result = null
try {
const arrMatch = rawOne.match(/\[[\s\S]*?\]/)
if (arrMatch) {
const arr = JSON.parse(arrMatch[0])
if (Array.isArray(arr) && arr.length > 0) result = arr[0]
}
if (!result) {
const objMatch = rawOne.match(/\{[\s\S]*\}/)
if (objMatch) result = JSON.parse(objMatch[0])
else result = JSON.parse(rawOne)
}
} catch (e) {
errors.push(`row_${row.id}_parse: ${String(e.message || e).slice(0, 60)}`)
newsLog('WARN', '研判解析失败', `row ${row.id}: raw=${rawOne.slice(0, 200)}`)
continue
}
if (!result || typeof result !== 'object') {
errors.push(`row_${row.id}_not_object`)
continue
}
const status = result.status === 'unrelated' ? 'unrelated' : 'related'
updateStmt.run(
status,
String(result.domestic || '').slice(0, 600),
String(result.overseas || '').slice(0, 600),
String(result.defensive || '').slice(0, 600),
row.id
)
analyzed++
} catch (e) {
const msg = String(e.message || e).slice(0, 100)
errors.push(`row_${row.id}_err: ${msg}`)
newsLog('ERROR', '研判请求失败', `row ${row.id}: ${msg}`)
}
}
continue
}
for (let j = 0; j < batch.length; j++) {
const row = batch[j]
const result = results[j]
if (!result || typeof result !== 'object') {
errors.push(`row_${row.id}_not_object`)
continue
}
const status = result.status === 'unrelated' ? 'unrelated' : 'related'
updateStmt.run(
status,
String(result.domestic || '').slice(0, 600),
String(result.overseas || '').slice(0, 600),
String(result.defensive || '').slice(0, 600),
row.id
)
analyzed++
}
} catch (e) {
const msg = String(e.message || e).slice(0, 100)
for (const row of batch) errors.push(`row_${row.id}_err: ${msg}`)
newsLog('ERROR', '批量研判请求失败', `batch ${i + 1}/${batches.length}: ${msg}`)
}
}
return { analyzed, errors }
}
const runScheduledFetch = async () => {
const cfg = getConfig()
if (!cfg.enabled) return
const date = yesterdayLocal()
try {
newsLog('INFO', '定时抓取任务开始', `date=${date}`)
const result = await runFetchForDate(date)
newsLog('INFO', '定时抓取任务完成', `date=${date}, saved=${result.saved}`)
} catch (e) {
newsLog('ERROR', '定时抓取任务失败', `date=${date}: ${e.message}`)
}
}
const runScheduledTranslate = async () => {
const cfg = getConfig()
if (!cfg.enabled) return
const date = yesterdayLocal()
try {
newsLog('INFO', '定时翻译任务开始', `date=${date}`)
await runTranslateForDate(date)
} catch (e) {
newsLog('ERROR', '定时翻译任务失败', `date=${date}: ${e.message}`)
}
}
const runScheduledAnalyze = async () => {
const cfg = getConfig()
if (!cfg.enabled) return
const date = yesterdayLocal()
try {
newsLog('INFO', '定时研判任务开始', `date=${date}`)
await runAnalyzeForDate(date)
} catch (e) {
newsLog('ERROR', '定时研判任务失败', `date=${date}: ${e.message}`)
}
}
const maybeRunScheduledStage = async (stage, runner, triggerKey) => {
if (schedulerState.running) {
newsLog('WARN', '定时任务跳过', `已有任务运行中 stage=${schedulerState.stage}, incoming=${stage}`)
return
}
if (schedulerState.lastTriggerKey === triggerKey) return
schedulerState.running = true
schedulerState.stage = stage
schedulerState.lastTriggerKey = triggerKey
try {
await runner()
} finally {
schedulerState.running = false
schedulerState.stage = ''
}
}
const startScheduler = () => {
const cfg = getConfig()
if (!cfg.enabled) return
if (!isAutomationInstance(cfg)) {
newsLog('INFO', 'global_news scheduler standby', `current_port=${getCurrentPort()}, automation_port=${getAutomationPort(cfg)}`)
return
}
if (scheduledTimer) clearInterval(scheduledTimer)
const checkAndRun = () => {
const now = getBeijingHourMinute()
const fetchHour = Number(cfg.fetch_schedule_hour ?? 1)
const fetchMinute = Number(cfg.fetch_schedule_minute ?? 0)
const translateHour = Number(cfg.translate_schedule_hour ?? 2)
const translateMinute = Number(cfg.translate_schedule_minute ?? 0)
const analyzeHour = Number(cfg.analyze_schedule_hour ?? 3)
const analyzeMinute = Number(cfg.analyze_schedule_minute ?? 0)
if (now.hour === fetchHour && now.minute === fetchMinute) {
maybeRunScheduledStage('fetch', runScheduledFetch, `${now.date}:fetch`).catch(() => {})
return
}
if (now.hour === translateHour && now.minute === translateMinute) {
maybeRunScheduledStage('translate', runScheduledTranslate, `${now.date}:translate`).catch(() => {})
return
}
if (now.hour === analyzeHour && now.minute === analyzeMinute) {
maybeRunScheduledStage('analyze', runScheduledAnalyze, `${now.date}:analyze`).catch(() => {})
}
}
scheduledTimer = setInterval(checkAndRun, 60000)
checkAndRun()
newsLog('INFO', 'global_news scheduler started', `tz=Asia/Shanghai fetch=${String(Number(cfg.fetch_schedule_hour ?? 1)).padStart(2, '0')}:${String(Number(cfg.fetch_schedule_minute ?? 0)).padStart(2, '0')} translate=${String(Number(cfg.translate_schedule_hour ?? 2)).padStart(2, '0')}:${String(Number(cfg.translate_schedule_minute ?? 0)).padStart(2, '0')} analyze=${String(Number(cfg.analyze_schedule_hour ?? 3)).padStart(2, '0')}:${String(Number(cfg.analyze_schedule_minute ?? 0)).padStart(2, '0')}`)
}
module.exports = { bindRoutes, startScheduler, runScheduledFetch, runScheduledTranslate, runScheduledAnalyze, fetchAllSources, saveToDb, translateNewsByDate, analyzeNewsByDate, todayLocal }