C# Winform WebView2 的生命周期详解：初始化、导航、销毁全流程

2026-05-13

🎯 你真的了解 WebView2 的"一生"吗？

在 WinForms 项目中嵌入 WebView2 控件，看起来不过是拖一个控件、加几行代码的事。但不少开发者在实际项目里踩过坑：页面加载完才能执行脚本，结果脚本压根没跑；窗口关了，进程还活着；导航事件顺序搞不清，拦截逻辑写错了位置……

这些问题的根源，几乎都指向同一个盲区——对 WebView2 生命周期的理解不够深入。

WebView2 并不是一个普通的 UI 控件，它背后运行着一个独立的 Chromium 浏览器进程，有自己完整的初始化流程、导航状态机和销毁机制。如果把它当普通控件用，迟早会在内存泄漏、进程残留、事件时序这三个地方摔跟头。

读完本文，你将掌握：

WebView2 从控件创建到 CoreWebView2 就绪的完整初始化流程
导航的六个关键事件及其触发顺序与适用场景
窗口关闭时如何正确释放 WebView2，避免进程残留

测试环境：.NET 6 / WinForms，Microsoft.Web.WebView2 1.0.2045.28，Windows 11 22H2。

🔬 一、初始化阶段：两步走，缺一不可

WebView2 的"双层结构"

很多人第一次用 WebView2 时，会直接在 Form_Load 里调用 webView21.CoreWebView2.Navigate(url)，然后迎来一个经典异常：


NullReferenceException: Object reference not set to an instance of an object.

原因很简单：WebView2 控件和 CoreWebView2 是两个不同层次的对象。

WebView2（控件层）：WinForms 控件，随窗体创建而存在，负责 UI 渲染区域。
CoreWebView2（引擎层）：Chromium 浏览器进程的托管包装，异步初始化，未完成前为 null。

这就像买了一台电视（控件），但显像管（引擎）还没装好，你不能直接换台。

正确的初始化姿势

初始化分两步：调用 EnsureCoreWebView2Async + 等待 CoreWebView2InitializationCompleted 事件。

csharp
public partial class MainForm : Form
{
    public MainForm()
    {
        InitializeComponent();
        // 窗体加载时启动异步初始化
        this.Load += MainForm_Load;
    }

    private async void MainForm_Load(object sender, EventArgs e)
    {
        // 第一步：触发 CoreWebView2 异步初始化
        // 可传入 WebView2EnvironmentOptions 自定义用户数据目录、启动参数等
        await webView21.EnsureCoreWebView2Async(null);

        // 到这里，CoreWebView2 已就绪，可以安全操作
        webView21.CoreWebView2.Navigate("https://example.com");
    }
}

EnsureCoreWebView2Async 内部做了什么？简单说，它会：

查找或创建 WebView2 运行时环境（CoreWebView2Environment）
启动独立的 msedgewebview2.exe 子进程
建立控件与浏览器进程之间的 IPC 通道
将 CoreWebView2 对象注入控件

整个过程是异步的，在低配机器上可能需要 300~800ms。在 await 完成之前，CoreWebView2 始终为 null，这是 NullReferenceException 的根本原因。

阅读全文

C# 条件语句全解析，从设备状态判断到报警分级，一次学透

编辑

2026-05-13

🔧 开篇钩子

注塑机料筒温度传感器数据已经读进来了，你打开 VS，准备写判断逻辑：温度正常就继续生产，温度偏高就报警，温度过高就紧急停机。

听起来很简单，结果写出来的代码要么判断顺序错了，要么 switch 写了一半发现不知道怎么匹配范围……

这种感觉，像极了拿着对的零件却装错了位置。

今天这篇，就把条件语句这块彻底理清楚。

📌 上节回顾

「上一节我们学了字符串操作，掌握了拼接、格式化和插值表达式（$"..." 写法）的用法。

