428 lines
18 KiB
Python
428 lines
18 KiB
Python
"""
|
||
文件下载 API
|
||
"""
|
||
import os
|
||
import zipfile
|
||
from datetime import datetime
|
||
from flask import request, jsonify, send_from_directory
|
||
from lib.db import execute_query
|
||
from lib.logger import log_info, log_error, log_warning
|
||
from . import download_bp
|
||
|
||
|
||
def get_reports_dir():
|
||
"""获取报表文件目录"""
|
||
return os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), 'public', 'reports')
|
||
|
||
|
||
def get_temp_dir():
|
||
"""获取临时zip文件目录"""
|
||
temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), 'public', 'temp_zip')
|
||
os.makedirs(temp_dir, exist_ok=True)
|
||
return temp_dir
|
||
|
||
|
||
def cleanup_old_temp_zips():
|
||
"""清理1小时前的临时zip文件"""
|
||
try:
|
||
temp_dir = get_temp_dir()
|
||
now = datetime.now().timestamp()
|
||
count = 0
|
||
for filename in os.listdir(temp_dir):
|
||
if filename.endswith('.zip'):
|
||
filepath = os.path.join(temp_dir, filename)
|
||
if now - os.path.getmtime(filepath) > 3600: # 1小时前的
|
||
os.remove(filepath)
|
||
count += 1
|
||
if count > 0:
|
||
log_info(f'[下载API] 清理了 {count} 个过期临时zip文件', 'download')
|
||
except Exception as e:
|
||
log_error(f'[下载API] 清理临时文件失败: {e}', 'download')
|
||
|
||
|
||
@download_bp.route('/download', methods=['GET'])
|
||
def download_file():
|
||
"""下载文件
|
||
支持两种方式：
|
||
1. path: 直接传文件路径
|
||
2. id: 传历史记录ID，从数据库查询文件路径
|
||
"""
|
||
try:
|
||
file_path = request.args.get('path')
|
||
history_id = request.args.get('id')
|
||
|
||
log_info(f'[下载API] 请求，path={file_path}, id={history_id}', 'download')
|
||
|
||
# 如果传了id，从历史记录表查询文件路径
|
||
if history_id:
|
||
result = execute_query(
|
||
'SELECT file_path, status FROM t_daily_report_history WHERE id = %s LIMIT 1',
|
||
(history_id,)
|
||
)
|
||
if not result:
|
||
log_error(f'[下载API] 历史记录不存在，id={history_id}', 'download')
|
||
return jsonify({'success': False, 'message': '历史记录不存在'}), 404
|
||
|
||
history = result[0]
|
||
if history.get('status') != 1:
|
||
log_error(f'[下载API] 生成失败的记录无法下载，id={history_id}', 'download')
|
||
return jsonify({'success': False, 'message': '该记录生成失败，无法下载'}), 400
|
||
|
||
file_path = history.get('file_path')
|
||
if not file_path:
|
||
log_error(f'[下载API] 文件路径为空，id={history_id}', 'download')
|
||
return jsonify({'success': False, 'message': '文件路径不能为空'}), 400
|
||
|
||
log_info(f'[下载API] 从历史记录查询到文件路径: {file_path}', 'download')
|
||
|
||
if not file_path:
|
||
return jsonify({'success': False, 'message': '文件路径不能为空'}), 400
|
||
|
||
# 安全检查：确保路径在 public/reports 目录下
|
||
if not file_path.startswith('/reports/') and not file_path.startswith('/public/reports/'):
|
||
return jsonify({'success': False, 'message': '无效的文件路径'}), 400
|
||
|
||
# 提取文件名
|
||
if file_path.startswith('/public/reports/'):
|
||
filename = file_path.replace('/public/reports/', '')
|
||
else:
|
||
filename = file_path.replace('/reports/', '')
|
||
|
||
reports_dir = get_reports_dir()
|
||
full_path = os.path.join(reports_dir, filename)
|
||
if not os.path.exists(full_path):
|
||
log_error(f'[下载API] 文件不存在: {full_path}', 'download')
|
||
return jsonify({'success': False, 'message': '文件不存在'}), 404
|
||
|
||
log_info(f'[下载API] 下载文件: {filename}', 'download')
|
||
return send_from_directory(reports_dir, filename, as_attachment=True)
|
||
except Exception as e:
|
||
log_error(f'[下载API] 错误: {e}', 'download')
|
||
import traceback
|
||
traceback.print_exc()
|
||
return jsonify({'success': False, 'message': str(e)}), 500
|
||
|
||
|
||
@download_bp.route('/download/batch', methods=['POST'])
|
||
def batch_download():
|
||
"""批量打包下载
|
||
接收 ids 数组，将对应的报表文件打包成zip后下载
|
||
"""
|
||
try:
|
||
data = request.get_json(silent=True) or {}
|
||
ids = data.get('ids', [])
|
||
|
||
log_info(f'[下载API] 批量下载，数量: {len(ids)}', 'download')
|
||
|
||
if not ids or len(ids) == 0:
|
||
return jsonify({'success': False, 'message': '请选择要下载的记录'}), 400
|
||
|
||
# 清理旧的临时文件
|
||
cleanup_old_temp_zips()
|
||
|
||
# 查询所有成功的历史记录（附带筛选条件以增强安全性）
|
||
placeholders = ', '.join(['%s'] * len(ids))
|
||
result = execute_query(
|
||
f'SELECT id, file_path, config_name, report_date, status, split_name FROM t_daily_report_history WHERE id IN ({placeholders})',
|
||
tuple(ids)
|
||
)
|
||
|
||
if not result:
|
||
return jsonify({'success': False, 'message': '未找到对应的历史记录'}), 404
|
||
|
||
total_matched = len(result)
|
||
success_records = [r for r in result if r.get('status') == 1]
|
||
skipped_failed = total_matched - len(success_records)
|
||
|
||
reports_dir = get_reports_dir()
|
||
temp_dir = get_temp_dir()
|
||
|
||
# 收集要打包的文件
|
||
files_to_zip = []
|
||
failed_names = []
|
||
used_names = set() # 检测 zip 内文件名冲突
|
||
|
||
for item in success_records:
|
||
file_path = item.get('file_path')
|
||
if not file_path:
|
||
failed_names.append(f"{item.get('config_name', '未知')} (无文件路径)")
|
||
continue
|
||
|
||
# 提取文件名
|
||
if file_path.startswith('/public/reports/'):
|
||
filename = file_path.replace('/public/reports/', '')
|
||
elif file_path.startswith('/reports/'):
|
||
filename = file_path.replace('/reports/', '')
|
||
else:
|
||
filename = os.path.basename(file_path)
|
||
|
||
full_path = os.path.join(reports_dir, filename)
|
||
if not os.path.exists(full_path):
|
||
failed_names.append(f"{item.get('config_name', '未知')}_{item.get('report_date', '')} (文件不存在)")
|
||
continue
|
||
|
||
# 使用配置名+报表日期+拆分值作为zip内的文件名，避免重名
|
||
config_name = item.get('config_name', '未知')
|
||
report_date = item.get('report_date', '')
|
||
split_name = item.get('split_name', '')
|
||
ext = os.path.splitext(filename)[1] or '.xlsx'
|
||
|
||
base_parts = [config_name]
|
||
if report_date:
|
||
base_parts.append(str(report_date))
|
||
if split_name:
|
||
base_parts.append(str(split_name))
|
||
base_name = '_'.join(base_parts)
|
||
zip_inner_name = f"{base_name}{ext}"
|
||
|
||
# 处理文件名冲突：如果同名，追加记录ID
|
||
if zip_inner_name in used_names:
|
||
zip_inner_name = f"{base_name}_{item.get('id', '')}{ext}"
|
||
used_names.add(zip_inner_name)
|
||
|
||
files_to_zip.append({
|
||
'full_path': full_path,
|
||
'inner_name': zip_inner_name
|
||
})
|
||
|
||
if len(files_to_zip) == 0:
|
||
return jsonify({'success': False, 'message': '没有可下载的文件'}), 400
|
||
|
||
# 生成zip文件名
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
zip_filename = f'批量下载_{timestamp}.zip'
|
||
zip_path = os.path.join(temp_dir, zip_filename)
|
||
|
||
# 打包
|
||
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||
for file_info in files_to_zip:
|
||
zf.write(file_info['full_path'], file_info['inner_name'])
|
||
|
||
log_info(f'[下载API] 批量打包完成，共 {len(files_to_zip)} 个文件，失败 {len(failed_names)} 个，跳过 {skipped_failed} 个失败记录', 'download')
|
||
if failed_names:
|
||
log_warning(f'[下载API] 缺失/无效文件列表: {failed_names[:20]}', 'download')
|
||
|
||
# 返回zip文件，通过响应头返回统计信息
|
||
from flask import send_file
|
||
response = send_file(
|
||
zip_path,
|
||
as_attachment=True,
|
||
download_name=zip_filename,
|
||
mimetype='application/zip'
|
||
)
|
||
response.headers['X-Total-Matched'] = str(total_matched)
|
||
response.headers['X-Files-Zipped'] = str(len(files_to_zip))
|
||
response.headers['X-Files-Missing'] = str(len(failed_names))
|
||
response.headers['X-Skipped-Failed'] = str(skipped_failed)
|
||
return response
|
||
except Exception as e:
|
||
log_error(f'[下载API] 批量下载错误: {e}', 'download')
|
||
import traceback
|
||
traceback.print_exc()
|
||
return jsonify({'success': False, 'message': str(e)}), 500
|
||
|
||
|
||
@download_bp.route('/download/batch-filter', methods=['POST'])
|
||
def batch_download_by_filter():
|
||
"""按筛选条件批量打包下载
|
||
接收筛选条件（config_name, start_date, end_date, status），
|
||
将所有匹配的成功记录的报表文件打包成zip后下载。
|
||
与历史记录列表的搜索条件保持同步。
|
||
"""
|
||
try:
|
||
data = request.get_json(silent=True) or {}
|
||
config_name = data.get('config_name', '').strip()
|
||
start_date = data.get('start_date', '').strip()
|
||
end_date = data.get('end_date', '').strip()
|
||
status = data.get('status', '')
|
||
|
||
log_info(f'[下载API] 按筛选批量下载，配置名称: {config_name or "全部"}，日期范围: {start_date or "开始"} ~ {end_date or "结束"}，状态: {status or "全部"}', 'download')
|
||
|
||
# 清理旧的临时文件
|
||
cleanup_old_temp_zips()
|
||
|
||
# 构建查询条件（与历史记录列表查询条件完全一致）
|
||
where_clauses = ['status = 1'] # 只下载成功的记录
|
||
params = []
|
||
|
||
if start_date:
|
||
where_clauses.append('DATE(start_time) >= %s')
|
||
params.append(start_date)
|
||
|
||
if end_date:
|
||
where_clauses.append('DATE(end_time) <= %s')
|
||
params.append(end_date)
|
||
|
||
# 注意：批量导出只导出成功记录（status=1），忽略用户传入的 status 筛选
|
||
# 因为失败的记录没有文件可下载
|
||
|
||
if config_name:
|
||
where_clauses.append('config_name LIKE %s')
|
||
params.append(f'%{config_name}%')
|
||
|
||
where_sql = ' AND '.join(where_clauses)
|
||
|
||
# 先查询总数，用于日志记录
|
||
count_result = execute_query(
|
||
f'SELECT COUNT(*) as total FROM t_daily_report_history WHERE {where_sql}',
|
||
tuple(params)
|
||
)
|
||
total_matched = count_result[0]['total'] if count_result else 0
|
||
log_info(f'[下载API] 按筛选匹配到 {total_matched} 条成功记录', 'download')
|
||
|
||
# 查询所有匹配的成功记录（按报表日期正序排列，方便用户查看）
|
||
result = execute_query(
|
||
f'SELECT id, file_path, config_name, report_date, status, split_name FROM t_daily_report_history WHERE {where_sql} ORDER BY report_date ASC, config_name ASC, id ASC',
|
||
tuple(params)
|
||
)
|
||
|
||
if not result:
|
||
return jsonify({'success': False, 'message': '未找到符合条件的历史记录'}), 404
|
||
|
||
reports_dir = get_reports_dir()
|
||
temp_dir = get_temp_dir()
|
||
|
||
# 收集要打包的文件
|
||
files_to_zip = []
|
||
failed_names = []
|
||
used_names = set() # 用于检测 zip 内文件名冲突
|
||
|
||
for item in result:
|
||
file_path = item.get('file_path')
|
||
if not file_path:
|
||
failed_names.append(f"{item.get('config_name', '未知')}_{item.get('report_date', '')} (无文件路径)")
|
||
continue
|
||
|
||
# 提取文件名
|
||
if file_path.startswith('/public/reports/'):
|
||
filename = file_path.replace('/public/reports/', '')
|
||
elif file_path.startswith('/reports/'):
|
||
filename = file_path.replace('/reports/', '')
|
||
else:
|
||
filename = os.path.basename(file_path)
|
||
|
||
full_path = os.path.join(reports_dir, filename)
|
||
if not os.path.exists(full_path):
|
||
failed_names.append(f"{item.get('config_name', '未知')}_{item.get('report_date', '')} (文件不存在)")
|
||
continue
|
||
|
||
# 构建 zip 内文件名：配置名_报表日期（避免重名）
|
||
config_name_val = item.get('config_name', '未知')
|
||
report_date = item.get('report_date', '')
|
||
split_name = item.get('split_name', '')
|
||
ext = os.path.splitext(filename)[1] or '.xlsx'
|
||
|
||
# 基础名称：配置名_报表日期
|
||
base_parts = [config_name_val]
|
||
if report_date:
|
||
base_parts.append(str(report_date))
|
||
if split_name:
|
||
base_parts.append(str(split_name))
|
||
base_name = '_'.join(base_parts)
|
||
zip_inner_name = f"{base_name}{ext}"
|
||
|
||
# 处理文件名冲突：如果同名，追加记录ID
|
||
if zip_inner_name in used_names:
|
||
zip_inner_name = f"{base_name}_{item.get('id', '')}{ext}"
|
||
used_names.add(zip_inner_name)
|
||
|
||
files_to_zip.append({
|
||
'full_path': full_path,
|
||
'inner_name': zip_inner_name
|
||
})
|
||
|
||
log_info(f'[下载API] 匹配 {total_matched} 条记录，成功打包 {len(files_to_zip)} 个文件，缺失 {len(failed_names)} 个文件', 'download')
|
||
if failed_names:
|
||
log_warning(f'[下载API] 缺失的文件列表: {failed_names[:20]}', 'download')
|
||
|
||
if len(files_to_zip) == 0:
|
||
return jsonify({'success': False, 'message': '没有可下载的文件（匹配的记录文件均不存在）'}), 400
|
||
|
||
# 生成zip文件名
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
zip_filename = f'批量导出_{timestamp}.zip'
|
||
zip_path = os.path.join(temp_dir, zip_filename)
|
||
|
||
# 打包
|
||
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||
for file_info in files_to_zip:
|
||
zf.write(file_info['full_path'], file_info['inner_name'])
|
||
|
||
log_info(f'[下载API] 按筛选批量打包完成，共 {len(files_to_zip)} 个文件，失败 {len(failed_names)} 个', 'download')
|
||
|
||
# 返回zip文件，通过自定义头部返回统计信息
|
||
from flask import send_file
|
||
response = send_file(
|
||
zip_path,
|
||
as_attachment=True,
|
||
download_name=zip_filename,
|
||
mimetype='application/zip'
|
||
)
|
||
# 通过响应头返回统计信息，便于前端诊断
|
||
response.headers['X-Total-Matched'] = str(total_matched)
|
||
response.headers['X-Files-Zipped'] = str(len(files_to_zip))
|
||
response.headers['X-Files-Missing'] = str(len(failed_names))
|
||
return response
|
||
except Exception as e:
|
||
log_error(f'[下载API] 按筛选批量下载错误: {e}', 'download')
|
||
import traceback
|
||
traceback.print_exc()
|
||
return jsonify({'success': False, 'message': str(e)}), 500
|
||
|
||
|
||
@download_bp.route('/download/batch-filter-count', methods=['POST'])
|
||
def batch_download_filter_count():
|
||
"""按筛选条件统计可导出的记录数
|
||
在批量导出前调用，让用户确认导出范围。
|
||
"""
|
||
try:
|
||
data = request.get_json(silent=True) or {}
|
||
config_name = data.get('config_name', '').strip()
|
||
start_date = data.get('start_date', '').strip()
|
||
end_date = data.get('end_date', '').strip()
|
||
|
||
# 构建查询条件（与批量导出完全一致，只统计成功记录）
|
||
where_clauses = ['status = 1']
|
||
params = []
|
||
|
||
if start_date:
|
||
where_clauses.append('DATE(start_time) >= %s')
|
||
params.append(start_date)
|
||
|
||
if end_date:
|
||
where_clauses.append('DATE(end_time) <= %s')
|
||
params.append(end_date)
|
||
|
||
if config_name:
|
||
where_clauses.append('config_name LIKE %s')
|
||
params.append(f'%{config_name}%')
|
||
|
||
where_sql = ' AND '.join(where_clauses)
|
||
|
||
# 统计总记录数
|
||
count_result = execute_query(
|
||
f'SELECT COUNT(*) as total FROM t_daily_report_history WHERE {where_sql}',
|
||
tuple(params)
|
||
)
|
||
total = count_result[0]['total'] if count_result else 0
|
||
|
||
# 统计日期范围
|
||
date_range = execute_query(
|
||
f'SELECT MIN(report_date) as min_date, MAX(report_date) as max_date FROM t_daily_report_history WHERE {where_sql}',
|
||
tuple(params)
|
||
)
|
||
min_date = date_range[0]['min_date'] if date_range else None
|
||
max_date = date_range[0]['max_date'] if date_range else None
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'total': total,
|
||
'min_date': str(min_date) if min_date else '',
|
||
'max_date': str(max_date) if max_date else ''
|
||
}
|
||
})
|
||
except Exception as e:
|
||
log_error(f'[下载API] 统计可导出记录数错误: {e}', 'download')
|
||
return jsonify({'success': False, 'message': str(e)}), 500
|