目录一 项目架构二 数据流三 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 ← 存入 SQLiteURL去重│▼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数据库表的结构Articlesid:主键titleurl(唯一索引去重)authorsource / source_namepublish_timecontent_text (正文)collected_at (采集日)summaries(AI摘要表)article_id (外键)summary (AI摘要文本)keywords (关键词)category (分类)daily_reports日报记录表report_datearticle_countoutput_file核心函数函数功能init_db()建表CREATE IF NOT EXISTS幂等安全insert_article()插入文章URL重复自动跳过INSERT OR IGNOREget_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 APIV3 失败时的降级方案├── 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, paramsparams) 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, paramsparams) return resp.json()两个 HTML 解析器_HTMLStripper _WechatParser───────────────── ─────────────────输入Base64解码后的HTML 输入微信文章页HTML输出纯文本 输出#js_content 内的正文过滤script/style/nav 定位idjs_content 的 div/footer/header/code/precollect_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 APIBase64 编码的 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_namename) 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_concurrent5)九 依赖和运行requirements.txthttpx0.25.0 # 异步 HTTP 客户端API调用 AI请求运行命令cd D:\桌面\da\供应链开发项目\方案\py测试 pip install httpx python main.py标准库依赖模块用途html.parser.HTMLParserHTML 标签剥离base64Base64 编解码sqlite3数据库操作asyncio异步并发控制jsonJSON 解析re正则清理uuid卡片ID生成pathlib.Path文件路径操作
目录一 项目架构二 数据流三 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 ← 存入 SQLiteURL去重│▼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数据库表的结构Articlesid:主键titleurl(唯一索引去重)authorsource / source_namepublish_timecontent_text (正文)collected_at (采集日)summaries(AI摘要表)article_id (外键)summary (AI摘要文本)keywords (关键词)category (分类)daily_reports日报记录表report_datearticle_countoutput_file核心函数函数功能init_db()建表CREATE IF NOT EXISTS幂等安全insert_article()插入文章URL重复自动跳过INSERT OR IGNOREget_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 APIV3 失败时的降级方案├── 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, paramsparams) 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, paramsparams) return resp.json()两个 HTML 解析器_HTMLStripper _WechatParser───────────────── ─────────────────输入Base64解码后的HTML 输入微信文章页HTML输出纯文本 输出#js_content 内的正文过滤script/style/nav 定位idjs_content 的 div/footer/header/code/precollect_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 APIBase64 编码的 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_namename) 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_concurrent5)九 依赖和运行requirements.txthttpx0.25.0 # 异步 HTTP 客户端API调用 AI请求运行命令cd D:\桌面\da\供应链开发项目\方案\py测试 pip install httpx python main.py标准库依赖模块用途html.parser.HTMLParserHTML 标签剥离base64Base64 编解码sqlite3数据库操作asyncio异步并发控制jsonJSON 解析re正则清理uuid卡片ID生成pathlib.Path文件路径操作