今天在这个基础上，我们进一步学习如何用条件语句，让程序根据不同的数据做出不同的判断和响应。」

💡 核心知识讲解

程序也需要"判断力"

你在工厂里做质检，看到产品尺寸偏大就打回返工，尺寸正常就放行，尺寸偏小就报废。

程序也一样——它需要根据数据的不同，走不同的处理路径。

C# 里实现这个能力，靠的就是条件语句。

最基础的三种：if、else if、switch。

if 语句：最基本的"是/否"判断

if 就是一个门卫：条件满足，放行；不满足，拦住。

csharp
if (deviceTemp > 80)
{
    // 温度超过80°C，触发报警
    TriggerAlarm();
}

只有 deviceTemp（设备温度）超过 80 时，TriggerAlarm()（触发报警）才会执行。

否则，这段代码直接跳过，什么都不做。

else if：多条件分支，像档位一样

实际工厂场景里，"是/否"往往不够用。

温度有正常区间、预警区间、危险区间，每个区间要做不同的事。

这时候就需要 else if 来扩展判断分支：

csharp
if (deviceTemp < 60)
{
    // 温度偏低，设备预热中
    ShowStatus("预热中");
}
else if (deviceTemp >= 60 && deviceTemp <= 80)
{
    // 温度正常，允许生产
    ShowStatus("正常生产");
}
else if (deviceTemp > 80 && deviceTemp <= 100)
{
    // 温度偏高，触发预警
    TriggerWarning();
}
else
{
    // 温度超过100°C，紧急停机
    EmergencyStop();
}

else（否则）是兜底分支，前面所有条件都不满足时才执行。

「记住：if 到 else if 的判断是从上往下依次检查，一旦某个条件命中，后面的分支就不再判断了。」

if vs else if：一个对比帮你记清楚

写法	执行特点	适用场景
多个独立 `if`	每个 `if` 都会被判断	条件互不影响时
`if` + `else if`	命中一个就停止	条件互斥、分级判断
`if` + `else`	非此即彼	只有两种结果时

温控报警这种分级场景，一定要用 if + else if，不要写多个独立 if——否则温度 105°C 时，报警和停机可能会同时触发。

阅读全文

从Excel到工业数据库：Python数据迁移实战

编辑

2026-05-13

Python

🏭 那些年，我们被Excel坑过的岁月

说真的，我在工厂车间里做数据系统的时候，见过太多这种场景了——

一台老旧的工控机，桌面上摆着十几个Excel文件，文件名叫"设备数据_最终版_v3_真的最终版.xlsx"。每次要查历史数据，就得打开七八个表格，手动复制粘贴，搞个把小时才能出一份报表。更要命的是，有时候数据还对不上，因为两个班次的操作员各自维护了一份，格式还不一样。

这不是个例。制造业里，Excel作为"数据库"使用的现象极其普遍。它轻便、直观，入门门槛低，但一旦数据量上去了、多人协作了、需要实时查询了——它的局限性就暴露得一干二净。

今天咱们就聊一件很多工程师都绕不开的事：怎么用Python把Excel里的历史数据，优雅地迁移到SQLite或MySQL里，同时还得保证数据不丢、格式不乱、迁移过程可追溯。

🔍 先把问题摸透，再动手写代码

在我接触过的工业数据迁移项目里，有三类问题反复出现，踩坑率极高。

第一类：数据格式的混乱程度超出想象。 同一列"温度"字段，有的行写的是85.3，有的写85.3℃，有的写约85度，甚至还有--表示传感器离线。这种"人工智能"录入方式，直接导致数值列无法直接入库。

第二类：时间戳格式五花八门。 2023/8/5、2023-08-05、8月5日 14:30……同一个Excel文件里可能混用三种格式，pandas读进来直接变成object类型，后续时序查询全部废掉。

