// 采集器: 家庭/设备发现 + 每小时采样 + 电量 + 日聚合 + vendor 补齐编排 import { prisma } from '../db.js' import { checkAuth, getApi, MihomeAuthError } from './client.js' import { getSpec } from './specs.js' import { dayStart, hourStart, addDays } from './time.js' import * as vendorStats from './vendorStats.js' const BACKFILL_DAYS = 45 // 停机/首次接入时, 云端统计补齐窗口 const round2 = (n) => (Number.isFinite(n) ? Math.round(n * 100) / 100 : null) // 互斥锁: 同一时刻只允许一个同步在跑 (cron 与手动/启动触发并发时合并) let runLock = null function withLock(fn) { if (runLock) return runLock runLock = Promise.resolve() .then(fn) .finally(() => { runLock = null }) return runLock } // ==================== 设备发现 ==================== async function syncHomes() { const api = getApi() const homes = await api.getHomesList() for (const h of homes) { const homeId = String(h.id) await prisma.mihomeHome.upsert({ where: { homeId }, create: { homeId, name: h.name || homeId }, update: { name: h.name || homeId }, }) } return homes.length } async function syncDevices() { const api = getApi() const list = await api.getDevicesList() let kept = 0 for (const d of list) { const spec = getSpec(d.model) if (!spec) continue // 非环境设备白名单, 跳过 const homeId = String(d.home_id || '') if (!homeId) continue const data = { homeId, model: d.model, name: String(d.name || d.did), tempSiid: spec.temp[0], tempPiid: spec.temp[1], humSiid: spec.hum[0], humPiid: spec.hum[1], battSiid: spec.batt[0], battPiid: spec.batt[1], } await prisma.mihomeDevice.upsert({ where: { did: d.did }, create: { did: d.did, ...data }, update: data, }) kept++ } return kept } async function listEnabledDevices() { return prisma.mihomeDevice.findMany({ where: { enabled: true } }) } // ==================== 采样 ==================== // 每小时采样: 批量读温湿度 -> upsert 小时档 async function sampleHour(devices) { const api = getApi() const bucket = hourStart() const props = [] for (const d of devices) { props.push({ did: d.did, siid: d.tempSiid, piid: d.tempPiid }) props.push({ did: d.did, siid: d.humSiid, piid: d.humPiid }) } if (props.length === 0) return 0 const results = [] try { const res = await api.getDevicesProp(props) results.push(...(Array.isArray(res) ? res : [res])) } catch (e) { // 整体失败: 不写入任何小时档, 缺档留给整点补采/vendor 补齐, 本次直接抛给上层记录 throw e } // did -> { temp, hum } const byDevice = new Map(devices.map((d) => [d.did, d])) const perDevice = new Map() for (const r of results) { if (!r || !r.did) continue const ok = Number(r.code) === 0 && r.value !== undefined && r.value !== null if (!ok) continue const d = byDevice.get(r.did) if (!d) continue const cur = perDevice.get(r.did) || {} if (r.siid === d.tempSiid && r.piid === d.tempPiid) { cur.temperatureAvg = round2(r.value) } else if (r.siid === d.humSiid && r.piid === d.humPiid) { cur.humidityAvg = round2(r.value) } perDevice.set(r.did, cur) } let written = 0 for (const d of devices) { const v = perDevice.get(d.did) if (!v || (v.temperatureAvg === undefined && v.humidityAvg === undefined)) continue await prisma.mihomeEnvHour.upsert({ where: { deviceId_hour: { deviceId: d.id, hour: bucket } }, create: { deviceId: d.id, hour: bucket, temperatureAvg: v.temperatureAvg ?? null, humidityAvg: v.humidityAvg ?? null, raw: null, // 自采 }, update: { temperatureAvg: v.temperatureAvg ?? null, humidityAvg: v.humidityAvg ?? null, raw: null, }, }) await prisma.mihomeDevice.update({ where: { id: d.id }, data: { lastSyncAt: new Date() }, }) written++ } return written } // 电量: 每天读一次 (batteryAt 不在今天就刷新), 标题展示用 async function refreshBattery(devices) { const api = getApi() const today = dayStart() for (const d of devices) { const isFresh = d.batteryAt && d.batteryAt.getTime() >= today.getTime() if (isFresh) continue try { const res = await api.getDevicesProp({ did: d.did, siid: d.battSiid, piid: d.battPiid }) if (res && Number(res.code) === 0 && typeof res.value === 'number') { await prisma.mihomeDevice.update({ where: { id: d.id }, data: { battery: Math.round(res.value), batteryAt: new Date() }, }) } } catch (e) { // 电量读不到不报错, 下次再试 } } } // ==================== 日聚合 ==================== // 按日档重算: 读该上海日小时内档 -> avg/min/max + 最后一次值 export async function aggregateDay(device, dayDate) { const start = dayStart(dayDate.getTime()) const end = addDays(start, 1) const hours = await prisma.mihomeEnvHour.findMany({ where: { deviceId: device.id, hour: { gte: start, lt: end } }, orderBy: { hour: 'asc' }, }) if (hours.length === 0) return false const temps = hours.filter((h) => h.temperatureAvg != null).map((h) => h.temperatureAvg) const hums = hours.filter((h) => h.humidityAvg != null).map((h) => h.humidityAvg) const last = hours[hours.length - 1] const avg = (arr) => (arr.length ? arr.reduce((a, b) => a + b, 0) / arr.length : null) const allVendor = hours.every((h) => h.raw && h.raw.vendor === true) await prisma.mihomeEnvSample.upsert({ where: { deviceId_date: { deviceId: device.id, date: start } }, create: { deviceId: device.id, date: start, temperature: temps.length ? temps[temps.length - 1] : null, humidity: hums.length ? hums[hums.length - 1] : null, temperatureAvg: round2(avg(temps)), humidityAvg: round2(avg(hums)), temperatureMin: round2(temps.length ? Math.min(...temps) : null), temperatureMax: round2(temps.length ? Math.max(...temps) : null), humidityMin: round2(hums.length ? Math.min(...hums) : null), humidityMax: round2(hums.length ? Math.max(...hums) : null), origin: allVendor ? 'vendor_history' : 'live', sampledAt: last.hour, raw: { bucketCount: hours.length }, }, update: { temperature: temps.length ? temps[temps.length - 1] : null, humidity: hums.length ? hums[hums.length - 1] : null, temperatureAvg: round2(avg(temps)), humidityAvg: round2(avg(hums)), temperatureMin: round2(temps.length ? Math.min(...temps) : null), temperatureMax: round2(temps.length ? Math.max(...temps) : null), humidityMin: round2(hums.length ? Math.min(...hums) : null), humidityMax: round2(hums.length ? Math.max(...hums) : null), origin: allVendor ? 'vendor_history' : 'live', sampledAt: last.hour, }, }) return true } // 补聚合缺口日: [since, today) 有小时档却没有日档的日子 export async function aggregateGapDays(device, since) { const today = dayStart() const existing = await prisma.mihomeEnvSample.findMany({ where: { deviceId: device.id, date: { gte: since, lt: today } }, select: { date: true }, }) const have = new Set(existing.map((s) => s.date.getTime())) let agg = 0 for (let t = since.getTime(); t < today.getTime(); t += 86400000) { if (!have.has(t)) { const ok = await aggregateDay(device, new Date(t)) if (ok) agg++ } } return agg } // ==================== 对外入口 ==================== // 每次同步: 发现 + 当前小时采样 + 今天/昨天聚合 + 电量 (轻量, 每小时跑) // 注意: 单独不包 withLock, 由 syncNow/dailyMaintenance 统一加锁, 避免嵌套自锁 async function syncLocked() { const auth = await checkAuth() if (auth.status !== 'ok') throw new MihomeAuthError(auth.lastError || '米家未授权或已过期') const began = Date.now() const homes = await syncHomes() const devices = await syncDevices() const kept = await listEnabledDevices() const sampled = await sampleHour(kept) await refreshBattery(kept) // 今天 + 昨天都聚合 (保证 00 点边界不丢) const nowMs = Date.now() const today = dayStart(nowMs) let agg = 0 for (const d of kept) { if (await aggregateDay(d, today)) agg++ const y = addDays(today, -1) if (await aggregateDay(d, y)) agg++ } return { ok: true, homes, devices, sampled, aggregated: agg, ms: Date.now() - began, } } export function syncNow() { return withLock(syncLocked) } // 每日维护: 轻量同步 + vendor 补齐历史缺口 (45 天窗口) + 缺口日聚合 export function dailyMaintenance() { return withLock(async () => { const r = await syncLocked() const auth = await checkAuth() if (auth.status !== 'ok') throw new MihomeAuthError(auth.lastError || '米家未授权或已过期') const since = addDays(dayStart(), -BACKFILL_DAYS) const devices = await listEnabledDevices() let hFilled = 0 let dFilled = 0 let agg = 0 for (const d of devices) { hFilled += await vendorStats.fillHourGaps(d, since) dFilled += await vendorStats.fillDayGaps(d, since) agg += await aggregateGapDays(d, since) } return { ...r, vendorHours: hFilled, vendorDays: dFilled, gapAggregated: agg } }) }