vent_deep_agent/tools/report_utils.py

# -*- coding: utf-8 -*-
"""
审查报告生成工具模块

功能：
- 将 Markdown 格式的审查报告转换为 Word (.docx) 文档
- 使用 pypandoc + 自定义样式模板（data/template.docx）
- 生成静态文件下载 URL

工具列表：
- convert_markdown_to_docx: Markdown → Word 文档转换
- get_docx_download_url: 生成 Word 文档的下载 URL
"""

import os
import re
import uuid
from pathlib import Path
from urllib.parse import quote


# ============================================================
# 配置常量
# ============================================================

# 项目根目录
PROJECT_ROOT = Path(__file__).parent.parent

# 样式模板路径
TEMPLATE_FILENAME = "template.docx"
TEMPLATE_PATH = str(PROJECT_ROOT / "data" / TEMPLATE_FILENAME)

# 报告输出目录（StaticFiles 挂载的静态目录）
REPORT_OUTPUT_DIR = PROJECT_ROOT / "static" / "reports"

# 服务器地址（读取 .env，无配置则用 localhost）
def _get_server_host() -> str:
    """从 .env 读取 SERVER_HOST，默认 localhost:8000。"""
    from dotenv import dotenv_values
    cfg = dotenv_values(str(PROJECT_ROOT / ".env"))
    host = cfg.get("HOST", "0.0.0.0")
    port = cfg.get("PORT", "8000")
    # 对外暴露用 localhost 或配置的域名
    return cfg.get("SERVER_HOST", f"http://{host}:{port}")


SERVER_HOST = _get_server_host()


# ============================================================
# Markdown → Word 转换
# ============================================================

def _normalize_markdown(md_content: str) -> str:
    """规范化 Markdown 内容，修复常见的 LLM 输出格式问题。

    修复以下问题：
    1. 标题行首有空格 → 去除（保证 ## 顶格）
    2. `##` 与标题文字之间缺少空格 → 补空格
    3. 标题前缺少空行 → 补空行
    4. 以 `**##` 或 `**###` 开头的伪标题 → 转为正常标题
    5. 全角 `＃` 替换为半角 `#`

    注意：不修改表格内部内容，避免破坏正确的 markdown 表格格式。

    Args:
        md_content: 原始 Markdown 内容

    Returns:
        str: 规范化后的 Markdown 内容
    """
    import re

    lines = md_content.split("\n")
    result = []
    in_table = False  # 跟踪是否在表格内

    for i, line in enumerate(lines):
        # 检测表格边界
        stripped_line = line.strip()
        if stripped_line.startswith("|") and stripped_line.endswith("|"):
            in_table = True
        elif stripped_line.startswith("|") and "---" in stripped_line:
            in_table = True  # 表格分隔行
        elif in_table and not stripped_line.startswith("|"):
            in_table = False

        # 1. 全角 # 替换为半角
        line = line.replace("\uff03", "#")

        # 表格内的行：跳过标题规范化处理，原样保留
        if in_table:
            result.append(line)
            continue

        # 2. 标题行首去空格 + 补空格：检测 ## 开头的行
        stripped = line.lstrip()
        # 匹配: 行首（可能有空格）的 # 标记，后面可选空格 + 文字
        if re.match(r"^#{1,6}(\s|$)", stripped) or re.match(r"^#{1,6}[^\s#]", stripped):
            # 确保 # 后面有空格（如 "##标题" → "## 标题"）
            fixed = re.sub(r"^(#{1,6})\s*", r"\1 ", stripped)
            fixed = fixed.rstrip()
            # 确保标题前有空行（避免标题粘在上文后）
            if result and result[-1].strip() != "":
                result.append("")
            result.append(fixed)
            continue

        # 3. 检测 **## 标题** 这类伪标题 → 转为 ## 标题
        bold_header = re.match(r"^\*\*(#{1,6})\s+(.+?)\*\*\s*$", line)
        if bold_header:
            fixed = f"{bold_header.group(1)} {bold_header.group(2).rstrip('*').strip()}"
            if result and result[-1].strip() != "":
                result.append("")
            result.append(fixed)
            continue

        # 4. 检测 **### 标题** 变体
        bold_header2 = re.match(r"^\*\*(#{1,6})\s*(.+?)\*\*\s*$", line)
        if bold_header2:
            fixed = f"{bold_header2.group(1)} {bold_header2.group(2).rstrip('*').strip()}"
            if result and result[-1].strip() != "":
                result.append("")
            result.append(fixed)
            continue

        result.append(line)

    return "\n".join(result)


def convert_markdown_to_docx(md_content: str, save_path: str) -> None:
    """将 Markdown 内容转换为 Word (.docx) 文档。

    使用 pypandoc 进行转换，应用 data/template.docx 作为样式模板。
    首次使用时会自动下载 pandoc。

    Args:
        md_content: Markdown 格式的报告内容
        save_path: 输出 .docx 文件的完整路径

    Raises:
        FileNotFoundError: 样式模板不存在
        RuntimeError: pandoc 转换失败
    """
    try:
        import pypandoc
    except ImportError:
        raise ImportError(
            "pypandoc 未安装。请执行: pip install pypandoc"
        )

    # 自动检测并下载 pandoc
    try:
        pypandoc.get_pandoc_path()
    except OSError:
        print("[报告] 未检测到 pandoc，正在自动下载...")
        pypandoc.download_pandoc()

    # 清理内容
    report_content = md_content.strip()
    if not report_content:
        raise ValueError("报告内容为空，无法生成文档")

    # 规范化 Markdown 格式（修复 LLM 输出中常见的排版问题）
    # report_content = _normalize_markdown(report_content)
    print("[报告] Markdown 规范化完成")

    # 检查模板
    if not os.path.exists(TEMPLATE_PATH):
        raise FileNotFoundError(
            f"样式模板不存在！路径：{TEMPLATE_PATH}\n"
            f"请将模板文件放置到 data/template.docx"
        )

    # 确保输出目录存在
    os.makedirs(os.path.dirname(save_path), exist_ok=True)

    # pandoc 转换参数
    extra_args = [
        "--standalone",
        f"--reference-doc={TEMPLATE_PATH}",
        "--from=gfm",               # GitHub Flavored Markdown，表格支持更好
        "--wrap=none",              # 禁止自动换行，保持表格单元格完整
        "--columns=120",            # 宽列宽，避免窄列强制断行
    ]

    try:
        pypandoc.convert_text(
            source=report_content,
            format="md",
            to="docx",
            outputfile=save_path,
            extra_args=extra_args,
            encoding="utf-8",
        )
        print(f"[报告] ✅ 转换成功！文件路径：{save_path}")
    except Exception as e:
        raise RuntimeError(f"pandoc 转换失败: {e}")


# ============================================================
# 下载 URL 生成
# ============================================================

def get_docx_download_url(word_filename: str) -> str:
    """生成 Word 文档的下载 URL。

    Args:
        word_filename: Word 文档文件名（不含路径），如 "report_abc123.docx"

    Returns:
        str: 完整下载 URL，如 "http://localhost:8000/static/reports/report_abc123.docx"
    """
    return f"{SERVER_HOST}/static/reports/{quote(word_filename)}"


def generate_report_filename(title: str = None) -> str:
    """生成审查报告文件名。

    Args:
        title: 报告标题（如煤矿名称），为空时使用默认名称

    Returns:
        str: 文件名，如 "XX煤矿配风计划审查报告_20260713_150132.docx"
    """
    from datetime import datetime
    date_str = datetime.now().strftime("%Y%m%d_%H%M%S")
    if title:
        # 清理标题中的非法文件名字符
        safe_title = re.sub(r'[\\/:*?"<>|]', '', title).strip()
        if safe_title:
            return f"{safe_title}配风计划审查报告_{date_str}.docx"
    return f"配风计划审查报告_{date_str}.docx"