第三类：多Sheet、多文件的数据孤岛。 按月份拆分的Excel，每个文件有12个Sheet，字段名还不完全一致（有的叫"压力值"，有的叫"压力"，有的叫"P_value"）。合并之前必须做字段映射，否则入库之后数据根本没法用。

搞清楚这三类问题，咱们的迁移方案就有了清晰的骨架。

🛠️ 环境准备，先把工具备齐

bash
pip install pandas openpyxl sqlalchemy pymysql tqdm

这几个包各有分工：pandas 负责读取和清洗Excel，openpyxl 是pandas读取.xlsx的底层引擎，sqlalchemy 提供统一的数据库抽象层（SQLite和MySQL都能用），pymysql 是MySQL的Python驱动，tqdm 用来显示迁移进度条——数据量大的时候，没有进度条真的会让人抓狂。

阅读全文

C# WinForms + OPC DA 网关适配器：从零搭建工业数据服务

编辑

2026-05-12

🏭 这个需求，比你想的要常见得多

做工控项目的同学，大概率遇到过这种场景——产线上跑着十几年的老设备，底层走 OPC DA 协议，新来的需求要求接云平台、做数据看板，甚至要对外提供 REST API。

一看现有代码：COM Interop、裸 object 类型、lock 满天飞，根本没有现代化接口可言。

重写？停产风险太大。凑合用？技术债越堆越高。

咱们这次换个思路——不动设备、不停产线，用 .NET 8 + WinForms 在上面套一层网关适配器，把 OPC DA 的 COM 调用包成 REST API 对外暴露，同时给运维人员一个可视化的桌面操作界面。

读完本文，你将拿到一套完整可运行的工程结构，包含：

OpcClientSdk 托管库接入（告别 COM Interop）
单点/批量读写 + 实时订阅三种数据获取模式
WinForms 可视化界面（完整 Designer 代码）
内嵌 WebApplication REST API 网关（SSE 实时推送）
JSON 配置持久化 + DI 容器完整接入

🔍 老问题，新视角：OPC DA 对接到底难在哪？

很多人第一次接触这块，踩的第一个坑不是协议，是线程模型。

OPC DA 基于 COM/DCOM，天生是单线程公寓（STA）模型。你在 .NET 的 Task 线程池里直接调它，轻则读出脏数据，重则进程直接崩。这就是为什么所有 COM 调用都必须收拢在 lock 里——不是强迫症，是救命符。

除此之外，还有几个藏得比较深的问题。

类型系统的鸿沟。 OPC DA 的数据全是 VARIANT，映射到 C# 就是 object。质量码（Quality）是个 short，但语义是位域——0xC0 才是 Good，很多人直接 quality > 0 判断，结果把 Uncertain 状态当 Good 用了，数据悄悄出错。

性能天花板。 逐点同步读取，500 个点位以上延迟就开始飙升。实测数据：1000 个点位逐一读取耗时约 8200ms，批量 SyncRead 压到 580ms，差了整整 14 倍。这个坑不踩不知道，踩了就忘不了。

可测试性为零。 COM Interop 的类没法 Mock，单元测试形同虚设。这也是为什么咱们这次引入 ITagReader 接口——不是为了炫技，是为了让代码将来能测、能换、能活。

传统方案用 OPCAutomation COM Interop，麻烦且依赖本机注册表。本文选用 Technosoftware 的 OpcClientSdk，纯托管库，NuGet 直装，.NET 8 原生支持，不需要 COM 注册，部署时少了一大堆环境依赖。

🏗️ 整体架构设计

先看清楚这套东西长什么样，再动手写代码。

WinForms 和 WebApplication 共享同一个 OpcDaGateway 单例，通过 DI 容器注入，两侧都能读写数据，互不干扰。这个设计的好处是：运维人员可以在桌面界面操作，业务系统可以通过 HTTP 接口调用，底层 OPC 连接只维护一条。

👨‍💻先看效果

🔧 核心实现解析

📌 接口设计：为迁移留后路

csharp
using AppOpcDaGateway.Models;

