Compare commits

...

17 Commits

Author SHA1 Message Date
v6ole 18f415b363 fix: 企微 API 通过代理绕过 IP 白名单限制,补充错误日志
- 新增 WECHAT_API_BASE_URL 配置项,支持企微 API 代理转发
- _send_message 增加 API 错误日志(errcode + errmsg)
- _get_access_token 增加失败日志
- handle_text 不再为空,返回菜单引导提示
- 防重入触发时通知用户等待时间
- menu.py API 调用统一使用配置化的 base URL

Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-12 08:43:03 +08:00
v6ole 103de16a8f fix: 白名单校验改为不区分大小写 2026-05-26 16:10:43 +08:00
v6ole a51161b5f3 feat: 企微菜单新增 AI 分析开关 + 白名单控制
- 新增 ai_whitelist 配置项,仅白名单用户可操作 AI 开关
- 新增 app/services/ai_state.py 运行时开关模块
- 系统管理菜单新增「AI 分析」按钮 (toggle_ai)
- 监控配置、系统状态显示 AI 状态和 AI 标记统计
- pipeline 改用运行时开关 is_ai_enabled()
2026-05-26 16:02:26 +08:00
v6ole 68215aa804 fix: 改进正文提取,支持政采云 SPA API 和大化县政府网 div.article-con
- 新增 _extract_zcy_content(): 通过政采云隐藏 API 获取公告正文
- 新增 _extract_dahuagov_content(): 大化县专用提取,支持 div.article-con
- 通用提取器新增更多选择器
2026-05-26 15:47:48 +08:00
v6ole 754214692e feat: 接入 DeepSeek AI 分析公告是否为中国电信可承接项目
- 新增 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
2026-05-26 11:45:51 +08:00
v6ole 136941d84e docs: 添加项目说明 README.md
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 18:16:56 +08:00
v6ole 4f16134955 fix: 最新公告防重入覆盖整个函数,推送改为时间升序间隔1秒
- 防重入锁移到函数入口,60秒内重复点击直接忽略,避免企微重试导致重复推送
- 取最新6条后按 publish_date 升序推送,企微向上滑即为时间正序
- 每条间隔1秒发送

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 17:59:10 +08:00
v6ole c01a3d427b fix: 存库前先标记 keyword_matched,修复历史数据全为 false 的问题
pipeline 原来先存库再过滤,导致 keyword_matched 永远是 False。
改为先标记再存库,最新公告查询才能正确区分两个来源。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 17:51:00 +08:00
v6ole dc3ca1f76c feat: 最新公告点击时触发爬取并优化查询逻辑
- 点击时先爬取一次(60秒防重入),爬取期间发送提示
- 查询改为:广西政采网关键词匹配 + 大化县政府网全部,按时间倒序取6条

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 17:38:58 +08:00
v6ole 6d1d512fca feat: 最新公告改为全来源最新6条按时间排序
去掉关键词过滤限制,两个来源的公告统一按 publish_date 倒序取6条。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 17:21:48 +08:00
v6ole b2d0503bc2 fix: 大化县政府网公告抓取详情页获取精确发布时间
列表页只有日期(00:00),改为并发抓取详情页解析 <meta name="PubDate">
获取精确时分秒,失败时保留列表页日期作为 fallback。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 17:16:33 +08:00
v6ole 77c6722a92 fix: 修复广西政采网公告时间时区错误
fromtimestamp() 在 UTC 服务器上会少 8 小时,改为明确指定 Asia/Shanghai 时区解析。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 10:21:26 +08:00
v6ole 048fca283d fix: 最新公告恢复关键词过滤 2026-05-10 14:29:33 +08:00
v6ole 0121017272 fix: 最新公告直接返回最新5条 + 显示来源网站 + 菜单调整
- latest_announcements 不再过滤 keyword_matched,返回全部最新5条
- 每条公告显示来源网站(广西政府采购网/大化县政府网采购公告)
- 菜单「今日公告」改为顶级按钮「最新公告」,移除子菜单重复项
2026-05-10 14:29:02 +08:00
v6ole 24d844bf44 fix: /crawl/sources 包含独立爬虫来源 + trigger 支持 spider_name
- sources 接口原来只返回 GXGP 子来源,遗漏了大化县政府网
- trigger 接口现在支持 spider_name 参数指定单个爬虫
2026-05-10 14:22:58 +08:00
v6ole 9f90e16661 feat(wechat): 新增企微菜单功能并修复重复触发问题
- 新增「查询」菜单组:监控配置、系统状态、今日工作日、最新公告
- 新增「系统管理」子菜单:暂停/恢复定时任务处理器
- 修复立即爬取被企微重试机制触发多次的问题(60秒防重入锁)
- docker-compose 补充 image 名称和 container_name

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-10 14:01:11 +08:00
v6ole d84c838e76 chore: 删除 dead code — markdown 文件生成功能
onu.md 和 markdown_generated 从未被实际使用,是旧项目遗留代码
2026-05-10 13:59:34 +08:00
21 changed files with 1037 additions and 33 deletions
+15 -3
View File
@@ -20,6 +20,9 @@ WECHAT_TOKEN=
WECHAT_ENCODING_AES_KEY=
WECHAT_PORT=18001
WECHAT_HOST=0.0.0.0
# 企微 API 代理(可选),用于绕过 IP 白名单限制
# 留空则直连 https://qyapi.weixin.qq.com
WECHAT_API_BASE_URL=https://qyapi.weixin.qq.com
# 定时任务
SCHEDULER_ENABLED=true
@@ -30,6 +33,15 @@ LOGHIVE_ENDPOINT=http://10.10.10.14:8000
LOGHIVE_PROJECT=gx-gp-notify
LOGHIVE_API_KEY=
# Markdown
MARKDOWN_ENABLED=true
MARKDOWN_OUTPUT_FILE=onu.md
# AI 分析 (DeepSeek)
AI_ENABLED=false
# AI 管理白名单(企微用户ID,逗号分隔),留空表示所有人可操作
AI_WHITELIST=
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...
+2
View File
@@ -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/
+108
View File
@@ -0,0 +1,108 @@
# GX-GP-Notify
广西政府采购公告监控系统,自动爬取广西政府采购网和大化县政府网的公告,通过企业微信推送关键词匹配的公告。
## 功能
- **定时爬取**:按 cron 规则自动爬取,工作日 8:00-21:00 每小时执行
- **关键词过滤**:广西政采网公告按关键词过滤(默认:`大化`),大化县政府网全量爬取
- **企微推送**:新公告实时推送到企业微信应用
- **去重**:基于内容 hash 去重,避免重复入库和推送
- **节假日感知**:非工作日自动跳过爬取
## 数据来源
| 来源 | 说明 | 过滤 |
|------|------|------|
| 广西政府采购网 | 采购公告、结果公告、合同公告等 13 个分类 | 关键词过滤 |
| 大化县政府网 | 大化县门户网站采购公告 | 全量爬取 |
## 企微菜单功能
| 菜单 | 说明 |
|------|------|
| 最新公告 | 触发一次爬取,推送两个来源合并后最新 6 条(时间升序) |
| 手动爬取 | 立即触发全量爬取并推送新公告 |
| 今日统计 | 查看今日新增和累计公告数 |
| 系统状态 | 查看待推送数量、定时任务状态 |
| 监控配置 | 查看关键词、爬取页数、定时规则 |
| 工作日状态 | 查看今天是否为工作日 |
| 同步节假日 | 同步当年节假日数据 |
| 暂停/恢复任务 | 控制定时爬取任务 |
## 快速部署
### 1. 配置环境变量
复制 `.env.example``.env` 并填写:
```env
# 数据库
DATABASE_URL=postgresql+asyncpg://user:password@host:5432/gx-gp-notify
# 企业微信
WECHAT_CORP_ID=your_corp_id
WECHAT_AGENT_ID=your_agent_id
WECHAT_SECRET=your_secret
WECHAT_TOKEN=your_token
WECHAT_ENCODING_AES_KEY=your_aes_key
# 爬虫
CRAWLER_KEYWORDS=["大化"]
CRAWLER_MAX_PAGES=10
SCHEDULER_CRON=0 8-21 * * 1-5
```
### 2. 启动服务
```bash
docker compose -f docker/docker-compose.yml up -d
```
### 3. 数据库迁移
```bash
docker compose -f docker/docker-compose.yml exec app alembic upgrade head
```
### 4. 配置企微回调
在企业微信后台将回调 URL 设置为:
```
http://your-server:18001/wechat/callback
```
## 开发
```bash
# 安装依赖
pip install -e ".[dev]"
# 运行测试
pytest
# 代码检查
ruff check .
```
## 技术栈
- **框架**FastAPI + uvicorn
- **数据库**PostgreSQL + SQLAlchemy (async) + Alembic
- **爬虫**httpx + BeautifulSoup4
- **定时任务**APScheduler
- **推送**:企业微信应用消息 API
- **部署**Docker
## 项目结构
```
app/
├── api/ # HTTP 接口(公告查询、手动触发爬取等)
├── crawler/ # 爬虫(广西政采网、大化县政府网)
├── models/ # 数据库模型
├── scheduler/ # 定时任务
├── services/ # 业务逻辑(pipeline、过滤、通知)
└── wechat/ # 企业微信(消息处理、推送客户端)
```
@@ -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')
+21 -7
View File
@@ -9,7 +9,10 @@ router = APIRouter()
@router.post("/crawl/trigger")
async def trigger_crawl(request: CrawlTriggerRequest):
service = get_crawl_service()
names = service.get_spider_names()
if request.spider_name:
names = [request.spider_name]
else:
names = service.get_spider_names()
all_results = []
total_stored = 0
@@ -46,9 +49,20 @@ async def crawl_sources():
from app.config import settings
sources = json.loads(settings.announcement_sources)
return {
"sources": [
{"code": code, "name": info["name"], "type": info["type"]}
for code, info in sources.items()
]
}
result = [
{"code": code, "name": info["name"], "type": info["type"]}
for code, info in sources.items()
]
# 加入独立爬虫来源(非 GXGP 子来源的独立 Spider
service = get_crawl_service()
for name in service.get_spider_names():
if name == "gxgp":
continue # gxgp 的子来源已在上面列出
spider = service.spiders.get(name)
if spider:
result.append({
"code": spider.source_code,
"name": spider.source_name,
"type": "independent",
})
return {"sources": result}
+37 -3
View File
@@ -1,5 +1,6 @@
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import field_validator
class Settings(BaseSettings):
@@ -28,6 +29,7 @@ class Settings(BaseSettings):
wechat_encoding_aes_key: str = ""
wechat_port: int = 18001
wechat_host: str = "0.0.0.0"
wechat_api_base_url: str = "https://qyapi.weixin.qq.com"
# 定时任务
scheduler_enabled: bool = True
@@ -38,9 +40,41 @@ class Settings(BaseSettings):
loghive_project: str = "gx-gp-notify"
loghive_api_key: str = ""
# Markdown
markdown_enabled: bool = True
markdown_output_file: str = "onu.md"
# AI 分析 (DeepSeek)
ai_enabled: bool = False
ai_whitelist: str = ""
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
-1
View File
@@ -9,7 +9,6 @@ class PipelineResult:
stored: int = 0
filtered: int = 0
notified: int = 0
markdown_generated: bool = False
@dataclass
+29 -2
View File
@@ -6,7 +6,7 @@ import httpx
from app.config import settings
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig
from app.crawler.parsers import parse_dahuagov_html
from app.crawler.parsers import parse_dahuagov_html, parse_dahuagov_detail_pubdate
class DahuagovSpider(BaseSpider):
@@ -56,7 +56,9 @@ class DahuagovSpider(BaseSpider):
crawled_at=start_time,
)
announcements = parse_dahuagov_html(html, start_time)
announcements = parse_dahuagov_html(html, start_time)
await self._fetch_detail_dates(client, announcements, headers)
duration = (datetime.now() - start_time).total_seconds()
return CrawlResult(
@@ -69,6 +71,31 @@ class DahuagovSpider(BaseSpider):
duration=duration,
)
async def _fetch_detail_dates(
self,
client: httpx.AsyncClient,
announcements: list[dict],
headers: dict,
) -> None:
"""并发抓取详情页,用 PubDate meta 更新精确发布时间"""
sem = asyncio.Semaphore(5)
async def fetch_one(ann: dict) -> None:
async with sem:
await asyncio.sleep(random.uniform(0.3, 0.8))
try:
resp = await client.get(ann["content_url"], headers=headers)
if resp.status_code == 200:
pub = parse_dahuagov_detail_pubdate(resp.text)
if pub:
ann["publish_date"] = pub
ann["is_today"] = pub.date() == datetime.now().date()
except Exception:
pass # 保留列表页日期作为 fallback
await asyncio.gather(*[fetch_one(ann) for ann in announcements])
async def _delay(self):
delay = random.uniform(1.0, 3.0)
await asyncio.sleep(delay)
+186 -1
View File
@@ -2,6 +2,9 @@ import hashlib
from datetime import datetime
from typing import Any
from urllib.parse import urljoin
from zoneinfo import ZoneInfo
_TZ = ZoneInfo("Asia/Shanghai")
from bs4 import BeautifulSoup
@@ -30,7 +33,7 @@ def parse_gxgp_api_response(
if not timestamp:
continue
try:
publish_date = datetime.fromtimestamp(int(timestamp) / 1000)
publish_date = datetime.fromtimestamp(int(timestamp) / 1000, tz=_TZ).replace(tzinfo=None)
except (ValueError, TypeError):
continue
@@ -131,6 +134,188 @@ def parse_dahuagov_html(html: str, crawled_at: datetime) -> list[dict[str, Any]]
return results
def parse_dahuagov_detail_pubdate(html: str) -> datetime | None:
"""从详情页 <meta name="PubDate"> 解析精确发布时间"""
soup = BeautifulSoup(html, "html.parser")
meta = soup.find("meta", attrs={"name": "PubDate"})
if not meta:
return None
content = meta.get("content", "").strip()
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
try:
return datetime.strptime(content, fmt)
except ValueError:
continue
return None
async def extract_page_content(url: str, timeout: int = 30) -> str | None:
"""抓取详情页并用 BeautifulSoup 提取正文纯文本"""
import urllib.parse
import httpx
# ---- 站点专用处理 ----
# ① 政采云 SPA (HTTP API, 无需渲染)
if "zfcg.gxzf.gov.cn" in url and "articleId=" in url:
return await _extract_zcy_content(url, timeout)
# ② 大化县政府网 (静态HTML)
if "www.gxdh.gov.cn" in url or "gxdh.gov.cn" in url:
return await _extract_dahuagov_content(url, timeout)
# ---- 通用 HTML 提取 ----
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",
"div.article-con", ".trs_editor_view", ".TRS_UEDITOR",
".trs_paper_default", ".article-content",
]
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:
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])
if len(text) > 50:
return text
return None
except Exception:
return None
async def _extract_zcy_content(url: str, timeout: int = 30) -> str | None:
"""从政采云 SPA 隐藏 API 提取公告正文"""
import urllib.parse
import httpx
from bs4 import BeautifulSoup
params = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
article_id = params.get("articleId", [None])[0]
parent_id = params.get("parentId", [None])[0]
if not article_id:
return None
api_url = "https://zfcg.gxzf.gov.cn/portal/detail"
if parent_id:
api_url += f"?articleId={article_id}&parentId={parent_id}"
else:
api_url += f"?articleId={article_id}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "application/json, text/plain, */*",
"Referer": url,
}
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.get(api_url, headers=headers)
if resp.status_code != 200:
return None
data = resp.json()
if not data.get("success"):
return None
content_html = data.get("result", {}).get("data", {}).get("content", "")
if not content_html:
return None
soup = BeautifulSoup(content_html, "html.parser")
for tag in soup.find_all(["script", "style"]):
tag.decompose()
text = soup.get_text(separator="\n", strip=True)
# 清理过短行和多余空白
lines = [l.strip() for l in text.split("\n") if len(l.strip()) > 5]
text = "\n".join(lines[:300])
return text if len(text) > 50 else None
except Exception:
return None
async def _extract_dahuagov_content(url: str, timeout: int = 30) -> str | None:
"""从大化县政府网详情页提取正文"""
import httpx
from bs4 import BeautifulSoup
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",
}
try:
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-con", ".trs_editor_view", ".TRS_UEDITOR",
".trs_paper_default", "div.content", "div.TRS_Editor",
"div.article-content", "div.Custom_UnionStyle",
"div#content", "div.main-content", "article", ".detail-content",
]
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:
lines = [l.strip() for l in text.split("\n") if l.strip()]
return "\n".join(lines[:300])
# 兜底
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])
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')}"
+2
View File
@@ -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
+1
View File
@@ -29,6 +29,7 @@ class AnnouncementListResponse(BaseModel):
class CrawlTriggerRequest(BaseModel):
spider_name: str | None = None
keywords: list[str] | None = None
sources: list[str] | None = None
manual: bool = False
+147
View File
@@ -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}")
+51
View File
@@ -0,0 +1,51 @@
"""AI 运行时状态 — 支持企微菜单动态开关"""
import logging
from app.config import settings
logger = logging.getLogger(__name__)
# 运行时覆盖值,None 表示使用 settings.ai_enabled
_runtime_override: bool | None = None
def is_ai_enabled() -> bool:
"""获取 AI 分析当前是否启用(考虑运行时覆盖)"""
if _runtime_override is not None:
return _runtime_override
return settings.ai_enabled
def set_ai_enabled(enabled: bool) -> bool:
"""设置 AI 分析运行时开关,返回是否真的发生了变化"""
global _runtime_override
current = is_ai_enabled()
if enabled == current:
return False
_runtime_override = enabled
status = "启用" if enabled else "禁用"
logger.info("AI 分析已通过企微菜单%s", status)
return True
def get_ai_status_text() -> str:
"""获取 AI 状态文本"""
return "已启用" if is_ai_enabled() else "已禁用"
def get_whitelist() -> list[str]:
"""获取 AI 白名单用户列表"""
raw = settings.ai_whitelist
if not raw:
return []
return [u.strip() for u in raw.split(",") if u.strip()]
def is_whitelisted(user_id: str) -> bool:
"""检查用户是否在 AI 白名单中(不区分大小写)"""
whitelist = get_whitelist()
if not whitelist:
# 白名单为空则所有人都可以操作
return True
return user_id.lower() in [u.lower() for u in whitelist]
+36
View File
@@ -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)
+69 -7
View File
@@ -1,8 +1,13 @@
import logging
from typing import Any
from app.config import settings
from app.crawler.base import PipelineConfig, PipelineResult
from app.services.ai_state import is_ai_enabled
from app.services.filter_service import dedup_by_hash
logger = logging.getLogger(__name__)
class PostCrawlPipeline:
def __init__(self, db_session, notification_service):
@@ -20,24 +25,32 @@ class PostCrawlPipeline:
if config.dedup_enabled:
announcements = dedup_by_hash(announcements)
# 2. Store to database
# 2. 先标记 keyword_matched,再存库
if config.filter_enabled and config.keywords:
for a in announcements:
a["keyword_matched"] = self._match_keywords(a, config.keywords)
# 3. Store to database
stored = await self._save_to_db(announcements)
result.stored = stored
to_notify = announcements
# 3. Filter
# 4. Filter
if config.filter_enabled and config.keywords:
before = len(to_notify)
to_notify = [a for a in to_notify
if self._match_keywords(a, config.keywords)]
to_notify = [a for a in to_notify if a.get("keyword_matched")]
result.filtered = before - len(to_notify)
# 4. Skip already-notified
# 5. Skip already-notified
if to_notify:
to_notify = await self._exclude_sent(to_notify)
# 5. Notify
# 5.5 AI 分析(可选,支持运行时开关)
if is_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)
elif config.notify_mode == "filtered":
@@ -46,7 +59,7 @@ class PostCrawlPipeline:
elif not config.filter_enabled:
result.notified = await self._send_notifications(to_notify)
# 6. Mark sent
# 7. Mark sent
if config.mark_sent and result.notified > 0:
await self._mark_sent(to_notify)
@@ -125,3 +138,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()
+15 -3
View File
@@ -1,9 +1,12 @@
import logging
import time
import httpx
from app.config import settings
logger = logging.getLogger(__name__)
class WeChatClient:
def __init__(self):
@@ -15,7 +18,7 @@ class WeChatClient:
if self._access_token and now < self._token_expires_at:
return self._access_token
url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
url = f"{settings.wechat_api_base_url}/cgi-bin/gettoken"
params = {
"corpid": settings.wechat_corp_id,
"corpsecret": settings.wechat_secret,
@@ -27,6 +30,7 @@ class WeChatClient:
self._access_token = data["access_token"]
self._token_expires_at = now + data.get("expires_in", 7200) - 300
return self._access_token
logger.error(f"获取 access_token 失败: {data}")
return None
async def send_text(self, content: str, to_user: str = "@all") -> bool:
@@ -59,9 +63,10 @@ class WeChatClient:
) -> bool:
token = await self._get_access_token()
if not token:
logger.error("无法获取 access_token,跳过消息发送")
return False
url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
url = f"{settings.wechat_api_base_url}/cgi-bin/message/send"
params = {"access_token": token}
body = {
"touser": to_user,
@@ -73,4 +78,11 @@ class WeChatClient:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.post(url, params=params, json=body)
data = response.json()
return data.get("errcode") == 0
errcode = data.get("errcode")
if errcode == 0:
return True
logger.error(
f"企业微信消息发送失败: errcode={errcode} errmsg={data.get('errmsg')} "
f"msgtype={msgtype} touser={to_user}"
)
return False
+249
View File
@@ -1,4 +1,6 @@
import asyncio
import logging
import time
import xml.etree.ElementTree as ET
from app.config import settings
@@ -7,6 +9,14 @@ from app.wechat.client import WeChatClient
logger = logging.getLogger(__name__)
# 防重入:记录最近一次触发爬取的时间戳,60秒内不重复执行
_last_crawl_time: float = 0.0
_crawl_lock = asyncio.Lock()
# 防重入:最新公告爬取,60秒内不重复执行
_last_latest_time: float = 0.0
_latest_lock = asyncio.Lock()
class WeChatMessageHandler:
def __init__(self):
@@ -65,9 +75,34 @@ class WeChatMessageHandler:
return await self._handle_trigger_crawl(from_user)
elif event_key == "sync_holidays":
return await self._handle_sync_holidays(from_user)
elif event_key == "monitor_config":
return await self._handle_monitor_config(from_user)
elif event_key == "system_status":
return await self._handle_system_status(from_user)
elif event_key == "workday_status":
return await self._handle_workday_status(from_user)
elif event_key == "latest_announcements":
return await self._handle_latest_announcements(from_user)
elif event_key == "pause_scheduler":
return await self._handle_pause_scheduler(from_user)
elif event_key == "resume_scheduler":
return await self._handle_resume_scheduler(from_user)
elif event_key == "toggle_ai":
return await self._handle_toggle_ai(from_user)
return None
async def handle_text(self, content: str, from_user: str) -> str | None:
"""文本消息:返回帮助提示"""
help_text = (
"请使用菜单操作:\n"
"---\n"
"📋 最新公告 - 获取最新公告\n"
"📊 查询 → 监控配置/系统状态/今日工作日\n"
"⚙️ 系统管理 → 立即爬取/同步节假日/AI 分析"
)
ok = await self.client.send_text(help_text, from_user)
if not ok:
logger.warning(f"发送帮助消息失败, touser={from_user}")
return None
async def _handle_today_stats(self, from_user: str) -> str | None:
@@ -97,8 +132,21 @@ class WeChatMessageHandler:
await self.client.send_text("查询失败,请稍后再试", from_user)
async def _handle_trigger_crawl(self, from_user: str) -> str | None:
global _last_crawl_time
from app.api.deps import get_crawl_service
# 防重入:企业微信会对同一事件重试多次,60秒内只执行一次
async with _crawl_lock:
now = time.monotonic()
if now - _last_crawl_time < 60:
remaining = int(60 - (now - _last_crawl_time))
logger.info(f"爬取请求被忽略(防重入),距上次 {now - _last_crawl_time:.1f}s")
await self.client.send_text(
f"爬取任务进行中,请 {remaining} 秒后再试", from_user
)
return None
_last_crawl_time = now
await self.client.send_text("开始爬取,请稍候...", from_user)
try:
@@ -136,3 +184,204 @@ class WeChatMessageHandler:
except Exception as e:
logger.error(f"同步节假日失败: {e}")
await self.client.send_text(f"同步失败: {e}", from_user)
async def _handle_monitor_config(self, from_user: str) -> str | None:
import json
from app.services.ai_state import get_ai_status_text
try:
keywords = settings.crawler_keywords
sources = json.loads(settings.announcement_sources)
source_names = "".join(v["name"] for v in sources.values())
text = (
f"📋 监控配置\n"
f"---\n"
f"监控关键词: {', '.join(keywords)}\n"
f"爬取页数: {settings.crawler_max_pages}\n"
f"定时规则: {settings.scheduler_cron}\n"
f"公告来源: {source_names}\n"
f"---\n"
f"🤖 AI 分析: {get_ai_status_text()}\n"
f"AI 模型: {settings.ai_model}\n"
f"重点标记: {settings.ai_analysis_title}"
)
await self.client.send_text(text, from_user)
except Exception as e:
logger.error(f"查询监控配置失败: {e}")
await self.client.send_text("查询失败,请稍后再试", from_user)
async def _handle_system_status(self, from_user: str) -> str | None:
from app.api.deps import get_db
from sqlalchemy import func, select
from app.models.announcement import Announcement
from app.services.ai_state import get_ai_status_text
try:
async for db in get_db():
total_result = await db.execute(
select(func.count()).select_from(Announcement)
)
total = total_result.scalar() or 0
today_result = await db.execute(
select(func.count()).where(
func.date(Announcement.publish_date) == func.current_date()
).select_from(Announcement)
)
today = today_result.scalar() or 0
unsent_result = await db.execute(
select(func.count()).where(
Announcement.is_sent == False, # noqa: E712
Announcement.keyword_matched == True, # noqa: E712
).select_from(Announcement)
)
unsent = unsent_result.scalar() or 0
# AI 标记统计
ai_relevant_result = await db.execute(
select(func.count()).where(
Announcement.ai_relevant == True, # noqa: E712
).select_from(Announcement)
)
ai_relevant = ai_relevant_result.scalar() or 0
scheduler_status = "已启用" if settings.scheduler_enabled else "已禁用"
text = (
f"📊 系统状态\n"
f"---\n"
f"累计公告: {total}\n"
f"今日新增: {today}\n"
f"待推送: {unsent}\n"
f"---\n"
f"定时任务: {scheduler_status}\n"
f"定时规则: {settings.scheduler_cron}\n"
f"---\n"
f"🤖 AI 分析: {get_ai_status_text()}\n"
f"AI 标记项目: {ai_relevant}"
)
await self.client.send_text(text, from_user)
except Exception as e:
logger.error(f"查询系统状态失败: {e}")
await self.client.send_text("查询失败,请稍后再试", from_user)
async def _handle_workday_status(self, from_user: str) -> str | None:
from app.services.holiday_service import now_in_china, is_workday
from app.api.deps import get_db
try:
async for db in get_db():
today = now_in_china()
workday = await is_workday(db, today)
status = "工作日,正常爬取" if workday else "非工作日,跳过爬取"
text = f"今天 {today.strftime('%Y-%m-%d %A')}\n{status}"
await self.client.send_text(text, from_user)
except Exception as e:
logger.error(f"查询工作日状态失败: {e}")
await self.client.send_text("查询失败,请稍后再试", from_user)
async def _handle_latest_announcements(self, from_user: str) -> str | None:
global _last_latest_time
from app.api.deps import get_db, get_crawl_service
from sqlalchemy import select, asc, or_
from app.models.announcement import Announcement
# 防重入:60秒内整个函数只执行一次(含推送)
async with _latest_lock:
now = time.monotonic()
if (now - _last_latest_time) < 60:
logger.info("最新公告请求被忽略(防重入)")
return None
_last_latest_time = now
try:
await self.client.send_text("正在获取最新公告,请稍候...", from_user)
service = get_crawl_service()
await service.run_all()
# 查:广西政采网关键词匹配 + 大化县政府网全部,取最新6条后按时间升序推送
async for db in get_db():
from sqlalchemy import select, desc, or_
result = await db.execute(
select(Announcement)
.where(or_(
Announcement.keyword_matched == True, # noqa: E712
Announcement.source_code == "dahuagov",
))
.order_by(desc(Announcement.publish_date))
.limit(6)
)
items = result.scalars().all()
if not items:
await self.client.send_text("暂无公告", from_user)
return None
# 按时间升序推送,企微里向上滑即为时间正序
for ann in sorted(items, key=lambda a: a.publish_date):
title = ann.title if len(ann.title) <= 128 else ann.title[:125] + "..."
purchase_name = ann.purchase_name or ""
if len(purchase_name) > 25:
purchase_name = purchase_name[:22] + "..."
time_str = ann.publish_date.strftime("%Y-%m-%d %H:%M") if ann.publish_date else "时间未知"
source_name = ann.source_name or ann.source_code or ""
description = f"{source_name} | {purchase_name} | {time_str}"
await self.client.send_textcard(title, description, ann.content_url or "", from_user)
await asyncio.sleep(1)
except Exception as e:
logger.error(f"查询最新公告失败: {e}")
await self.client.send_text("查询失败,请稍后再试", from_user)
async def _handle_pause_scheduler(self, from_user: str) -> str | None:
try:
from app.scheduler.jobs import scheduler
if scheduler.running:
scheduler.pause()
await self.client.send_text("定时任务已暂停", from_user)
else:
await self.client.send_text("定时任务未在运行", from_user)
except Exception as e:
logger.error(f"暂停定时任务失败: {e}")
await self.client.send_text(f"操作失败: {e}", from_user)
async def _handle_toggle_ai(self, from_user: str) -> str | None:
from app.services.ai_state import (
get_ai_status_text,
get_whitelist,
is_ai_enabled,
is_whitelisted,
set_ai_enabled,
)
# 白名单校验
whitelist = get_whitelist()
if whitelist and not is_whitelisted(from_user):
await self.client.send_text(
f"⚠️ 你没有权限操作 AI 分析开关\n"
f"当前仅以下用户可操作:\n{', '.join(whitelist)}",
from_user,
)
return None
current = is_ai_enabled()
changed = set_ai_enabled(not current)
if not changed:
await self.client.send_text(
f"AI 分析当前已是「{get_ai_status_text()}」状态,无需切换",
from_user,
)
return None
await self.client.send_text(
f"✅ AI 分析已切换为「{get_ai_status_text()}\n"
f"下次爬取触发时生效",
from_user,
)
async def _handle_resume_scheduler(self, from_user: str) -> str | None:
try:
from app.scheduler.jobs import scheduler
scheduler.resume()
await self.client.send_text("定时任务已恢复", from_user)
except Exception as e:
logger.error(f"恢复定时任务失败: {e}")
await self.client.send_text(f"操作失败: {e}", from_user)
+30 -5
View File
@@ -9,9 +9,29 @@ logger = logging.getLogger(__name__)
MENU = {
"button": [
{
"name": "今日公告",
"name": "最新公告",
"type": "click",
"key": "today_stats",
"key": "latest_announcements",
},
{
"name": "查询",
"sub_button": [
{
"name": "监控配置",
"type": "click",
"key": "monitor_config",
},
{
"name": "系统状态",
"type": "click",
"key": "system_status",
},
{
"name": "今日工作日",
"type": "click",
"key": "workday_status",
},
],
},
{
"name": "系统管理",
@@ -26,6 +46,11 @@ MENU = {
"type": "click",
"key": "sync_holidays",
},
{
"name": "AI 分析",
"type": "click",
"key": "toggle_ai",
},
],
},
]
@@ -46,7 +71,7 @@ class MenuManager:
if not token:
return False
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/create"
url = f"{settings.wechat_api_base_url}/cgi-bin/menu/create"
params = {"access_token": token, "agentid": int(settings.wechat_agent_id)}
async with httpx.AsyncClient(timeout=15) as client:
@@ -63,7 +88,7 @@ class MenuManager:
if not token:
return False
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/delete"
url = f"{settings.wechat_api_base_url}/cgi-bin/menu/delete"
params = {"access_token": token, "agentid": int(settings.wechat_agent_id)}
async with httpx.AsyncClient(timeout=15) as client:
@@ -76,7 +101,7 @@ class MenuManager:
if not token:
return None
url = "https://qyapi.weixin.qq.com/cgi-bin/menu/get"
url = f"{settings.wechat_api_base_url}/cgi-bin/menu/get"
params = {"access_token": token, "agentid": int(settings.wechat_agent_id)}
async with httpx.AsyncClient(timeout=15) as client:
+4
View File
@@ -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 \
+5
View File
@@ -3,6 +3,11 @@ services:
build:
context: ..
dockerfile: docker/Dockerfile
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"
env_file:
-1
View File
@@ -53,7 +53,6 @@ def test_pipeline_result_defaults():
assert result.stored == 0
assert result.filtered == 0
assert result.notified == 0
assert result.markdown_generated is False
def test_generate_content_hash():