import os from pathlib import Path from datetime import datetime from typing import Dict, Any from jinja2 import Environment, FileSystemLoader class PDFService: """PDF报告生成服务""" def __init__(self): # 模板目录 self.template_dir = Path(__file__).parent.parent / "templates" self.template_dir.mkdir(exist_ok=True) # 输出目录 self.output_dir = Path(__file__).parent.parent / "generated_reports" self.output_dir.mkdir(exist_ok=True) # 初始化Jinja2环境 self.jinja_env = Environment(loader=FileSystemLoader(str(self.template_dir))) def generate_report( self, filename: str, analysis: Dict[str, Any], llm_type: str = "Coze Workflow" ) -> str: """ 生成PDF报告 Args: filename: 原始文件名 analysis: 分析结果 llm_type: 使用的LLM类型 Returns: 生成的PDF文件路径 """ try: # 准备模板数据 template_data = self._prepare_template_data(filename, analysis, llm_type) # 渲染HTML html_content = self._render_html(template_data) # 生成PDF pdf_path = self._generate_pdf(html_content, filename) return pdf_path except Exception as e: raise Exception(f"PDF生成失败: {str(e)}") def _prepare_template_data( self, filename: str, analysis: Dict[str, Any], llm_type: str ) -> Dict[str, Any]: """准备模板数据""" # 处理 key_findings key_findings = analysis.get("key_findings", []) if key_findings: # 如果是对象数组，提取文本 key_findings = [ item.get("finding", item.get("text", str(item))) if isinstance(item, dict) else str(item) for item in key_findings ] # 处理 abnormal_items abnormal_items = analysis.get("abnormal_items", []) if abnormal_items: processed_items = [] for item in abnormal_items: if isinstance(item, dict): processed_items.append(item) else: processed_items.append({"name": str(item)}) abnormal_items = processed_items # 处理 risk_assessment risk_assessment = analysis.get("risk_assessment", "未提供") if isinstance(risk_assessment, dict): # 如果是对象，转换为文本 parts = [] if risk_assessment.get("high_risk"): parts.append(f"【高风险】{'; '.join(risk_assessment['high_risk'])}") if risk_assessment.get("medium_risk"): parts.append(f"【中风险】{'; '.join(risk_assessment['medium_risk'])}") if risk_assessment.get("low_risk"): parts.append(f"【低风险】{'; '.join(risk_assessment['low_risk'])}") risk_assessment = "\n".join(parts) if parts else "未检测到明确风险" # 处理 recommendations recommendations = analysis.get("recommendations", []) if recommendations: recommendations = [ item.get("recommendation", item.get("text", str(item))) if isinstance(item, dict) else str(item) for item in recommendations ] return { "filename": filename, "analysis_date": datetime.now().strftime("%Y年%m月%d日"), "llm_type": llm_type, "summary": analysis.get("summary", "暂无摘要"), "key_findings": key_findings, "abnormal_items": abnormal_items, "risk_assessment": risk_assessment, "recommendations": recommendations, "generation_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } def _render_html(self, template_data: Dict[str, Any]) -> str: """渲染HTML模板""" template = self.jinja_env.get_template("report_template.html") return template.render(**template_data) def _generate_pdf(self, html_content: str, original_filename: str) -> str: """将HTML转换为PDF""" try: # 生成PDF文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") base_name = Path(original_filename).stem pdf_filename = f"{base_name}_分析报告_{timestamp}.pdf" pdf_path = self.output_dir / pdf_filename # 尝试使用 WeasyPrint（推荐，质量更好） try: from weasyprint import HTML, CSS HTML(string=html_content).write_pdf( str(pdf_path), stylesheets=[CSS(string='@page { size: A4; margin: 1cm; }')] ) except ImportError: # 降级到 xhtml2pdf（更简单，无需额外依赖） print(" WeasyPrint 未安装，使用 xhtml2pdf 生成PDF...") from xhtml2pdf import pisa with open(pdf_path, "wb") as pdf_file: pisa_status = pisa.CreatePDF(html_content, dest=pdf_file) if pisa_status.err: raise Exception("xhtml2pdf 生成失败") return str(pdf_path) except Exception as e: raise Exception(f"PDF转换失败: {str(e)}") def get_pdf_file(self, pdf_path: str) -> bytes: """读取PDF文件内容""" if not os.path.exists(pdf_path): raise FileNotFoundError("PDF文件不存在") with open(pdf_path, "rb") as f: return f.read() def generate_comprehensive_report( self, patient_name: str, template_data: Dict[str, Any] ) -> str: """ 生成综合健康报告（多份报告整合） Args: patient_name: 患者姓名 template_data: 包含所有报告数据和分析结果的字典 Returns: 生成的PDF文件路径 """ try: # 准备综合报告模板数据 comprehensive_data = self._prepare_comprehensive_data(patient_name, template_data) # 渲染HTML（使用综合报告模板） html_content = self._render_comprehensive_html(comprehensive_data) # 生成PDF timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") pdf_filename = f"{patient_name}_综合健康报告_{timestamp}.pdf" pdf_path = self.output_dir / pdf_filename try: from weasyprint import HTML, CSS HTML(string=html_content).write_pdf( str(pdf_path), stylesheets=[CSS(string='@page { size: A4; margin: 1cm; }')] ) except ImportError: # 如果 WeasyPrint 不可用，使用 xhtml2pdf from xhtml2pdf import pisa with open(pdf_path, "wb") as pdf_file: pisa_status = pisa.CreatePDF(html_content, dest=pdf_file) if pisa_status.err: raise Exception("xhtml2pdf 生成失败") return str(pdf_path) except Exception as e: raise Exception(f"综合报告生成失败: {str(e)}") def _prepare_comprehensive_data( self, patient_name: str, template_data: Dict[str, Any] ) -> Dict[str, Any]: """准备综合报告模板数据""" analysis = template_data.get("analysis", {}) # 处理分析结果（与单报告相同的逻辑） key_findings = analysis.get("key_findings", []) if key_findings: key_findings = [ item.get("finding", item.get("text", str(item))) if isinstance(item, dict) else str(item) for item in key_findings ] abnormal_items = analysis.get("abnormal_items", []) if abnormal_items: processed_items = [] for item in abnormal_items: if isinstance(item, dict): processed_items.append(item) else: processed_items.append({"name": str(item)}) abnormal_items = processed_items risk_assessment = analysis.get("risk_assessment", "未提供") if isinstance(risk_assessment, dict): parts = [] if risk_assessment.get("high_risk"): parts.append(f"【高风险】{'; '.join(risk_assessment['high_risk'])}") if risk_assessment.get("medium_risk"): parts.append(f"【中风险】{'; '.join(risk_assessment['medium_risk'])}") if risk_assessment.get("low_risk"): parts.append(f"【低风险】{'; '.join(risk_assessment['low_risk'])}") risk_assessment = "\n".join(parts) if parts else "未检测到明确风险" recommendations = analysis.get("recommendations", []) if recommendations: recommendations = [ item.get("recommendation", item.get("text", str(item))) if isinstance(item, dict) else str(item) for item in recommendations ] return { "patient_name": patient_name, "report_count": template_data.get("report_count", 0), "report_list": template_data.get("report_list", []), "generation_date": template_data.get("generation_date", datetime.now().strftime("%Y年%m月%d日")), "summary": analysis.get("summary", "暂无摘要"), "key_findings": key_findings, "abnormal_items": abnormal_items, "risk_assessment": risk_assessment, "recommendations": recommendations, "generation_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } def _render_comprehensive_html(self, template_data: Dict[str, Any]) -> str: """渲染综合报告HTML模板""" template = self.jinja_env.get_template("comprehensive_report_template.html") return template.render(**template_data)