namespace AppOpcDaGateway.Services;

public interface ITagReader : IDisposable
{
    bool IsConnected { get; }
    Task<TagValue> ReadTagAsync(string tagName, CancellationToken ct = default);
    Task<List<TagValue>> ReadTagsAsync(string[] tagNames, CancellationToken ct = default);
    Task<TagValue> WriteTagAsync(string tagName, object? value, CancellationToken ct = default);
    Task<List<TagValue>> WriteTagsAsync(Dictionary<string, object?> tagValues, CancellationToken ct = default);
    event EventHandler<TagValue> TagDataChanged;
    event EventHandler<string> ConnectionStatusChanged;
}

阅读全文

Peewee：被严重低估的 Python 轻量级 ORM

编辑

2026-05-12

Python

🤔 你真的需要 SQLAlchemy 吗？

在不少 Python 项目里，咱们下意识就会拉上 SQLAlchemy——毕竟名气大、生态全。但你有没有算过，一个只需要管理十几张表的内部工具或数据采集服务，光是 SQLAlchemy 的初始化配置就要写多少行？Session 管理、Engine 绑定、Base 声明……还没开始写业务逻辑，头已经大了。

我在做一个 Windows 上位机数据记录模块时，最初也是习惯性地用 SQLAlchemy，结果光是数据库连接层就折腾了半天。后来换成 Peewee，整个模型定义加连接管理压缩到不足 30 行，查询逻辑清晰得像在读英语句子。

Peewee 是一个极简的 Python ORM，代码库不超过 6000 行，支持 SQLite、MySQL、PostgreSQL，在轻量级应用、嵌入式数据库场景、快速原型开发中有着无可替代的优势。读完本文，你将掌握：

Peewee 的核心模型设计与字段映射
关联查询与批量操作的正确姿势
在 Windows 上位机或中间件项目中落地的实战模板

🔍 问题深度剖析：ORM 选型的隐性成本

很多开发者在选 ORM 时只看"功能是否齐全"，却忽略了另一个维度——认知负担与维护成本。

以一个典型的设备数据采集服务为例，需求很简单：每隔 5 秒把传感器数值写入 SQLite，偶尔按时间范围查询。这种场景下，SQLAlchemy 的 Session 生命周期管理、连接池配置、事务上下文……每一个知识点都是额外的学习成本，而且在多线程环境下稍有不慎就会出现 DetachedInstanceError 或连接泄漏。

问题根源在于工具与场景的错配。 重型 ORM 为复杂的企业级应用设计，内置了大量在小型项目中根本用不到的抽象层。这些抽象层不仅增加了初始化开销，还让代码变得难以追踪——一个简单的 INSERT 操作背后可能经历了三四层封装。

在实测中（测试环境：Windows 11，Python 3.11，SQLite 本地文件，10 万条记录批量写入），Peewee 的 bulk_create 耗时约 1.2 秒，而等价的 SQLAlchemy Core 写法耗时约 1.8 秒，ORM 层写法则接近 3.5 秒。差距在数据量增大后会进一步拉开。

💡 核心要点提炼：Peewee 的设计哲学

Peewee 的底层逻辑非常直接：模型即表，字段即列，查询即链式调用。它没有 SQLAlchemy 那种"工作单元"模式，也没有复杂的 identity map，每次查询就是一次干净的数据库交互。

🏗️ 模型定义：直觉优先

python
from peewee import *

# 连接 SQLite 数据库（Windows 路径兼容）
db = SqliteDatabase('sensor_data.db')

class BaseModel(Model):
    class Meta:
        database = db

class Device(BaseModel):
    """设备信息表"""
    name = CharField(max_length=64, unique=True)
    location = CharField(max_length=128, null=True)
    created_at = DateTimeField(constraints=[SQL('DEFAULT CURRENT_TIMESTAMP')])

    class Meta:
        table_name = 'devices'

