"""Report generation: Excel and PDF export."""
from io import BytesIO
from decimal import Decimal
from datetime import datetime
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment
from openpyxl.utils import get_column_letter
# WeasyPrint optional for PDF
try:
from weasyprint import HTML, CSS
HAS_WEASYPRINT = True
except ImportError:
HAS_WEASYPRINT = False
async def build_excel_report(case, transactions: list) -> bytes:
"""Build Excel workbook: summary sheet + transaction detail sheet. Returns file bytes."""
wb = Workbook()
ws_summary = wb.active
ws_summary.title = "汇总"
ws_summary.append(["案件编号", case.case_number])
ws_summary.append(["受害人", case.victim_name])
ws_summary.append(["总损失", str(case.total_loss)])
ws_summary.append(["交易笔数", len(transactions)])
total_out = sum(
(t.amount if isinstance(t.amount, Decimal) else Decimal(str(t.amount)))
for t in transactions
if t.transaction_type in ("转出", "消费", "付款", "提现") or "转出" in (t.transaction_type or "") or "消费" in (t.transaction_type or "")
)
total_in = sum(
(t.amount if isinstance(t.amount, Decimal) else Decimal(str(t.amount)))
for t in transactions
if t.transaction_type in ("转入", "收款", "充值") or "转入" in (t.transaction_type or "") or "收款" in (t.transaction_type or "")
)
ws_summary.append(["转出合计", str(total_out)])
ws_summary.append(["转入合计", str(total_in)])
ws_summary.append(["净损失", str(total_out - total_in)])
for row in range(1, 8):
ws_summary.cell(row=row, column=1).font = Font(bold=True)
ws_detail = wb.create_sheet("交易明细")
headers = ["APP来源", "类型", "金额", "币种", "对方名称", "对方账号", "订单号", "交易时间", "备注", "置信度"]
ws_detail.append(headers)
for t in transactions:
ws_detail.append([
t.app_source,
t.transaction_type or "",
str(t.amount),
t.currency or "CNY",
t.counterparty_name or "",
t.counterparty_account or "",
t.order_number or "",
t.transaction_time.isoformat() if t.transaction_time else "",
t.remark or "",
t.confidence or "",
])
for col in range(1, len(headers) + 1):
ws_detail.cell(row=1, column=col).font = Font(bold=True)
for col in range(1, ws_detail.max_column + 1):
ws_detail.column_dimensions[get_column_letter(col)].width = 16
buf = BytesIO()
wb.save(buf)
buf.seek(0)
return buf.getvalue()
def _pdf_html(case, transactions: list) -> str:
rows = []
for t in transactions:
time_str = t.transaction_time.strftime("%Y-%m-%d %H:%M") if t.transaction_time else ""
rows.append(
f"| {t.app_source} | {t.transaction_type or ''} | {t.amount} | "
f"{t.counterparty_name or ''} | {t.counterparty_account or ''} | {time_str} |
"
)
table_rows = "\n".join(rows)
return f"""
案件报告
资金追踪报告
案件编号：{case.case_number}
受害人：{case.victim_name}
总损失：{case.total_loss}
交易笔数：{len(transactions)}
交易明细
| APP | 类型 | 金额 | 对方名称 | 对方账号 | 时间 |
|---|
{table_rows}
"""
async def build_pdf_report(case, transactions: list) -> bytes:
"""Build PDF report. Returns file bytes. Falls back to empty PDF if weasyprint not available."""
if not HAS_WEASYPRINT:
return b"%PDF-1.4 (WeasyPrint not installed)"
html_str = _pdf_html(case, transactions)
html = HTML(string=html_str)
buf = BytesIO()
html.write_pdf(buf)
buf.seek(0)
return buf.getvalue()
async def build_excel_report_path(case, transactions: list, path: str) -> str:
"""Write Excel to file path; return path."""
data = await build_excel_report(case, transactions)
with open(path, "wb") as f:
f.write(data)
return path
async def build_pdf_report_path(case, transactions: list, path: str) -> str:
"""Write PDF to file path; return path."""
data = await build_pdf_report(case, transactions)
with open(path, "wb") as f:
f.write(data)
return path