feat: 优化PDF报告生成逻辑并添加调试脚本

移除对weasyprint_available的冗余检查,使用更高效的Count查询统计家庭事项
添加debug_pdf_email.py调试脚本用于直接测试PDF生成和邮件发送功能
This commit is contained in:
2026-01-19 23:24:04 +08:00
parent e3c9d6f17d
commit c22c001ee0
4 changed files with 286 additions and 4 deletions

View File

@@ -355,8 +355,7 @@ def celery_send_pdf_report_email(self):
logger.error(f"[任务 {task_id}] ❌ WeasyPrint库无法导入: {e}") logger.error(f"[任务 {task_id}] ❌ WeasyPrint库无法导入: {e}")
raise ValueError("WeasyPrint库无法导入PDF功能将不可用") raise ValueError("WeasyPrint库无法导入PDF功能将不可用")
if not weasyprint_available:
raise ValueError("WeasyPrint库不可用无法生成PDF报告")
# 从数据库获取邮件配置 # 从数据库获取邮件配置
from core.models import SystemConfig, ReadingRecord, InsightRecord, TodayPlan, FamilyTask from core.models import SystemConfig, ReadingRecord, InsightRecord, TodayPlan, FamilyTask
@@ -415,7 +414,8 @@ def celery_send_pdf_report_email(self):
# 获取家庭事项统计 # 获取家庭事项统计
logger.info(f"[任务 {task_id}] 📊 获取家庭事项统计...") logger.info(f"[任务 {task_id}] 📊 获取家庭事项统计...")
family_task_stats = FamilyTask.objects.values('type').annotate(count=FamilyTask.objects.values('id').filter(type=F('type')).count()) from django.db.models import Count
family_task_stats = FamilyTask.objects.values('type').annotate(count=Count('id'))
logger.info(f"[任务 {task_id}] ✅ 家庭事项统计: {len(family_task_stats)} 个类型") logger.info(f"[任务 {task_id}] ✅ 家庭事项统计: {len(family_task_stats)} 个类型")
# 准备上下文数据 # 准备上下文数据
@@ -445,7 +445,6 @@ def celery_send_pdf_report_email(self):
# 使用WeasyPrint生成PDF设置超时 # 使用WeasyPrint生成PDF设置超时
logger.info(f"[任务 {task_id}] 📄 正在生成PDF...") logger.info(f"[任务 {task_id}] 📄 正在生成PDF...")
from weasyprint import HTML
HTML(string=html_string).write_pdf(pdf_path, timeout=60) # 设置60秒超时 HTML(string=html_string).write_pdf(pdf_path, timeout=60) # 设置60秒超时
logger.info(f"[任务 {task_id}] ✅ PDF报告生成成功: {pdf_path}") logger.info(f"[任务 {task_id}] ✅ PDF报告生成成功: {pdf_path}")

View File

@@ -168,4 +168,21 @@
}); });
}); });
</script> </script>
<div class="card mt-4">
<div class="card-header bg-info text-white">
<h5 class="card-title mb-0">历史PDF报告</h5>
</div>
<div class="card-body">
<div class="row">
{% for date in historical_dates %}
<div class="col-md-3 mb-2">
<a href="{% url 'generate_pdf_report' date|date:'Y-m-d' %}" class="btn btn-outline-primary btn-sm btn-block">
{{ date }} PDF
</a>
</div>
{% endfor %}
</div>
</div>
</div>
{% endblock %} {% endblock %}

View File

@@ -435,6 +435,12 @@ def generate_report(request):
# 获取家庭事项统计 # 获取家庭事项统计
family_task_stats = FamilyTask.objects.values('type').annotate(count=Count('id')) family_task_stats = FamilyTask.objects.values('type').annotate(count=Count('id'))
# 获取过去30天的日期列表用于生成历史PDF链接
historical_dates = []
for i in range(1, 31):
historical_date = today - timedelta(days=i)
historical_dates.append(historical_date)
context = { context = {
'today': today, 'today': today,
'yesterday': yesterday, 'yesterday': yesterday,
@@ -442,6 +448,7 @@ def generate_report(request):
'yesterday_insight': yesterday_insight, 'yesterday_insight': yesterday_insight,
'today_plan': today_plan, 'today_plan': today_plan,
'family_task_stats': family_task_stats, 'family_task_stats': family_task_stats,
'historical_dates': historical_dates,
} }
return render(request, 'core/report.html', context) return render(request, 'core/report.html', context)