class SensorRecord(BaseModel):
    """传感器记录表"""
    device = ForeignKeyField(Device, backref='records', on_delete='CASCADE')
    temperature = FloatField()
    humidity = FloatField(null=True)
    recorded_at = DateTimeField(index=True)

    class Meta:
        table_name = 'sensor_records'

模型定义清晰到不需要注释就能看懂结构。backref='records' 这一个参数，就完成了反向关联的声明——后续可以直接用 device.records 遍历该设备的所有记录。

🔗 连接管理：上下文即安全

Peewee 推荐使用上下文管理器处理连接，这在 Windows 上位机的多线程环境中尤为重要：

python
# 建表（仅首次运行或迁移时执行）
with db:
    db.create_tables([Device, SensorRecord], safe=True)

# 日常操作统一用 atomic() 事务上下文
def save_record(device_name: str, temp: float, humidity: float, ts):
    with db.atomic():
        device, _ = Device.get_or_create(name=device_name)
        SensorRecord.create(
            device=device,
            temperature=temp,
            humidity=humidity,
            recorded_at=ts
        )

db.atomic() 既是事务边界，也是异常回滚的保障。如果块内抛出异常，事务自动回滚，不会留下脏数据。

🚀 解决方案设计

方案一：基础 CRUD 与链式查询

适用场景： 单表操作、条件筛选、排序分页，覆盖 80% 的日常需求。

python
import os  
  
from peewee import *  
from datetime import datetime, timedelta  
import logging  
import random  
  
# 配置日志  
logging.basicConfig(level=logging.INFO)  
logger = logging.getLogger(__name__)  
  
# 连接 SQLite 数据库（Windows 路径兼容）  
db = SqliteDatabase('sensor_data.db')  
  
  
class BaseModel(Model):  
    class Meta:  
        database = db  
  
  
class Device(BaseModel):  
    """设备信息表"""  
    name = CharField(max_length=64, unique=True)  
    location = CharField(max_length=128, null=True)  
    status = CharField(max_length=20, default='active')  # active, inactive, maintenance  
    created_at = DateTimeField(constraints=[SQL('DEFAULT CURRENT_TIMESTAMP')])  
  
    class Meta:  
        table_name = 'devices'  
  
    def __str__(self):  
        return f"Device({self.name}, {self.location})"  
  
  
class SensorRecord(BaseModel):  
    """传感器记录表"""  
    device = ForeignKeyField(Device, backref='records', on_delete='CASCADE')  
    temperature = FloatField()  
    humidity = FloatField(null=True)  
    pressure = FloatField(null=True)  # 新增气压字段  
    recorded_at = DateTimeField(index=True)  
  
    class Meta:  
        table_name = 'sensor_records'  
        indexes = (  
            # 复合索引，提升查询性能  
            (('device', 'recorded_at'), False),  
        )  
  
    def __str__(self):  
        return f"Record({self.device.name}, {self.temperature}°C, {self.recorded_at})"  
  
  
# --- 数据库初始化 ---def init_database(reset=False):  
    """初始化数据库，创建表结构"""  
    try:  
        # 如果需要重置，删除现有数据库文件  
        if reset and os.path.exists('sensor_data.db'):  
            os.remove('sensor_data.db')  
            logger.info("已删除现有数据库文件")  
  
        db.connect()  
        db.create_tables([Device, SensorRecord], safe=True)  
        logger.info("数据库初始化完成")  
    except Exception as e:  
        logger.error(f"数据库初始化失败: {e}")  
        raise  
    finally:  
        if not db.is_closed():  
            db.close()  
  
  
# --- 设备管理 ---
def create_device(name: str, location: str = None) -> Device:  
    """创建设备"""  
    try:  
        device = Device.create(name=name, location=location)  
        logger.info(f"设备创建成功: {device}")  
        return device  
    except IntegrityError:  
        logger.warning(f"设备 {name} 已存在")  
        return Device.get(Device.name == name)  
  
  