def extract_mine_name(text: str) -> str | None:
    """从 PDF 文本中提取煤矿/公司名称。

    在文本前 5000 字中查找含"煤矿/煤业/矿业/能源/公司"的行，
    提取其中最可能的企业名称。

    Args:
        text: PDF 提取的文本内容

    Returns:
        str | None: 提取到的名称，未找到返回 None
    """
    import re
    head = text[:5000] if len(text) > 5000 else text

    # 先找含关键字的行（每行独立匹配，避免跨行粘连）
    lines = head.split('\n')
    candidates = []

    for line in lines:
        line = line.strip()
        if not line or len(line) < 4:
            continue

        # 匹配模式：任意前缀 + 煤矿/煤业/矿业 + 可选后缀
        m = re.search(
            r'([\u4e00-\u9fa5\w（）()\u3000]+?'
            r'(?:煤矿|煤业|矿业|能源)'
            r'(?:有限责任公司|股份有限公司|集团有限公司|有限公司|集团公司|集团|公司)?)',
            line
        )
        if m:
            name = m.group(1).strip()
            if len(name) >= 4:
                candidates.append(name)

        # 匹配 XX矿（非矿山/矿区/矿井/矿务局/矿长）
        m2 = re.search(
            r'([\u4e00-\u9fa5\w（）()\u3000]{4,20}矿)'
            r'(?!山|区|井|业|务|长|灯|车|石|泥|泉水)',
            line
        )
        if m2:
            name = m2.group(1).strip()
            if len(name) >= 4 and name not in candidates:
                candidates.append(name)

    if not candidates:
        return None

    # 返回最长匹配（通常最完整）
    candidates.sort(key=len, reverse=True)
    best = candidates[0]

    # 清理末尾多余字符
    best = re.sub(r'[，。；：、！？\s]+$', '', best)
    # 如果以"公司"结尾且前面有"有限/责任/集团"，确保完整性
    # 如果名称太长（>30字），截取到最后一个关键字处
    if len(best) > 40:
        # 尝试截断到 煤矿/煤业/矿业/公司 处
        trunc = re.match(r'(.{1,30}(?:煤矿|煤业|矿业|集团|公司))', best)
        if trunc:
            best = trunc.group(1)

    return best if len(best) >= 4 else None


def save_review_report(md_content: str, title: str = None) -> tuple[str, str]:
    """保存审查报告为 Word 文档，返回 (文件路径, 下载URL)。

    一站式函数：生成文件名 → 转换 docx → 返回路径和下载链接。

    Args:
        md_content: Markdown 格式的审查报告
        title: 报告标题（煤矿名称），用于生成文件名

    Returns:
        tuple[str, str]: (本地文件路径, 下载URL)
    """
    filename = generate_report_filename(title)
    save_path = str(REPORT_OUTPUT_DIR / filename)
    convert_markdown_to_docx(md_content, save_path)
    download_url = get_docx_download_url(save_path)
    return save_path, download_url


# ═══════════════════════════════════════════════════════════════
# Agent 工具：保存 Markdown 报告
# ═══════════════════════════════════════════════════════════════

async def save_report(content: str, filename: str = "") -> str:
    """将 Markdown 格式的分析报告保存为 .md 文件，并返回下载链接。
    
    当你需要生成通风分析报告、风量计算报告、数据解读报告等输出文档时调用此工具。
    直接传入完整的 Markdown 内容即可，工具会自动生成文件名并保存到报告目录。
    
    注意：你只需要调用一次本工具，将完整报告内容传入。不要尝试用 write_file
    或 edit_file 等文件系统工具来写文件，那些操作已被权限限制。
    
    Args:
        content: 完整的 Markdown 格式报告内容
        filename: 可选，文件名（不含 .md 扩展名）。
                  不传则自动生成：report_YYYYMMDD_HHMMSS.md
    
    Returns:
        JSON 格式，包含 success、file_path（本地路径）、filename、
        download_url（可通过浏览器访问的下载链接）。
    """
    import json
    from datetime import datetime, timezone, timedelta

    if not content or not content.strip():
        return json.dumps({"error": "报告内容不能为空"}, ensure_ascii=False)

    os.makedirs(str(REPORT_OUTPUT_DIR), exist_ok=True)

    # 生成文件名
    if filename and filename.strip():
        safe_name = filename.strip().replace("/", "_").replace("\\", "_")
        safe_name = re.sub(r"[<>:\"|?*]", "_", safe_name)
        md_filename = f"{safe_name}.md"
    else:
        now = datetime.now(timezone(timedelta(hours=8)))
        md_filename = now.strftime("report_%Y%m%d_%H%M%S.md")

    file_path = str(REPORT_OUTPUT_DIR / md_filename)

    try:
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(content.strip())

        # 构建下载 URL（文件名需 URL 编码，处理中文/特殊字符）
        download_url = f"{SERVER_HOST}/static/reports/{quote(md_filename)}"

        return json.dumps({
            "success": True,
            "file_path": file_path,
            "filename": md_filename,
            "download_url": download_url,
            "message": f"报告已保存为 {md_filename}，可通过 {download_url} 下载",
        }, ensure_ascii=False)
    except Exception as e:
        return json.dumps({
            "success": False,
            "error": f"保存报告失败: {str(e)}",
        }, ensure_ascii=False)