【Python】RSSの更新を検知したらメールで通知する方法

<rss version="2.0">
  <channel>
    <title>Example RSS Feed</title>
    <link>http://www.example.com/</link>
    <description>This is an example RSS feed</description>
    <item>
      <title>First Article</title>
      <link>http://www.example.com/first-article</link>
      <description>This is the first article</description>
      <pubDate>Mon, 01 Jan 2023 00:00:00 +0000</pubDate>
    </item>
    <item>
      <title>Second Article</title>
      <link>http://www.example.com/second-article</link>
      <description>This is the second article</description>
      <pubDate>Tue, 02 Jan 2023 00:00:00 +0000</pubDate>
    </item>
  </channel>
</rss>

pip install feedparser

import feedparser
# RSSフィードのURLを指定
rss_url = "https://example.com/rss"
feed = feedparser.parse(rss_url)
# フィードのタイトルを表示
print(feed.feed.title)

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
# SMTPサーバーの設定
smtp_server = "smtp.example.com"
smtp_port = 587
smtp_user = "[email protected]"
smtp_password = "your_password"
# メールの設定
msg = MIMEMultipart()
msg['From'] = smtp_user
msg['To'] = "[email protected]"
msg['Subject'] = "RSSフィードの更新通知"
# メールの本文
body = "RSSフィードが更新されました。"
msg.attach(MIMEText(body, 'plain'))
# メールの送信
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()
server.login(smtp_user, smtp_password)
server.sendmail(smtp_user, "[email protected]", msg.as_string())
server.quit()

pip install requests

pip install schedule

pip install beautifulsoup4

import feedparser
# RSSフィードのURLを指定
rss_url = "https://example.com/rss"
# RSSフィードを取得して解析
feed = feedparser.parse(rss_url)
# フィードのタイトルを表示
print("フィードのタイトル:", feed.feed.title)
# 各エントリーのタイトルとリンクを表示
for entry in feed.entries:
    print("エントリーのタイトル:", entry.title)
    print("エントリーのリンク:", entry.link)

import feedparser
# TechCrunchのRSSフィードのURLを指定
rss_url = "http://feeds.feedburner.com/TechCrunch/"
# RSSフィードを取得して解析
feed = feedparser.parse(rss_url)
# フィードのタイトルを表示
print("フィードのタイトル:", feed.feed.title)
# 各エントリーのタイトルとリンクを表示
for entry in feed.entries:
    print("エントリーのタイトル:", entry.title)
    print("エントリーのリンク:", entry.link)

import json
# 前回の更新情報を保存する関数
def save_last_update(data, filename='last_update.json'):
    with open(filename, 'w') as file:
        json.dump(data, file)
# 前回の更新情報を読み込む関数
def load_last_update(filename='last_update.json'):
    try:
        with open(filename, 'r') as file:
            return json.load(file)
    except FileNotFoundError:
        return None

