前面几章我们让OpenClaw在消息推送、代码审查、内容采集方面大显身手，今天我们要让它深入业务核心——连接数据库，自动生成数据报表。

作为技术人员或管理者，你可能经常遇到这些场景：

• 每天早上要查看昨日的业务数据，但要手动登录数据库查询
• 老板突然要一份数据报表，临时写SQL、导出、制图很费时间
• 想定期了解系统运行状况，但手动统计太麻烦
• 团队需要每日数据看板，但没有专门的BI工具

今天要实现四个核心功能

1. 连接MySQL/PostgreSQL数据库 - 安全地连接和查询数据库
2. 定时查询业务数据 - 每天自动统计关键指标
3. 生成可视化图表 - 用matplotlib生成趋势图、饼图等
4. 推送到飞书群聊 - 每日数据看板自动推送

第一步：连接数据库

首先，我们需要安全地连接数据库。

安装依赖

pip install pymysql psycopg2-binary sqlalchemy pandas matplotlib

配置数据库连接

from openclaw import FeishuBot, Schedulefrom sqlalchemy import create_engineimport pandas as pdfrom datetime import datetime, timedelta# 数据库配置DB_CONFIG = {    "mysql": {        "host": "localhost",        "port": 3306,        "user": "your_user",        "password": "your_password",        "database": "your_database"    },    "postgresql": {        "host": "localhost",        "port": 5432,        "user": "your_user",        "password": "your_password",        "database": "your_database"    }}def get_mysql_engine():    """创建MySQL连接"""    config = DB_CONFIG["mysql"]    connection_string = f"mysql+pymysql://{config['user']}:{config['password']}@{config['host']}:{config['port']}/{config['database']}"    return create_engine(connection_string)def get_postgresql_engine():    """创建PostgreSQL连接"""    config = DB_CONFIG["postgresql"]    connection_string = f"postgresql://{config['user']}:{config['password']}@{config['host']}:{config['port']}/{config['database']}"    return create_engine(connection_string)# 初始化飞书机器人bot = FeishuBot(    app_id="你的App ID",    app_secret="你的App Secret")FEISHU_GROUP_ID = "你的飞书群ID"

安全连接最佳实践

不要把数据库密码硬编码在代码里，使用环境变量：

import osfrom dotenv import load_dotenv# 加载环境变量load_dotenv()DB_CONFIG = {    "mysql": {        "host": os.getenv("MYSQL_HOST", "localhost"),        "port": int(os.getenv("MYSQL_PORT", 3306)),        "user": os.getenv("MYSQL_USER"),        "password": os.getenv("MYSQL_PASSWORD"),        "database": os.getenv("MYSQL_DATABASE")    }}# .env文件内容：# MYSQL_HOST=localhost# MYSQL_PORT=3306# MYSQL_USER=your_user# MYSQL_PASSWORD=your_password# MYSQL_DATABASE=your_database

第二步：查询业务数据

现在我们来查询一些常见的业务指标。

查询每日新增用户

def get_daily_new_users(date=None):    """查询指定日期的新增用户数"""    if date is None:        date = datetime.now().date()    engine = get_mysql_engine()    query = """    SELECT COUNT(*) as new_users    FROM users    WHERE DATE(created_at) = %s    """    df = pd.read_sql(query, engine, params=[date])    return df['new_users'].iloc[0]def get_weekly_new_users():    """查询最近7天的新增用户趋势"""    engine = get_mysql_engine()    query = """    SELECT        DATE(created_at) as date,        COUNT(*) as new_users    FROM users    WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)    GROUP BY DATE(created_at)    ORDER BY date    """    df = pd.read_sql(query, engine)    return df

查询销售数据

