GX-gp-notify/app/services/ai_analyzer.py

import json
import logging
from dataclasses import dataclass
from typing import Any

import httpx

from app.config import settings
from app.crawler.parsers import extract_page_content

logger = logging.getLogger(__name__)


@dataclass
class AiResult:
    """DeepSeek 分析结果"""
    is_relevant: bool = False
    reason: str = ""
    business_type: str = ""
    error: str | None = None
    content_snippet: str | None = None  # 提取到的正文前 200 字，供入库参考


class AiAnalyzer:
    """AI 分析器 — 调用 DeepSeek 判断公告是否为中国电信可承接项目"""

    def __init__(self):
        self.api_key = settings.ai_api_key
        self.base_url = settings.ai_base_url.rstrip("/")
        self.model = settings.ai_model
        self.timeout = settings.ai_timeout
        self.prompt_template = settings.ai_prompt_template

    async def analyze(self, announcement: dict[str, Any]) -> AiResult:
        """分析单条公告"""
        if not self.api_key:
            return AiResult(error="AI_API_KEY 未配置")

        # 1. 获取公告正文
        content_url = announcement.get("content_url", "")
        content = None
        content_snippet = None
        if content_url:
            content = await extract_page_content(content_url, self.timeout)

        if content:
            content_snippet = content[:200]
        else:
            logger.warning("无法获取公告正文: %s", content_url)

        # 2. 构建 prompt
        prompt = self.prompt_template.format(
            title=announcement.get("title", ""),
            purchase_name=announcement.get("purchase_name", ""),
            announcement_type=announcement.get("announcement_type", ""),
            content=content or "（无法获取正文，请仅根据标题和采购人信息判断）",
        )

        # 3. 调用 DeepSeek API
        try:
            result = await self._call_deepseek(prompt)
            if result.error:
                return AiResult(error=result.error, content_snippet=content_snippet)
            return AiResult(
                is_relevant=result.is_relevant,
                reason=result.reason,
                business_type=result.business_type,
                content_snippet=content_snippet,
            )
        except Exception as e:
            logger.exception("AI 分析异常")
            return AiResult(error=str(e), content_snippet=content_snippet)

    async def analyze_batch(
        self, announcements: list[dict[str, Any]], max_concurrent: int = 3
    ) -> list[AiResult]:
        """批量分析，控制并发数"""
        import asyncio

        sem = asyncio.Semaphore(max_concurrent)

        async def _limited(ann: dict[str, Any]) -> AiResult:
            async with sem:
                return await self.analyze(ann)

        tasks = [_limited(ann) for ann in announcements]
        return await asyncio.gather(*tasks)

    async def _call_deepseek(self, prompt: str) -> AiResult:
        """调用 DeepSeek Chat API"""
        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload = {
            "model": self.model,
            "messages": [
                {
                    "role": "system",
                    "content": "你是一个专业的政府采购项目分析师。请根据公告信息判断是否为中国电信可以承接的项目，并用 JSON 格式回答。",
                },
                {"role": "user", "content": prompt},
            ],
            "temperature": 0.3,  # 低温度，提高判断一致性
            "max_tokens": 512,
        }

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(url, headers=headers, json=payload)
            if response.status_code != 200:
                return AiResult(
                    error=f"API 请求失败 (HTTP {response.status_code}): {response.text[:200]}"
                )

            data = response.json()
            choices = data.get("choices", [])
            if not choices:
                return AiResult(error="API 返回空 choices")

            content = choices[0].get("message", {}).get("content", "")
            return self._parse_response(content)

    @staticmethod
    def _parse_response(content: str) -> AiResult:
        """从 LLM 回复中提取 JSON 结果"""
        # 清理可能的 markdown 代码块标记
        content = content.strip()
        if content.startswith("```"):
            # 移除 ```json 或 ``` 包裹
            lines = content.split("\n")
            if lines[0].strip().startswith("```"):
                lines = lines[1:]
            if lines and lines[-1].strip() == "```":
                lines = lines[:-1]
            content = "\n".join(lines).strip()

        try:
            result = json.loads(content)
            return AiResult(
                is_relevant=bool(result.get("is_relevant", False)),
                reason=str(result.get("reason", "")),
                business_type=str(result.get("business_type", "")),
            )
        except (json.JSONDecodeError, ValueError) as e:
            logger.warning("JSON 解析失败: %s\n原始内容: %s", e, content[:200])
            return AiResult(error=f"JSON 解析失败: {e}")