如何编写自动化新闻简报脚本

wen 实用脚本 2026-08-01 198

本文目录导读：

如何编写自动化新闻简报脚本

基础架构设计
多源新闻抓取实现
智能过滤与分类
邮件发送功能
主程序 - 定时调度
部署与优化建议
扩展功能想法

我来帮你设计一个自动化新闻简报脚本,以下是完整方案：

基础架构设计

# news_briefing.py
import requests
from datetime import datetime
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from bs4 import BeautifulSoup
import schedule
import time
class NewsBriefingSystem:
    def __init__(self):
        self.sources = []
        self.articles = []
        self.config = self.load_config()
    def load_config(self):
        """加载配置文件"""
        try:
            with open('config.json', 'r', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            return {
                "email_settings": {},
                "sources": []
            }
    def fetch_news(self):
        """获取新闻"""
        raise NotImplementedError("子类需要实现此方法")
    def filter_news(self, keywords):
        """按关键词过滤新闻"""
        filtered = []
        for article in self.articles:
            if any(keyword.lower() in article['title'].lower() 
                   or keyword.lower() in article['summary'].lower() 
                   for keyword in keywords):
                filtered.append(article)
        return filtered
    def generate_briefing(self, news_list):
        """生成简报"""
        briefing = f"""
        ===== 新闻简报 =====
        日期：{datetime.now().strftime('%Y-%m-%d %H:%M')}
        类别：每日重要新闻
        =================================
        """
        for i, news in enumerate(news_list, 1):
            briefing += f"""
        {i}. {news['title']}
        来源：{news['source']}
        时间：{news['time']}
        {news['summary']}
        链接：{news['link']}
        ---------------------------------"""
        return briefing

多源新闻抓取实现

# multi_source_fetcher.py
import requests
from bs4 import BeautifulSoup
import feedparser
from datetime import datetime
import re
class NewsFetcher:
    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 fetch_from_rss(self, rss_urls):
        """从RSS源获取新闻"""
        articles = []
        for url in rss_urls:
            try:
                feed = feedparser.parse(url)
                for entry in feed.entries[:5]:  # 每个源取5条
                    articles.append({
                        'title': entry.title,
                        'summary': entry.get('summary', ''),
                        'link': entry.link,
                        'source': url,
                        'time': entry.get('published', datetime.now().isoformat())
                    })
            except Exception as e:
                print(f"Error fetching from {url}: {e}")
        return articles
    def fetch_from_website(self, website_urls):
        """从网站抓取新闻"""
        articles = []
        for url in website_urls:
            try:
                response = self.session.get(url)
                soup = BeautifulSoup(response.content, 'html.parser')
                # 根据实际页面结构调整选择器
                news_items = soup.select('.news-item, article, .story-item')
                for item in news_items[:5]:
                    title_tag = item.select_one('h2, h3, .title a')
                    link_tag = item.select_one('a')
                    summary_tag = item.select_one('.summary, .description, p')
                    if title_tag and link_tag:
                        articles.append({
                            'title': title_tag.text.strip(),
                            'summary': summary_tag.text.strip() if summary_tag else '',
                            'link': link_tag['href'],
                            'source': url,
                            'time': datetime.now().isoformat()
                        })
            except Exception as e:
                print(f"Error fetching from {url}: {e}")
        return articles
    def fetch_from_api(self, api_endpoints):
        """从API获取新闻"""
        articles = []
        for endpoint in api_endpoints:
            try:
                response = self.session.get(endpoint['url'], 
                                          params=endpoint.get('params', {}))
                if response.status_code == 200:
                    data = response.json()
                    # 根据API响应结构调整解析逻辑
                    if 'articles' in data:
                        for article in data['articles'][:5]:
                            articles.append({
                                'title': article.get('title', ''),
                                'summary': article.get('description', ''),
                                'link': article.get('url', ''),
                                'source': article.get('source', {}).get('name', ''),
                                'time': article.get('publishedAt', '')
                            })
            except Exception as e:
                print(f"Error fetching from API {endpoint.get('url')}: {e}")
        return articles
# 使用示例
if __name__ == "__main__":
    fetcher = NewsFetcher()
    # 配置来源
    rss_sources = [
        "http://rss.cnn.com/rss/cnn_topstories.rss",
        "http://feeds.bbci.co.uk/news/rss.xml",
        "http://feeds.foxnews.com/foxnews/top-stories"
    ]
    website_sources = [
        "https://news.google.com/",
        "https://www.reuters.com/"
    ]
    # 获取新闻
    all_news = []
    all_news.extend(fetcher.fetch_from_rss(rss_sources))
    all_news.extend(fetcher.fetch_from_website(website_sources))
    print(f"共获取 {len(all_news)} 条新闻")

智能过滤与分类

# news_filter.py
from datetime import datetime, timedelta
import re
from collections import Counter
class NewsFilter:
    def __init__(self):
        self.categories = {
            'technology': ['AI', '区块链', 'tech', '科技', 'digital'],
            'finance': ['economics', 'stocks', '市场', '金融', '经济'],
            'international': ['global', 'international', '国际', '外交'],
            'politics': ['politics', 'election', '政治', '选举'],
            'sports': ['sports', 'football', '体育', '足球']
        }
    def filter_by_recency(self, articles, hours=24):
        """按时间过滤"""
        cutoff = datetime.now() - timedelta(hours=hours)
        recent = []
        for article in articles:
            try:
                pub_time = datetime.fromisoformat(article['time'].replace('Z', '+00:00'))
                if pub_time > cutoff:
                    recent.append(article)
            except:
                # 时间解析失败，默认保留
                recent.append(article)
        return recent
    def filter_by_keywords(self, articles, keywords):
        """按关键词过滤"""
        filtered = []
        for article in articles:
            text = f"{article['title']} {article['summary']}".lower()
            if any(keyword.lower() in text for keyword in keywords):
                filtered.append(article)
        return filtered
    def categorize_news(self, articles):
        """新闻分类"""
        categorized = {key: [] for key in self.categories}
        categorized['other'] = []
        for article in articles:
            text = f"{article['title']} {article['summary']}".lower()
            category_found = False
            for category, keywords in self.categories.items():
                if any(keyword.lower() in text for keyword in keywords):
                    categorized[category].append(article)
                    category_found = True
                    break
            if not category_found:
                categorized['other'].append(article)
        return categorized
    def remove_duplicates(self, articles):
        """去重"""
        seen_titles = set()
        unique_articles = []
        for article in articles:
            title_hash = article['title'].strip().lower()
            if title_hash not in seen_titles:
                seen_titles.add(title_hash)
                unique_articles.append(article)
        return unique_articles
    def extract_hot_topics(self, articles, top_n=5):
        """提取热门主题"""
        keywords = []
        stop_words = {'的', '了', '和', '是', 'in', 'the', 'a', 'an'}
        for article in articles:
            words = re.findall(r'\w+', article['title'].lower())
            keywords.extend([w for w in words if w not in stop_words])
        counter = Counter(keywords)
        return counter.most_common(top_n)

邮件发送功能

# email_sender.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
import os
from pathlib import Path
class EmailSender:
    def __init__(self, smtp_config):
        self.smtp_host = smtp_config.get('host', 'smtp.gmail.com')
        self.smtp_port = smtp_config.get('port', 587)
        self.username = smtp_config.get('username')
        self.password = smtp_config.get('password')
        self.use_ssl = smtp_config.get('use_ssl', False)
    def send_email(self, to_emails, subject, body):
        """发送邮件"""
        msg = MIMEMultipart('alternative')
        msg['Subject'] = Header(subject, 'utf-8')
        msg['From'] = self.username
        msg['To'] = ', '.join(to_emails)
        # HTML格式邮件
        html_part = MIMEText(self.to_html(body), 'html', 'utf-8')
        msg.attach(html_part)
        try:
            if self.use_ssl:
                server = smtplib.SMTP_SSL(self.smtp_host, 465)
            else:
                server = smtplib.SMTP(self.smtp_host, self.smtp_port)
                server.starttls()
            server.login(self.username, self.password)
            server.send_message(msg)
            server.quit()
            print("邮件发送成功!")
            return True
        except Exception as e:
            print(f"邮件发送失败: {e}")
            return False
    def to_html(self, text):
        """将文本转换为HTML"""
        # 简单转换，可以根据需要美化
        html = f"""
        <html>
          <body>
            <pre style="font-family: Arial, sans-serif; font-size: 14px;">
              {text}
            </pre>
          </body>
        </html>
        """
        return html

主程序 - 定时调度

# main.py
import schedule
import time
from datetime import datetime
import json
from typing import List, Dict
# 导入之前的模块
from multi_source_fetcher import NewsFetcher
from news_filter import NewsFilter
from email_sender import EmailSender
class DailyNewsBriefing:
    def __init__(self, config_file='config.json'):
        with open(config_file, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
        self.fetcher = NewsFetcher()
        self.filter = NewsFilter()
        self.email_sender = EmailSender(self.config['smtp'])
        self.news_list = []
    def update_news(self):
        """更新新闻"""
        print(f"[{datetime.now()}] 开始更新新闻...")
        # 1. 获取新闻
        all_news = []
        # RSS源
        if self.config['sources'].get('rss'):
            rss_news = self.fetcher.fetch_from_rss(
                self.config['sources']['rss']
            )
            all_news.extend(rss_news)
        # 网站
        if self.config['sources'].get('websites'):
            web_news = self.fetcher.fetch_from_website(
                self.config['sources']['websites']
            )
            all_news.extend(web_news)
        # API
        if self.config['sources'].get('apis'):
            api_news = self.fetcher.fetch_from_api(
                self.config['sources']['apis']
            )
            all_news.extend(api_news)
        # 2. 清洗数据
        all_news = self.filter.remove_duplicates(all_news)
        all_news = self.filter.filter_by_recency(all_news, 24)
        # 3. 根据偏好过滤
        if self.config.get('keywords'):
            all_news = self.filter.filter_by_keywords(
                all_news, self.config['keywords']
            )
        self.news_list = all_news
        print(f"更新完成，共 {len(all_news)} 条新闻")
    def generate_and_send(self):
        """生成并发送简报"""
        if not self.news_list:
            print("没有新闻数据")
            return
        # 生成简报文本
        briefing = self.generate_briefing()
        # 发送邮件
        subject = f"每日新闻简报 - {datetime.now().strftime('%Y-%m-%d')}"
        to_emails = self.config['email_recipients']
        self.email_sender.send_email(to_emails, subject, briefing)
    def generate_briefing(self):
        """生成简报内容"""
        briefing = f"""
        ===== 每日新闻简报 =====
        日期：{datetime.now().strftime('%Y-%m-%d %H:%M')}
        获取新闻数：{len(self.news_list)}
        ===========================
        """
        for idx, news in enumerate(self.news_list, 1):
            briefing += f"""
        {idx}. {news['title']}
        来源：{news['source']}
        时间：{news['time'][:16]}
        {news['summary'][:200]}...
        链接：{news['link']}
        ---------------------------
        """
        return briefing
    def run_daily(self, schedule_time="08:00"):
        """每日定时运行"""
        # 立即运行一次
        self.update_news()
        self.generate_and_send()
        # 设置定时任务
        schedule.every().day.at(schedule_time).do(
            lambda: (self.update_news(), self.generate_and_send())
        )
        print(f"定时任务已设置，每天 {schedule_time} 发送简报")
        # 主循环
        while True:
            schedule.run_pending()
            time.sleep(60)
# 配置文件示例
CONFIG_EXAMPLE = {
    "sources": {
        "rss": [
            "http://rss.cnn.com/rss/cnn_topstories.rss",
            "http://feeds.bbci.co.uk/news/rss.xml"
        ],
        "websites": [
            "https://news.ycombinator.com/",
            "https://www.reddit.com/r/worldnews/"
        ],
        "apis": []
    },
    "keywords": ["AI", "technology", "climate"],
    "smtp": {
        "host": "smtp.gmail.com",
        "port": 587,
        "username": "your_email@gmail.com",
        "password": "your_app_password",
        "use_ssl": False
    },
    "email_recipients": ["receiver1@example.com", "receiver2@example.com"]
}
if __name__ == "__main__":
    # 创建配置文件
    with open('config.json', 'w', encoding='utf-8') as f:
        json.dump(CONFIG_EXAMPLE, f, ensure_ascii=False, indent=2)
    # 启动程序
    briefing_system = DailyNewsBriefing()
    briefing_system.run_daily("08:00")

部署与优化建议

Docker部署

# Dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "main.py"]

依赖配置

# requirements.txt
requests>=2.26.0
beautifulsoup4>=4.10.0
feedparser>=6.0.8
schedule>=1.1.0
python-dotenv>=0.19.0
# 环境变量管理（.env文件）
# SMTP_USER=your_email@gmail.com
# SMTP_PASSWORD=your_app_password

日志记录

import logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('news_briefing.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)