Compare commits

...

4 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
8 changed files with 279 additions and 14 deletions
+5
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
@@ -32,6 +35,8 @@ LOGHIVE_API_KEY=
# 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
+2
View File
@@ -29,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
@@ -41,6 +42,7 @@ class Settings(BaseSettings):
# 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"
+120 -4
View File
@@ -151,8 +151,21 @@ def parse_dahuagov_detail_pubdate(html: str) -> datetime | 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",
@@ -172,15 +185,16 @@ async def extract_page_content(url: str, timeout: int = 30) -> str | None:
"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: # 至少50字才算有效正文
if len(text) > 50:
return text
# 兜底:取 body 内所有文本
@@ -189,9 +203,8 @@ async def extract_page_content(url: str, timeout: int = 30) -> str | None:
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行
text = "\n".join(lines[:200])
if len(text) > 50:
return text
@@ -200,6 +213,109 @@ async def extract_page_content(url: str, timeout: int = 30) -> str | None:
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')}"
+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]
+3 -2
View File
@@ -3,6 +3,7 @@ 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__)
@@ -45,8 +46,8 @@ class PostCrawlPipeline:
if to_notify:
to_notify = await self._exclude_sent(to_notify)
# 5.5 AI 分析(可选)
if settings.ai_enabled and to_notify:
# 5.5 AI 分析(可选,支持运行时开关
if is_ai_enabled() and to_notify:
await self._ai_analyze(to_notify)
# 6. Notify
+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
+75 -2
View File
@@ -87,9 +87,22 @@ class WeChatMessageHandler:
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:
@@ -126,7 +139,11 @@ class WeChatMessageHandler:
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
@@ -170,15 +187,22 @@ class WeChatMessageHandler:
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}"
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:
@@ -189,6 +213,7 @@ class WeChatMessageHandler:
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():
@@ -212,13 +237,27 @@ class WeChatMessageHandler:
)
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}"
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:
@@ -304,6 +343,40 @@ class WeChatMessageHandler:
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
+8 -3
View File
@@ -46,6 +46,11 @@ MENU = {
"type": "click",
"key": "sync_holidays",
},
{
"name": "AI 分析",
"type": "click",
"key": "toggle_ai",
},
],
},
]
@@ -66,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:
@@ -83,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:
@@ -96,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: