新增米家环境监控模块 (采集/聚合/历史/前端 EnvView) + CI 挂载米家凭证目录并自动应用迁移

server/mihome/collector.js

@@ -0,0 +1,285 @@
+// 采集器: 家庭/设备发现 + 每小时采样 + 电量 + 日聚合 + 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 }
+  })
+}