This commit is contained in:
MarkLo 2025-12-16 00:18:15 +08:00
parent b47e79f44a
commit 7d5155052b
3 changed files with 427 additions and 115 deletions

View File

@ -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}"
}
)

View File

@ -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

View File

@ -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