def get_or_create_device(name: str, location: str = None) -> Device:  
    """获取或创建设备"""  
    device, created = Device.get_or_create(  
        name=name,  
        defaults={'location': location}  
    )    if created:  
        logger.info(f"新设备创建: {device}")  
    return device  
  
  
def list_devices(status: str = None):  
    """列出所有设备"""  
    query = Device.select()  
    if status:  
        query = query.where(Device.status == status)  
    return list(query)  
  
  
def update_device_status(device_name: str, status: str):  
    """更新设备状态"""  
    updated = (Device  
               .update(status=status)  
               .where(Device.name == device_name)  
               .execute())  
    if updated:  
        logger.info(f"设备 {device_name} 状态更新为 {status}")  
    return updated > 0  
  
  
# --- 数据写入 ---
def batch_insert(records: list[dict]):  
    """  
    批量写入，推荐使用 bulk_create 替代循环 create    测试环境：Windows 11 / Python 3.11 / SQLite  
    1万条：bulk_create ≈ 0.12s，循环 create ≈ 1.8s    """    if not records:  
        return 0  
  
    try:  
        with db.atomic():  
            # 确保设备存在  
            device_names = {r.get('device_name') or r.get('device') for r in records}  
            for name in device_names:  
                if name:  
                    get_or_create_device(name)  
  
            # 转换记录格式  
            sensor_records = []  
            for r in records:  
                device_name = r.get('device_name') or r.get('device')  
                if isinstance(device_name, str):  
                    device = Device.get(Device.name == device_name)  
                else:  
                    device = device_name  
  
                record = SensorRecord(  
                    device=device,  
                    temperature=r['temperature'],  
                    humidity=r.get('humidity'),  
                    pressure=r.get('pressure'),  
                    recorded_at=r.get('recorded_at', datetime.now())  
                )  
                sensor_records.append(record)  
  
            SensorRecord.bulk_create(sensor_records, batch_size=500)  
            logger.info(f"批量插入 {len(records)} 条记录成功")  
            return len(records)  
    except Exception as e:  
        logger.error(f"批量插入失败: {e}")  
        raise  
  
  
def add_single_record(device_name: str, temperature: float,  
                      humidity: float = None, pressure: float = None,  
                      recorded_at: datetime = None):  
    """添加单条记录"""  
    device = get_or_create_device(device_name)  
    record = SensorRecord.create(  
        device=device,  
        temperature=temperature,  
        humidity=humidity,  
        pressure=pressure,  
        recorded_at=recorded_at or datetime.now()  
    )  
    logger.info(f"记录添加成功: {record}")  
    return record  
  
  
# --- 查询功能 ---
def query_recent(device_name: str, hours: int = 24):  
    """查询指定设备最近 N 小时的记录"""  
    since = datetime.now() - timedelta(hours=hours)  
    return (  
        SensorRecord  
        .select(SensorRecord, Device)  
        .join(Device)  
        .where(  
            Device.name == device_name,  
            SensorRecord.recorded_at >= since  
        )  
        .order_by(SensorRecord.recorded_at.desc())  
        .limit(1000)  
    )  
  
  
def query_by_date_range(device_name: str = None,  
                        start_date: datetime = None,  
                        end_date: datetime = None):  
    """按日期范围查询"""  
    query = SensorRecord.select(SensorRecord, Device).join(Device)  
  
    conditions = []  
    if device_name:  
        conditions.append(Device.name == device_name)  
    if start_date:  
        conditions.append(SensorRecord.recorded_at >= start_date)  
    if end_date:  
        conditions.append(SensorRecord.recorded_at <= end_date)  
  
    if conditions:  
        query = query.where(*conditions)  
  
    return query.order_by(SensorRecord.recorded_at.desc())  
  
  
