大家好！我是CSDN的Python新手博主～ 上一篇我们完成了看板的跨系统同步与数据备份，解决了数据互通与安全问题，但企业用户反馈三大核心痛点：① 可视化能力弱，仅支持基础图表，无法多维度分析数据关联关系；② 权限管控粗放，仅分管理员/普通用户，无法按部门、数据范围限制访问，易造成数据越权查看；③ 报表导出不灵活，格式固定且无定制化模板，领导汇报、跨部门同步需手动调整格式。今天就带来超落地的新手实战项目——办公看板集成可视化升级+精细化权限管控+定制化报表导出！
本次基于之前的“跨系统安全看板”代码，新增3大核心功能：① 可视化升级（新增组合图表、漏斗图、热力图，支持多维度筛选与钻取）；② 精细化权限管控（实现行级权限、按钮级权限、部门权限关联，适配企业组织架构）；③ 定制化报表导出（支持Excel/PDF模板定制，动态填充数据，保留格式与样式，支持定时导出分发）。全程基于现有技术栈（Flask+MySQL+ECharts+Jinja2），新增图表配置模块、权限管理引擎、报表模板工具，代码注释详细，新手只需配置权限规则与报表模板，跟着步骤复制操作就能成功，让看板适配多角色决策与汇报需求～
一、本次学习目标
掌握ECharts复杂图表实现技巧，搭建多维度可视化体系（组合图、漏斗图、热力图），支持数据筛选、钻取与联动；
学会精细化权限设计，实现行级权限（按部门筛选数据）、按钮级权限（控制操作可见性）、角色-权限-部门关联，符合企业组织架构；
理解定制化报表导出逻辑，基于模板引擎实现Excel/PDF动态填充，支持样式保留、多sheet导出、水印嵌入，适配汇报场景；
实现权限与可视化、报表功能联动，确保不同角色看到对应数据图表、仅能操作权限内功能、导出权限范围内报表；
搭建报表定时导出与分发机制，支持邮件推送、OSS存储，自动完成汇报材料分发，提升办公效率。
二、前期准备
- 安装核心依赖库
安装核心依赖（报表导出、模板渲染、邮件推送）
pip3 install openpyxl xlsxwriter reportlab python-dotenv jinja2 email-validator -i https://pypi.tuna.tsinghua.edu.cn/simple
确保已有依赖正常（Flask、ECharts、APScheduler等）
pip3 install --upgrade flask flask-login flask-sqlalchemy apscheduler requests pandas pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple
说明：可视化升级基于ECharts原生扩展，无需额外依赖；精细化权限通过数据库设计与Flask装饰器实现；报表导出用openpyxl（Excel模板）、reportlab（PDF生成），模板渲染依赖Jinja2（Flask自带）。
- 第三方服务与配置准备
权限规则配置：梳理企业组织架构（部门、角色），定义权限维度（行级：部门数据可见；按钮级：同步、导出、删除等操作权限），确定角色-权限映射关系（如销售经理可看全部门数据，普通销售仅看个人数据）；
报表模板准备：设计Excel/PDF报表模板（含表头、固定样式、动态数据占位符），Excel模板保存至/app/templates/report/excel，PDF模板通过Jinja2语法定义；
邮件服务配置：开启邮箱SMTP服务（如QQ邮箱、企业邮箱），记录SMTP服务器地址、端口、授权码，用于报表定时推送；
前端资源准备：更新ECharts至最新版，引入漏斗图、热力图扩展组件，准备权限控制相关前端组件（按钮显示/隐藏、数据筛选过滤）。
- 数据库表优化与创建
– 连接MySQL数据库（替换为你的数据库信息）
mysql -u office_user -p -h 47.108.xxx.xxx office_data
– 创建部门表（department）
CREATE TABLE department (
id INT AUTO_INCREMENT PRIMARY KEY,
dept_name VARCHAR(50) NOT NULL COMMENT ‘部门名称’,
parent_id INT DEFAULT 0 COMMENT ‘父部门ID（0为顶级部门）’,
dept_desc VARCHAR(255) NULL COMMENT ‘部门描述’,
is_enable TINYINT(1) DEFAULT 1 COMMENT ‘是否启用（1-启用，0-禁用）’,
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT ‘创建时间’,
update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT ‘更新时间’,
UNIQUE KEY uk_dept_name (dept_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT=‘部门表’;
– 创建权限表（permission）
CREATE TABLE permission (
id INT AUTO_INCREMENT PRIMARY KEY,
perm_name VARCHAR(50) NOT NULL COMMENT ‘权限名称（如erp_sync、data_export、delete_data）’,
perm_desc VARCHAR(255) NULL COMMENT ‘权限描述’,
perm_type ENUM(‘button’, ‘menu’, ‘data’) NOT NULL COMMENT ‘权限类型（按钮/菜单/数据）’,
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT ‘创建时间’,
UNIQUE KEY uk_perm_name (perm_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT=‘权限表’;
– 创建角色-权限关联表（role_permission）
CREATE TABLE role_permission (
id INT AUTO_INCREMENT PRIMARY KEY,
role_id INT NOT NULL COMMENT ‘角色ID’,
perm_id INT NOT NULL COMMENT ‘权限ID’,
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT ‘创建时间’,
UNIQUE KEY uk_role_perm (role_id, perm_id),
FOREIGN KEY (role_id) REFERENCES role(id) ON DELETE CASCADE,
FOREIGN KEY (perm_id) REFERENCES permission(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT=‘角色-权限关联表’;
– 创建用户-部门关联表（user_department）
CREATE TABLE user_department (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL COMMENT ‘用户ID’,
dept_id INT NOT NULL COMMENT ‘部门ID’,
is_leader TINYINT(1) DEFAULT 0 COMMENT ‘是否为部门负责人（1-是，0-否）’,
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT ‘创建时间’,
UNIQUE KEY uk_user_dept (user_id, dept_id),
FOREIGN KEY (user_id) REFERENCES user(id) ON DELETE CASCADE,
FOREIGN KEY (dept_id) REFERENCES department(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT=‘用户-部门关联表’;
– 创建报表配置表（report_config）
CREATE TABLE report_config (
id INT AUTO_INCREMENT PRIMARY KEY,
report_name VARCHAR(100) NOT NULL COMMENT ‘报表名称’,
report_type ENUM(‘excel’, ‘pdf’) NOT NULL COMMENT ‘报表类型’,
template_path VARCHAR(255) NOT NULL COMMENT ‘模板路径’,
export_fields TEXT NOT NULL COMMENT ‘导出字段（JSON格式）’,
is_timed TINYINT(1) DEFAULT 0 COMMENT ‘是否定时导出（1-是，0-否）’,
timed_rule VARCHAR(50) NULL COMMENT ‘定时规则（Cron表达式）’,
receiver_emails TEXT NULL COMMENT ‘接收人邮箱（逗号分隔）’,
is_enable TINYINT(1) DEFAULT 1 COMMENT ‘是否启用（1-启用，0-禁用）’,
create_by VARCHAR(50) NOT NULL COMMENT ‘创建人’,
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT ‘创建时间’,
update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT ‘更新时间’,
UNIQUE KEY uk_report_name (report_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT=‘报表配置表’;
– 初始化基础数据
– 1. 插入部门数据
INSERT INTO department (dept_name, parent_id, dept_desc)
VALUES (‘销售部’, 0, ‘负责产品销售与客户对接’), (‘技术部’, 0, ‘负责系统开发与维护’), (‘行政部’, 0, ‘负责日常行政与后勤’);
– 2. 插入权限数据
INSERT INTO permission (perm_name, perm_desc, perm_type)
VALUES
(‘erp_sync’, ‘ERP数据同步’, ‘button’),
(‘data_export’, ‘数据导出’, ‘button’),
(‘delete_data’, ‘数据删除’, ‘button’),
(‘view_all_data’, ‘查看全量数据’, ‘data’),
(‘view_dept_data’, ‘查看部门数据’, ‘data’);
– 3. 关联角色-权限（管理员角色拥有所有权限，普通角色拥有部分权限）
– 假设已有角色表：leader（管理员）、sales（销售）、tech（技术）
INSERT INTO role_permission (role_id, perm_id)
SELECT 1, id FROM permission; – 管理员拥有所有权限
INSERT INTO role_permission (role_id, perm_id)
SELECT 2, id FROM permission WHERE perm_name IN (‘data_export’, ‘view_dept_data’); – 销售角色权限
三、实战：可视化升级+精细化权限+定制报表集成
- 第一步：可视化升级，搭建多维度图表体系
{% extends “base.html” %}
{% block content %}
销售额与订单量趋势
订单转化漏斗
部门-时段销售额热力图
{% endblock %}
-- coding: utf-8 --
chart_api.py 图表数据接口脚本
from flask import Blueprint, request, jsonify
from flask_login import login_required, current_user
from models import Order, Department, UserDepartment
from datetime import datetime, timedelta
import pandas as pd
from sqlalchemy import func
chart_bp = Blueprint(“chart”,name)
时间范围映射（转换为查询条件）
TIME_RANGE_MAP = {
“today”: datetime.now().replace(hour=0, minute=0, second=0, microsecond=0),
“yesterday”: (datetime.now() - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0),
“week”: datetime.now() - timedelta(days=7),
“month”: datetime.now().replace(day=1, hour=0, minute=0, second=0, microsecond=0),
“quarter”: datetime.now().replace(month=((datetime.now().month-1)//3)*3 +1, day=1, hour=0, minute=0, second=0, microsecond=0)
}
@chart_bp.route(“/api/chart/data”, methods=[“POST”])
@login_required
def get_chart_data():
“”“获取图表数据（按筛选条件+权限过滤）”“”
data = request.get_json()
time_range = data.get(“timeRange”, “today”)
dept_id = data.get(“deptId”)
status = data.get(“status”, “all”)
# 1. 构建查询条件（时间+状态） start_time = TIME_RANGE_MAP[time_range] query = Order.query.filter(Order.create_time >= start_time) if status != "all": query = query.filter(Order.status == status) # 2. 权限过滤（行级权限） if not current_user.has_perm("view_all_data"): # 普通用户仅能查看所属部门数据 user_dept_ids = [dept.id for dept in current_user.departments] query = query.filter(Order.dept_id.in_(user_dept_ids)) elif dept_id != "all": # 管理员按筛选部门过滤 query = query.filter(Order.dept_id == dept_id) # 3. 查询数据并处理 orders = query.all() df = pd.DataFrame([{ "create_time": order.create_time.strftime("%m-%d %H"), "sales": order.amount, "order_id": order.id, "status": order.status, "dept_id": order.dept_id, "dept_name": order.department.dept_name if order.department else "未知部门" } for order in orders]) if df.empty: return jsonify({"success": True, "comboData": {}, "funnelData": [], "heatmapData": {}}) # 4. 组合图数据（按小时分组） combo_group = df.groupby("create_time").agg({ "sales": "sum", "order_id": "count" }).reset_index() combo_data = { "xData": combo_group["create_time"].tolist(), "salesData": combo_group["sales"].tolist(), "orderData": combo_group["order_id"].tolist() } # 5. 漏斗图数据（按状态分组） status_map = {"pending": "待付款", "paid": "已付款", "shipped": "已发货", "completed": "已完成", "cancelled": "已取消"} funnel_group = df.groupby("status").agg({"order_id": "count"}).reset_index() funnel_data = [ {"name": status_map.get(row["status"], row["status"]), "value": row["order_id"]} for _, row in funnel_group.iterrows() ] # 6. 热力图数据（部门-时段分组） df["hour_range"] = pd.cut( pd.to_datetime(df["create_time"]).dt.hour, bins=[0,3,6,9,12,15,18,21,24], labels=["0-3点", "3-6点", "6-9点", "9-12点", "12-15点", "15-18点", "18-21点", "21-24点"], right=False ) heat_group = df.groupby(["dept_name", "hour_range"])["sales"].sum().reset_index() dept_data = heat_group["dept_name"].unique().tolist() heat_data = [ [row["dept_name"], row["hour_range"], round(row["sales"], 2)] for _, row in heat_group.iterrows() ] heatmap_data = {"deptData": dept_data, "heatData": heat_data} return jsonify({ "success": True, "comboData": combo_data, "funnelData": funnel_data, "heatmapData": heatmap_data })- 第二步：搭建精细化权限体系，实现多维度管控
-- coding: utf-8 --
permission.py 精细化权限管理脚本
from flask import g, redirect, url_for, abort
from functools import wraps
from models import Permission, RolePermission, UserDepartment, Department
from flask_login import current_user
====================== 模型扩展（在原有User/Role模型中补充） ======================
1. User模型扩展（添加权限、部门关联方法）
class User:
# 原有字段与方法省略…
@property def departments(self): """获取用户所属部门列表""" user_dept_relations = UserDepartment.query.filter_by(user_id=self.id).all() dept_ids = [rel.dept_id for rel in user_dept_relations] return Department.query.filter(Department.id.in_(dept_ids), Department.is_enable == 1).all() @property def is_dept_leader(self): """判断用户是否为部门负责人""" return UserDepartment.query.filter_by( user_id=self.id, is_leader=1, department.is_enable == 1 ).join(Department).first() is not None def has_perm(self, perm_name): """判断用户是否拥有指定权限""" # 管理员默认拥有所有权限 if self.role.role_name == "leader": return True # 查询角色关联的权限 perm = Permission.query.filter_by(perm_name=perm_name).first() if not perm: return False return RolePermission.query.filter_by( role_id=self.role.id, perm_id=perm.id ).first() is not None2. Role模型扩展（添加权限关联）
class Role:
# 原有字段与方法省略…
@property def permissions(self): """获取角色拥有的权限列表""" role_perm_relations = RolePermission.query.filter_by(role_id=self.id).all() perm_ids = [rel.perm_id for rel in role_perm_relations] return Permission.query.filter(Permission.id.in_(perm_ids)).all()====================== 权限校验装饰器（按钮级/数据级权限控制） ======================
def permission_required(perm_name):
“”"
按钮级权限装饰器：无权限则禁止访问
:param perm_name: 权限名称（如data_export、erp_sync）
“”"
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if not current_user.is_authenticated:
return redirect(url_for(“auth.login”))
if not current_user.has_perm(perm_name):
abort(403) # 无权限，返回403
return f(*args, **kwargs)
return decorated_function
return decorator
def data_permission_required():
“”"
行级权限装饰器：过滤用户可访问的数据
需在视图函数中通过g.filtered_dept_ids获取可访问部门ID
“”"
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if not current_user.is_authenticated:
return redirect(url_for(“auth.login”))
# 管理员可访问全部门数据
if current_user.has_perm(“view_all_data”):
g.filtered_dept_ids = [dept.id for dept in Department.query.filter_by(is_enable=1).all()]
else:
# 普通用户仅可访问所属部门数据
g.filtered_dept_ids = [dept.id for dept in current_user.departments]
return f(*args, **kwargs)
return decorated_function
return decorator
====================== 前端权限渲染工具（模板全局函数） ======================
def init_permission_template_filters(app):
“”“初始化模板权限过滤函数，用于前端按钮显示/隐藏”“”
@app.template_filter(“has_perm”)
def has_perm_filter(perm_name):
“”“模板中判断用户是否拥有权限：{{ ‘data_export’|has_perm }}”“”
if not current_user.is_authenticated:
return False
return current_user.has_perm(perm_name)
@app.template_global() def get_user_departments(): """模板中获取用户所属部门：{{ get_user_departments() }}""" if not current_user.is_authenticated: return [] return current_user.departments在app.py中新增以下内容
from permission import init_permission_template_filters, permission_required, data_permission_required
from chart_api import chart_bp
注册蓝图
app.register_blueprint(chart_bp)
初始化模板权限函数
init_permission_template_filters(app)
示例：使用权限装饰器控制接口访问
@app.route(“/data/delete”, methods=[“POST”])
@login_required
@permission_required(“delete_data”) # 按钮级权限控制
def delete_data():
“”“删除数据接口，仅拥有delete_data权限的用户可访问”“”
# 业务逻辑省略…
return jsonify({“success”: True, “msg”: “删除成功”})
@app.route(“/data/list”)
@login_required
@data_permission_required() # 行级权限控制，过滤部门数据
def get_data_list():
“”“获取数据列表，按用户权限过滤部门数据”“”
# 通过g.filtered_dept_ids获取可访问部门ID
data = Order.query.filter(Order.dept_id.in_(g.filtered_dept_ids)).all()
# 数据处理与返回省略…
return jsonify({“success”: True, “data”: data})
- 第三步：定制化报表导出，实现模板化与定时分发
-- coding: utf-8 --
report_export.py 定制化报表导出脚本
import os
import datetime
import pandas as pd
from openpyxl import load_workbook
from openpyxl.styles import Font, Alignment
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
from reportlab.platypus import SimpleDocTemplate, Paragraph, Table, TableStyle
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib import colors
from flask import Blueprint, request, send_file, jsonify
from flask_login import login_required, current_user
from dotenv import load_dotenv
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from models import ReportConfig, Order, Department
from permission import permission_required
from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import func
加载环境变量
load_dotenv()
report_bp = Blueprint(“report”,name)
====================== 配置信息（新手修改这里） ======================
报表模板路径
EXCEL_TEMPLATE_PATH = “/app/templates/report/excel”
PDF_TEMPLATE_PATH = “/app/templates/report/pdf”
临时文件存储路径
TEMP_REPORT_PATH = “/app/temp/report”
os.makedirs(TEMP_REPORT_PATH, exist_ok=True)
邮件配置
SMTP_SERVER = os.getenv(“SMTP_SERVER”) # 如smtp.qq.com
SMTP_PORT = int(os.getenv(“SMTP_PORT”, 465))
SMTP_USER = os.getenv(“SMTP_USER”)
SMTP_PASS = os.getenv(“SMTP_PASS”) # 授权码
====================== 核心功能：Excel定制化导出（基于模板） ======================
def export_excel_by_template(report_config, filter_params=None):
“”"
基于Excel模板导出报表
:param report_config: 报表配置对象
:param filter_params: 筛选参数（时间、部门等）
:return: 临时文件路径
“”"
# 1. 读取模板文件
template_file = os.path.join(EXCEL_TEMPLATE_PATH, report_config.template_path)
if not os.path.exists(template_file):
raise Exception(f"Excel模板文件不存在：{template_file}")
wb = load_workbook(template_file)
ws = wb.active # 默认使用第一个工作表
# 2. 解析导出字段（JSON格式→字典） export_fields = eval(report_config.export_fields) # 字段映射：数据库字段→Excel表头 db_fields = list(export_fields.keys()) excel_headers = list(export_fields.values()) # 3. 查询数据（结合筛选参数与权限） query = Order.query if filter_params: # 时间筛选 if "timeRange" in filter_params: from permission import TIME_RANGE_MAP start_time = TIME_RANGE_MAP.get(filter_params["timeRange"], datetime.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)) query = query.filter(Order.create_time >= start_time) # 部门筛选（权限过滤） if not current_user.has_perm("view_all_data"): user_dept_ids = [dept.id for dept in current_user.departments] query = query.filter(Order.dept_id.in_(user_dept_ids)) elif "deptId" in filter_params and filter_params["deptId"] != "all": query = query.filter(Order.dept_id == filter_params["deptId"]) # 4. 处理数据并填充到模板 orders = query.all() data_list = [] for order in orders: row_data = [] for field in db_fields: # 处理关联字段（如部门名称） if field == "dept_name": value = order.department.dept_name if order.department else "" else: value = getattr(order, field, "") # 格式化时间字段 if isinstance(value, datetime.datetime): value = value.strftime("%Y-%m-%d %H:%M:%S") row_data.append(value) data_list.append(row_data) # 5. 填充数据到Excel（从第3行开始，第1-2行为表头/标题） start_row = 3 for row_idx, row_data in enumerate(data_list, start=start_row): for col_idx, value in enumerate(row_data, start=1): cell = ws.cell(row=row_idx, column=col_idx, value=value) # 设置单元格样式（对齐、字体） cell.alignment = Alignment(horizontal="center", vertical="center") cell.font = Font(name="微软雅黑", size=10) # 6. 填充报表基本信息（标题、时间、生成人） ws["A1"] = report_config.report_name # 报表标题 ws["A1"].font = Font(name="微软雅黑", size=16, bold=True) ws["A1"].alignment = Alignment(horizontal="center") ws.merge_cells("A1:{}{}".format(chr(64 + len(excel_headers)), 1)) # 合并标题单元格 ws["A2"] = f"生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" ws["B2"] = f"生成人：{current_user.name}" ws["C2"] = f"数据条数：{len(data_list)}" # 7. 保存临时文件 temp_file_name = f"{report_config.report_name}_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}.xlsx" temp_file_path = os.path.join(TEMP_REPORT_PATH, temp_file_name) wb.save(temp_file_path) return temp_file_path====================== 核心功能：PDF定制化导出（基于Jinja2模板） ======================
def export_pdf_by_template(report_config, filter_params=None):
“”"
基于Jinja2模板导出PDF报表
:param report_config: 报表配置对象
:param filter_params: 筛选参数
:return: 临时文件路径
“”"
# 1. 准备PDF文件
temp_file_name = f"{report_config.report_name}_{datetime.datetime.now().strftime(‘%Y%m%d%H%M%S’)}.pdf"
temp_file_path = os.path.join(TEMP_REPORT_PATH, temp_file_name)
doc = SimpleDocTemplate(temp_file_path, pagesize=A4)
elements = []
styles = getSampleStyleSheet()
# 2. 添加报表标题 title_style = ParagraphStyle( name="Title", fontName="Helvetica-Bold", fontSize=16, alignment=1, spaceAfter=20 ) title = Paragraph(report_config.report_name, title_style) elements.append(title) # 3. 添加报表基本信息 info_style = ParagraphStyle(name="Info", fontSize=10, spaceAfter=10) info_text = f"生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | 生成人：{current_user.name}" elements.append(Paragraph(info_text, info_style)) # 4. 查询并处理数据（逻辑同Excel导出） export_fields = eval(report_config.export_fields) db_fields = list(export_fields.keys()) excel_headers = list(export_fields.values()) query = Order.query # 筛选逻辑省略（同Excel导出） orders = query.all() data_list = [[export_fields[field] for field in db_fields]] # 表头 for order in orders: row_data = [] for field in db_fields: value = getattr(order, field, "") if isinstance(value, datetime.datetime): value = value.strftime("%Y-%m-%d %H:%M:%S") row_data.append(str(value)) data_list.append(row_data) # 5. 添加表格数据 table = Table(data_list, repeatRows=1) table_style = TableStyle([ ("BACKGROUND", (0,0), (-1,0), colors.lightblue), ("TEXTCOLOR", (0,0), (-1,0), colors.black), ("ALIGN", (0,0), (-1,-1), "CENTER"), ("FONTNAME", (0,0), (-1,0), "Helvetica-Bold"), ("FONTSIZE", (0,0), (-1,-1), 10), ("GRID", (0,0), (-1,-1), 1, colors.black), ("ROWBACKGROUNDS", (0,1), (-1,-1), [colors.white, colors.lightgrey]) ]) table.setStyle(table_style) elements.append(table) # 6. 生成PDF doc.build(elements) return temp_file_path====================== 核心功能：邮件推送报表 ======================
def send_report_email(report_path, report_name, receiver_emails):
“”“发送报表邮件（带附件）”“”
if not receiver_emails or not os.path.exists(report_path):
return False
# 1. 构建邮件 msg = MIMEMultipart() msg["Subject"] = f"【自动报表】{report_name}（{datetime.datetime.now().strftime('%Y-%m-%d')}）" msg["From"] = SMTP_USER msg["To"] = ",".join(receiver_emails) # 2. 邮件正文 text = f"您好，附件为{report_name}，请查收。\n生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" msg.attach(MIMEText(text, "plain", "utf-8")) # 3. 添加附件 with open(report_path, "rb") as f: part = MIMEApplication(f.read(), Name=os.path.basename(report_path)) part["Content-Disposition"] = f'attachment; filename="{os.path.basename(report_path)}"' msg.attach(part) # 4. 发送邮件 try: with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server: server.login(SMTP_USER, SMTP_PASS) server.sendmail(SMTP_USER, receiver_emails, msg.as_string()) return True except Exception as e: print(f"邮件发送失败：{str(e)}") return False====================== 接口：手动导出定制报表 ======================
@report_bp.route(“/report/export/custom”, methods=[“POST”])
@login_required
@permission_required(“data_export”)
def export_custom_report():
“”“手动导出定制报表（按配置）”“”
data = request.get_json()
report_id = data.get(“report_id”)
filter_params = data.get(“filter_params”, {})
if not report_id: return jsonify({"success": False, "error": "需指定报表ID"}) report_config = ReportConfig.query.get(report_id) if not report_config or not report_config.is_enable: return jsonify({"success": False, "error": "报表配置不存在或已禁用"}) try: # 按报表类型导出 if report_config.report_type == "excel": temp_path = export_excel_by_template(report_config, filter_params) else: temp_path = export_pdf_by_template(report_config, filter_params) # 发送文件给前端下载 return send_file( temp_path, as_attachment=True, download_name=os.path.basename(temp_path), mimetype="application/vnd.openxmlformats-off
)
