用 DuckDB 搭建自动日报系统，每月多赚 5000 元

import duckdb
from datetime import datetime, timedelta

# 直接连接 MySQL，无需导出到本地
con = duckdb.connect('mysql://user:***@host:3306/shop', read_only=True)

# 7日汇总分析
daily_report = con.execute("""
    SELECT 
        DATE(order_time) AS dt,
        COUNT(*) AS orders,
        SUM(amount) AS revenue,
        AVG(amount) AS avg_order_value,
        COUNT(DISTINCT user_id) AS active_users
    FROM orders
    WHERE order_time >= CURRENT_DATE - INTERVAL '7 days'
    GROUP BY DATE(order_time)
    ORDER BY dt DESC
""").fetchdf()

print(daily_report)

import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg')  # 无头模式，服务器友好

fig, axes = plt.subplots(1, 2, figsize=(12, 5))

# 左图：7日营收趋势
axes[0].plot(daily_report['dt'], daily_report['revenue'], 
             marker='o', color='#2E86AB', linewidth=2)
axes[0].fill_between(daily_report['dt'], daily_report['revenue'], 
                     alpha=0.2, color='#2E86AB')
axes[0].set_title('7-Day Revenue Trend')
axes[0].set_xlabel('Date')
axes[0].set_ylabel('Revenue (¥)')
axes[0].tick_params(axis='x', rotation=45)

# 右图：订单量 vs 客单价散点图（气泡大小=活跃用户数）
scatter = axes[1].scatter(
    daily_report['orders'], 
    daily_report['avg_order_value'], 
    s=daily_report['active_users'] * 2, 
    alpha=0.6, 
    c=daily_report['revenue'],
    cmap='viridis'
)
axes[1].set_title('Orders vs Avg Order Value')
axes[1].set_xlabel('Order Count')
axes[1].set_ylabel('Avg Order Value')
plt.colorbar(scatter, ax=axes[1], label='Revenue')

plt.tight_layout()
plt.savefig('/tmp/daily_report.png', dpi=120, bbox_inches='tight')

from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def format_report(df):
    """将 DataFrame 格式化为可读的日报文本"""
    today = df.iloc[0]
    yesterday = df.iloc[1] if len(df) > 1 else None
    
    report = f"""📊 今日销售日报 · {today['dt'].strftime('%Y-%m-%d')}

营收：¥{today['revenue']:,.2f}
订单数：{int(today['orders'])} 单
客单价：¥{today['avg_order_value']:.2f}
活跃用户：{int(today['active_users'])} 人

"""
    if yesterday is not None:
        rev_change = (today['revenue'] - yesterday['revenue']) / yesterday['revenue'] * 100
        sign = '📈' if rev_change > 0 else '📉'
        report += f"⚡ 环比昨日：{sign} {abs(rev_change):.1f}%"
    
    return report

def send_report(report_text, image_path):
    """发送邮件（可替换为 Telegram Bot API）"""
    msg = MIMEMultipart()
    msg['Subject'] = f"日报 · {datetime.now().strftime('%m-%d')}"
    msg.attach(MIMEText(report_text, 'plain', 'utf-8'))
    
    with open(image_path, 'rb') as f:
        from email.mime.image import MIMEImage
        img = MIMEImage(f.read())
        img.add_header('Content-ID', '<report>')
        msg.attach(img)
    
    # smtp 发送逻辑省略，替换为你自己的配置
    print(f"Report ready → {image_path}")
    print(report_text)

# 执行
report_text = format_report(daily_report)
send_report(report_text, '/tmp/daily_report.png')

import requests

def send_telegram(chat_id, text, photo_path=None):
    bot_token = "YOUR_BOT_TOKEN"
    url = f"https://api.telegram.org/bot{bot_token}"
    
    # 发送文本
    requests.post(f"{url}/sendMessage", json={
        "chat_id": chat_id,
        "text": text,
        "parse_mode": "HTML"
    })
    
    # 发送图片
    if photo_path:
        with open(photo_path, 'rb') as f:
            requests.post(f"{url}/sendPhoto", json={"chat_id": chat_id}, files={"photo": f})

import schedule
import time

def generate_daily_report(client_id):
    """为单个客户生成日报"""
    con = duckdb.connect(f'mysql://user:***@host:3306/{client_id}', read_only=True)
    # ... 执行查询、生成图表、发送报告 ...
    con.close()

# 每天 9:00 自动生成
schedule.every().day.at("09:00").do(generate_daily_report, client_id="client_001")
schedule.every().day.at("09:05").do(generate_daily_report, client_id="client_002")
schedule.every().day.at("09:10").do(generate_daily_report, client_id="client_003")

# 72小时轮询，低负载
while True:
    schedule.run_pending()
    time.sleep(60)