import sqlite3
# データベースの初期化
def init_db(db_name='rss_updates.db'):
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS updates (
            feed_url TEXT PRIMARY KEY,
            last_update TEXT
        )
    ''')
    conn.commit()
    conn.close()
# 前回の更新情報を保存する関数
def save_last_update(feed_url, last_update, db_name='rss_updates.db'):
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    cursor.execute('''
        INSERT OR REPLACE INTO updates (feed_url, last_update)
        VALUES (?, ?)
    ''', (feed_url, last_update))
    conn.commit()
    conn.close()
# 前回の更新情報を読み込む関数
def load_last_update(feed_url, db_name='rss_updates.db'):
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    cursor.execute('''
        SELECT last_update FROM updates WHERE feed_url = ?
    ''', (feed_url,))
    result = cursor.fetchone()
    conn.close()
    return result[0] if result else None

import feedparser
from datetime import datetime
# RSSフィードのURL
feed_url = 'https://example.com/rss'
# 前回の更新情報を読み込む
last_update = load_last_update(feed_url)
# RSSフィードを取得
feed = feedparser.parse(feed_url)
# 更新を検知する
new_entries = []
for entry in feed.entries:
    entry_time = datetime(*entry.updated_parsed[:6])
    if last_update is None or entry_time > datetime.fromisoformat(last_update):
        new_entries.append(entry)
# 新しいエントリーがあれば、前回の更新情報を更新
if new_entries:
    save_last_update(feed.entries[0].updated, feed_url)

# 前回の更新情報を読み込む
last_update_ids = load_last_update(feed_url)
# RSSフィードを取得
feed = feedparser.parse(feed_url)
# 更新を検知する
new_entries = []
current_ids = [entry.id for entry in feed.entries]
for entry in feed.entries:
    if last_update_ids is None or entry.id not in last_update_ids:
        new_entries.append(entry)
# 新しいエントリーがあれば、前回の更新情報を更新
if new_entries:
    save_last_update(current_ids, feed_url)

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
# GmailのSMTPサーバー情報
smtp_server = "smtp.gmail.com"
smtp_port = 587
username = "[email protected]"
password = "your_password"
# メールの送信先
to_email = "[email protected]"
# メールの内容
subject = "RSSフィードの更新通知"
body = "新しい記事が公開されました。"
# メールの作成
msg = MIMEMultipart()
msg["From"] = username
msg["To"] = to_email
msg["Subject"] = subject
msg.attach(MIMEText(body, "plain"))
# SMTPサーバーに接続してメールを送信
try:
    server = smtplib.SMTP(smtp_server, smtp_port)
    server.starttls()  # TLS暗号化を開始
    server.login(username, password)
    server.sendmail(username, to_email, msg.as_string())
    print("メールが送信されました")
except Exception as e:
    print(f"メールの送信に失敗しました: {e}")
finally:
    server.quit()

# YahooメールのSMTPサーバー情報
smtp_server = "smtp.mail.yahoo.com"
smtp_port = 587
username = "[email protected]"
password = "your_password"

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
# メールの送信先
to_email = "[email protected]"
# メールの内容
subject = "RSSフィードの更新通知"
body = "新しい記事が公開されました。"
# メールの作成
msg = MIMEMultipart()
msg["From"] = username
msg["To"] = to_email
msg["Subject"] = subject
msg.attach(MIMEText(body, "plain"))

# HTMLメールの内容
html_body = """
<html>
  <body>
    <h1>RSSフィードの更新通知</h1>
    <p>新しい記事が公開されました。<br>
       <a href="https://example.com">記事を読む</a>
    </p>
  </body>
</html>
"""
# HTMLメールの作成
msg.attach(MIMEText(html_body, "html"))

import feedparser
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import os
import json
# RSSフィードのURL
RSS_URL = "https://example.com/rss"
# 前回の更新情報を保存するファイル
LAST_UPDATE_FILE = "last_update.json"
# メール設定
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
SMTP_USER = "[email protected]"
SMTP_PASSWORD = "your_password"
TO_EMAIL = "[email protected]"
def get_last_update():
    if os.path.exists(LAST_UPDATE_FILE):
        with open(LAST_UPDATE_FILE, 'r') as file:
            return json.load(file)
    return None
def save_last_update(last_update):
    with open(LAST_UPDATE_FILE, 'w') as file:
        json.dump(last_update, file)
def send_email(subject, body):
    msg = MIMEMultipart()
    msg['From'] = SMTP_USER
    msg['To'] = TO_EMAIL
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'plain'))
    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(SMTP_USER, SMTP_PASSWORD)
        server.sendmail(SMTP_USER, TO_EMAIL, msg.as_string())
def check_rss_update():
    feed = feedparser.parse(RSS_URL)
    last_update = get_last_update()
    if not last_update or feed.entries[0].published != last_update['published']:
        latest_entry = feed.entries[0]
        subject = f"New RSS Update: {latest_entry.title}"
        body = f"Title: {latest_entry.title}\nLink: {latest_entry.link}\nPublished: {latest_entry.published}"
        send_email(subject, body)
        save_last_update({'published': latest_entry.published})
if __name__ == "__main__":
    check_rss_update()

def get_last_update():
    try:
        if os.path.exists(LAST_UPDATE_FILE):
            with open(LAST_UPDATE_FILE, 'r') as file:
                return json.load(file)
    except Exception as e:
        print(f"Error reading last update file: {e}")
    return None
def save_last_update(last_update):
    try:
        with open(LAST_UPDATE_FILE, 'w') as file:
            json.dump(last_update, file)
    except Exception as e:
        print(f"Error saving last update file: {e}")
def send_email(subject, body):
    try:
        msg = MIMEMultipart()
        msg['From'] = SMTP_USER
        msg['To'] = TO_EMAIL
        msg['Subject'] = subject
        msg.attach(MIMEText(body, 'plain'))
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SMTP_USER, SMTP_PASSWORD)
            server.sendmail(SMTP_USER, TO_EMAIL, msg.as_string())
    except Exception as e:
        print(f"Error sending email: {e}")
def check_rss_update():
    try:
        feed = feedparser.parse(RSS_URL)
        last_update = get_last_update()
        if not last_update or feed.entries[0].published != last_update['published']:
            latest_entry = feed.entries[0]
            subject = f"New RSS Update: {latest_entry.title}"
            body = f"Title: {latest_entry.title}\nLink: {latest_entry.link}\nPublished: {latest_entry.published}"
            send_email(subject, body)
            save_last_update({'published': latest_entry.published})
    except Exception as e:
        print(f"Error checking RSS update: {e}")
if __name__ == "__main__":
    check_rss_update()

import logging
# ログの設定
logging.basicConfig(filename='rss_update.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def get_last_update():
    try:
        if os.path.exists(LAST_UPDATE_FILE):
            with open(LAST_UPDATE_FILE, 'r') as file:
                return json.load(file)
    except Exception as e:
        logging.error(f"Error reading last update file: {e}")
    return None
def save_last_update(last_update):
    try:
        with open(LAST_UPDATE_FILE, 'w') as file:
            json.dump(last_update, file)
    except Exception as e:
        logging.error(f"Error saving last update file: {e}")
def send_email(subject, body):
    try:
        msg = MIMEMultipart()
        msg['From'] = SMTP_USER
        msg['To'] = TO_EMAIL
        msg['Subject'] = subject
        msg.attach(MIMEText(body, 'plain'))
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SMTP_USER, SMTP_PASSWORD)
            server.sendmail(SMTP_USER, TO_EMAIL, msg.as_string())
        logging.info(f"Email sent: {subject}")
    except Exception as e:
        logging.error(f"Error sending email: {e}")
def check_rss_update():
    try:
        feed = feedparser.parse(RSS_URL)
        last_update = get_last_update()
        if not last_update or feed.entries[0].published != last_update['published']:
            latest_entry = feed.entries[0]
            subject = f"New RSS Update: {latest_entry.title}"
            body = f"Title: {latest_entry.title}\nLink: {latest_entry.link}\nPublished: {latest_entry.published}"
            send_email(subject, body)
            save_last_update({'published': latest_entry.published})
            logging.info(f"RSS update detected: {latest_entry.title}")
    except Exception as e:
        logging.error(f"Error checking RSS update: {e}")
if __name__ == "__main__":
    check_rss_update()

crontab -e

0 9 * * * /usr/bin/python3 /path/to/your_script.py

import time
import feedparser
import smtplib
def check_rss_and_send_email():
    # RSSフィードの取得と更新の検知
    # メールの送信
    pass
# 10分ごとに実行
interval = 600
while True:
    check_rss_and_send_email()
    time.sleep(interval)

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
# メールサーバーの設定
smtp_server = "smtp.gmail.com"
smtp_port = 587
smtp_user = "[email protected]"
smtp_password = "your_password"
# メールの内容を作成
msg = MIMEMultipart()
msg['From'] = smtp_user
msg['To'] = "[email protected]"
msg['Subject'] = "RSSフィードの更新通知"
body = "RSSフィードに新しい更新があります。"
msg.attach(MIMEText(body, 'plain'))
# SMTPサーバーに接続し、メールを送信
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()  # TLSを開始
server.login(smtp_user, smtp_password)
server.sendmail(smtp_user, "[email protected]", msg.as_string())
server.quit()

import os
smtp_user = os.getenv("SMTP_USER")
smtp_password = os.getenv("SMTP_PASSWORD")

import feedparser
rss_url = "https://example.com/rss"
feed = feedparser.parse(rss_url)

import os
api_key = os.getenv("API_KEY")

# .gitignore
config.ini