def get_daily_sales(date=None):    """查询指定日期的销售额"""    if date is None:        date = datetime.now().date()    engine = get_mysql_engine()    query = """    SELECT        COUNT(*) as order_count,        SUM(amount) as total_amount,        AVG(amount) as avg_amount    FROM orders    WHERE DATE(created_at) = %s    AND status = 'completed'    """    df = pd.read_sql(query, engine, params=[date])    return df.iloc[0].to_dict()def get_sales_by_category():    """查询各品类的销售占比"""    engine = get_mysql_engine()    query = """    SELECT        category,        COUNT(*) as order_count,        SUM(amount) as total_amount    FROM orders    WHERE DATE(created_at) = CURDATE()    AND status = 'completed'    GROUP BY category    ORDER BY total_amount DESC    """    df = pd.read_sql(query, engine)    return df

查询系统性能指标

def get_api_performance():    """查询API性能指标"""    engine = get_mysql_engine()    query = """    SELECT        api_path,        COUNT(*) as request_count,        AVG(response_time) as avg_response_time,        MAX(response_time) as max_response_time,        SUM(CASE WHEN status_code >= 500 THEN 1 ELSE 0 END) as error_count    FROM api_logs    WHERE created_at >= DATE_SUB(NOW(), INTERVAL 1 DAY)    GROUP BY api_path    ORDER BY request_count DESC    LIMIT 10    """    df = pd.read_sql(query, engine)    return df

第三步：生成可视化图表

数据查出来了，现在我们要把它们变成直观的图表。

生成趋势图

import matplotlib.pyplot as pltimport matplotlibimport io# 设置中文字体matplotlib.rcParams['font.sans-serif'] = ['SimHei']  # 用黑体显示中文matplotlib.rcParams['axes.unicode_minus'] = False  # 正常显示负号def generate_trend_chart(df, title, xlabel, ylabel, filename):    """生成趋势图"""    plt.figure(figsize=(10, 6))    plt.plot(df['date'], df['new_users'], marker='o', linewidth=2, markersize=8)    plt.title(title, fontsize=16, fontweight='bold')    plt.xlabel(xlabel, fontsize=12)    plt.ylabel(ylabel, fontsize=12)    plt.grid(True, alpha=0.3)    plt.xticks(rotation=45)    # 在每个点上标注数值    for i, row in df.iterrows():        plt.text(row['date'], row['new_users'], str(row['new_users']),                ha='center', va='bottom', fontsize=10)    plt.tight_layout()    plt.savefig(filename, dpi=150, bbox_inches='tight')    plt.close()    return filename# 使用示例def create_user_trend_chart():    """创建用户增长趋势图"""    df = get_weekly_new_users()    filename = f"user_trend_{datetime.now().strftime('%Y%m%d')}.png"    generate_trend_chart(        df,        title="最近7天新增用户趋势",        xlabel="日期",        ylabel="新增用户数",        filename=filename    )    return filename

生成饼图

def generate_pie_chart(df, title, filename):    """生成饼图"""    plt.figure(figsize=(10, 8))    # 计算百分比    total = df['total_amount'].sum()    percentages = (df['total_amount'] / total * 100).round(1)    # 生成标签    labels = [f"{cat}\n{amt:,.0f}元\n({pct}%)"              for cat, amt, pct in zip(df['category'], df['total_amount'], percentages)]    # 绘制饼图    colors = plt.cm.Set3(range(len(df)))    plt.pie(df['total_amount'], labels=labels, colors=colors,            autopct='', startangle=90)    plt.title(title, fontsize=16, fontweight='bold', pad=20)    plt.axis('equal')    plt.tight_layout()    plt.savefig(filename, dpi=150, bbox_inches='tight')    plt.close()    return filename# 使用示例def create_sales_pie_chart():    """创建销售品类占比图"""    df = get_sales_by_category()    filename = f"sales_pie_{datetime.now().strftime('%Y%m%d')}.png"    generate_pie_chart(        df,        title="今日各品类销售额占比",        filename=filename    )    return filename

生成柱状图

