# -*- coding: utf-8 -*-
"""
标准库文档生成器（零 pip 依赖）
使用 zipfile + xml.etree.ElementTree 生成 .docx / .pptx / .xlsx 文件。
当 python-docx / python-pptx / openpyxl 未安装时作为 fallback。
依赖：Python 标准库 only（zipfile, xml.etree.ElementTree, io, os, datetime）
OOXML 格式说明：
.docx / .pptx / .xlsx 均为 ZIP 打包的 XML 文件（Office Open XML / ECMA-376）
无需 lxml，标准库 xml.etree.ElementTree 可完全满足。
"""
from __future__ import annotations
import zipfile
import io
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, List, Optional
try:
from report_exporter import EXPORT_DIR
except Exception:
EXPORT_DIR = Path(__file__).parent.parent / "data" / "exports"
# ─── 辅助工具 ────────────────────────────────────────────────────────────────
def _esc(s: Any) -> str:
"""XML 文本转义"""
if s is None:
return ""
return str(s).replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
def _attr(obj: Any, key: str, default: Any = "") -> Any:
"""兼容 object.attr 和 dict['attr'] 两种访问方式"""
if isinstance(obj, dict):
return obj.get(key, default)
return getattr(obj, key, default)
def _row_esc(s: Any) -> str:
"""单元格文本（允许换行）"""
return _esc(str(s) if s is not None else "")
# ─── DOCX 生成器 ────────────────────────────────────────────────────────────
class StdlibDocxWriter:
"""用标准库生成最小化 .docx（Word 2007+ 兼容）"""
def export_comprehensive_report(self, data: Dict[str, Any],
stock_code: str,
output_path: str = "") -> str:
"""
生成 Word 文档（.docx）
结构：
[Content_Types].xml
_rels/.rels
word/_rels/document.xml.rels
word/document.xml ← 标题 + 概览表 + 列表
word/styles.xml
"""
if not output_path:
os.makedirs(EXPORT_DIR, exist_ok=True)
output_path = str(EXPORT_DIR / f"{stock_code}_report.docx")
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
summary = data.get("summary", {})
stock_name = summary.get("stock_name", stock_code)
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# ── 构建 document.xml ────────────────────────────────────────────────
body_parts: List[str] = []
# 标题
body_parts.append(
f''
f''
f'{_esc(stock_name)} ({stock_code}) 综合分析报告'
)
# 概览标题
body_parts.append(
f''
f'报告概览'
)
# 概览表格（2列，7行）
info_rows = [
("股票代码", stock_code),
("股票名称", stock_name),
("定期报告", f"{summary.get('periodic_count', 0)} 份"),
("券商研报", f"{summary.get('broker_count', 0)} 份"),
("公告", f"{summary.get('announcement_count', 0)} 份"),
("买入评级研报", "是" if summary.get('has_buy_rating') else "否"),
("生成时间", now),
]
tbl_xml = (
''
''
''
''
''
''
''
''
)
for k, v in info_rows:
tbl_xml += (
''
f'{_esc(k)}'
''
f'{_esc(v)}'
)
tbl_xml += ''
body_parts.append(tbl_xml)
# 定期报告
periodic = data.get("periodic_reports", [])
if periodic:
body_parts.append(
f''
f'定期报告 ({len(periodic)} 份)'
)
for p in periodic:
title = _row_esc(_attr(p, 'title', ''))
date = str(_attr(p, 'publish_date', ''))[:10]
url = _row_esc(_attr(p, 'url', ''))
body_parts.append(
# v1.1 修复：补 闭合标签（之前漏了，导致下载保存时 Word 严格解析失败）
f''
f'{date} - {title}'
)
if url:
body_parts.append(
f''
f''
f'链接: {url}'
)
# 券商研报
broker = data.get("broker_reports", [])
if broker:
rating_stats = {"买入": 0, "增持": 0, "中性": 0, "减持": 0}
for b in broker:
r = _attr(b, 'rating', '')
if r in rating_stats:
rating_stats[r] += 1
body_parts.append(
f''
f'券商研报 ({len(broker)} 份)'
)
body_parts.append(
f'评级分布: '
f"买入{rating_stats['买入']} 增持{rating_stats['增持']} "
f"中性{rating_stats['中性']} 减持{rating_stats['减持']}"
)
for b in broker:
title = _row_esc(_attr(b, 'title', ''))
date = str(_attr(b, 'publish_date', ''))[:10]
broker_name = _row_esc(_attr(b, 'broker_name', ''))
analyst = _row_esc(_attr(b, 'analyst', ''))
rating = _row_esc(_attr(b, 'rating', ''))
tp = _attr(b, 'target_price', 0)
url = _row_esc(_attr(b, 'url', ''))
body_parts.append(
f''
f'{date} [{rating}] {broker_name}'
f'：{title}'
)
if analyst:
body_parts.append(
f''
f'分析师: {analyst}'
)
if tp and float(tp) > 0:
body_parts.append(
f''
f'目标价: ¥{float(tp):.2f}'
)
# v4.4.0: 添加原文链接
if url:
body_parts.append(
f''
f''
f'链接: {url}'
)
# 公告
announcements = data.get("announcements", [])
if announcements:
body_parts.append(
f''
f'公告 ({len(announcements)} 条)'
)
for a in announcements[:30]:
title = _row_esc(_attr(a, 'title', ''))
date = str(_attr(a, 'publish_date', ''))[:10]
ann_type = _row_esc(_attr(a, 'announcement_type', ''))
url = _row_esc(_attr(a, 'url', ''))
body_parts.append(
f''
f'{date} [{ann_type}]'
f' {title}'
)
if url:
body_parts.append(
f''
f''
f'链接: {url}'
)
body_xml = '\n'.join(body_parts)
# ── 构建各文件 ────────────────────────────────────────────────────────
ct_xml = (
'\n'
''
''
''
''
''
''
)
rels_xml = (
'\n'
''
''
''
)
doc_rels_xml = (
'\n'
''
''
''
)
styles_xml = (
'\n'
''
''
# v1.1 修复：完整字体定义（4 个属性齐全 + hint + lang）
# - ascii/hAnsi: 拉丁字符（用 universal 的 Times New Roman）
# - eastAsia: 东亚字符（用 universal 的 宋体 SimSun，几乎所有 Windows/Mac 都有）
# - cs: 复杂脚本（阿拉伯/泰文等）
# - hint: 告诉 Word 优先按 eastAsia 处理 CJK 字符
# - lang: 显式语言，让 Word 知道这段是中文
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
)
numbering_xml = (
'\n'
''
''
''
''
''
''
''
''
''
)
doc_body = (
'\n'
''
''
f'{body_xml}'
''
''
''
)
numbering_rels = (
'\n'
''
''
''
)
# ── 写入 ZIP ─────────────────────────────────────────────────────────
with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zf:
zf.writestr('[Content_Types].xml', ct_xml)
zf.writestr('_rels/.rels', rels_xml)
zf.writestr('word/_rels/document.xml.rels', doc_rels_xml)
zf.writestr('word/document.xml', doc_body)
zf.writestr('word/styles.xml', styles_xml)
zf.writestr('word/numbering.xml', numbering_xml)
zf.writestr('word/_rels/numbering.xml.rels', numbering_rels)
print(f"[标准库 Docx] 已生成: {output_path}")
return output_path
# ─── PPTX 生成器 ────────────────────────────────────────────────────────────
class StdlibPptxWriter:
"""用标准库生成最小化 .pptx（PowerPoint 2007+ 兼容）"""
def export_comprehensive_report(self, data: Dict[str, Any],
stock_code: str,
output_path: str = "") -> str:
"""
生成 PowerPoint 文档（.pptx）
结构：
[Content_Types].xml
_rels/.rels
ppt/_rels/presentation.xml.rels
ppt/presentation.xml
ppt/slideLayouts/slideLayout1.xml (title)
ppt/slideLayouts/slideLayout2.xml (title+body)
ppt/slides/slide1.xml (标题页)
ppt/slides/slide2.xml (概览)
ppt/slides/slide3.xml (定期报告)
ppt/slides/slide4.xml (券商研报)
ppt/slides/slide5.xml (公告)
ppt/theme/theme1.xml
"""
if not output_path:
os.makedirs(EXPORT_DIR, exist_ok=True)
output_path = str(EXPORT_DIR / f"{stock_code}_report.pptx")
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
summary = data.get("summary", {})
stock_name = summary.get("stock_name", stock_code)
now = datetime.now().strftime('%Y-%m-%d %H:%M')
slides: List[str] = []
# ── Slide 1: 标题页 ────────────────────────────────────────────────
slides.append(
'\n'
''
''
''
''
''
''
''
''
''
''
''
'{}'
''
''
''
'生成时间: {}'
''
''.format(
_esc(f"{stock_name} ({stock_code}) 综合报告"), _esc(now)
)
)
# ── Slide 2: 概览 ─────────────────────────────────────────────────
overview_lines = [
f"定期报告: {summary.get('periodic_count', 0)} 份",
f"券商研报: {summary.get('broker_count', 0)} 份",
f"公告: {summary.get('announcement_count', 0)} 份",
f"买入评级研报: {'是' if summary.get('has_buy_rating') else '否'}",
f"最新定期报告: {summary.get('latest_periodic_date', 'N/A')}",
f"最新券商研报: {summary.get('latest_broker_date', 'N/A')}",
f"最新公告: {summary.get('latest_announcement_date', 'N/A')}",
]
body_content = "\n".join(
f'{_esc(line)}'
for line in overview_lines
)
slides.append(
'\n'
''
''
''
''
''
''
''
''
'报告概览'
''
''
''
''
+ body_content +
''
''
)
# ── Slide 3: 定期报告 ───────────────────────────────────────────────
periodic = data.get("periodic_reports", [])
if periodic:
p_lines = [f"共 {len(periodic)} 份定期报告", ""]
for p in periodic[:10]:
p_url = str(_attr(p, 'url', ''))
p_entry = f"• {str(_attr(p,'publish_date',''))[:10]} - {_esc(_attr(p,'title',''))[:50]}"
if p_url:
p_entry += f" [{p_url}]" # v4.4.0: PPT 添加原文链接
p_lines.append(p_entry)
content = "\n".join(f'{_esc(l)}' for l in p_lines)
slides.append(_make_content_slide("定期报告", content, 3))
# ── Slide 4: 券商研报 ───────────────────────────────────────────────
broker = data.get("broker_reports", [])
if broker:
r_stats = {"买入": 0, "增持": 0, "中性": 0, "减持": 0}
for b in broker:
r = _attr(b, 'rating', '')
if r in r_stats:
r_stats[r] += 1
b_lines = [
f"共 {len(broker)} 份研报",
f"评级分布: 买入{r_stats['买入']} 增持{r_stats['增持']} 中性{r_stats['中性']} 减持{r_stats['减持']}",
""
]
for b in broker[:8]:
b_url = str(_attr(b, 'url', ''))
b_entry = f"• [{_attr(b,'rating','')}] {_esc(_attr(b,'broker_name',''))}: {_esc(_attr(b,'title',''))[:40]}"
if b_url:
b_entry += f" [{b_url}]" # v4.4.0: PPT 添加原文链接
b_lines.append(b_entry)
content = "\n".join(f'{_esc(l)}' for l in b_lines)
slides.append(_make_content_slide("券商研报", content, 4))
# ── Slide 5: 公告 ───────────────────────────────────────────────────
announcements = data.get("announcements", [])
if announcements:
a_lines = [f"共 {len(announcements)} 条公告", ""]
for a in announcements[:10]:
a_url = str(_attr(a, 'url', ''))
a_entry = f"• {str(_attr(a,'publish_date',''))[:10]} - {_esc(_attr(a,'title',''))[:50]}"
if a_url:
a_entry += f" [{a_url}]" # v4.4.0: PPT 添加原文链接
a_lines.append(a_entry)
content = "\n".join(f'{_esc(l)}' for l in a_lines)
slides.append(_make_content_slide("公告", content, 5))
# ── 构建 PPTX ───────────────────────────────────────────────────────
slide_xmls = "\n".join(slides)
slide_ids = "".join(
f''
for i in range(len(slides))
)
slide_rels = "".join(
f''
for i in range(len(slides))
)
ct_xml = (
'\n'
''
''
''
''
''
''
''
''
)
for i in range(len(slides)):
ct_xml += (
f''
)
ct_xml += ''
pres_rels = (
'\n'
''
''
+ "".join(
f''
for i in range(len(slides))
) +
''
''
)
pres_xml = (
'\n'
''
''
'' + slide_ids + ''
''
''
''
)
master_xml = (
'\n'
''
''
''
''
''
''
''
)
layout_xml = (
'\n'
''
''
''
''
''
''
''
)
theme_xml = (
'\n'
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
# v1.1 修复：CJK 字体从空改为"宋体"（universal），避免 Mac/Linux/旧 Windows 下载后乱码
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
)
with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zf:
zf.writestr('[Content_Types].xml', ct_xml)
zf.writestr('_rels/.rels', (
'\n'
''
''
''
))
zf.writestr('ppt/_rels/presentation.xml.rels', pres_rels)
zf.writestr('ppt/presentation.xml', pres_xml)
zf.writestr('ppt/slideMasters/slideMaster1.xml', master_xml)
zf.writestr('ppt/slideLayouts/slideLayout1.xml', layout_xml)
zf.writestr('ppt/slideLayouts/slideLayout2.xml', layout_xml)
zf.writestr('ppt/theme/theme1.xml', theme_xml)
for i, slide_xml in enumerate(slides):
zf.writestr(f'ppt/slides/slide{i+1}.xml', slide_xml)
zf.writestr('ppt/_rels/slideLayout1.xml.rels', (
'\n'
''
''
''
))
zf.writestr('ppt/_rels/slideLayout2.xml.rels', (
'\n'
''
''
''
))
print(f"[标准库 Pptx] 已生成: {output_path}")
return output_path
def _make_content_slide(title: str, content_xml: str, slide_id: int) -> str:
"""生成内容页幻灯片 XML"""
return (
'\n'
''
''
''
''
''
''
''
''
'{}'
''
''
''
''
+ content_xml +
''
''
''.format(_esc(title))
)
# ─── XLSX 生成器 ────────────────────────────────────────────────────────────
class StdlibXlsxWriter:
"""用标准库生成最小化 .xlsx（Excel 2007+ 兼容）"""
def export_comprehensive_report(self, data: Dict[str, Any],
stock_code: str,
output_path: str = "") -> str:
"""
生成 Excel 文档（.xlsx）
结构：
[Content_Types].xml
_rels/.rels
xl/workbook.xml
xl/_rels/workbook.xml.rels
xl/worksheets/sheet1.xml (概览)
xl/worksheets/sheet2.xml (定期报告)
xl/worksheets/sheet3.xml (券商研报)
xl/worksheets/sheet4.xml (公告)
xl/sharedStrings.xml
xl/styles.xml
"""
if not output_path:
os.makedirs(EXPORT_DIR, exist_ok=True)
output_path = str(EXPORT_DIR / f"{stock_code}_report.xlsx")
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
summary = data.get("summary", {})
stock_name = summary.get("stock_name", stock_code)
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
sheets: List[Dict] = []
# ── Sheet 1: 报告概览 ───────────────────────────────────────────────
s1_rows = [
["项目", "内容"],
["股票代码", stock_code],
["股票名称", stock_name],
["定期报告", f"{summary.get('periodic_count', 0)} 份"],
["券商研报", f"{summary.get('broker_count', 0)} 份"],
["公告", f"{summary.get('announcement_count', 0)} 份"],
["买入评级研报", "是" if summary.get('has_buy_rating') else "否"],
["最新定期报告", str(summary.get('latest_periodic_date', 'N/A'))],
["最新券商研报", str(summary.get('latest_broker_date', 'N/A'))],
["最新公告", str(summary.get('latest_announcement_date', 'N/A'))],
["生成时间", now],
]
sheets.append(("报告概览", s1_rows))
# ── Sheet 2: 定期报告 ───────────────────────────────────────────────
periodic = data.get("periodic_reports", [])
s2_rows = [["日期", "标题", "报告类型", "URL"]]
for p in periodic:
s2_rows.append([
str(_attr(p, 'publish_date', ''))[:10],
_row_esc(_attr(p, 'title', '')),
_row_esc(_attr(p, 'report_type', '')),
_row_esc(_attr(p, 'url', '')),
])
sheets.append(("定期报告", s2_rows if len(s2_rows) > 1 else [["无数据"]]))
# ── Sheet 3: 券商研报 ───────────────────────────────────────────────
broker = data.get("broker_reports", [])
s3_rows = [["日期", "标题", "券商", "分析师", "评级", "目标价", "目标涨幅", "URL"]]
for b in broker:
s3_rows.append([
str(_attr(b, 'publish_date', ''))[:10],
_row_esc(_attr(b, 'title', '')),
_row_esc(_attr(b, 'broker_name', '')),
_row_esc(_attr(b, 'analyst', '')),
_row_esc(_attr(b, 'rating', '')),
f"{float(_attr(b, 'target_price', 0)):.2f}" if _attr(b, 'target_price', 0) else "",
f"{float(_attr(b, 'target_change_pct', 0)):.2f}%" if _attr(b, 'target_change_pct', 0) else "",
_row_esc(_attr(b, 'url', '')),
])
sheets.append(("券商研报", s3_rows if len(s3_rows) > 1 else [["无数据"]]))
# ── Sheet 4: 公告 ───────────────────────────────────────────────────
announcements = data.get("announcements", [])
s4_rows = [["日期", "标题", "公告类型", "URL"]]
for a in announcements:
s4_rows.append([
str(_attr(a, 'publish_date', ''))[:10],
_row_esc(_attr(a, 'title', '')),
_row_esc(_attr(a, 'announcement_type', '')),
_row_esc(_attr(a, 'url', '')),
])
sheets.append(("公告", s4_rows if len(s4_rows) > 1 else [["无数据"]]))
# ── 构建共享字符串 ─────────────────────────────────────────────────
all_strings: List[str] = []
str_map: Dict[str, int] = {}
def _si(s: str) -> str:
nonlocal all_strings, str_map
s = str(s)
if s not in str_map:
str_map[s] = len(all_strings)
all_strings.append(s)
return str(str_map[s])
def _cell_row(vals: List[Any], row_idx: int, style: int = 0) -> str:
cells = ""
for col_idx, v in enumerate(vals):
c = chr(65 + col_idx) # A, B, C...
sv = _si(v) if isinstance(v, str) else str(v)
if isinstance(v, str):
cells += f'{sv}'
else:
cells += f'{sv}'
return f'{cells}
'
sheet_xmls: List[str] = []
for sheet_idx, (sheet_name, rows) in enumerate(sheets):
sheet_data = ""
for ri, row in enumerate(rows, 1):
sheet_data += _cell_row(row, ri, 1 if ri == 1 else 0)
sheet_xmls.append(
'\n'
''
'' + sheet_data + ''
)
shared_strings_xml = (
'\n'
''
+ "".join(f'{_esc(s)}' for s in all_strings)
+ ''
)
styles_xml = (
'\n'
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
''
)
sheet_rels = "".join(
f''
for i in range(len(sheets))
) + ''
workbook_xml = (
'\n'
''
''
+ "".join(
f''
for i, (name, _) in enumerate(sheets)
)
+ ''
)
ct_xml = (
'\n'
''
''
''
''
''
''
+ "".join(
f''
for i in range(len(sheets))
)
+ ''
)
# ── 写入 ZIP ─────────────────────────────────────────────────────────
with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zf:
zf.writestr('[Content_Types].xml', ct_xml)
zf.writestr('_rels/.rels', (
'\n'
''
''
''
))
zf.writestr('xl/_rels/workbook.xml.rels', (
'\n'
''
+ sheet_rels +
''
))
zf.writestr('xl/workbook.xml', workbook_xml)
zf.writestr('xl/styles.xml', styles_xml)
zf.writestr('xl/sharedStrings.xml', shared_strings_xml)
for i, sheet_xml in enumerate(sheet_xmls):
zf.writestr(f'xl/worksheets/sheet{i+1}.xml', sheet_xml)
print(f"[标准库 Xlsx] 已生成: {output_path}")
return output_path