钉钉机器人接AI进行消息筛选和推送

目录

一 项目架构

二 数据流

三 config.py——配置中心

四 database.py——SQlite数据库

表的结构

核心函数

设计要点

五 jintian_client.py ——客户端

核心类和方法

V3API调用和V2API降级

两个 HTML 解析器

collect_column() 采集流程

技术保障

六 ai_summarizer.py — AI 摘要

AISummarizer 类

Prompt 设计

batch_summarize() 并发控制

分类体系

七 robot_output.py — 机器人输出模拟

RobotOutput 类

卡片 JSON 结构

八 main.py — 主流程

7 步流水线

关键逻辑

九 依赖和运行

requirements.txt

运行命令

标准库依赖

---

一 项目架构

py测试/
├── config.py ← 配置中心（API密钥、专栏列表、AI参数）
├── database.py ← SQLite 数据库（3张表，永久存储）
├── jintian_client.py ← 今天看看 API 客户端（采集+解码+兜底）
├── ai_summarizer.py ← DeepSeek AI 摘要生成器
├── robot_output.py ← 钉钉机器人输出模拟（卡片+文档+预览）
├── main.py ← 主流程编排（7个步骤）
├── requirements.txt ← 依赖（httpx）
├── data/
│ └── ivd_news.db ← SQLite 数据库文件（自动生成）
└── output/
├── IVD日报_xxx.md ← 日报文件
├── 钉钉卡片_xxx.json ← 群消息卡片 JSON
└── 钉钉知识库文档_xxx.md ← 知识库文档 Markdown

二 数据流

jintian_client.py ← 采集 + Base64解码 + HTML清洗
│
▼
database.py ← 存入 SQLite（URL去重）
│
▼
ai_summarizer.py ← DeepSeek AI 生成摘要/关键词/分类
│
▼
robot_output.py ← 生成钉钉卡片JSON + 知识库文档
│
▼
output/ ← 保存文件 + 控制台预览

三 config.py——配置中心

# API（数据来源） JINTIAN_USER = "webo.com" # API用户邮箱 JINTIAN_TOKEN = "sa" # API密钥 # 12个行业专栏 COLUMNS = [ {"slug": "VsHMRliSah", "name": "数码智能"}, {"slug": "PeWGZlUGyq", "name": "未来已至"}, # ... 共12个 ] # DeepSeek AI AI_API_KEY = "sk-f7f87a9dw8eba482932a52b9lo59b4ec1" AI_API_URL = "https://api.deepseek.com" AI_MODEL = "deepseek-v4-pro" # 标题过滤关键字（含这些词的文章被过滤掉） FILTER_KEYWORDS = ["2"]

• 专栏采用slug(API标识)+name(中文名)配对存储
• 在这里可以设置每个专栏的最多采集文章保证数据采集的广泛性

  # 每专栏最大采集文章数 MAX_ARTICLES_PER_COLUMN = 20

四 database.py——SQlite数据库

表的结构

Articles

id:主键

title
url(唯一索引，去重)

author

source / source_name

publish_time

content_text (正文)

collected_at (采集日)

summaries(AI摘要表)

article_id (外键)

summary (AI摘要文本)

keywords (关键词)

category (分类)

daily_reports（日报记录表）

report_date

article_count

output_file

核心函数

函数	功能
init_db()	建表（CREATE IF NOT EXISTS，幂等安全）
insert_article()	插入文章，URL重复自动跳过（INSERT OR IGNORE）
get_today_articles()	查今日文章，LEFT JOIN 摘要表一并返回
get_articles_without_summary()	查还没生成AI摘要的文章
save_summary()	存AI摘要（INSERT OR REPLACE 幂等）
get_stats()	数据库统计：总文章数、总摘要数、今日新增

设计要点

• row_factory = sqlite3.Row使查询结果可用字典访问
• INSERT OR IGNORE+ URL唯一索引 = 自动去重
• INSERT OR REPLACE保证摘要重复执行不报错

五 jintian_client.py ——客户端

作用:调用API获取文章，解码Base64内容，清洗HTML

核心类和方法

JintianClient（异步 httpx 客户端）
│
├── fetch_v3() ← 调用 V3 API（返回含 content 字段的数据）
├── fetch_v2() ← 调用 V2 API（V3 失败时的降级方案）
├── decode_content() ← Base64 解码 → HTML 清洗 → 纯文本
├── fetch_from_wechat() ← 兜底：直接请求微信原文 URL 解析正文
└── collect_column() ← 采集单个专栏（翻页+解码+兜底，核心方法）

V3API调用和V2API降级

# ── V3 API ── async def fetch_v3(self, slug: str, page: int = 1, page_size: int = 30) -> dict: url = f"{self.api_base}/api3/query/adv/get_topics_by_one_columnV3" params = {"user": self.user, "token": self.token, "slug": slug, "page": page, "page_size": page_size} resp = await self.client.get(url, params=params) return resp.json() # ── V2 降级 ── async def fetch_v2(self, slug: str, page: int = 1, page_size: int = 30) -> dict: url = f"{self.api_base}/api3/query/adv/get_topics_by_one_columnV2" params = {"user": self.user, "token": self.token, "slug": slug, "page": page, "page_size": page_size} resp = await self.client.get(url, params=params) return resp.json()

两个 HTML 解析器

_HTMLStripper _WechatParser
───────────────── ─────────────────
输入：Base64解码后的HTML 输入：微信文章页HTML
输出：纯文本 输出：#js_content 内的正文
过滤：script/style/nav 定位：id="js_content" 的 div
/footer/header
/code/pre

