From 754214692e95c9806db6861b2ef9793a82784cf6 Mon Sep 17 00:00:00 2001 From: v6ole Date: Tue, 26 May 2026 11:45:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E5=85=A5=20DeepSeek=20AI=20?= =?UTF-8?q?=E5=88=86=E6=9E=90=E5=85=AC=E5=91=8A=E6=98=AF=E5=90=A6=E4=B8=BA?= =?UTF-8?q?=E4=B8=AD=E5=9B=BD=E7=94=B5=E4=BF=A1=E5=8F=AF=E6=89=BF=E6=8E=A5?= =?UTF-8?q?=E9=A1=B9=E7=9B=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 ai_enabled/ai_api_key/ai_base_url/ai_model 等配置项，通过 .env 管理 - 新增 app/services/ai_analyzer.py — DeepSeek API 调用 + 详情页正文提取 - 新增 extract_page_content() 从详情页抓取正文供 AI 分析 - Announcement 模型新增 ai_relevant / ai_analysis 字段 - 流水线集成 AI 分析步骤（关键词匹配后、通知前） - AI 标记为可承接的项目额外发送 markdown 着重通知 - 创建 alembic 迁移版本 6e8f4c2d1b0a --- .env.example | 11 ++ .gitignore | 2 + .../6e8f4c2d1b0a_add_ai_analysis_columns.py | 30 ++++ app/config.py | 35 +++++ app/crawler/parsers.py | 51 ++++++ app/models/announcement.py | 2 + app/services/ai_analyzer.py | 147 ++++++++++++++++++ app/services/notification_service.py | 36 +++++ app/services/pipeline.py | 57 +++++++ docker/Dockerfile | 4 + docker/docker-compose.yml | 5 +- 11 files changed, 379 insertions(+), 1 deletion(-) create mode 100644 alembic/versions/6e8f4c2d1b0a_add_ai_analysis_columns.py create mode 100644 app/services/ai_analyzer.py diff --git a/.env.example b/.env.example index 1d8dfa4..8950fee 100644 --- a/.env.example +++ b/.env.example @@ -29,3 +29,14 @@ SCHEDULER_CRON=0 8-21 * * * LOGHIVE_ENDPOINT=http://10.10.10.14:8000 LOGHIVE_PROJECT=gx-gp-notify LOGHIVE_API_KEY= + +# AI 分析 (DeepSeek) +AI_ENABLED=false +AI_API_KEY=sk-your-deepseek-api-key +AI_BASE_URL=https://api.deepseek.com/v1 +AI_MODEL=deepseek-chat +AI_TIMEOUT=30 +AI_ANALYSIS_TITLE=🔔 中国电信可承接项目 +# 自定义分析提示词（可选），不设置则用代码内置默认值 +# 使用 \n 表示换行，支持 {title} {purchase_name} {announcement_type} {content} 四个占位符 +# AI_PROMPT_TEMPLATE=你是一个政府采购项目分析师...\n\n--- 公告信息 ---\n标题：{title}\n... diff --git a/.gitignore b/.gitignore index 2666c69..19ec8d4 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,5 @@ gx_gp_monitor/config/config.yaml onu.md vendor/build/ vendor/*.egg-info/ +# Added by code-review-graph +.code-review-graph/ diff --git a/alembic/versions/6e8f4c2d1b0a_add_ai_analysis_columns.py b/alembic/versions/6e8f4c2d1b0a_add_ai_analysis_columns.py new file mode 100644 index 0000000..82a5a9b --- /dev/null +++ b/alembic/versions/6e8f4c2d1b0a_add_ai_analysis_columns.py @@ -0,0 +1,30 @@ +"""add_ai_analysis_columns + +Revision ID: 6e8f4c2d1b0a +Revises: 1f59799a5083 +Create Date: 2026-05-21 10:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '6e8f4c2d1b0a' +down_revision: Union[str, Sequence[str], None] = '1f59799a5083' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """添加 AI 分析相关字段""" + op.add_column('announcements', sa.Column('ai_relevant', sa.Boolean(), nullable=True)) + op.add_column('announcements', sa.Column('ai_analysis', sa.Text(), nullable=True)) + + +def downgrade() -> None: + """回滚""" + op.drop_column('announcements', 'ai_analysis') + op.drop_column('announcements', 'ai_relevant') diff --git a/app/config.py b/app/config.py index e9cbadb..65c86f1 100644 --- a/app/config.py +++ b/app/config.py @@ -1,5 +1,6 @@ from pydantic_settings import BaseSettings, SettingsConfigDict +from pydantic import field_validator class Settings(BaseSettings): @@ -38,6 +39,40 @@ class Settings(BaseSettings): loghive_project: str = "gx-gp-notify" loghive_api_key: str = "" + # AI 分析 (DeepSeek) + ai_enabled: bool = False + ai_api_key: str = "" + ai_base_url: str = "https://api.deepseek.com/v1" + ai_model: str = "deepseek-chat" + ai_timeout: int = 30 + ai_analysis_title: str = "🔔 中国电信可承接项目" + ai_prompt_template: str = ( + "你是一个政府采购项目分析师，专门帮助中国电信识别可以承接的项目。\n\n" + "可承接范围包括但不限于：\n" + "- 通信工程、光缆建设、基站建设\n" + "- 信息化系统建设、系统集成\n" + "- 云计算、大数据、政务云\n" + "- 物联网、智慧城市、智慧园区\n" + "- 安防监控、视频会议、应急通信\n" + "- 网络运维、网络优化、IDC 服务\n" + "- 5G 应用、专线服务\n\n" + "--- 公告信息 ---\n" + "标题：{title}\n" + "采购人：{purchase_name}\n" + "公告类型：{announcement_type}\n\n" + "--- 公告正文 ---\n" + "{content}\n\n" + "请用 JSON 格式回答：\n" + '{{"is_relevant": true/false, "reason": "简要判断理由", "business_type": "业务分类"}}' + ) + + @field_validator("ai_prompt_template", mode="before") + @classmethod + def convert_newlines(cls, v: str) -> str: + """将 .env 中字面 \\n 转换为真实换行""" + if isinstance(v, str) and "\\n" in v: + return v.replace("\\n", "\n") + return v # 公告来源（JSON 字符串，从环境变量读取） announcement_sources: str = '{"ZcyAnnouncement1":{"category_id":66485,"name":"采购公告","type":"purchase"},"ZcyAnnouncement2":{"category_id":66485,"name":"结果公告","type":"result"},"ZcyAnnouncement3":{"category_id":66485,"name":"合同公告","type":"contract"},"ZcyAnnouncement4":{"category_id":66485,"name":"更正公告","type":"correction"},"ZcyAnnouncement5":{"category_id":66485,"name":"招标文件预公示","type":"pre_announcement"},"ZcyAnnouncement6":{"category_id":66485,"name":"单一来源公示","type":"single_source"},"ZcyAnnouncement7":{"category_id":66485,"name":"电子卖场公示","type":"electronic_market"},"ZcyAnnouncement10":{"category_id":66485,"name":"履约验收公示","type":"acceptance"},"ZcyAnnouncement11":{"category_id":66485,"name":"工程类公告","type":"engineering"},"ZcyAnnouncement20":{"category_id":66485,"name":"框架协议征集公告","type":"framework_agreement"},"ZcyAnnouncement21":{"category_id":66485,"name":"框架协议入围结果公告","type":"framework_result"},"ZcyAnnouncement23":{"category_id":66485,"name":"框架协议成交结果汇总公告","type":"framework_summary"},"61-266648":{"category_id":66485,"name":"采购意向公开","type":"intention"}}' # noqa: E501 diff --git a/app/crawler/parsers.py b/app/crawler/parsers.py index b9c32d9..f79c855 100644 --- a/app/crawler/parsers.py +++ b/app/crawler/parsers.py @@ -149,6 +149,57 @@ def parse_dahuagov_detail_pubdate(html: str) -> datetime | None: return None +async def extract_page_content(url: str, timeout: int = 30) -> str | None: + """抓取详情页并用 BeautifulSoup 提取正文纯文本""" + import httpx + + try: + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "Accept": "text/html,application/xhtml+xml", + "Accept-Language": "zh-CN,zh;q=0.9", + } + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: + resp = await client.get(url, headers=headers) + if resp.status_code != 200: + return None + + soup = BeautifulSoup(resp.text, "html.parser") + + # 尝试多种常见正文容器选择器 + selectors = [ + "div.article-content", "div.content", "div.TRS_Editor", + "div.Custom_UnionStyle", "div.pages_content", "div#content", + "div.main-content", "article", ".article", ".detail-content", + ".text-content", ".news-content", ".detail-article", + ] + for selector in selectors: + container = soup.select_one(selector) + if container: + # 移除脚本和样式 + for tag in container.find_all(["script", "style"]): + tag.decompose() + text = container.get_text(separator="\n", strip=True) + if len(text) > 50: # 至少50字才算有效正文 + return text + + # 兜底：取 body 内所有文本 + body = soup.find("body") + if body: + for tag in body.find_all(["script", "style", "nav", "footer", "header"]): + tag.decompose() + text = body.get_text(separator="\n", strip=True) + # 移除过长的空白行 + lines = [l.strip() for l in text.split("\n") if l.strip()] + text = "\n".join(lines[:200]) # 最多取前200行 + if len(text) > 50: + return text + + return None + except Exception: + return None + + def _generate_hash(ann: dict[str, Any]) -> str: content = ( f"{ann['title']}|{ann['publish_date'].strftime('%Y-%m-%d')}" diff --git a/app/models/announcement.py b/app/models/announcement.py index aa338f4..0b569b9 100644 --- a/app/models/announcement.py +++ b/app/models/announcement.py @@ -25,6 +25,8 @@ class Announcement(Base): is_new: Mapped[bool] = mapped_column(Boolean, default=True) is_sent: Mapped[bool] = mapped_column(Boolean, default=False) keyword_matched: Mapped[bool] = mapped_column(Boolean, default=False) + ai_relevant: Mapped[bool | None] = mapped_column(Boolean, nullable=True, default=None) + ai_analysis: Mapped[str | None] = mapped_column(Text, nullable=True, default=None) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now()) # noqa: E501 diff --git a/app/services/ai_analyzer.py b/app/services/ai_analyzer.py new file mode 100644 index 0000000..73816bc --- /dev/null +++ b/app/services/ai_analyzer.py @@ -0,0 +1,147 @@ +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}") diff --git a/app/services/notification_service.py b/app/services/notification_service.py index a17fa66..58746b5 100644 --- a/app/services/notification_service.py +++ b/app/services/notification_service.py @@ -17,6 +17,7 @@ class NotificationService: sent = 0 for ann in announcements: try: + # 发送普通 textcard title = ann.get("title", "") if len(title) > 128: title = title[:125] + "..." @@ -36,7 +37,42 @@ class NotificationService: if await self.client.send_textcard(title, description, url): sent += 1 + + # AI 标记为可承接的，额外发送着重通知 + ai_result = ann.get("ai_result") + if ai_result and ai_result.get("is_relevant"): + await self._send_ai_emphasis(ann, ai_result) + except Exception: continue return sent + + async def _send_ai_emphasis( + self, ann: dict[str, Any], ai_result: dict[str, Any] + ) -> bool: + """发送 AI 分析的着重通知（markdown 格式）""" + title = ann.get("title", "") + purchase_name = ann.get("purchase_name", "") + pub_date = ann.get("publish_date") + time_str = pub_date.strftime("%Y-%m-%d %H:%M") if pub_date else "时间未知" + url = ann.get("content_url", "") + + reason = ai_result.get("reason", "") + business_type = ai_result.get("business_type", "") + + # 企业微信 markdown 格式 + md = ( + f"{settings.ai_analysis_title}\n" + f"---\n" + f"**标题：** [{title}]({url})\n" + f"> 采购人：{purchase_name}\n" + f"> 发布时间：{time_str}\n\n" + f"**🤖 AI 分析：**\n" + f"> {reason}\n\n" + f"**🏷 业务分类：** {business_type}\n" + f"---\n" + f"[📄 查看公告原文]({url})" + ) + + return await self.client.send_markdown(md) diff --git a/app/services/pipeline.py b/app/services/pipeline.py index ac4b829..0168a5c 100644 --- a/app/services/pipeline.py +++ b/app/services/pipeline.py @@ -1,8 +1,12 @@ +import logging from typing import Any +from app.config import settings from app.crawler.base import PipelineConfig, PipelineResult from app.services.filter_service import dedup_by_hash +logger = logging.getLogger(__name__) + class PostCrawlPipeline: def __init__(self, db_session, notification_service): @@ -41,6 +45,10 @@ class PostCrawlPipeline: if to_notify: to_notify = await self._exclude_sent(to_notify) + # 5.5 AI 分析（可选） + if settings.ai_enabled and to_notify: + await self._ai_analyze(to_notify) + # 6. Notify if config.notify_mode == "all": result.notified = await self._send_notifications(to_notify) @@ -129,3 +137,52 @@ class PostCrawlPipeline: def _match_keywords(announcement: dict[str, Any], keywords: list[str]) -> bool: text = f"{announcement.get('title', '')} {announcement.get('purchase_name', '')}" return any(kw in text for kw in keywords) + + async def _ai_analyze(self, announcements: list[dict[str, Any]]) -> None: + """对公告列表执行 AI 分析，将结果附加到每条公告的 ai_result 字段，并更新数据库""" + from app.services.ai_analyzer import AiAnalyzer + + analyzer = AiAnalyzer() + logger.info("AI 分析开始：共 %d 条公告", len(announcements)) + + results = await analyzer.analyze_batch(announcements) + + ai_updates = [] + for ann, ai_result in zip(announcements, results): + ann["ai_result"] = { + "is_relevant": ai_result.is_relevant, + "reason": ai_result.reason, + "business_type": ai_result.business_type, + } + if ai_result.error: + logger.warning("AI 分析失败 [%s]: %s", ann.get("title", "")[:30], ai_result.error) + else: + ai_updates.append({ + "content_hash": ann["content_hash"], + "ai_relevant": ai_result.is_relevant, + "ai_analysis": ai_result.reason, + }) + if ai_result.is_relevant: + logger.info("AI 标记可承接项目: %s (%s)", ann.get("title", "")[:40], ai_result.business_type) + + # 批量更新数据库中的 AI 分析结果 + if ai_updates: + await self._update_ai_results(ai_updates) + + async def _update_ai_results(self, updates: list[dict[str, Any]]) -> None: + """批量更新公告的 AI 分析结果到数据库""" + from sqlalchemy import update + + from app.models.announcement import Announcement + + for u in updates: + stmt = ( + update(Announcement) + .where(Announcement.content_hash == u["content_hash"]) + .values( + ai_relevant=u["ai_relevant"], + ai_analysis=u["ai_analysis"], + ) + ) + await self.db.execute(stmt) + await self.db.commit() diff --git a/docker/Dockerfile b/docker/Dockerfile index c7d0475..3a8427d 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -1,5 +1,9 @@ FROM python:3.12-slim +LABEL maintainer="GX-gp-notify Dev Team" \ + version="2.0.0" \ + description="广西政府采购网公告监控系统" + WORKDIR /app RUN apt-get update && apt-get install -y --no-install-recommends \ diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index aa9f8c1..90c2315 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -3,7 +3,10 @@ services: build: context: .. dockerfile: docker/Dockerfile - image: gx-gp-notify:latest + tags: + - gx-gp-notify:2.0.0 + - gx-gp-notify:latest + image: gx-gp-notify:2.0.0 container_name: gx-gp-notify ports: - "18001:8000"