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

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`	文件路径操作