From 7d5155052b40ee68a90b6493b7bc15d008a6ebda Mon Sep 17 00:00:00 2001 From: MarkLo Date: Tue, 16 Dec 2025 00:18:15 +0800 Subject: [PATCH] --- backend/app/api/routes.py | 37 +-- backend/app/services/download_service.py | 136 +++------ backend/app/services/pdf_generator.py | 369 +++++++++++++++++++++++ 3 files changed, 427 insertions(+), 115 deletions(-) diff --git a/backend/app/api/routes.py b/backend/app/api/routes.py index 85dc0767..69ab3aad 100644 --- a/backend/app/api/routes.py +++ b/backend/app/api/routes.py @@ -175,7 +175,12 @@ async def get_tickers(): @router.post("/download/reports") async def download_reports(request: DownloadRequest): """ - Download analyst reports as PDF or ZIP + Download all analyst reports as a single combined PDF + + Creates a professional PDF with: + - Cover page: Ticker + Analysis Date + - Table of Contents: Price chart + Analyst list + - All analyst reports as separate sections Supports two modes: 1. Task-based: If task_id is provided, lookup reports from task manager @@ -185,7 +190,7 @@ async def download_reports(request: DownloadRequest): request: Download request with ticker, date, analysts, and either task_id or direct reports Returns: - PDF file (single analyst) or ZIP file (multiple analysts) + Single combined PDF file (e.g., AVGO_Combined_Report_2025-12-16.pdf) """ from fastapi.responses import Response from backend.app.services.download_service import download_service @@ -261,27 +266,8 @@ async def download_reports(request: DownloadRequest): if not reports_to_download: raise HTTPException(status_code=404, detail="No reports found for selected analysts") - # Single report - return PDF - if len(reports_to_download) == 1: - pdf_bytes, filename = download_service.create_single_pdf( - analyst_name=reports_to_download[0]["analyst_name"], - ticker=request.ticker, - analysis_date=request.analysis_date, - report_content=reports_to_download[0]["report_content"], - price_data=price_data, - price_stats=price_stats, - ) - - return Response( - content=pdf_bytes, - media_type="application/pdf", - headers={ - "Content-Disposition": f"attachment; filename={filename}" - } - ) - - # Multiple reports - return ZIP - zip_bytes, filename = download_service.create_multiple_pdfs_zip( + # Always generate combined PDF + pdf_bytes, filename = download_service.create_combined_pdf( ticker=request.ticker, analysis_date=request.analysis_date, reports=reports_to_download, @@ -290,9 +276,10 @@ async def download_reports(request: DownloadRequest): ) return Response( - content=zip_bytes, - media_type="application/zip", + content=pdf_bytes, + media_type="application/pdf", headers={ "Content-Disposition": f"attachment; filename={filename}" } ) + diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 7c0e23c8..0ae139b2 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -1,9 +1,8 @@ """ Download Service for Analyst Reports -Handles single PDF and multiple PDF ZIP downloads +Generates combined PDF reports with all analyst analyses """ import io -import zipfile from typing import List, Dict, Optional from datetime import datetime @@ -43,59 +42,7 @@ class DownloadService: """Initialize download service""" self.pdf_generator = PDFGenerator() - def _get_english_name(self, analyst_name: str) -> str: - """ - 獲取分析師的英文名稱 - - Args: - analyst_name: 中文分析師名稱 - - Returns: - 英文分析師名稱 - """ - # 使用對照表，如果找不到則使用原名稱並替換空格 - return ANALYST_NAME_MAPPING.get(analyst_name, analyst_name.replace(" ", "_")) - - def create_single_pdf( - self, - analyst_name: str, - ticker: str, - analysis_date: str, - report_content: str, - price_data: list = None, - price_stats: dict = None, - ) -> tuple[bytes, str]: - """ - Create a PDF for a single analyst report - - Args: - analyst_name: Name of the analyst - ticker: Stock ticker symbol - analysis_date: Date of analysis (YYYY-MM-DD) - report_content: Markdown formatted report content - price_data: Optional list of price data for cover page - price_stats: Optional price statistics for cover page - - Returns: - Tuple of (PDF bytes, filename) - """ - # Generate PDF - pdf_bytes = self.pdf_generator.generate_analyst_report_pdf( - analyst_name=analyst_name, - ticker=ticker, - analysis_date=analysis_date, - report_content=report_content, - price_data=price_data, - price_stats=price_stats, - ) - - # Generate filename with English name: TICKER_English_Name_DATE.pdf - english_name = self._get_english_name(analyst_name) - filename = f"{ticker}_{english_name}_{analysis_date}.pdf" - - return pdf_bytes, filename - - def create_multiple_pdfs_zip( + def create_combined_pdf( self, ticker: str, analysis_date: str, @@ -104,53 +51,62 @@ class DownloadService: price_stats: dict = None, ) -> tuple[bytes, str]: """ - Create a ZIP file containing multiple analyst report PDFs + Create a single combined PDF containing all analyst reports + + Features: + - Cover page with ticker and analysis date + - Table of contents with price chart and analyst list + - All analyst reports as separate sections Args: ticker: Stock ticker symbol analysis_date: Date of analysis (YYYY-MM-DD) reports: List of dicts with keys 'analyst_name' and 'report_content' - price_data: Optional list of price data for cover page - price_stats: Optional price statistics for cover page + price_data: Optional list of price data for chart + price_stats: Optional price statistics for TOC Returns: - Tuple of (ZIP bytes, filename) + Tuple of (PDF bytes, filename) """ - # Create in-memory ZIP file - zip_buffer = io.BytesIO() + # Define the preferred order for analysts + analyst_order = [ + '市場分析師', + '基本面分析師', + '社群媒體分析師', + '新聞分析師', + '看漲研究員', + '看跌研究員', + '激進分析師', + '保守分析師', + '中立分析師', + '研究經理', + '風險經理', + '交易員', + ] - with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file: - for report in reports: - analyst_name = report.get('analyst_name', 'Unknown') - report_content = report.get('report_content', '') - - # Skip if no content - if not report_content: - continue - - # Generate PDF for this analyst - pdf_bytes = self.pdf_generator.generate_analyst_report_pdf( - analyst_name=analyst_name, - ticker=ticker, - analysis_date=analysis_date, - report_content=report_content, - price_data=price_data, - price_stats=price_stats, - ) - - # Add to ZIP with English filename - english_name = self._get_english_name(analyst_name) - pdf_filename = f"{ticker}_{english_name}_{analysis_date}.pdf" - zip_file.writestr(pdf_filename, pdf_bytes) + # Sort reports by preferred order + def get_order(report): + analyst_name = report.get('analyst_name', '') + try: + return analyst_order.index(analyst_name) + except ValueError: + return len(analyst_order) # Unknown analysts go to the end - # Get ZIP content - zip_bytes = zip_buffer.getvalue() - zip_buffer.close() + sorted_reports = sorted(reports, key=get_order) - # Generate ZIP filename: TICKER_DATE.zip - zip_filename = f"{ticker}_{analysis_date}.zip" + # Generate combined PDF + pdf_bytes = self.pdf_generator.generate_combined_report_pdf( + ticker=ticker, + analysis_date=analysis_date, + reports=sorted_reports, + price_data=price_data, + price_stats=price_stats, + ) - return zip_bytes, zip_filename + # Generate filename: TICKER_Combined_Report_DATE.pdf + filename = f"{ticker}_Combined_Report_{analysis_date}.pdf" + + return pdf_bytes, filename # Singleton instance diff --git a/backend/app/services/pdf_generator.py b/backend/app/services/pdf_generator.py index 0ae7b44a..8f735cd8 100644 --- a/backend/app/services/pdf_generator.py +++ b/backend/app/services/pdf_generator.py @@ -728,3 +728,372 @@ class PDFGenerator: text = text.replace(emoji, unicode_symbol) return text + + def generate_combined_report_pdf( + self, + ticker: str, + analysis_date: str, + reports: list, + price_data: list = None, + price_stats: dict = None, + ) -> bytes: + """ + Generate a combined PDF containing all analyst reports with cover page and table of contents + + Args: + ticker: Stock ticker symbol + analysis_date: Date of analysis + reports: List of dicts with 'analyst_name' and 'report_content' + price_data: Optional list of price data dicts + price_stats: Optional dict with price statistics + + Returns: + PDF file content as bytes + """ + from reportlab.platypus import TableOfContents, Paragraph, Spacer, PageBreak, Image + from reportlab.lib.styles import ParagraphStyle + from reportlab.lib.colors import HexColor + from reportlab.lib.units import cm + from reportlab.lib.pagesizes import A4 + from reportlab.platypus import SimpleDocTemplate + from reportlab.lib.enums import TA_CENTER, TA_LEFT + + buffer = io.BytesIO() + + # Create PDF document + doc = SimpleDocTemplate( + buffer, + pagesize=A4, + rightMargin=1.5*cm, + leftMargin=1.5*cm, + topMargin=1.5*cm, + bottomMargin=1.5*cm, + ) + + elements = [] + styles = self._get_styles() + + # === COVER PAGE === + elements.extend(self._create_cover_page(ticker, analysis_date, styles)) + elements.append(PageBreak()) + + # === TABLE OF CONTENTS PAGE === + elements.extend(self._create_toc_page(ticker, analysis_date, reports, price_data, price_stats, styles)) + elements.append(PageBreak()) + + # === ANALYST REPORTS === + for i, report in enumerate(reports): + analyst_name = report.get('analyst_name', 'Unknown') + report_content = report.get('report_content', '') + + if not report_content: + continue + + # Add analyst report section + elements.extend(self._create_analyst_section( + analyst_name=analyst_name, + ticker=ticker, + analysis_date=analysis_date, + report_content=report_content, + styles=styles, + )) + + # Page break between analysts (except for the last one) + if i < len(reports) - 1: + elements.append(PageBreak()) + + # Build PDF + doc.build(elements) + + pdf_content = buffer.getvalue() + buffer.close() + + return pdf_content + + def _get_styles(self): + """Get all paragraph styles for the combined PDF""" + from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle + from reportlab.lib.enums import TA_CENTER, TA_LEFT + from reportlab.lib.colors import HexColor + + styles = getSampleStyleSheet() + + custom_styles = { + 'cover_title': ParagraphStyle( + 'CoverTitle', + parent=styles['Heading1'], + fontName=self.primary_font, + fontSize=48, + textColor=HexColor('#1a1a1a'), + alignment=TA_CENTER, + spaceAfter=20, + wordWrap='CJK', + ), + 'cover_subtitle': ParagraphStyle( + 'CoverSubtitle', + parent=styles['Normal'], + fontName=self.primary_font, + fontSize=24, + textColor=HexColor('#666666'), + alignment=TA_CENTER, + spaceAfter=40, + wordWrap='CJK', + ), + 'cover_info': ParagraphStyle( + 'CoverInfo', + parent=styles['Normal'], + fontName=self.primary_font, + fontSize=14, + textColor=HexColor('#888888'), + alignment=TA_CENTER, + spaceAfter=10, + wordWrap='CJK', + ), + 'toc_title': ParagraphStyle( + 'TOCTitle', + parent=styles['Heading1'], + fontName=self.primary_font, + fontSize=28, + textColor=HexColor('#1a1a1a'), + alignment=TA_CENTER, + spaceAfter=30, + wordWrap='CJK', + ), + 'toc_section': ParagraphStyle( + 'TOCSection', + parent=styles['Heading2'], + fontName=self.primary_font, + fontSize=16, + textColor=HexColor('#2c3e50'), + spaceAfter=15, + spaceBefore=20, + wordWrap='CJK', + ), + 'toc_item': ParagraphStyle( + 'TOCItem', + parent=styles['Normal'], + fontName=self.primary_font, + fontSize=12, + textColor=HexColor('#444444'), + spaceAfter=8, + leftIndent=20, + wordWrap='CJK', + ), + 'section_title': ParagraphStyle( + 'SectionTitle', + parent=styles['Heading1'], + fontName=self.primary_font, + fontSize=24, + textColor=HexColor('#1a1a1a'), + spaceAfter=20, + alignment=TA_CENTER, + wordWrap='CJK', + ), + 'section_subtitle': ParagraphStyle( + 'SectionSubtitle', + parent=styles['Normal'], + fontName=self.primary_font, + fontSize=12, + textColor=HexColor('#666666'), + spaceAfter=15, + alignment=TA_CENTER, + wordWrap='CJK', + ), + 'heading': ParagraphStyle( + 'CustomHeading', + parent=styles['Heading2'], + fontName=self.primary_font, + fontSize=16, + textColor=HexColor('#2c3e50'), + spaceAfter=12, + spaceBefore=16, + wordWrap='CJK', + ), + 'body': ParagraphStyle( + 'CustomBody', + parent=styles['Normal'], + fontName=self.primary_font, + fontSize=9, + leading=14, + textColor=HexColor('#333333'), + spaceAfter=8, + wordWrap='CJK', + splitLongWords=True, + ), + 'stats_label': ParagraphStyle( + 'StatsLabel', + parent=styles['Normal'], + fontName=self.primary_font, + fontSize=10, + textColor=HexColor('#666666'), + spaceAfter=2, + wordWrap='CJK', + ), + 'stats_value': ParagraphStyle( + 'StatsValue', + parent=styles['Normal'], + fontName=self.primary_font, + fontSize=14, + textColor=HexColor('#1a1a1a'), + spaceAfter=10, + wordWrap='CJK', + ), + } + + return custom_styles + + def _create_cover_page(self, ticker: str, analysis_date: str, styles: dict) -> list: + """Create cover page elements""" + from reportlab.platypus import Spacer + from reportlab.lib.units import cm + + elements = [] + + # Add vertical space to center content + elements.append(Spacer(1, 6*cm)) + + # Main title: Stock ticker + elements.append(Paragraph(ticker, styles['cover_title'])) + + # Subtitle: Analysis date + elements.append(Paragraph(analysis_date, styles['cover_subtitle'])) + + # Additional info + elements.append(Spacer(1, 2*cm)) + elements.append(Paragraph("TradingAgentsX 分析報告", styles['cover_info'])) + elements.append(Paragraph("AI 驅動的多角度投資分析", styles['cover_info'])) + + return elements + + def _create_toc_page( + self, + ticker: str, + analysis_date: str, + reports: list, + price_data: list, + price_stats: dict, + styles: dict + ) -> list: + """Create table of contents page with price chart""" + from reportlab.platypus import Spacer, Image + from reportlab.lib.units import cm + from reportlab.lib.colors import HexColor + from reportlab.lib.styles import ParagraphStyle + + elements = [] + + # TOC Title + elements.append(Paragraph("目錄", styles['toc_title'])) + + # === Price Chart Section === + if price_data and len(price_data) >= 5: + elements.append(Paragraph("價格走勢圖 \u0026 交易量柱狀圖", styles['toc_section'])) + + try: + chart_bytes = self._generate_price_chart(price_data, ticker) + if chart_bytes: + chart_buffer = io.BytesIO(chart_bytes) + chart_img = Image(chart_buffer, width=16*cm, height=9.6*cm) + elements.append(chart_img) + elements.append(Spacer(1, 0.5*cm)) + except Exception as e: + print(f"Chart generation failed in TOC: {e}") + + # === Price Statistics === + if price_stats: + growth_rate = price_stats.get('growth_rate', 0) + growth_color = '#22c55e' if growth_rate >= 0 else '#ef4444' + growth_text = f"+{growth_rate:.2f}%" if growth_rate >= 0 else f"{growth_rate:.2f}%" + + growth_style = ParagraphStyle( + 'GrowthRate', + fontName=self.primary_font, + fontSize=16, + textColor=HexColor(growth_color), + spaceAfter=5, + ) + + elements.append(Paragraph(f"總報酬率：{growth_text}", growth_style)) + elements.append(Paragraph( + f"分析期間：{price_stats.get('duration_days', 'N/A')} 天 " + f"({price_stats.get('start_date', 'N/A')} ~ {price_stats.get('end_date', 'N/A')})", + styles['toc_item'] + )) + elements.append(Spacer(1, 0.3*cm)) + + # === Analyst List Section === + elements.append(Paragraph("分析師報告", styles['toc_section'])) + + # Group analysts by category + analyst_categories = { + '分析師組': ['市場分析師', '基本面分析師', '社群媒體分析師', '新聞分析師'], + '研究員組': ['看漲研究員', '看跌研究員'], + '風險辯論組': ['激進分析師', '保守分析師', '中立分析師'], + '決策組': ['研究經理', '風險經理', '交易員'], + } + + # Flatten list for page number tracking + analyst_order = [] + for category, analysts in analyst_categories.items(): + analyst_order.extend(analysts) + + # Create TOC items grouped by category + for category, analysts in analyst_categories.items(): + category_analysts = [r for r in reports if r.get('analyst_name') in analysts] + if category_analysts: + elements.append(Paragraph(f"• {category}", styles['toc_item'])) + for report in category_analysts: + analyst_name = report.get('analyst_name', 'Unknown') + elements.append(Paragraph(f" - {analyst_name}", styles['toc_item'])) + + return elements + + def _create_analyst_section( + self, + analyst_name: str, + ticker: str, + analysis_date: str, + report_content: str, + styles: dict + ) -> list: + """Create a single analyst report section""" + from reportlab.platypus import Spacer + from reportlab.lib.units import cm + + elements = [] + + # Section title + elements.append(Paragraph(analyst_name, styles['section_title'])) + + # Subtitle with ticker and date + elements.append(Paragraph(f"{ticker} | {analysis_date}", styles['section_subtitle'])) + elements.append(Spacer(1, 0.5*cm)) + + # Process report content + report_content = self._replace_emojis(report_content) + content = self._clean_markdown(report_content) + + # Split into paragraphs + paragraphs = content.split('\n') + + for para in paragraphs: + para = para.strip() + if not para: + elements.append(Spacer(1, 0.2*cm)) + continue + + # Check heading levels + if para.startswith('### '): + text = para[4:] + elements.append(Paragraph(text, styles['heading'])) + elif para.startswith('## '): + text = para[3:] + elements.append(Paragraph(text, styles['heading'])) + elif para.startswith('# '): + text = para[2:] + elements.append(Paragraph(text, styles['heading'])) + else: + text = self._escape_html(para) + elements.append(Paragraph(text, styles['body'])) + + return elements