分析与自动化

从 cqdataframe 导入 数据表, 读取CSV, 范围数组

# 创建数据表
表 = 数据表({"姓名": ["小明", "小红", "小刚"], "年龄": [18, 20, 25]})
print(表.形状())          # (3, 2)
print(表.列们())          # ['姓名', '年龄']

# 查看与统计
print(表.查看头部(2))
print(表.平均值())        # 仅数值列

# 过滤 / 排序
筛选 = 表.过滤(表.列("年龄") > 19)    # 年龄大于 19 的行
按年龄 = 表.排序("年龄", 升序=False)

# 读写文件
表.保存CSV("数据.csv")
表2 = 读取CSV("数据.csv")

# numpy 便捷函数
arr = 范围数组(0, 10)     # array([0..9])

从 cqemail 导入 创建邮件, 发送邮件, 接收邮件, 邮件服务器

# 发送
邮件对象 = 创建邮件(
    主题="会议通知",
    正文="下午 3 点开会。",
    发件人="me@example.com",
    收件人=["you@example.com"],
)
邮件对象.添加附件("会议纪要.docx")
发送邮件("smtp.example.com", 465, "me@example.com", "邮箱密码", 邮件对象)

# 使用服务器对象（可复用连接）
with 邮件服务器("smtp.example.com", 465, "me@example.com", "密码") as 服务器:
    服务器.发送(邮件对象)

# 接收
消息们 = 接收邮件("imap.example.com", 993, 用户名="me@example.com", 密码="密码", 数量=10)
for 消息 in 消息们:
    print(消息.主题, 消息.发件人)
    print(消息.正文文本)
    if 消息.附件们:
        消息.保存附件("下载附件.bin")

从 cqschedule 导入 调度器, 每, 在, 启动调度器

调度 = 调度器()

# 间隔任务
调度.间隔任务(心跳, 秒=30)                    # 每 30 秒
调度.间隔任务(备份, 小时=1)                   # 每小时

# cron 定时
调度.定时任务(打卡, 小时=9, 分钟=30)           # 每天 09:30
调度.定时任务(日报, 星期="mon-fri", 小时=18)   # 工作日 18:00

# 一次性任务
调度.日期任务(到期提醒, 时间(2026, 12, 31, 23, 59))
调度.启动()                                   # 后台运行，不阻塞主线程

# 链式装饰器写法（全局调度器）
@每(5).秒
def 心跳():
    print("心跳")

@在("14:00").每天
def 打卡():
    print("打卡")

启动调度器()                                  # 前台运行（脚本末尾阻塞）

从 cqtest 导入 测试, 参数化, 断言等于, 期望异常, 跳过, 运行

@测试
def 测试加法():
    断言等于(1 + 1, 2)

@参数化("数值, 期望", [(1, 1), (2, 4), (3, 9)])
def 测试平方(数值, 期望):
    断言等于(数值 ** 2, 期望)

@测试
def 测试异常():
    with 期望异常(ValueError):
        int("abc")

@跳过("待实现")
def 测试待办():
    跳过

运行()        # 运行当前模块中所有 @测试，打印通过/失败统计