proposa-ai/bid_proposal/requirement_analyzer.py

"""
requirement_analyzer.py - 招标要求分析模块

使用 OpenAI API 分析招标文件内容，识别并分类其中的各项要求。
"""

import json
import logging
import time
from typing import Optional

from .config import get_config
from .models import AnalysisResult, BidDocument, Requirement
from .utils import chunk_text, table_to_markdown

logger = logging.getLogger(__name__)


class AnalysisError(Exception):
    """AI 分析失败异常"""
    pass


class RequirementAnalyzer:
    """招标文件需求分析器。

    使用 OpenAI 分析招标文件中的各项要求，包括评分标准、服务要求、
    资质要求等，返回结构化的分析结果。
    """

    CATEGORY_MAP = {
        "evaluation": "评审标准",
        "service": "服务要求",
        "qualification": "资质要求",
        "other": "其他",
    }
    REVERSE_CATEGORY_MAP = {v: k for k, v in CATEGORY_MAP.items()}

    def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4o"):
        """初始化分析器。

        Args:
            api_key: OpenAI API Key，默认从环境变量读取
            model: 使用的模型名称
        """
        try:
            from openai import OpenAI
        except ImportError:
            raise ImportError("请安装 openai: pip install openai")

        key = api_key or get_config().openai_api_key
        if not key:
            raise ValueError(
                "未提供 OpenAI API Key。请通过参数传入或设置 OPENAI_API_KEY 环境变量。"
            )
        self.client = OpenAI(api_key=key)
        self.model = model

    def analyze(self, doc: BidDocument) -> AnalysisResult:
        """对招标文件进行 AI 分析，提取所有要求。

        Args:
            doc: 招标文件结构化内容

        Returns:
            AnalysisResult: 分析结果（结构化需求列表 + 摘要）

        Raises:
            AnalysisError: AI 分析失败
        """
        full_text = doc.full_text.strip()
        if not full_text:
            raise AnalysisError("招标文件内容为空，无法进行分析")

        logger.info("开始分析招标文件: %s (%d 字符)", doc.pdf_path, len(full_text))

        config = get_config()
        max_length = config.max_text_length

        if len(full_text) <= max_length:
            prompt = self._build_prompt(doc)
            result = self._call_api(prompt)
            logger.info(
                "分析完成: 评审标准 %d 项, 服务要求 %d 项, 资质要求 %d 项, 其他 %d 项",
                len(result.evaluation_criteria),
                len(result.service_requirements),
                len(result.qualification_requirements),
                len(result.other_requirements),
            )
            return result

        # 文本过长，分块处理
        logger.info("文本过长 (%d 字符，超过 %d)，进行分块分析", len(full_text), max_length)
        return self._analyze_by_chunks(doc, full_text, max_length)

    def _analyze_by_chunks(
        self, doc: BidDocument, full_text: str, max_length: int
    ) -> AnalysisResult:
        """将长文本分块后分别分析，再合并结果。"""
        text_chunks = chunk_text(full_text, max_length)
        logger.info("文本已切分为 %d 个块", len(text_chunks))

        all_reqs: list[Requirement] = []
        summaries: list[str] = []
        project_names: list[str] = []

        for i, chunk in enumerate(text_chunks):
            logger.info("分析第 %d/%d 块...", i + 1, len(text_chunks))
            partial_prompt = self._build_chunk_prompt(chunk, i + 1, len(text_chunks))
            try:
                result = self._call_api(partial_prompt)
                all_reqs.extend(result.all_requirements)
                summaries.append(result.summary)
                if result.project_name:
                    project_names.append(result.project_name)
            except AnalysisError as e:
                logger.warning("第 %d 块分析失败: %s，跳过该块", i + 1, e)

        # 合并结果，按类别分类
        combined = AnalysisResult(
            summary="\n".join(summaries),
            project_name=project_names[0] if project_names else "",
            evaluation_criteria=[r for r in all_reqs if r.category == "evaluation"],
            service_requirements=[r for r in all_reqs if r.category == "service"],
            qualification_requirements=[r for r in all_reqs if r.category == "qualification"],
            other_requirements=[r for r in all_reqs if r.category == "other"],
        )
        logger.info(
            "分块分析完成: %d 项要求",
            combined.total_requirements,
        )
        return combined

    def _build_chunk_prompt(self, chunk: str, chunk_num: int, total_chunks: int) -> str:
        """为单个文本块构建分析 prompt。"""
        return f"""=== 招标文件内容（第 {chunk_num}/{total_chunks} 部分）===

{chunk}

---

请分析以上招标文件内容片段，提取其中包含的所有要求。"""

    def _build_prompt(self, doc: BidDocument) -> str:
        """构建包含招标文件完整内容的 prompt。

        将文本和表格按页组织，方便 AI 理解内容结构。
        """
        sections: list[str] = ["=== 招标文件内容 ==="]

        # 构建表格索引：页码 → 表格索引
        table_idx_by_page: dict[int, list[int]] = {}
        for ti, table in enumerate(doc.tables):
            page_num = getattr(table, 'page_num', 0) + 1  # page_num是0-indexed
            table_idx_by_page.setdefault(page_num, []).append(ti)

        for page in doc.pages:
            page_num = page.page_num
            sections.append(f"\n--- 第 {page_num} 页 ---")

            if page.text:
                sections.append(page.text)
            else:
                sections.append("（本页无文本内容）")

            # 添加本页中的表格
            page_table_indices = table_idx_by_page.get(page_num, [])
            for ti in page_table_indices:
                if ti < len(doc.tables):
                    table = doc.tables[ti]
                    md_table = table_to_markdown(table)
                    if md_table and md_table != "（空表格）":
                        sections.append(f"\n本页中的表格：\n{md_table}")

        return "\n".join(sections)

    def _call_api(self, user_prompt: str) -> AnalysisResult:
        """调用 OpenAI API 进行需求分析，使用结构化输出。"""
        system_prompt = """你是一个专业的招标文件分析专家。请仔细阅读招标文件内容，提取并分类其中的所有要求。

请识别以下类别的要求：
1. **evaluation（评审标准）**：评审因素、评分项、分值分配、评审标准等
2. **service（服务要求）**：服务范围、服务内容、服务标准、人员配置等
3. **qualification（资质要求）**：投标人资格条件、资质证书、注册资金、业绩要求等
4. **other（其他）**：商务条款、技术规范、预算金额、付款方式、服务期限等

对于每一项要求，尽可能提取其标题(title)、详细描述(description)、分值(score)和评分标准详情(detail)。"""

        max_retries = 2
        last_error: Optional[Exception] = None

        for attempt in range(max_retries + 1):
            try:
                response = self.client.chat.completions.create(
                    model=self.model,
                    messages=[
                        {"role": "system", "content": system_prompt},
                        {"role": "user", "content": user_prompt},
                    ],
                    temperature=get_config().openai_temperature,
                    max_tokens=get_config().openai_max_tokens,
                    timeout=get_config().openai_timeout,
                    response_format={
                        "type": "json_schema",
                        "json_schema": {
                            "name": "analysis_result",
                            "strict": True,
                            "schema": {
                                "type": "object",
                                "properties": {
                                    "summary": {
                                        "type": "string",
                                        "description": "招标文件概要总结，不超过300字",
                                    },
                                    "project_name": {
                                        "type": "string",
                                        "description": "项目名称",
                                    },
                                    "requirements": {
                                        "type": "array",
                                        "items": {
                                            "type": "object",
                                            "properties": {
                                                "category": {
                                                    "type": "string",
                                                    "enum": ["evaluation", "service", "qualification", "other"],
                                                    "description": "要求类别",
                                                },
                                                "title": {
                                                    "type": "string",
                                                    "description": "要求标题",
                                                },
                                                "description": {
                                                    "type": "string",
                                                    "description": "详细描述",
                                                },
                                                "score": {
                                                    "type": ["number", "null"],
                                                    "description": "分值（如明确有分值）",
                                                },
                                                "detail": {
                                                    "type": "string",
                                                    "description": "评分标准详细说明",
                                                },
                                            },
                                            "required": ["category", "title", "description", "score", "detail"],
                                            "additionalProperties": False,
                                        },
                                    },
                                },
                                "required": ["summary", "project_name", "requirements"],
                                "additionalProperties": False,
                            },
                        },
                    },
                )

                content = response.choices[0].message.content
                if not content:
                    raise AnalysisError("API 返回了空响应")

                data = json.loads(content)

                # 按类别分类构建 Requirement
                all_reqs: list[Requirement] = []
                for req in data.get("requirements", []):
                    category = req.get("category", "other")
                    # 校验category合法性
                    if category not in ("evaluation", "service", "qualification", "other"):
                        category = "other"
                    all_reqs.append(Requirement(
                        category=category,
                        title=req.get("title", ""),
                        description=req.get("description", ""),
                        score=req.get("score"),
                        detail=req.get("detail", ""),
                    ))

                result = AnalysisResult(
                    summary=data.get("summary", ""),
                    project_name=data.get("project_name", ""),
                    evaluation_criteria=[r for r in all_reqs if r.category == "evaluation"],
                    service_requirements=[r for r in all_reqs if r.category == "service"],
                    qualification_requirements=[r for r in all_reqs if r.category == "qualification"],
                    other_requirements=[r for r in all_reqs if r.category == "other"],
                )
                logger.debug("API 分析成功: %d 项要求", result.total_requirements)
                return result

            except Exception as e:
                last_error = e
                if attempt < max_retries:
                    wait = 2 ** (attempt + 1)
                    logger.warning(
                        "API 调用失败（第 %d 次），%ds 后重试: %s",
                        attempt + 1, wait, e,
                    )
                    time.sleep(wait)
                else:
                    logger.error("API 调用最终失败: %s", e)

        raise AnalysisError(f"AI 分析失败，已重试 {max_retries} 次: {last_error}")