259
debug_pdf_email.py Normal file
View File

@@ -0,0 +1,259 @@
#!/usr/bin/env python
"""
PDF邮件发送调试脚本
用于在生产服务器上直接调试PDF生成和邮件发送功能无需依赖Celery
"""
import os
import sys
import time
from pathlib import Path
from loguru import logger
# 配置日志
logger.remove()
logger.add(
sys.stdout,
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
level="DEBUG"
)
logger.add(
"debug_pdf_email.log",
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
level="DEBUG",
rotation="1 day",
retention="7 days",
encoding="utf-8"
)
logger.info("=== PDF邮件发送调试脚本开始 ===")
logger.info(f"Python版本: {sys.version}")
logger.info(f"当前目录: {os.getcwd()}")
# 初始化Django环境
def init_django():
"""初始化Django环境"""
logger.info("初始化Django环境...")
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
import django
django.setup()
logger.info("✅ Django环境初始化成功")
# 测试WeasyPrint安装
def test_weasyprint():
"""测试WeasyPrint是否正确安装"""
logger.info("测试WeasyPrint安装...")
try:
from weasyprint import HTML
logger.info("✅ WeasyPrint导入成功")
return True
except ImportError as e:
logger.error(f"❌ WeasyPrint导入失败: {e}")
logger.error("解决方案: 运行 'pip install weasyprint' 安装WeasyPrint库")
return False
except Exception as e:
logger.error(f"❌ WeasyPrint测试失败: {e}")
return False
# 测试生成PDF
def test_pdf_generation():
"""测试生成PDF"""
logger.info("测试生成PDF...")
try:
from django.conf import settings
from django.template.loader import render_to_string
from weasyprint import HTML
from core.models import SystemConfig, ReadingRecord, InsightRecord, TodayPlan, FamilyTask
from django.utils import timezone
from datetime import timedelta
# 准备数据
today = timezone.now().date()
yesterday = today - timedelta(days=1)
logger.info(f"报告日期: {today}, 昨日: {yesterday}")
# 获取数据
yesterday_reading = ReadingRecord.objects.filter(date=yesterday)
yesterday_insight = InsightRecord.objects.filter(date=yesterday)
today_plan = TodayPlan.objects.filter(date=today)
logger.info(f"昨日阅读记录: {yesterday_reading.count()}")
logger.info(f"昨日感悟记录: {yesterday_insight.count()}")
logger.info(f"今日计划: {today_plan.count()}")
# 准备上下文
context = {
'today': today,
'yesterday': yesterday,
'yesterday_reading': yesterday_reading,
'yesterday_insight': yesterday_insight,
'today_plan': today_plan,
}
# 渲染模板
logger.info("渲染HTML模板...")
html_string = render_to_string('core/report_pdf.html', context)
logger.info("✅ HTML模板渲染成功")
# 生成PDF
pdf_file = f"debug_report_{today.strftime('%Y-%m-%d')}.pdf"
pdf_path = os.path.join(settings.REPORTS_ROOT, pdf_file)
logger.info(f"生成PDF文件: {pdf_path}")
os.makedirs(settings.REPORTS_ROOT, exist_ok=True)
HTML(string=html_string).write_pdf(pdf_path, timeout=60)
logger.info(f"✅ PDF生成成功文件大小: {os.path.getsize(pdf_path) / 1024:.2f} KB")
return pdf_path
except Exception as e:
logger.error(f"❌ PDF生成失败: {e}")
import traceback
logger.error(f"错误详情:\n{traceback.format_exc()}")
return None
# 测试邮件发送
def test_email_sending(pdf_path=None):
"""测试邮件发送"""
logger.info("测试邮件发送...")
try:
from django.core.mail import EmailMessage
from django.core.mail.backends.smtp import EmailBackend
from core.models import SystemConfig
from django.utils import timezone
# 获取配置
config = SystemConfig.get_config()
# 验证配置
required_fields = ['smtp_server', 'smtp_username', 'smtp_password', 'recipient_email']
missing_fields = []
for field in required_fields:
if not getattr(config, field):
missing_fields.append(field)
if missing_fields:
logger.error(f"❌ 缺少必要的邮件配置: {', '.join(missing_fields)}")
logger.error("解决方案: 登录管理后台,配置邮件设置")
return False
# 准备邮件
from_email = config.sender_email or config.smtp_username
to_email = config.recipient_email
subject = f"[调试] 家庭日报 {timezone.now().date()} - PDF报告"
body = f"""
这是一封调试邮件用于测试PDF生成和邮件发送功能。
发送时间: {timezone.now().strftime('%Y-%m-%d %H:%M:%S')}
报告日期: {timezone.now().date()}
这封邮件是从调试脚本直接发送的没有经过Celery队列。
"""
# 创建SMTP连接
logger.info(f"创建SMTP连接: {config.smtp_server}:{config.smtp_port}")
logger.info(f"发件人: {from_email}")
logger.info(f"收件人: {to_email}")
backend = EmailBackend(
host=config.smtp_server,
port=config.smtp_port or 587,
username=config.smtp_username,
password=config.smtp_password,
use_tls=True,
fail_silently=False,
timeout=30
)
# 创建邮件
email = EmailMessage(
subject=subject,
body=body,
from_email=from_email,
to=[to_email],
connection=backend
)
# 添加附件
if pdf_path and os.path.exists(pdf_path):
logger.info(f"添加PDF附件: {pdf_path}")
with open(pdf_path, 'rb') as f:
email.attach(os.path.basename(pdf_path), f.read(), 'application/pdf')
# 发送邮件
logger.info("发送邮件...")
sent_count = email.send(fail_silently=False)
logger.info(f"邮件发送返回: {sent_count}")
if sent_count > 0:
logger.success("✅ 邮件发送成功!")
return True
else:
logger.error("❌ 邮件发送失败")
return False
except Exception as e:
logger.error(f"❌ 邮件发送失败: {e}")
import traceback
logger.error(f"错误详情:\n{traceback.format_exc()}")
return False
# 测试Redis连接
def test_redis_connection():
"""测试Redis连接"""
logger.info("测试Redis连接...")
try:
import redis
from django.conf import settings
redis_url = settings.CELERY_BROKER_URL
logger.info(f"Redis URL: {redis_url}")
# 测试连接
client = redis.from_url(redis_url, socket_connect_timeout=10, socket_timeout=10)
pong = client.ping()
logger.info(f"Redis ping: {pong}")
logger.info(f"Redis版本: {client.info().get('redis_version', '未知')}")
client.close()
logger.info("✅ Redis连接成功")
return True
except Exception as e:
logger.error(f"❌ Redis连接失败: {e}")
return False
# 主函数
def main():
"""主函数"""
logger.info("=== PDF邮件发送调试开始 ===")
# 初始化Django
init_django()
# 测试Redis连接
logger.info("\n1. 测试Redis连接")
test_redis_connection()
# 测试WeasyPrint
logger.info("\n2. 测试WeasyPrint")
if not test_weasyprint():
logger.error("WeasyPrint测试失败无法继续")
return 1
# 测试生成PDF
logger.info("\n3. 测试生成PDF")
pdf_path = test_pdf_generation()
if not pdf_path:
logger.error("PDF生成失败无法继续邮件发送测试")
else:
# 测试发送邮件
logger.info("\n4. 测试发送邮件")
test_email_sending(pdf_path)
logger.info("\n=== PDF邮件发送调试结束 ===")
return 0
if __name__ == "__main__":
sys.exit(main())