在日常开发中我们经常需要获取最新的新闻资讯来支持业务决策或内容创作但手动浏览多个新闻源既耗时又低效。本文基于AI Agent技术手把手教你构建一个自动读取新闻的Skill实现新闻内容的智能抓取、解析和格式化输出。这个项目特别适合需要实时新闻监控的内容创作者、数据分析师和产品经理通过将AI能力与新闻采集相结合可以大幅提升信息获取效率。下面将从基础概念到完整实现详细讲解如何打造这样一个实用的自动化工具。1. AI Skill开发基础与环境准备1.1 什么是AI SkillAI Skill可以理解为AI代理的技能包它封装了特定的任务处理逻辑让AI能够执行专业化的操作。在Claude Code、Cursor Agent等AI编程工具中Skill通常以文件夹形式存在包含技能描述、示例代码和配置文件。与传统的插件不同Skill更注重任务完成的完整性和用户体验通常遵循统一的协议规范便于在不同AI代理间共享和复用。1.2 开发环境要求为了顺利开发新闻读取Skill需要准备以下环境操作系统要求Windows 10/11, macOS Monterey及以上, 或主流Linux发行版确保具备命令行操作权限AI代理工具任选其一Claude Code CLI (claude)Cursor Agent (cursor-agent)OpenAI Codex (codex)Gemini CLI (gemini)开发工具链Node.js 18 或 Python 3.8Git版本管理文本编辑器或IDEVSCode推荐1.3 项目结构规划一个标准的Skill文件夹结构如下news-reader-skill/ ├── SKILL.md # 技能描述文件 ├── example.html # 输出示例 ├── assets/ # 静态资源 │ └── style.css ├── references/ # 参考文档 └── scripts/ # 辅助脚本2. 新闻源接入与数据获取2.1 选择新闻数据源新闻读取Skill的核心是可靠的数据源。根据使用场景和技术要求可以选择以下几种类型的新闻源RSS/Atom订阅源适合技术博客、新闻网站的标准格式# 示例解析RSS新闻源 import feedparser from datetime import datetime def parse_rss_feed(feed_url): 解析RSS订阅源获取新闻内容 try: feed feedparser.parse(feed_url) articles [] for entry in feed.entries: article { title: entry.title, link: entry.link, summary: entry.summary, published: entry.published_parsed, source: feed.feed.title } articles.append(article) return articles except Exception as e: print(f解析RSS源失败: {e}) return []新闻API服务提供结构化的新闻数据# 使用新闻API的示例配置 NEWS_API_CONFIG { base_url: https://newsapi.org/v2, endpoints: { top_headlines: /top-headlines, everything: /everything }, params: { pageSize: 20, language: zh } }网页爬虫方案直接抓取新闻网站内容import requests from bs4 import BeautifulSoup import time class NewsScraper: def __init__(self): self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 }) def scrape_news_site(self, url, selectors): 根据CSS选择器抓取新闻内容 try: response self.session.get(url, timeout10) response.raise_for_status() soup BeautifulSoup(response.content, html.parser) articles [] # 根据选择器提取新闻条目 news_items soup.select(selectors[article]) for item in news_items: title_elem item.select_one(selectors[title]) link_elem item.select_one(selectors[link]) summary_elem item.select_one(selectors[summary]) if title_elem and link_elem: article { title: title_elem.get_text().strip(), link: link_elem.get(href), summary: summary_elem.get_text().strip() if summary_elem else , timestamp: time.time() } articles.append(article) return articles except Exception as e: print(f网页抓取失败: {e}) return []2.2 数据清洗与标准化不同新闻源的数据格式各异需要进行标准化处理import re from datetime import datetime class NewsProcessor: def __init__(self): self.clean_patterns [ (r[^], ), # 移除HTML标签 (r\s, ), # 合并多个空格 (r^[^a-zA-Z0-9\u4e00-\u9fa5], ), # 移除开头特殊字符 (r[^a-zA-Z0-9\u4e00-\u9fa5]$, ) # 移除结尾特殊字符 ] def clean_text(self, text): 清理和标准化文本内容 if not text: return cleaned text for pattern, replacement in self.clean_patterns: cleaned re.sub(pattern, replacement, cleaned) return cleaned.strip() def standardize_date(self, date_str): 标准化日期格式 date_formats [ %Y-%m-%dT%H:%M:%SZ, %a, %d %b %Y %H:%M:%S %z, %Y-%m-%d %H:%M:%S, %d/%m/%Y %H:%M ] for fmt in date_formats: try: return datetime.strptime(date_str, fmt) except ValueError: continue return datetime.now()3. Skill核心功能实现3.1 创建SKILL.md文件SKILL.md是Skill的描述文件定义了技能的基本信息和执行逻辑--- mode:>class HTMLGenerator: def __init__(self): self.template !DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title{title}/title style {styles} /style /head body div classcontainer header h1今日新闻摘要/h1 p更新时间: {update_time}/p /header main {content} /main /div /body /html self.styles * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, PingFang SC, Hiragino Sans GB, Microsoft YaHei, Helvetica Neue, Helvetica, Arial, sans-serif; line-height: 1.6; color: #333; background: #f5f5f5; } .container { max-width: 800px; margin: 0 auto; padding: 20px; background: white; min-height: 100vh; } header { text-align: center; margin-bottom: 32px; padding-bottom: 16px; border-bottom: 2px solid #eaeaea; } h1 { color: #2c3e50; margin-bottom: 8px; font-size: 28px; } .article { margin-bottom: 24px; padding: 16px; border-left: 4px solid #3498db; background: #f8f9fa; } .article h2 { color: #2c3e50; margin-bottom: 8px; font-size: 18px; } .article .meta { color: #7f8c8d; font-size: 14px; margin-bottom: 8px; } .article .summary { color: #555; line-height: 1.5; } media (max-width: 768px) { .container { padding: 16px; } h1 { font-size: 24px; } } def generate_news_html(self, articles: List[Dict]) - str: 生成新闻HTML页面 from datetime import datetime content_html for article in articles: article_html f article classarticle h2{article.get(title, )}/h2 div classmeta span来源: {article.get(source, )}/span span | /span span时间: {article.get(published, )}/span /div div classsummary {article.get(summary, )} /div a href{article.get(link, #)} target_blank阅读原文/a /article content_html article_html return self.template.format( title智能新闻摘要, stylesself.styles, contentcontent_html, update_timedatetime.now().strftime(%Y-%m-%d %H:%M:%S) )4. 完整实战案例4.1 项目初始化与配置首先创建项目目录结构# 创建Skill目录 mkdir news-reader-skill cd news-reader-skill # 创建必要的文件和目录 touch SKILL.md touch example.html mkdir -p assets scripts references # 初始化Python环境如果使用Python python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows pip install requests beautifulsoup4 feedparser aiohttp4.2 实现主程序入口创建主执行文件main.py#!/usr/bin/env python3 新闻阅读器Skill主程序 import asyncio import sys import json from datetime import datetime from news_processor import NewsProcessor from html_generator import HTMLGenerator from news_reader import NewsReaderSkill class NewsReaderApp: def __init__(self): self.reader NewsReaderSkill() self.generator HTMLGenerator() self.processor NewsProcessor() async def run(self, output_filenews_output.html): 运行新闻读取任务 print(开始获取新闻内容...) try: # 获取新闻数据 articles await self.reader.fetch_all_news() # 处理和数据清洗 processed_articles [] for article in articles: cleaned_article { title: self.processor.clean_text(article.get(title, )), summary: self.processor.clean_text(article.get(summary, )), source: article.get(source, 未知来源), link: article.get(link, #), published: article.get(published, 未知时间) } processed_articles.append(cleaned_article) # 生成HTML页面 html_content self.generator.generate_news_html(processed_articles) # 保存输出文件 with open(output_file, w, encodingutf-8) as f: f.write(html_content) print(f新闻生成完成共处理 {len(processed_articles)} 篇文章) print(f输出文件: {output_file}) return processed_articles except Exception as e: print(f程序执行出错: {e}) return [] def main(): 主函数 app NewsReaderApp() # 支持命令行参数指定输出文件 output_file sys.argv[1] if len(sys.argv) 1 else news_output.html # 运行异步任务 articles asyncio.run(app.run(output_file)) # 在控制台也显示摘要 if articles: print(\n 最新新闻摘要 ) for i, article in enumerate(articles[:5], 1): print(f{i}. {article[title]} - {article[source]}) if __name__ __main__: main()4.3 创建配置文件添加配置文件config.json{ news_sources: [ { name: 技术博客RSS, type: rss, url: https://example.com/tech.rss, enabled: true, update_interval: 3600 }, { name: 财经新闻API, type: api, url: https://api.example.com/finance, enabled: true, update_interval: 1800 } ], output: { format: html, template: modern, auto_open: true }, processing: { max_articles: 50, summary_length: 200, language: zh } }4.4 运行与测试创建测试脚本test_news_reader.pyimport asyncio import os from main import NewsReaderApp async def test_basic_functionality(): 测试基本功能 print(测试新闻阅读器基本功能...) app NewsReaderApp() # 测试新闻获取 articles await app.reader.fetch_all_news() print(f获取到 {len(articles)} 篇新闻) # 测试HTML生成 if articles: html_content app.generator.generate_news_html(articles[:3]) print(HTML生成成功长度:, len(html_content)) # 保存测试文件 with open(test_output.html, w, encodingutf-8) as f: f.write(html_content) print(测试文件已保存: test_output.html) return len(articles) 0 async def test_performance(): 测试性能 import time start_time time.time() app NewsReaderApp() articles await app.reader.fetch_all_news() elapsed time.time() - start_time print(f性能测试: 获取 {len(articles)} 篇新闻耗时 {elapsed:.2f} 秒) return elapsed 10.0 # 应该在10秒内完成 if __name__ __main__: # 运行测试 async def run_tests(): test1 await test_basic_functionality() test2 await test_performance() print(f\n测试结果:) print(f基本功能测试: {通过 if test1 else 失败}) print(f性能测试: {通过 if test2 else 失败}) asyncio.run(run_tests())5. 集成AI代理与技能部署5.1 创建AI Skill适配器为了让新闻阅读器能够被AI代理调用需要创建适配器文件# skill_adapter.py import json import subprocess from pathlib import Path class NewsReaderSkillAdapter: def __init__(self, skill_path: str): self.skill_path Path(skill_path) self.config self.load_config() def load_config(self) - dict: 加载Skill配置 config_file self.skill_path / skill_config.json if config_file.exists(): with open(config_file, r, encodingutf-8) as f: return json.load(f) return {} def execute(self, prompt: str) - str: 执行Skill并返回结果 try: # 解析用户指令 command self.parse_command(prompt) # 执行新闻读取任务 result subprocess.run([ python, main.py, --output, command[output_file], --sources, ,.join(command[sources]) ], capture_outputTrue, textTrue, cwdself.skill_path) if result.returncode 0: return self.format_success_response(command[output_file]) else: return self.format_error_response(result.stderr) except Exception as e: return fSkill执行错误: {str(e)} def parse_command(self, prompt: str) - dict: 解析用户指令 # 简单的指令解析逻辑 if 科技 in prompt or 技术 in prompt: sources [technology] elif 财经 in prompt or 经济 in prompt: sources [finance] else: sources [technology, finance] # 默认所有源 return { sources: sources, output_file: news_output.html } def format_success_response(self, output_file: str) - str: 格式化成功响应 return f 新闻读取任务已完成 生成的新闻摘要已保存到: {output_file} 文件包含最新的新闻内容可以直接在浏览器中打开查看。 使用提示: - 文件为独立的HTML格式包含完整的样式和布局 - 支持响应式设计在手机和电脑上都能良好显示 - 包含原文链接方便深入阅读 下一步操作建议: 1. 在浏览器中打开 {output_file} 查看新闻 2. 可以修改配置来定制新闻源和显示样式 3. 设置定时任务实现自动更新 5.2 部署到AI代理环境创建部署脚本deploy_skill.py#!/usr/bin/env python3 Skill部署脚本 import shutil import json from pathlib import Path def deploy_to_agent(skill_path: str, agent_type: str claude): 部署Skill到指定的AI代理 skill_name Path(skill_path).name deploy_paths { claude: Path.home() / .claude / skills / skill_name, cursor: Path.home() / .cursor / skills / skill_name, codex: Path.home() / .codex / skills / skill_name } if agent_type not in deploy_paths: print(f不支持的代理类型: {agent_type}) return False target_path deploy_paths[agent_type] try: # 确保目标目录存在 target_path.parent.mkdir(parentsTrue, exist_okTrue) # 复制Skill文件 if target_path.exists(): shutil.rmtree(target_path) shutil.copytree(skill_path, target_path) # 更新代理的技能注册表 update_skill_registry(agent_type, skill_name, str(target_path)) print(fSkill已成功部署到 {agent_type}) print(f位置: {target_path}) return True except Exception as e: print(f部署失败: {e}) return False def update_skill_registry(agent_type: str, skill_name: str, skill_path: str): 更新AI代理的技能注册表 registry_files { claude: Path.home() / .claude / skill_registry.json, cursor: Path.home() / .cursor / skill_registry.json } if agent_type not in registry_files: return registry_file registry_files[agent_type] # 读取或创建注册表 if registry_file.exists(): with open(registry_file, r, encodingutf-8) as f: registry json.load(f) else: registry {} # 更新注册信息 registry[skill_name] { path: skill_path, type: news_reader, version: 1.0.0, description: 智能新闻阅读器 - 自动获取和展示多源新闻内容 } # 写回注册表 with open(registry_file, w, encodingutf-8) as f: json.dump(registry, f, indent2, ensure_asciiFalse) if __name__ __main__: # 部署到所有支持的代理 agents [claude, cursor, codex] skill_path input(请输入Skill路径: ).strip() for agent in agents: print(f\n部署到 {agent}...) deploy_to_agent(skill_path, agent)6. 高级功能与优化6.1 实现新闻分类与标签系统class NewsCategorizer: def __init__(self): self.categories { technology: [AI, 编程, 科技, 互联网, 软件, 硬件], finance: [股票, 经济, 金融, 投资, 市场, 货币], sports: [体育, 比赛, 运动员, 赛事, 冠军], entertainment: [电影, 音乐, 明星, 娱乐, 综艺] } def categorize_article(self, title: str, content: str) - list: 对新闻文章进行分类 text f{title} {content}.lower() matched_categories [] for category, keywords in self.categories.items(): score sum(1 for keyword in keywords if keyword.lower() in text) if score 0: matched_categories.append((category, score)) # 按匹配度排序 matched_categories.sort(keylambda x: x[1], reverseTrue) return [cat for cat, score in matched_categories[:2]] # 返回前2个分类6.2 添加定时任务与自动更新import schedule import time from threading import Thread class NewsScheduler: def __init__(self, news_reader): self.reader news_reader self.running False self.thread None def start_daily_update(self, hour: int 9, minute: int 0): 启动每日定时更新 schedule.every().day.at(f{hour:02d}:{minute:02d}).do( self.update_news ) self.running True self.thread Thread(targetself._run_scheduler) self.thread.daemon True self.thread.start() print(f已启动定时任务每天 {hour:02d}:{minute:02d} 自动更新新闻) def _run_scheduler(self): 运行调度器 while self.running: schedule.run_pending() time.sleep(60) # 每分钟检查一次 async def update_news(self): 执行新闻更新任务 print(开始定时新闻更新...) try: articles await self.reader.fetch_all_news() # 生成新的HTML文件 # 可以添加通知功能如发送邮件或桌面通知 print(f定时更新完成获取 {len(articles)} 篇新闻) except Exception as e: print(f定时更新失败: {e}) def stop(self): 停止定时任务 self.running False if self.thread: self.thread.join(timeout5)6.3 实现新闻去重与质量过滤class NewsFilter: def __init__(self): self.seen_titles set() self.min_title_length 10 self.max_title_length 200 def is_duplicate(self, article: dict) - bool: 检查是否为重复新闻 title article.get(title, ).strip().lower() # 简单的标题相似度检查 for seen_title in self.seen_titles: similarity self.calculate_similarity(title, seen_title) if similarity 0.8: # 80%相似度认为是重复 return True self.seen_titles.add(title) return False def calculate_similarity(self, text1: str, text2: str) - float: 计算文本相似度 words1 set(text1.split()) words2 set(text2.split()) if not words1 or not words2: return 0.0 intersection words1.intersection(words2) union words1.union(words2) return len(intersection) / len(union) def is_quality_article(self, article: dict) - bool: 检查文章质量 title article.get(title, ) summary article.get(summary, ) # 检查标题长度 if not (self.min_title_length len(title) self.max_title_length): return False # 检查内容完整性 if len(summary) 50: # 摘要太短 return False # 检查是否包含关键信息 if not any(keyword in title.lower() for keyword in [发布, 宣布, 开发, 推出, 发现]): return False return True7. 常见问题与解决方案7.1 网络请求问题排查问题现象新闻获取失败连接超时解决方案class RobustNewsFetcher: def __init__(self): self.retry_count 3 self.timeout 30 async def fetch_with_retry(self, session, url, retriesNone): 带重试机制的请求 retries retries or self.retry_count for attempt in range(retries): try: async with session.get(url, timeoutself.timeout) as response: if response.status 200: return await response.text() else: print(f请求失败状态码: {response.status}) except asyncio.TimeoutError: print(f请求超时第 {attempt 1} 次重试) except Exception as e: print(f请求错误: {e}) if attempt retries - 1: await asyncio.sleep(2 ** attempt) # 指数退避 return None7.2 内容解析异常处理问题现象HTML解析失败返回空内容解决方案def safe_html_parse(html_content, fallback_parserhtml.parser): 安全的HTML解析 try: from bs4 import BeautifulSoup return BeautifulSoup(html_content, fallback_parser) except Exception as e: print(fHTML解析失败: {e}) # 尝试使用更宽松的解析器 try: return BeautifulSoup(html_content, html5lib) except: # 最后尝试简单的字符串处理 return None def extract_text_safe(element): 安全提取文本内容 if element is None: return try: return element.get_text().strip() except Exception as e: print(f文本提取失败: {e}) return 7.3 性能优化建议优化方案1使用缓存减少重复请求import diskcache class CachedNewsFetcher: def __init__(self, cache_dir.news_cache, ttl3600): self.cache diskcache.Cache(cache_dir) self.ttl ttl # 缓存有效期秒 async def fetch_cached(self, session, url): 带缓存的请求 cache_key fnews_{hash(url)} # 检查缓存 cached self.cache.get(cache_key) if cached: return cached # 执行请求 content await self.fetch_news(session, url) if content: # 缓存结果 self.cache.set(cache_key, content, self.ttl) return content优化方案2并行处理提高效率async def parallel_process_articles(articles, process_func, batch_size5): 并行处理文章列表 results [] for i in range(0, len(articles), batch_size): batch articles[i:i batch_size] # 创建并行任务 tasks [process_func(article) for article in batch] batch_results await asyncio.gather(*tasks, return_exceptionsTrue) # 过滤成功的结果 successful_results [ result for result in batch_results if not isinstance(result, Exception) ] results.extend(successful_results) return results8. 生产环境部署建议8.1 安全注意事项API密钥管理import os from dotenv import load_dotenv class SecureConfig: def __init__(self): load_dotenv() # 加载环境变量 def get_api_key(self, service_name): 安全获取API密钥 env_var f{service_name.upper()}_API_KEY key os.getenv(env_var) if not key: raise ValueError(f未找到{service_name}的API密钥请设置{env_var}环境变量) return key def validate_config(self): 验证配置完整性 required_vars [NEWS_API_KEY, RSS_FEED_URL] missing_vars [var for var in required_vars if not os.getenv(var)] if missing_vars: print(警告缺少以下环境变量:) for var in missing_vars: print(f - {var}) return False return True8.2 监控与日志记录完整的日志系统import logging from logging.handlers import RotatingFileHandler import json from datetime import datetime class NewsLogger: def __init__(self, log_dirlogs): self.log_dir Path(log_dir) self.log_dir.mkdir(exist_okTrue) self.setup_logging() def setup_logging(self): 配置日志系统 logger logging.getLogger(news_reader) logger.setLevel(logging.INFO) # 文件处理器自动轮转 file_handler RotatingFileHandler( self.log_dir / news_reader.log, maxBytes10*1024*1024, # 10MB backupCount5 ) file_handler.setFormatter(logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s )) # 控制台处理器 console_handler logging.StreamHandler() console_handler.setFormatter(logging.Formatter( %(levelname)s: %(message)s )) logger.addHandler(file_handler) logger.addHandler(console_handler) self.logger logger def log_news_update(self, article_count, duration): 记录新闻更新日志 log_entry { timestamp: datetime.now().isoformat(), event: news_update, article_count: article_count, duration_seconds: duration, status: success } self.logger.info(json.dumps(log_entry))通过以上完整的实现方案你已经构建了一个功能完善的自动新闻读取Skill。这个工具不仅能够自动获取和处理新闻内容还具备了生产环境所需的稳定性、安全性和可维护性。