proposa-ai/bid_proposal/task_a1.md

A1 工作包：招标文件读取与需求分析

任务概述

实现 bid_proposal 包的前半部分：招标文件读取（bid_reader.py）和需求分析（requirement_analyzer.py），以及两个模块共享的基础设施（models.py、config.py、__init__.py）。

输出文件

文件	说明
bid_proposal/__init__.py	包入口（导出所有公共 API）
bid_proposal/models.py	所有数据类定义 + 自定义异常
bid_proposal/config.py	配置管理（API Key、模型名）
bid_proposal/bid_reader.py	招标文件读取
bid_proposal/requirement_analyzer.py	需求分析

---

模块实现详细要求

1. models.py — 数据类定义

定义以下数据类（使用 from __future__ import annotations）：

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional


# ---- 异常层次 ----
class BidProposalError(Exception):
    """所有 bid_proposal 异常的基类。"""
    pass

class BidReadError(BidProposalError):
    """PDF 读取/解析失败。"""
    pass

class AnalysisError(BidProposalError):
    """AI 分析失败（API 错误、响应格式异常等）。"""
    pass

class GenerationError(BidProposalError):
    """内容生成失败。"""
    pass

class WriteError(BidProposalError):
    """DOCX 输出失败。"""
    pass


# ---- 招标文件数据结构 ----
@dataclass
class BidPage:
    """招标文件中的一页"""
    page_num: int           # 页码（1-indexed）
    text: str               # 页面文本内容


@dataclass
class BidDocument:
    """招标文件结构化内容"""
    pdf_path: str                       # PDF 文件路径
    file_name: str                      # 文件名（不含路径）
    total_pages: int                    # PDF 总页数
    pages: list[BidPage]               # 所有页面文本
    full_text: str                      # 全文拼接文本（供 AI 分析使用）
    tables: list                        # list[TableInfo] — 来自 pdf_table_to_docx
    table_count: int                    # 表格数量


# ---- 需求分析数据结构 ----
@dataclass
class Requirement:
    """单个招标要求项"""
    category: str                       # 类别: evaluation / service / qualification / other
    title: str                          # 要求标题
    description: str                    # 详细描述
    score: Optional[float] = None       # 分值（评分项专用）
    detail: str = ""                    # 评分标准详细说明


@dataclass
class AnalysisResult:
    """招标文件分析结果"""
    summary: str                                    # 招标概要
    project_name: str = ""                          # 项目名称
    evaluation_criteria: list[Requirement] = field(default_factory=list)   # 评审标准（评分项）
    service_requirements: list[Requirement] = field(default_factory=list)  # 服务要求
    qualification_requirements: list[Requirement] = field(default_factory=list)  # 资质要求
    other_requirements: list[Requirement] = field(default_factory=list)    # 其他要求
    raw_response: str = ""                          # AI 原始响应内容（调试用）


# ---- 投标内容数据结构（A2 也会用到，由 A1 先行定义） ----
@dataclass
class ProposalSection:
    """投标文件的一个章节"""
    title: str                          # 章节标题
    level: int = 1                      # 层级（1=一级标题, 2=二级标题, ...）
    content: str = ""                   # 章节正文
    requirement_ref: str = ""           # 对应的招标要求引用


@dataclass
class ProposalDocument:
    """完整的投标文件内容"""
    title: str                          # 文档标题
    project_name: str = ""              # 项目名称
    sections: list[ProposalSection] = field(default_factory=list)  # 所有章节
    summary: str = ""                   # 投标概要

2. config.py — 配置管理

import os


class Config:
    """配置管理。"""
    
    @staticmethod
    def get_openai_api_key() -> str:
        """获取 OpenAI API Key。
        
        Returns:
            str: API Key
        
        Raises:
            ValueError: 未设置 OPENAI_API_KEY 环境变量
        """
        api_key = os.environ.get("OPENAI_API_KEY")
        if not api_key:
            raise ValueError(
                "未设置 OPENAI_API_KEY 环境变量。\n"
                "请通过 export OPENAI_API_KEY='sk-...' 设置"
            )
        return api_key
    
    @staticmethod
    def get_default_model() -> str:
        """获取默认模型名。"""
        return os.environ.get("OPENAI_MODEL", "gpt-4o")

3. bid_reader.py — 招标文件读取

核心函数

def read_bid_pdf(pdf_path: str) -> BidDocument:
    """读取招标 PDF，提取文本和表格。
    
    流程：
    1. 使用 pymupdf (fitz) 逐页提取文本
    2. 使用 pdf_table_to_docx.PDFTableExtractor 提取表格
    3. 合并为 BidDocument 返回
    
    Args:
        pdf_path: 招标文件 PDF 路径
    
    Returns:
        BidDocument: 结构化招标内容（含文本 + 表格）
    
    Raises:
        FileNotFoundError: PDF 文件不存在
        BidReadError: PDF 读取/解析失败
    """

实现步骤

1. 验证输入：

   if not os.path.exists(pdf_path):
      raise FileNotFoundError(f"PDF 文件不存在: {pdf_path}")
   

2. 提取文本（使用 pymupdf）：

   import fitz  # pymupdf Python 绑定
      
   doc = fitz.open(pdf_path)
   pages_data = []
   all_text_parts = []
      
   for page_num in range(doc.page_count):
      page = doc[page_num]
      text = page.get_text("text")
      pages_data.append(BidPage(page_num=page_num + 1, text=text))
      all_text_parts.append(text)
      
   full_text = "\n".join(all_text_parts)
   

3. 提取表格（复用 pdf_table_to_docx）：

   from pdf_table_to_docx.extractor import PDFTableExtractor
      
   try:
      extractor = PDFTableExtractor(pdf_path)
      tables = extractor.extract()
      table_count = len(tables)
   except Exception as e:
      logger.warning(f"表格提取失败（不影响文本提取）: {e}")
      tables = []
      table_count = 0
   

4. 构造返回：

   return BidDocument(
      pdf_path=os.path.abspath(pdf_path),
      file_name=os.path.basename(pdf_path),
      total_pages=len(pages_data),
      pages=pages_data,
      full_text=full_text,
      tables=tables,
      table_count=table_count,
   )
   

注意事项

• 编码问题：中文 PDF 需要确保 text 提取正确，pymupdf 默认处理 UTF-8
• 大型 PDF：超大文件（>100MB）注意内存使用
• 表格提取失败：降级处理，仅返回文本内容，记录警告日志
• 加密 PDF：fitz.open() 对加密 PDF 需要密码，捕获异常抛 BidReadError

日志记录

import logging
logger = logging.getLogger(__name__)
# 记录关键步骤：文件大小、页数、表格数等

4. requirement_analyzer.py — 需求分析

核心类

from openai import OpenAI
from .config import Config
from .models import BidDocument, AnalysisResult, Requirement, AnalysisError


class RequirementAnalyzer:
    """招标文件需求分析器。"""
    
    def __init__(self, api_key: str | None = None, model: str = "gpt-4o"):
        """初始化分析器。
        
        Args:
            api_key: OpenAI API Key，默认从环境变量读取
            model: 使用的模型名称（默认 gpt-4o）
        
        Raises:
            ValueError: API Key 未提供且环境变量未设置
        """
        api_key = api_key or Config.get_openai_api_key()
        self.client = OpenAI(api_key=api_key, timeout=120)
        self.model = model
    
    def analyze(self, doc: BidDocument) -> AnalysisResult:
        """对招标文件进行 AI 分析，提取所有要求。
        
        Args:
            doc: 招标文件结构化内容
        
        Returns:
            AnalysisResult: 分析结果（结构化需求列表 + 摘要）
        
        Raises:
            AnalysisError: AI 分析失败
        """

System Prompt 设计

SYSTEM_PROMPT = """你是一个专业的招标文件分析专家。请分析以下招标文件内容，提取并分类所有要求。

请严格按照以下 JSON Schema 输出（仅输出 JSON，不要添加任何其他文字）：

{
  "project_name": "项目名称（字符串）",
  "summary": "招标内容概要，不超过200字（字符串）",
  "evaluation_criteria": [
    {
      "title": "评审项名称",
      "description": "详细要求描述",
      "score": 分值（数字，或null如果没有明确分值）,
      "detail": "评分标准详细说明"
    }
  ],
  "service_requirements": [
    {
      "title": "服务要求名称",
      "description": "详细要求描述",
      "score": null,
      "detail": ""
    }
  ],
  "qualification_requirements": [...],
  "other_requirements": [...]
}

分类说明：
- evaluation（评审标准）：投标评审表中的评分项，通常带有明确分值
- service（服务要求）：需要投标人响应的具体服务内容或技术需求
- qualification（资质要求）：投标人必须满足的资格条件，如证书、注册资金等
- other（其他要求）：不属于以上三类的其他要求

注意：
- 尽量全面地提取所有要求，不要遗漏重要内容
- description 应尽量保留原文的关键信息
- 没有明确分值的项，score 设为 null
- 每个类别下的条目数量不限，按实际内容提取"""

实现步骤

1. 构造用户消息：

   # 将全文发送给 AI
   user_message = f"请分析以下招标文件：\n\n{doc.full_text}"
      
   # 如果文件过大（>100K字符），只发送前 100K 字符
   # 并在末尾说明截断
   MAX_CHARS = 100000
   if len(user_message) > MAX_CHARS:
      user_message = user_message[:MAX_CHARS]
      user_message += "\n\n[注意：原文过长，以上为前 100K 字符的截取内容]"
   

2. 调用 OpenAI API：

   with retry(max_attempts=3, delay=1.0, backoff=2.0):
      response = self.client.chat.completions.create(
          model=self.model,
          messages=[
              {"role": "system", "content": SYSTEM_PROMPT},
              {"role": "user", "content": user_message},
          ],
          response_format={"type": "json_object"},
          temperature=0.3,
      )
   