def query_temperature_range(device_name: str, min_temp: float, max_temp: float):  
    """查询温度范围内的记录"""  
    return (  
        SensorRecord  
        .select(SensorRecord, Device)  
        .join(Device)  
        .where(  
            Device.name == device_name,  
            SensorRecord.temperature.between(min_temp, max_temp)  
        )  
        .order_by(SensorRecord.recorded_at.desc())  
    )  
  
  
# --- 统计分析 ---
def get_stats(device_name: str, days: int = 7):  
    """获取统计信息"""  
    from peewee import fn  
    since = datetime.now() - timedelta(days=days)  
  
    stats = (  
        SensorRecord  
        .select(  
            fn.AVG(SensorRecord.temperature).alias('avg_temp'),  
            fn.MAX(SensorRecord.temperature).alias('max_temp'),  
            fn.MIN(SensorRecord.temperature).alias('min_temp'),  
            fn.AVG(SensorRecord.humidity).alias('avg_humidity'),  
            fn.MAX(SensorRecord.humidity).alias('max_humidity'),  
            fn.MIN(SensorRecord.humidity).alias('min_humidity'),  
            fn.COUNT(SensorRecord.id).alias('total_records')  
        )  
        .join(Device)  
        .where(  
            Device.name == device_name,  
            SensorRecord.recorded_at >= since  
        )  
        .dicts()  
        .first()  
    )  
  
    return stats  
  
  
def get_hourly_stats(device_name: str, date: datetime = None):  
    """获取按小时统计的数据"""  
    from peewee import fn  
    if not date:  
        date = datetime.now().date()  
  
    start_time = datetime.combine(date, datetime.min.time())  
    end_time = start_time + timedelta(days=1)  
  
    return (  
        SensorRecord  
        .select(  
            fn.strftime('%H', SensorRecord.recorded_at).alias('hour'),  
            fn.AVG(SensorRecord.temperature).alias('avg_temp'),  
            fn.COUNT(SensorRecord.id).alias('count')  
        )  
        .join(Device)  
        .where(  
            Device.name == device_name,  
            SensorRecord.recorded_at.between(start_time, end_time)  
        )  
        .group_by(fn.strftime('%H', SensorRecord.recorded_at))  
        .order_by(fn.strftime('%H', SensorRecord.recorded_at))  
        .dicts()  
    )  
  
  
def get_daily_extremes(device_name: str, days: int = 30):  
    """获取每日最高/最低温度"""  
    from peewee import fn  
    since = datetime.now() - timedelta(days=days)  
  
    return (  
        SensorRecord  
        .select(  
            fn.DATE(SensorRecord.recorded_at).alias('date'),  
            fn.MAX(SensorRecord.temperature).alias('max_temp'),  
            fn.MIN(SensorRecord.temperature).alias('min_temp'),  
            fn.AVG(SensorRecord.temperature).alias('avg_temp')  
        )  
        .join(Device)  
        .where(  
            Device.name == device_name,  
            SensorRecord.recorded_at >= since  
        )  
        .group_by(fn.DATE(SensorRecord.recorded_at))  
        .order_by(fn.DATE(SensorRecord.recorded_at).desc())  
        .dicts()  
    )  
  
  
# --- 数据清理 ---
def cleanup_old_records(days: int = 90):  
    """清理旧记录"""  
    cutoff = datetime.now() - timedelta(days=days)  
    deleted = (SensorRecord  
               .delete()  
               .where(SensorRecord.recorded_at < cutoff)  
               .execute())  
    logger.info(f"删除了 {deleted} 条旧记录（{days} 天前）")  
    return deleted  
  
  
def delete_device_records(device_name: str):  
    """删除指定设备的所有记录"""  
    try:  
        with db.atomic():  
            device = Device.get(Device.name == device_name)  
            deleted = (SensorRecord  
                       .delete()  
                       .where(SensorRecord.device == device)  
                       .execute())  
            logger.info(f"删除设备 {device_name} 的 {deleted} 条记录")  
            return deleted  
    except Device.DoesNotExist:  
        logger.warning(f"设备 {device_name} 不存在")  
        return 0  
  
  