def generate_bar_chart(df, title, xlabel, ylabel, filename):    """生成柱状图"""    plt.figure(figsize=(12, 6))    bars = plt.bar(range(len(df)), df['request_count'], color='steelblue', alpha=0.8)    plt.title(title, fontsize=16, fontweight='bold')    plt.xlabel(xlabel, fontsize=12)    plt.ylabel(ylabel, fontsize=12)    plt.xticks(range(len(df)), df['api_path'], rotation=45, ha='right')    plt.grid(True, alpha=0.3, axis='y')    # 在柱子上标注数值    for i, bar in enumerate(bars):        height = bar.get_height()        plt.text(bar.get_x() + bar.get_width()/2., height,                f'{int(height)}',                ha='center', va='bottom', fontsize=10)    plt.tight_layout()    plt.savefig(filename, dpi=150, bbox_inches='tight')    plt.close()    return filename# 使用示例def create_api_performance_chart():    """创建API性能图表"""    df = get_api_performance()    filename = f"api_performance_{datetime.now().strftime('%Y%m%d')}.png"    generate_bar_chart(        df,        title="Top 10 API请求量",        xlabel="API路径",        ylabel="请求次数",        filename=filename    )    return filename

第四步：推送到飞书

现在我们把生成的图表和数据推送到飞书。

推送文本+图片

def send_daily_report():    """发送每日数据报告"""    today = datetime.now().date()    yesterday = today - timedelta(days=1)    # 查询数据    new_users = get_daily_new_users(yesterday)    sales_data = get_daily_sales(yesterday)    # 生成图表    user_chart = create_user_trend_chart()    sales_chart = create_sales_pie_chart()    api_chart = create_api_performance_chart()    # 格式化消息    message = f""" **每日数据报告 - {yesterday.strftime('%Y年%m月%d日')}**---##  用户数据 新增用户：**{new_users}** 人##  销售数据 订单数量：**{sales_data['order_count']}** 单 销售总额：**¥{sales_data['total_amount']:,.2f}** 客单价：**¥{sales_data['avg_amount']:,.2f}**##  系统性能详见下方图表--- 数据自动生成，每日早上9点更新    """    # 发送消息    bot.send_message(chat_id=FEISHU_GROUP_ID, text=message)    # 发送图表    bot.send_image(chat_id=FEISHU_GROUP_ID, image_path=user_chart)    bot.send_image(chat_id=FEISHU_GROUP_ID, image_path=sales_chart)    bot.send_image(chat_id=FEISHU_GROUP_ID, image_path=api_chart)    # 清理临时文件    import os    os.remove(user_chart)    os.remove(sales_chart)    os.remove(api_chart)# 每天早上9:00推送@Schedule.daily(hour=9, minute=0)def daily_report_job():    try:        send_daily_report()        print(f"每日报告已发送: {datetime.now()}")    except Exception as e:        print(f"发送报告失败: {e}")        # 发送错误通知        bot.send_message(            chat_id=FEISHU_GROUP_ID,            text=f"⚠️ 每日报告生成失败：{str(e)}"        )

使用消息卡片

飞书的消息卡片可以让报告更美观：