3. 解析响应：

   raw = response.choices[0].message.content
   data = json.loads(raw)
      
   evaluation = [Requirement(category="evaluation", **item) for item in data.get("evaluation_criteria", [])]
   service = [Requirement(category="service", **item) for item in data.get("service_requirements", [])]
   qualification = [Requirement(category="qualification", **item) for item in data.get("qualification_requirements", [])]
   other = [Requirement(category="other", **item) for item in data.get("other_requirements", [])]
      
   return AnalysisResult(
      summary=data.get("summary", ""),
      project_name=data.get("project_name", ""),
      evaluation_criteria=evaluation,
      service_requirements=service,
      qualification_requirements=qualification,
      other_requirements=other,
      raw_response=raw,
   )
   

4. 重试逻辑：

   import time
      
   def _call_with_retry(self, messages, max_attempts=3):
      last_error = None
      for attempt in range(max_attempts):
          try:
              return self.client.chat.completions.create(...)
          except Exception as e:
              last_error = e
              if attempt < max_attempts - 1:
                  wait = (2 ** attempt)
                  logger.warning(f"API 调用失败（第{attempt+1}次），{wait}秒后重试: {e}")
                  time.sleep(wait)
      raise AnalysisError(f"AI 分析失败（已重试{max_attempts}次）: {last_error}")
   

注意事项

• JSON 解析异常：有时 AI 返回的 JSON 可能格式不规范，需要做鲁棒性处理
• 空结果处理：如果某个分类没有提取到任何项，返回空列表而非 None
• 长文本截断：超过 token 限制时做截断处理，并在结果中标注
• 超时设置：API 客户端设置 120 秒超时
• 温度设置：使用 temperature=0.3 保证输出的一致性

---

5. __init__.py — 包入口

"""bid_proposal - AI 驱动的投标文件生成工具。

基于招标文件 PDF，自动分析需求并生成投标文件。
"""

from .models import (
    BidPage, BidDocument, Requirement, AnalysisResult,
    ProposalSection, ProposalDocument,
    BidProposalError, BidReadError, AnalysisError,
    GenerationError, WriteError,
)
from .bid_reader import read_bid_pdf
from .requirement_analyzer import RequirementAnalyzer

__all__ = [
    "BidPage", "BidDocument", "Requirement", "AnalysisResult",
    "ProposalSection", "ProposalDocument",
    "BidProposalError", "BidReadError", "AnalysisError",
    "GenerationError", "WriteError",
    "read_bid_pdf",
    "RequirementAnalyzer",
]
__version__ = "0.2.0"

---

接口契约（A1 → A2）

A1 工作包完成后的输出接口：

# --- 从 models.py ---
BidDocument          # 招标数据结构
Requirement          # 需求项
AnalysisResult       # 分析结果
ProposalSection      # 投标章节（A2 也会使用）
ProposalDocument     # 投标文档（A2 也会使用）
BidProposalError     # 异常基类
AnalysisError        # 分析异常

# --- 从 bid_reader.py ---
read_bid_pdf(pdf_path: str) -> BidDocument

# --- 从 requirement_analyzer.py ---
RequirementAnalyzer(api_key?: str, model?: str)
RequirementAnalyzer.analyze(doc: BidDocument) -> AnalysisResult

# --- 从 config.py ---
Config.get_openai_api_key() -> str
Config.get_default_model() -> str

---

验收标准

#	标准	验证方式
1	对测试 PDF 调用 read_bid_pdf() 能正确提取文本	assert doc.full_text 非空
2	对测试 PDF 调用 read_bid_pdf() 能提取表格	assert doc.table_count > 0
3	对不存在的 PDF 抛 FileNotFoundError	pytest.raises
4	对损坏的 PDF 抛 BidReadError	pytest.raises
5	API Key 缺失时抛 ValueError	pytest.raises
6	RequirementAnalyzer.analyze() 输出正确的结构	验证 AnalysisResult 各字段类型
7	分析结果能区分三类需求	检查 evaluation_criteria / service_requirements / qualification_requirements
8	Mock API 测试覆盖率 >= 80%	pytest --cov

---

测试建议

创建 bid_proposal/tests/test_a1.py，包含以下测试：

# 1. test_read_bid_pdf_success
#    - 使用测试 PDF 验证提取文本和表格
# 2. test_read_bid_pdf_not_found
#    - 文件不存在场景
# 3. test_requirement_analyzer_success
#    - Mock OpenAI API 响应，验证分析结果
# 4. test_requirement_analyzer_api_error
#    - API 调用失败后的重试行为
# 5. test_requirement_analyzer_no_api_key
#    - 未设置 API Key 时的错误提示
# 6. test_analysis_result_structure
#    - 分析结果的字段类型验证

---

参考文档

• 架构文档：bid_proposal/ARCHITECTURE.md
• 现有代码参考：pdf_table_to_docx/extractor.py、pdf_table_to_docx/table_parser.py
• 测试数据：data/松江区机关事务管理局物业管理服务招标文件.pdf
• 相关任务：A2 工作包（task_a2.md）