collect_column()采集流程

while 还没采够 20 篇: try: result = fetch_v3(slug, page) # ① 优先 V3 except: result = fetch_v2(slug, page) # ② V3 失败降级 V2 for 每篇文章: if 发布时间 < 昨天: return # ③ 时间边界，停止翻页 text = decode_content(content) # ④ Base64解码 + HTML清洗 if not text: text = fetch_from_wechat(url) # ⑤ 兜底：微信原文抓取 articles.append({title, url, content_text, ...}) page += 1

技术保障

层	数据来源	成功率
V3 API	Base64 编码的 HTML 内容	~95%
V2 降级	V3 网络失败时用 V2（无内容）	保底不报错
微信原文	直接请求 mp.weixin.qq.com	成功率低（微信反爬）

六 ai_summarizer.py — AI 摘要

作用:调用 AI为每篇文章生成 150 字摘要、关键词和分类

AISummarizer类

AISummarizer
│
└── summarize(title, content) → {"summary", "keywords", "category"}
│
├─ 构造 Prompt（告诉 AI 角色 + 文章内容 + 输出格式）
│//这里以deepseek的模型为例
├─ POST https://api.deepseek.com/v1/chat/completions
│ model: deepseek-v4-pro
│ temperature: 0.3
│ max_tokens: 600
│
├─ 解析返回的 JSON
│ {"summary": "150字摘要", "keywords": "关键词", "category": "企业新闻"}
│
└─ 失败降级：返回原文前 150 字作为兜底摘要

Prompt 设计

你是IVD（体外诊断）行业资深分析师。请分析以下文章并返回JSON。

文章标题：{title}
文章内容：{content[:4000]} ← 最多传 4000 字给 AI

返回格式：
{
"summary": "150字以内的中文摘要，提炼核心事实、数据和观点",
"keywords": "3-5个中文关键词，逗号分隔",
"category": "政策法规/技术前沿/市场动态/企业新闻/学术研究/其他"
}

batch_summarize()并发控制

asyncio.Semaphore(5) ← 最多同时 5 个并发请求

分类体系

分类	示例
政策法规	药监局发布新规、卫健委文件
技术前沿	新产品发布、技术突破
市场动态	市场报告、招投标、业绩
企业新闻	融资、合作、人事变动
学术研究	论文、指南、共识

七 robot_output.py — 机器人输出模拟

作用: 生成钉钉群消息卡片 JSON 和知识库文档 Markdown，控制台预览

RobotOutput类

RobotOutput(articles, title_date)
│
├── build_card_json() ← 钉钉群消息卡片 JSON
│ 包含：概览信息 + Top 3 标题 + AI摘要预览 + 查看详情链接
│
├── build_doc_markdown() ← 钉钉知识库文档 Markdown
│ 包含：概览统计表 + 按分类分章节 + 每篇文章(标题/来源/摘要/关键词)
│
├── save_all() ← 保存到 output/ 目录
│ 钉钉卡片_xxx.json + 钉钉知识库文档_xxx.md
│
└── print_preview() ← 控制台画框预览
模拟钉钉卡片外观 + 分类统计柱状图

卡片 JSON 结构

{
"cardTemplateId": "StandardCard",
"cardData": {
"header": {"title": "IVD 供应链资讯日报已更新"},
"contents": [
{"type": "markdown", "text": "概览信息"},
{"type": "divider"},
{"type": "text", "text": "📰 今日核心动态"},
{"type": "markdown", "text": "Top 3 标题 + AI摘要"},
{"type": "divider"},
{"type": "markdown", "text": "查看完整日报链接"}
]
}
}

八 main.py — 主流程

作用：编排全部 7 个步骤，控制执行顺序和错误处理

7 步流水线

步骤1: 采集 → 循环12个专栏，每个专栏翻页采集最多20篇，写入数据库
步骤2: 读取 → 从数据库查询今日所有文章（含已有AI摘要）
步骤3: 过滤 → 标题关键字匹配（招聘/激励等），过滤掉并打印
步骤4: AI摘要 → 跳过已有摘要的，只对新文章调用 DeepSeek
步骤5: 生成日报 → 按分类组织 Markdown，保存 output/IVD日报_xxx.md
步骤6: 机器人 → 生成卡片JSON + 知识库文档 + 控制台预览
步骤7: 统计 → 打印数据库总量、今日新增、分类分布

关键逻辑

# 步骤1：逐个专栏采集 for col in COLUMNS: articles = await client.collect_column(slug, source_name=name) for article in articles: insert_article(article) # URL重复自动跳过 # 步骤3：关键字过滤 for a in articles: if any(kw in a["title"] for kw in FILTER_KEYWORDS): removed.append(a) # 含"招聘""激励"等 → 过滤 else: filtered.append(a) # 正常 → 保留 # 步骤4：只对新文章生成AI摘要 need_ai = [a for a in filtered if not a.get("ai_summary")] if need_ai: await batch_summarize(need_ai, max_concurrent=5)

九 依赖和运行

requirements.txt

httpx>=0.25.0 # 异步 HTTP 客户端（API调用 + AI请求）

运行命令

cd "D:\桌面\da\供应链开发项目\方案\py测试" pip install httpx python main.py

标准库依赖

模块	用途
html.parser.HTMLParser	HTML 标签剥离
base64	Base64 编解码
sqlite3	数据库操作
asyncio	异步并发控制
json	JSON 解析
re	正则清理
uuid	卡片ID生成
pathlib.Path	文件路径操作