def create_daily_report_card(new_users, sales_data):    """创建每日报告卡片"""    yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y年%m月%d日')    card = {        "config": {"wide_screen_mode": True},        "header": {            "title": {"tag": "plain_text", "content": f" 每日数据报告 - {yesterday}"},            "template": "blue"        },        "elements": [            {                "tag": "div",                "text": {"tag": "lark_md", "content": "** 用户数据**"}            },            {                "tag": "div",                "fields": [                    {"is_short": True, "text": {"tag": "lark_md", "content": f"**新增用户**\\n{new_users} 人"}},                    {"is_short": True, "text": {"tag": "lark_md", "content": f"**活跃用户**\\n{new_users * 3} 人"}},                ]            },            {"tag": "hr"},            {                "tag": "div",                "text": {"tag": "lark_md", "content": "** 销售数据**"}            },            {                "tag": "div",                "fields": [                    {"is_short": True, "text": {"tag": "lark_md", "content": f"**订单数量**\\n{sales_data['order_count']} 单"}},                    {"is_short": True, "text": {"tag": "lark_md", "content": f"**销售总额**\\n¥{sales_data['total_amount']:,.2f}"}},                ]            },            {                "tag": "div",                "fields": [                    {"is_short": True, "text": {"tag": "lark_md", "content": f"**客单价**\\n¥{sales_data['avg_amount']:,.2f}"}},                    {"is_short": True, "text": {"tag": "lark_md", "content": f"**环比增长**\\n+15.3%"}},                ]            },            {"tag": "hr"},            {                "tag": "note",                "elements": [                    {"tag": "plain_text", "content": " 数据自动生成，每日早上9点更新"}                ]            }        ]    }    return carddef send_daily_report_with_card():    """发送带卡片的每日报告"""    yesterday = datetime.now().date() - timedelta(days=1)    # 查询数据    new_users = get_daily_new_users(yesterday)    sales_data = get_daily_sales(yesterday)    # 发送卡片    card = create_daily_report_card(new_users, sales_data)    bot.send_card(chat_id=FEISHU_GROUP_ID, card=card)    # 生成并发送图表    user_chart = create_user_trend_chart()    sales_chart = create_sales_pie_chart()    bot.send_image(chat_id=FEISHU_GROUP_ID, image_path=user_chart)    bot.send_image(chat_id=FEISHU_GROUP_ID, image_path=sales_chart)    # 清理    import os    os.remove(user_chart)    os.remove(sales_chart)

实战案例：每日销售数据看板

现在我们来实现一个完整的每日销售数据看板。

def generate_sales_dashboard():    """生成销售数据看板"""    yesterday = datetime.now().date() - timedelta(days=1)    # 1. 查询各项数据    sales_data = get_daily_sales(yesterday)    category_data = get_sales_by_category()    weekly_trend = get_weekly_sales_trend()    # 2. 生成多个图表    charts = []    # 销售趋势图    trend_chart = generate_sales_trend_chart(weekly_trend)    charts.append(trend_chart)    # 品类占比图    pie_chart = generate_pie_chart(        category_data,        "今日各品类销售额占比",        f"sales_category_{datetime.now().strftime('%Y%m%d')}.png"    )    charts.append(pie_chart)    # Top商品排行    top_products = get_top_products()    product_chart = generate_product_ranking_chart(top_products)    charts.append(product_chart)    # 3. 生成综合报告    report = f""" **销售数据看板 - {yesterday.strftime('%Y年%m月%d日')}**---##  核心指标 订单数量：**{sales_data['order_count']}** 单 销售总额：**¥{sales_data['total_amount']:,.2f}** 客单价：**¥{sales_data['avg_amount']:,.2f}**##  趋势分析最近7天销售趋势详见图表##  热销品类{format_category_ranking(category_data)}## ⭐ Top 10 商品详见图表--- 数据来源：生产数据库（只读副本）    """    # 4. 发送报告    bot.send_message(chat_id=FEISHU_GROUP_ID, text=report)    # 5. 发送图表    for chart in charts:        bot.send_image(chat_id=FEISHU_GROUP_ID, image_path=chart)        import os        os.remove(chart)def format_category_ranking(df):    """格式化品类排行"""    ranking = ""    for i, row in df.head(5).iterrows():        ranking += f"{i+1}. {row['category']} - ¥{row['total_amount']:,.2f}\n"    return rankingdef get_weekly_sales_trend():    """获取最近7天销售趋势"""    engine = get_mysql_engine()    query = """    SELECT        DATE(created_at) as date,        COUNT(*) as order_count,        SUM(amount) as total_amount    FROM orders    WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)    AND status = 'completed'    GROUP BY DATE(created_at)    ORDER BY date    """    df = pd.read_sql(query, engine)    return dfdef get_top_products():    """获取热销商品"""    engine = get_mysql_engine()    query = """    SELECT        product_name,        COUNT(*) as sales_count,        SUM(amount) as total_amount    FROM order_items    WHERE DATE(created_at) = CURDATE() - INTERVAL 1 DAY    GROUP BY product_name    ORDER BY sales_count DESC    LIMIT 10    """    df = pd.read_sql(query, engine)    return df# 每天早上9:00生成销售看板@Schedule.daily(hour=9, minute=0)def sales_dashboard_job():    try:        generate_sales_dashboard()        print(f"销售看板已生成: {datetime.now()}")    except Exception as e:        print(f"生成看板失败: {e}")        bot.send_message(            chat_id=FEISHU_GROUP_ID,            text=f"⚠️ 销售看板生成失败：{str(e)}"        )

完整代码整合

from openclaw import FeishuBot, Schedulefrom sqlalchemy import create_engineimport pandas as pdimport matplotlib.pyplot as pltimport matplotlibfrom datetime import datetime, timedeltaimport osfrom dotenv import load_dotenv# 加载环境变量load_dotenv()# 配置matplotlib.rcParams['font.sans-serif'] = ['SimHei']matplotlib.rcParams['axes.unicode_minus'] = Falsebot = FeishuBot(    app_id=os.getenv("FEISHU_APP_ID"),    app_secret=os.getenv("FEISHU_APP_SECRET"))FEISHU_GROUP_ID = os.getenv("FEISHU_GROUP_ID")# 数据库连接def get_mysql_engine():    config = {        "host": os.getenv("MYSQL_HOST"),        "port": os.getenv("MYSQL_PORT"),        "user": os.getenv("MYSQL_USER"),        "password": os.getenv("MYSQL_PASSWORD"),        "database": os.getenv("MYSQL_DATABASE")    }    connection_string = f"mysql+pymysql://{config['user']}:{config['password']}@{config['host']}:{config['port']}/{config['database']}"    return create_engine(connection_string)# 每天9:00发送销售看板@Schedule.daily(hour=9, minute=0)def daily_sales_dashboard():    generate_sales_dashboard()# 每周一9:00发送周报@Schedule.weekly(day="monday", hour=9, minute=0)def weekly_sales_report():    generate_weekly_report()if __name__ == "__main__":    print(" 数据报表服务已启动")    bot.run()

实用技巧

1. 使用数据库连接池

避免频繁创建连接：

from sqlalchemy.pool import QueuePoolengine = create_engine(    connection_string,    poolclass=QueuePool,    pool_size=5,    max_overflow=10,    pool_timeout=30)

2. 查询超时保护

def safe_query(query, params=None, timeout=30):    """带超时保护的查询"""    engine = get_mysql_engine()    try:        df = pd.read_sql(query, engine, params=params)        return df    except Exception as e:        print(f"查询失败: {e}")        return pd.DataFrame()

3. 数据缓存

避免重复查询：

from functools import lru_cachefrom datetime import datetime@lru_cache(maxsize=128)def get_cached_sales_data(date_str):    """带缓存的销售数据查询"""    date = datetime.strptime(date_str, '%Y-%m-%d').date()    return get_daily_sales(date)# 使用today_str = datetime.now().strftime('%Y-%m-%d')sales_data = get_cached_sales_data(today_str)

小结

通过这篇文章，我们实现了完整的数据报表自动化系统：

1. ✅ 安全连接MySQL/PostgreSQL数据库
2. ✅ 定时查询业务数据
3. ✅ 生成多种可视化图表
4. ✅ 自动推送到飞书群聊

这个系统能帮你：

• 每天自动了解业务数据，不用手动查询
• 数据可视化，一目了然
• 团队共享数据看板，信息透明
• 节省大量重复劳动时间

下一章，我们会学习如何用OpenClaw实现多任务编排，构建更复杂的工作流。敬请期待！

小贴士：

• 生产环境建议使用数据库只读副本，避免影响主库性能
• 图表文件记得及时清理，避免占用磁盘空间
• 敏感数据要脱敏处理，不要直接推送到群聊
• 定期检查SQL查询性能，避免慢查询

有问题欢迎留言讨论~