iboard/server/mihome/vendorStats.js

// /v2/user/statistics 云端聚合历史: 探测可用性 + 小时/日缺口回填
// 实测: 炎米 ym001 可用 (stat_hour_v3/stat_day_v3/stat_month_v3); 青萍/VRF 返回 -8 invalid params
import { prisma } from '../db.js'
import { getApi } from './client.js'
import { dayStart, hourStart, addDays, addHours } from './time.js'
import { getSpec } from './specs.js'

// 型号 -> 是否支持统计 (探测一次后缓存; undefined=未探测)
const capabilityCache = new Map()

export function capabilityOf(model) {
  if (capabilityCache.has(model)) return capabilityCache.get(model)
  return undefined
}

// 用同型号任一设备探测该型号统计可用性
async function ensureCapability(device) {
  if (capabilityCache.has(device.model)) return capabilityCache.get(device.model)
  let supported = false
  try {
    const api = getApi()
    const res = await api.getStatistics({
      did: device.did,
      key: `${device.tempSiid}.${device.tempPiid}`,
      data_type: 'stat_day_v3',
      limit: 1,
      time_start: Math.floor(Date.now() / 1000) - 86400,
      time_end: Math.floor(Date.now() / 1000),
    })
    supported = Array.isArray(res)
  } catch (e) {
    supported = false
  }
  capabilityCache.set(device.model, supported)
  console.log(`[mihome] 统计探测 ${device.model}: ${supported ? '可用' : '不可用'}`)
  return supported
}

// 解析 "[28.8]" / "[28.1,29.200001]" -> [28.8, 29.2]
function parseStatValue(value) {
  if (typeof value !== 'string') return []
  const m = value.match(/\[([^\]]*)\]/)
  if (!m) return []
  return m[1]
    .split(',')
    .map((s) => parseFloat(s.trim()))
    .filter((n) => Number.isFinite(n))
}

// 拉某设备某属性的小时均值序列 (stat_hour_v3), 默认带排序
async function fetchStatHours(device, key, fromTs, toTs) {
  const api = getApi()
  const res = await api.getStatistics({
    did: device.did,
    key,
    data_type: 'stat_hour_v3',
    limit: 2000,
    time_start: fromTs,
    time_end: toTs,
  })
  if (!Array.isArray(res)) return []
  return res
    .filter((r) => r && r.time)
    .sort((a, b) => a.time - b.time)
    .map((r) => ({ time: new Date(r.time * 1000), values: parseStatValue(r.value) }))
}

// 回填小时缺口: 查 [since, now) 内缺失的整点桶, 从云端补齐 (origin 标记 raw.vendor=true)
// 返回新增小时桶数量
export async function fillHourGaps(device, since) {
  if (!(await ensureCapability(device))) return 0
  const now = hourStart()
  if (since >= now) return 0

  const existing = await prisma.mihomeEnvHour.findMany({
    where: {
      deviceId: device.id,
      hour: { gte: since, lte: addHours(now, -1) },
    },
    select: { hour: true },
  })
  const have = new Set(existing.map((h) => h.hour.getTime()))

  // 实际缺失桶数
  let missingCount = 0
  for (let t = since.getTime(); t <= addHours(now, -1).getTime(); t += 3600000) {
    if (!have.has(t)) missingCount++
  }
  if (missingCount === 0) return 0

  const fromTs = Math.floor(since.getTime() / 1000)
  const toTs = Math.floor(now.getTime() / 1000)
  let filled = 0

  for (const [siid, piid, col, num] of [
    [device.tempSiid, device.tempPiid, 'temperatureAvg', 0],
    [device.humSiid, device.humPiid, 'humidityAvg', 1],
  ]) {
    const rows = await fetchStatHours(device, `${siid}.${piid}`, fromTs, toTs)
    for (const r of rows) {
      const bucket = r.time.getTime()
      if (r.values.length <= num || have.has(bucket) || bucket > now.getTime()) continue
      // 只补严格已过去的整点
      have.add(bucket)
      try {
        await prisma.mihomeEnvHour.upsert({
          where: { deviceId_hour: { deviceId: device.id, hour: r.time } },
          create: { deviceId: device.id, hour: r.time, [col]: r.values[num], raw: { vendor: true } },
          update: { [col]: r.values[num], raw: { vendor: true } },
        })
        filled++
      } catch (e) {
        if (e?.code !== 'P2002') throw e
      }
    }
  }
  if (filled > 0) console.log(`[mihome] vendor 补齐小时档 ${device.did}: +${filled}`)
  return filled
}

// 回填日缺口: 对仍无日档的日子, 若云端有日 [min,max] 则补一条 vendor_history
export async function fillDayGaps(device, since) {
  if (!(await ensureCapability(device))) return 0
  const today = dayStart()
  const existingSamples = await prisma.mihomeEnvSample.findMany({
    where: { deviceId: device.id, date: { gte: since, lt: today } },
    select: { date: true },
  })
  const haveDay = new Set(existingSamples.map((s) => s.date.getTime()))

  const fromTs = Math.floor(since.getTime() / 1000)
  const toTs = Math.floor(Date.now() / 1000)
  let filled = 0

  for (const [siid, piid] of [
    [device.tempSiid, device.tempPiid],
    [device.humSiid, device.humPiid],
  ]) {
    const api = getApi()
    const res = await api.getStatistics({
      did: device.did,
      key: `${siid}.${piid}`,
      data_type: 'stat_day_v3',
      limit: 120,
      time_start: fromTs,
      time_end: toTs,
    })
    if (!Array.isArray(res)) continue
    const isTemp = siid === device.tempSiid && piid === device.tempPiid
    for (const r of res) {
      if (!r || !r.time) continue
      const day = new Date(r.time * 1000)
      const vals = parseStatValue(r.value)
      const bucket = dayStart(day.getTime()).getTime()
      if (haveDay.has(bucket) || bucket >= today.getTime() || vals.length < 2) continue
      haveDay.add(bucket)
      const data = isTemp
        ? { temperatureMin: vals[0], temperatureMax: vals[1] }
        : { humidityMin: vals[0], humidityMax: vals[1] }
      try {
        await prisma.mihomeEnvSample.upsert({
          where: { deviceId_date: { deviceId: device.id, date: dayStart(day.getTime()) } },
          create: {
            deviceId: device.id,
            date: dayStart(day.getTime()),
            origin: 'vendor_history',
            ...data,
          },
          update: { origin: 'vendor_history', ...data },
        })
        filled++
      } catch (e) {
        if (e?.code !== 'P2002') throw e
      }
    }
  }
  if (filled > 0) console.log(`[mihome] vendor 补齐日档 ${device.did}: +${filled}`)
  return filled
}