# --- 工具函数 ---
def generate_sample_data(device_name: str, days: int = 7, interval_minutes: int = 30):  
    """生成示例数据"""  
    device = get_or_create_device(device_name, f"测试位置_{device_name}")  
  
    records = []  
    start_time = datetime.now() - timedelta(days=days)  
    current_time = start_time  
  
    base_temp = 25.0  
    base_humidity = 60.0  
  
    while current_time < datetime.now():  
        # 模拟温度变化（加入日夜周期）  
        hour = current_time.hour  
        day_factor = abs(hour - 12) / 12  # 中午12点最热  
        temp_variation = random.uniform(-2, 2)  
        temperature = base_temp - day_factor * 5 + temp_variation  
  
        # 模拟湿度变化  
        humidity_variation = random.uniform(-10, 10)  
        humidity = max(20, min(90, base_humidity + humidity_variation))  
  
        # 模拟气压  
        pressure = random.uniform(1000, 1020)  
  
        records.append({  
            'device_name': device_name,  
            'temperature': round(temperature, 1),  
            'humidity': round(humidity, 1),  
            'pressure': round(pressure, 1),  
            'recorded_at': current_time  
        })  
  
        current_time += timedelta(minutes=interval_minutes)  
  
    batch_insert(records)  
    logger.info(f"为设备 {device_name} 生成了 {len(records)} 条示例数据")  
  
  
def export_to_dict(device_name: str, hours: int = 24):  
    """导出数据为字典格式"""  
    records = query_recent(device_name, hours)  
    return [  
        {            'device_name': r.device.name,  
            'temperature': r.temperature,  
            'humidity': r.humidity,  
            'pressure': r.pressure,  
            'recorded_at': r.recorded_at.isoformat()  
        }        for r in records  
    ]  
  
  
# --- 使用示例 ---
def demo():  
    """演示功能"""  
    print("=== 传感器数据管理系统演示 ===\n")  
  
    # 初始化数据库  
    init_database(reset=True)  
  
    # 创建测试设备  
    devices = ['温室A', '温室B', '仓库1']  
    for device_name in devices:  
        create_device(device_name, f"{device_name}位置")  
  
    # 生成示例数据  
    for device_name in devices:  
        generate_sample_data(device_name, days=3, interval_minutes=60)  
  
    # 查询演示  
    print("1. 最近24小时数据：")  
    recent_records = query_recent('温室A', 24)  
    for r in recent_records[:5]:  # 只显示前5条  
        print(f"  {r.recorded_at}: {r.temperature}°C, {r.humidity}%")  
  
    print(f"\n2. 温室A 统计信息：")  
    stats = get_stats('温室A')  
    if stats:  
        print(f"  平均温度: {stats['avg_temp']:.1f}°C")  
        print(f"  最高温度: {stats['max_temp']:.1f}°C")  
        print(f"  最低温度: {stats['min_temp']:.1f}°C")  
        print(f"  记录总数: {stats['total_records']}")  
  
    print(f"\n3. 温室A 每日极值：")  
    daily_extremes = get_daily_extremes('温室A', 7)  
    for day in daily_extremes:  
        print(f"  {day['date']}: 最高{day['max_temp']:.1f}°C, 最低{day['min_temp']:.1f}°C")  
  
    print(f"\n4. 所有设备列表：")  
    for device in list_devices():  
        record_count = device.records.count()  
        print(f"  {device.name} ({device.location}) - {record_count} 条记录")  
  
  
if __name__ == "__main__":  
    demo()

踩坑预警： 查询时如果不用 .join() 而是在循环里访问 record.device.name，会触发经典的 N+1 查询问题——100 条记录就是 101 次数据库请求。务必在 select 时把关联表一起取出来。

阅读全文