本文目录导读：
我来详细介绍如何编写报表自动推送脚本,包括多种实现方式和完整示例。
总体架构设计
graph TB
A[数据源] --> B[报表生成模块]
B --> C[推送渠道选择]
C --> D[邮件推送]
C --> E[企业微信推送]
C --> F[钉钉推送]
C --> G[Slack推送]
B --> H[调度执行]
H --> I[Crontab/Windows任务]
H --> J[定时触发]
Python完整示例代码
1 基础框架
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
报表自动推送脚本
支持多种数据源和推送渠道
"""
import pandas as pd
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import requests
import json
import logging
from datetime import datetime, timedelta
import os
import sqlalchemy
from typing import Dict, List, Optional
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class ReportGenerator:
"""报表生成器"""
def __init__(self, config: Dict):
self.config = config
def from_database(self, query: str, db_connection: str) -> pd.DataFrame:
"""从数据库获取数据"""
try:
engine = sqlalchemy.create_engine(db_connection)
df = pd.read_sql_query(query, engine)
logger.info(f"从数据库获取数据成功，共{len(df)}条记录")
return df
except Exception as e:
logger.error(f"数据库查询失败: {e}")
raise
def from_csv(self, file_path: str) -> pd.DataFrame:
"""从CSV文件获取数据"""
try:
df = pd.read_csv(file_path)
logger.info(f"从CSV文件读取数据成功，共{len(df)}条记录")
return df
except Exception as e:
logger.error(f"CSV文件读取失败: {e}")
raise
def from_excel(self, file_path: str, sheet_name: str = 'Sheet1') -> pd.DataFrame:
"""从Excel文件获取数据"""
try:
df = pd.read_excel(file_path, sheet_name=sheet_name)
logger.info(f"从Excel文件读取数据成功，共{len(df)}条记录")
return df
except Exception as e:
logger.error(f"Excel文件读取失败: {e}")
raise
def generate_report(self, data: pd.DataFrame, format: str = 'excel',
output_path: Optional[str] = None) -> str:
"""生成报表文件"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
if not output_path:
output_path = f"report_{timestamp}"
try:
if format == 'excel':
file_path = f"{output_path}.xlsx"
with pd.ExcelWriter(file_path, engine='openpyxl') as writer:
data.to_excel(writer, sheet_name='Report', index=False)
# 自动调整列宽
worksheet = writer.sheets['Report']
for column in data.columns:
column_width = max(data[column].astype(str).map(len).max(),
len(str(column))) + 2
worksheet.column_dimensions[column].width = column_width
elif format == 'csv':
file_path = f"{output_path}.csv"
data.to_csv(file_path, index=False, encoding='utf-8-sig')
elif format == 'html':
file_path = f"{output_path}.html"
html_content = data.to_html(index=False)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(html_content)
logger.info(f"报表文件已生成: {file_path}")
return file_path
except Exception as e:
logger.error(f"报表生成失败: {e}")
raise
class ReportPusher:
"""报表推送器"""
def __init__(self, config: Dict):
self.config = config
def push_email(self, subject: str, body: str,
attachments: List[str], recipients: List[str]) -> bool:
"""通过邮件推送报表"""
try:
msg = MIMEMultipart()
msg['From'] = self.config['email']['sender']
msg['To'] = ', '.join(recipients)
msg['Subject'] = subject
# 添加邮件正文
msg.attach(MIMEText(body, 'html', 'utf-8'))
# 添加附件
for file_path in attachments:
with open(file_path, 'rb') as attachment:
part = MIMEBase('application', 'octet-stream')
part.set_payload(attachment.read())
encoders.encode_base64(part)
part.add_header(
'Content-Disposition',
f'attachment; filename= {os.path.basename(file_path)}'
)
msg.attach(part)
# 发送邮件
with smtplib.SMTP_SSL(self.config['email']['smtp_server'],
self.config['email']['smtp_port']) as server:
server.login(self.config['email']['username'],
self.config['email']['password'])
server.send_message(msg)
logger.info(f"邮件推送成功，收件人: {recipients}")
return True
except Exception as e:
logger.error(f"邮件推送失败: {e}")
return False
def push_wechat(self, message: str, file_path: Optional[str] = None) -> bool:
"""通过企业微信推送报表"""
try:
# 企业微信机器人Webhook地址
webhook_url = self.config['wechat']['webhook_url']
if file_path:
# 上传文件
media_id = self._upload_file_to_wechat(file_path)
if media_id:
data = {
"msgtype": "file",
"file": {
"media_id": media_id
}
}
else:
data = {
"msgtype": "text",
"text": {
"content": f"报表已生成，但文件上传失败: {message}"
}
}
else:
data = {
"msgtype": "markdown",
"markdown": {
"content": message
}
}
response = requests.post(webhook_url, json=data)
if response.status_code == 200:
logger.info("企业微信推送成功")
return True
else:
logger.error(f"企业微信推送失败: {response.text}")
return False
except Exception as e:
logger.error(f"企业微信推送异常: {e}")
return False
def push_dingtalk(self, message: str, file_path: Optional[str] = None) -> bool:
"""通过钉钉推送报表"""
try:
webhook_url = self.config['dingtalk']['webhook_url']
if file_path:
# 钉钉需要先上传文件到临时存储
# 这里简化处理，发送文件链接
data = {
"msgtype": "markdown",
"markdown": {
"title": "报表推送",
"text": f"## 报表通知\n{message}\n> 文件路径: {file_path}"
}
}
else:
data = {
"msgtype": "text",
"text": {
"content": message
}
}
response = requests.post(webhook_url, json=data)
if response.status_code == 200:
logger.info("钉钉推送成功")
return True
else:
logger.error(f"钉钉推送失败: {response.text}")
return False
except Exception as e:
logger.error(f"钉钉推送异常: {e}")
return False
def _upload_file_to_wechat(self, file_path: str) -> Optional[str]:
"""上传文件到企业微信（获取media_id）"""
try:
# 需要access_token，这里简化处理
# 实际需要先通过corpid和corpsecret获取access_token
return None
except Exception as e:
logger.error(f"上传文件到企业微信失败: {e}")
return None
class ReportScheduler:
"""报表调度器"""
def __init__(self, report_generator: ReportGenerator,
report_pusher: ReportPusher, config: Dict):
self.generator = report_generator
self.pusher = report_pusher
self.config = config
def daily_report(self, query: str, recipients: List[str],
push_methods: List[str] = ['email']) -> Dict:
"""执行每日报表推送"""
try:
logger.info("开始执行每日报表推送")
# 1. 获取数据
data = self.generator.from_database(query, self.config['database'])
# 2. 生成报表
report_file = self.generator.generate_report(data, format='excel')
# 3. 推送报表
results = {}
for method in push_methods:
if method == 'email':
subject = f"每日报表 - {datetime.now().strftime('%Y-%m-%d')}"
body = """
<html>
<body>
<h2>每日报表推送</h2>
<p>您好，这是系统自动推送的每日报表。</p>
<p>报表日期: {date}</p>
<p>数据条数: {count}</p>
<p>报表文件请查看附件。</p>
</body>
</html>
""".format(
date=datetime.now().strftime('%Y-%m-%d'),
count=len(data)
)
success = self.pusher.push_email(
subject=subject,
body=body,
attachments=[report_file],
recipients=recipients
)
results['email'] = success
elif method == 'wechat':
message = f"## 每日报表通知\n" \
f"> 日期: {datetime.now().strftime('%Y-%m-%d')}\n" \
f"> 数据条数: {len(data)}\n" \
f"> 文件: {report_file}"
success = self.pusher.push_wechat(message)
results['wechat'] = success
elif method == 'dingtalk':
message = f"## 每日报表通知\n" \
f"日期: {datetime.now().strftime('%Y-%m-%d')}\n" \
f"数据条数: {len(data)}\n" \
f"文件: {report_file}"
success = self.pusher.push_dingtalk(message)
results['dingtalk'] = success
logger.info(f"每日报表推送完成，结果: {results}")
return results
except Exception as e:
logger.error(f"每日报表推送失败: {e}")
raise
class ReportSystem:
"""报表系统主类"""
def __init__(self, config_file: str = 'config.json'):
self.config = self._load_config(config_file)
self.generator = ReportGenerator(self.config)
self.pusher = ReportPusher(self.config)
self.scheduler = ReportScheduler(self.generator, self.pusher, self.config)
def _load_config(self, config_file: str) -> Dict:
"""加载配置文件"""
try:
with open(config_file, 'r', encoding='utf-8') as f:
config = json.load(f)
logger.info(f"配置文件加载成功: {config_file}")
return config
except Exception as e:
logger.error(f"配置文件加载失败: {e}")
# 返回默认配置
return self._default_config()
def _default_config(self) -> Dict:
"""默认配置"""
return {
"database": {
"connection_string": "mysql+pymysql://user:password@localhost/dbname"
},
"email": {
"smtp_server": "smtp.gmail.com",
"smtp_port": 465,
"sender": "your-email@gmail.com",
"username": "your-email@gmail.com",
"password": "your-password"
},
"wechat": {
"webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-key"
},
"dingtalk": {
"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=your-token"
},
"schedule": {
"time": "09:00",
"days": ["monday", "tuesday", "wednesday", "thursday", "friday"]
}
}
def run_once(self, query: str, recipients: List[str],
push_methods: List[str] = ['email']) -> Dict:
"""单次执行报表推送"""
return self.scheduler.daily_report(query, recipients, push_methods)
def setup_cron(self) -> str:
"""生成Crontab配置"""
schedule_time = self.config['schedule']['time']
hour, minute = schedule_time.split(':')
cron_config = f"""
# 报表自动推送任务
{minute} {hour} * * 1-5 cd /path/to/script && python3 report_system.py
"""
return cron_config
def main():
"""主函数"""
import argparse
parser = argparse.ArgumentParser(description='报表自动推送系统')
parser.add_argument('--config', default='config.json', help='配置文件路径')
parser.add_argument('--query', help='SQL查询语句')
parser.add_argument('--recipients', nargs='+', help='收件人列表')
parser.add_argument('--methods', nargs='+', default=['email'],
choices=['email', 'wechat', 'dingtalk'],
help='推送方式')
args = parser.parse_args()
# 初始化系统
system = ReportSystem(args.config)
# 执行推送
if args.query and args.recipients:
results = system.run_once(
query=args.query,
recipients=args.recipients,
push_methods=args.methods
)
print(f"推送结果: {results}")
else:
# 显示Crontab配置
print("请提供查询语句和收件人列表")
print("示例: python3 report_system.py --query \"SELECT * FROM sales\" --recipients user1@email.com --methods email wechat")
if __name__ == "__main__":
main()
2 配置文件示例 (config.json)
{
"database": {
"connection_string": "mysql+pymysql://username:password@localhost:3306/database_name",
"queries": {
"daily_sales": "SELECT date, product, amount, quantity FROM sales WHERE date = CURRENT_DATE",
"weekly_report": "SELECT DATE_FORMAT(date, '%Y-%u') as week, SUM(amount) as total_amount FROM sales GROUP BY week ORDER BY week DESC LIMIT 1"
}
},
"email": {
"smtp_server": "smtp.gmail.com",
"smtp_port": 465,
"sender": "reports@company.com",
"username": "reports@company.com",
"password": "your-app-password",
"recipients": {
"daily": ["manager@company.com", "team@company.com"],
"weekly": ["all-staff@company.com"]
}
},
"wechat": {
"enabled": true,
"webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-webhook-key"
},
"dingtalk": {
"enabled": false,
"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=your-token"
},
"schedule": {
"daily_time": "09:00",
"weekly_day": "Monday",
"weekly_time": "10:00",
"timezone": "Asia/Shanghai"
},
"output": {
"format": "excel",
"directory": "./reports",
"retention_days": 30,
"auto_delete": true
}
}
3 定时任务配置
Linux Crontab 配置：
# 每天9点执行日报 0 9 * * 1-5 /usr/bin/python3 /path/to/report_system.py --config /path/to/config.json 2>&1 /var/log/report.log # 每周一10点执行周报 0 10 * * 1 /usr/bin/python3 /path/to/report_system.py --config /path/to/config_weekly.json 2>&1 /var/log/report_weekly.log
Windows 任务计划程序：
@echo off C:\Python39\python.exe D:\scripts\report_system.py --config D:\scripts\config.json
高级功能实现
1 数据可视化报表
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg') # 非交互模式
class VisualizationReport:
"""可视化报表"""
def create_charts(self, data: pd.DataFrame) -> List[str]:
"""创建图表"""
chart_files = []
# 1. 折线图
plt.figure(figsize=(10, 6))
plt.plot(data['date'], data['amount'], marker='o')
plt.title('销售趋势')
plt.xlabel('日期')
plt.ylabel('销售额')
plt.xticks(rotation=45)
plt.tight_layout()
line_chart = 'sales_trend.png'
plt.savefig(line_chart)
chart_files.append(line_chart)
plt.close()
# 2. 饼图
plt.figure(figsize=(8, 8))
category_sum = data.groupby('category')['amount'].sum()
plt.pie(category_sum.values, labels=category_sum.index, autopct='%1.1f%%')
plt.title('分类占比')
pie_chart = 'category_distribution.png'
plt.savefig(pie_chart)
chart_files.append(pie_chart)
plt.close()
return chart_files
2 异常检测与告警
class AlertSystem:
"""告警系统"""
def check_threshold(self, data: pd.DataFrame, thresholds: Dict) -> List[str]:
"""检查数据阈值"""
alerts = []
for column, config in thresholds.items():
current_value = data[column].iloc[-1]
if current_value > config['max']:
alerts.append(
f"告警: {column} 当前值 {current_value} 超过上限 {config['max']}"
)
elif current_value < config['min']:
alerts.append(
f"告警: {column} 当前值 {current_value} 低于下限 {config['min']}"
)
return alerts
def compare_period(self, current: pd.DataFrame,
previous: pd.DataFrame) -> List[str]:
"""周期对比"""
alerts = []
current_total = current['amount'].sum()
previous_total = previous['amount'].sum()
change_rate = (current_total - previous_total) / previous_total * 100
if abs(change_rate) > 20: # 变化超过20%触发告警
direction = "增长" if change_rate > 0 else "下降"
alerts.append(
f"告警: 销售额环比{direction} {abs(change_rate):.1f}%"
)
return alerts
3 完整使用示例
# 使用示例
def example_usage():
"""使用示例"""
# 1. 基本使用
system = ReportSystem('config.json')
# 执行日报推送
result = system.run_once(
query="SELECT * FROM sales WHERE date = CURRENT_DATE",
recipients=['manager@company.com'],
push_methods=['email', 'wechat']
)
# 2. 高级用法
from datetime import date, timedelta
# 获取昨天和上周同期的数据
yesterday = date.today() - timedelta(days=1)
last_week = yesterday - timedelta(days=7)
generator = ReportGenerator(system.config)
# 生成对比报告
current_data = generator.from_database(
f"SELECT * FROM sales WHERE date = '{yesterday}'",
system.config['database']['connection_string']
)
previous_data = generator.from_database(
f"SELECT * FROM sales WHERE date = '{last_week}'",
system.config['database']['connection_string']
)
# 创建可视化报表
viz = VisualizationReport()
charts = viz.create_charts(current_data)
# 检查告警
alert_system = AlertSystem()
alerts = alert_system.compare_period(current_data, previous_data)
if alerts:
# 发送告警通知
pusher = ReportPusher(system.config)
for alert in alerts:
pusher.push_wechat(alert)
if __name__ == "__main__":
example_usage()
部署与维护
1 依赖安装
# 安装依赖 pip install pandas openpyxl sqlalchemy pymysql matplotlib requests schedule # 或使用requirements.txt pip install -r requirements.txt
2 日志配置
import logging.handlers
def setup_logging():
"""配置日志系统"""
logger = logging.getLogger('report_system')
logger.setLevel(logging.INFO)
# 文件日志
file_handler = logging.handlers.RotatingFileHandler(
'report_system.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5,
encoding='utf-8'
)
file_handler.setLevel(logging.INFO)
# 控制台日志
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 设置格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
这个完整的报表自动推送系统包含了：
- 多数据源支持（数据库、CSV、Excel）
- 多格式输出（Excel、CSV、HTML、图片）
- 多渠道推送（邮件、企业微信、钉钉）
- 定时任务调度
- 异常检测与告警
- 完整的日志系统
你可以根据实际需求选择需要的功能模块进行集成。
