在日常开发中，我们经常需要获取最新的新闻资讯来支持业务决策或内容创作，但手动浏览多个新闻源既耗时又低效。本文基于AI Agent技术，手把手教你构建一个自动读取新闻的Skill，实现新闻内容的智能抓取、解析和格式化输出。
这个项目特别适合需要实时新闻监控的内容创作者、数据分析师和产品经理，通过将AI能力与新闻采集相结合，可以大幅提升信息获取效率。下面将从基础概念到完整实现，详细讲解如何打造这样一个实用的自动化工具。
1. AI Skill开发基础与环境准备
1.1 什么是AI Skill
AI 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.8+
- Git版本管理
- 文本编辑器或IDE（VSCode推荐）
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, timeout=10) 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 lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>{title}</title> <style> {styles} </style> </head> <body> <div class="container"> <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 class="article"> <h2>{article.get('title', '')}</h2> <div class="meta"> <span>来源: {article.get('source', '')}</span> <span> | </span> <span>时间: {article.get('published', '')}</span> </div> <div class="summary"> {article.get('summary', '')} </div> <a href="{article.get('link', '#')}" target="_blank">阅读原文</a> </article> """ content_html += article_html return self.template.format( title="智能新闻摘要", styles=self.styles, content=content_html, update_time=datetime.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_file='news_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', encoding='utf-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.py：
import 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', encoding='utf-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', encoding='utf-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_output=True, text=True, cwd=self.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 f"Skill执行错误: {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(parents=True, exist_ok=True) # 复制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(f"Skill已成功部署到 {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', encoding='utf-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', encoding='utf-8') as f: json.dump(registry, f, indent=2, ensure_ascii=False) 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(key=lambda x: x[1], reverse=True) 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(target=self._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(timeout=5)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, retries=None): """带重试机制的请求""" retries = retries or self.retry_count for attempt in range(retries): try: async with session.get(url, timeout=self.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_parser='html.parser'): """安全的HTML解析""" try: from bs4 import BeautifulSoup return BeautifulSoup(html_content, fallback_parser) except Exception as e: print(f"HTML解析失败: {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', ttl=3600): self.cache = diskcache.Cache(cache_dir) self.ttl = ttl # 缓存有效期（秒） async def fetch_cached(self, session, url): """带缓存的请求""" cache_key = f"news_{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_size=5): """并行处理文章列表""" 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_exceptions=True) # 过滤成功的结果 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_dir='logs'): self.log_dir = Path(log_dir) self.log_dir.mkdir(exist_ok=True) 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', maxBytes=10*1024*1024, # 10MB backupCount=5 ) 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。这个工具不仅能够自动获取和处理新闻内容，还具备了生产环境所需的稳定性、安全性和可维护性。
