気になるニュースサイトが増えると、巡回するだけで時間が溶けます。そこでPythonのrequestsでRSSを取得し、feedparserで解析して、sqlite3へ新着だけを重複なく溜めるCLIツールを作りました。collectで集めてlistで読む2サブコマンド構成で、--demoを付ければ外部通信なしで動作を確認できます。
重複排除はuidのUNIQUE制約とINSERT OR IGNOREに任せる設計です。実際に4回のコマンドで動かした結果まで、手順を追いながら解説します。
requestsの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。
動画の内容をテキストで確認する
オープニング。requests・feedparser・SQLiteを使ってRSSニュース収集CLIを作るカリキュラムを始めます。概要紹介。
requests・feedparser・SQLiteの役割と使い方を学ぶRSSニュース収集CLIの要件を整理する 完成コードと実行結果を確認する 最後に実コマンドとファイル状態で完成挙動を確かめる 具体的にやること。
requests.SessionへUser-Agentを設定してRSSをGETする タイムアウト10秒を指定してフィードを取得する 取得に失敗したフィードだけをスキップして次へ進むfeedparser.parseでタイトルとリンクと公開日時と概要を取り出すentryのidかlinkからuidを決める 実装環境・必須アプリ。
OS：Windows 11 Pro Python：3.13.3シェル：PowerShell 5.1必須アプリ：コードエディター、ターミナル、エクスプローラー パッケージ：pip、requests、feedparser requests・feedparser・SQLiteとは。
requests：requestsは、URLを渡すだけでWebページやRSSのXMLを受け取れるHTTP通信のライブラリfeedparser：feedparserは、RSSやAtomのXMLを辞書のように読み取れる解析ライブラリ requestsで作るRSS収集CLIの要点。
INSERT OR IGNOREでUNIQUE違反の行を無視rowcountが正の値のときだけ新着として加算 解析件数から新着件数を引いて重複スキップ件数を算出 RSSニュース収集CLIの要件定義。
collect --demoで2本のフィードから解析5件と新着5件を表示する2サイクル目に解析6件と新着1件と重複スキップ5件を表示する 最終行へ保存先と今回の新着件数と累計件数を出すlistで公開日時の新しい順に指定件数を表示する 保存が空のときはcollectでの取得を促す案内を返す --feedも--demoも無い場合に指定方法を案内する INTRO: Monaco EditorでRSSニュース収集CLIを実装。
コードを1行ずつ入力し、補完と自動インデントを使いながら実行結果を確認します。LINE 001: モジュールdocstringの開始。このファイル全体がRSSニュース収集CLIであることを説明するdocstringの開始行です。
ファイルの目的を最初に示す役割を持ちます。LINE 003: 取得と解析の説明文。requestsでRSSフィードをHTTP取得し、feedparserで記事を解析する処理があることを説明しています。
この後に定義される各関数の役割の概要となります。LINE 004: 保存処理の説明文。SQLiteへ新着記事を重複なく保存する処理があることを説明しています。
次の行の一覧表示処理の説明につながります。LINE 005: 定期取得と一覧表示の説明文。複数サイクルで定期的に取得し、保存済みニュースを一覧表示する処理があることを説明しています。
それぞれの処理を関数に分けて実装している方針を伝えます。LINE 006: docstringの終了。ファイル全体の説明文であるdocstringを閉じる行です。
この後から実際のコードが始まります。LINE 008: 将来の型ヒント構文を有効化。annotationsを先読みで有効にすることで、型ヒントの記述を後方互換性の高い形にしています。
関数の引数や戻り値の型指定に影響します。LINE 010: argparseモジュールの読み込み。コマンドライン引数を解析するためのargparseモジュールを読み込んでいます。
CLIの引数定義に使われます。LINE 011: sqlite3モジュールの読み込み。SQLiteデータベースを操作するためのsqlite3モジュールを読み込んでいます。
記事の保存や取得に使われます。LINE 012: threadingモジュールの読み込み。デモ用HTTPサーバーを別スレッドで動かすためのthreadingモジュールを読み込んでいます。
メイン処理をブロックせずにサーバーを起動できます。LINE 013: timeモジュールの読み込み。日時のフォーマットやサイクル間の待機処理に使うtimeモジュールを読み込んでいます。
sleepや文字列変換に利用されます。LINE 014: 日時関連クラスの読み込み。現在時刻や時間差、タイムゾーンを扱うためにdatetime、timedelta、timezoneを読み込んでいます。
デモデータの日時計算に使われます。LINE 015: RFC822形式日時変換関数の読み込み。datetimeをRSSのpubDate形式の文字列に変換するformat_datetime関数を読み込んでいます。
デモRSSの生成で使われます。LINE 016: 簡易HTTPサーバー関連クラスの読み込み。デモ用のRSS配信サーバーを作るためにBaseHTTPRequestHandlerとThreadingHTTPServerを読み込んでいます。
外部通信なしで動作確認するための土台になります。LINE 018: feedparserモジュールの読み込み。RSS/AtomフィードのXMLを解析するためのfeedparserライブラリを読み込んでいます。
記事の抽出処理で使われます。LINE 019: requestsモジュールの読み込み。HTTP通信を行うためのrequestsライブラリを読み込んでいます。
RSSフィードの取得処理で使われます。LINE 023: 既定のDBファイル名を定義。SQLiteのデータベースファイル名の既定値を定義しています。
CLIで--dbオプションが指定されなかった場合に使われます。LINE 024: 既定のサイクル数を定義。取得サイクルの回数の既定値を定義しています。
--cyclesオプションが指定されなかった場合に使われます。LINE 025: 既定の待機秒数を定義。サイクル間の待機秒数の既定値を定義しています。
--intervalオプションが指定されなかった場合に使われます。LINE 026: 既定の表示件数を定義。一覧表示する記事件数の既定値を定義しています。
--limitオプションが指定されなかった場合に使われます。LINE 027: HTTPリクエストのタイムアウト秒数を定義。requestsでフィードを取得する際のタイムアウト秒数を定義しています。
応答が遅い場合に処理が止まり続けないようにする役割を持ちます。LINE 028: User-Agent文字列を定義。HTTPリクエスト送信時に名乗るUser-Agent文字列を定義しています。
サーバー側でクライアントを識別しやすくする役割を持ちます。RUN 1/9: 基本設定の値を確かめる。既定のデータベース名やサイクル数、タイムアウト、User-Agentを定義した段階です。
printで読み出し、後の関数が参照する初期値を把握します。CHECK 1/9: 途中実行に成功。既定のDB news.db / 既定の表示件数20 既定のサイクル1 / 既定の間隔60.0秒 タイムアウト10.0秒 / User-Agent rss-news-collector/1.0 RETURN 01: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 032: デモフィードのパス対応表を開始。デモ配信するパスとフィード名を対応させる辞書DEMO_FEEDSの定義を開始しています。
この辞書がデモサーバーの配信内容の元になります。LINE 033: ビジネス系デモフィードのパス。/biz.xmlというパスに「サンプルビジネス通信」というフィード名を対応させています。
デモ用に用意された1本目のフィードです。LINE 034: テック系デモフィードのパス。/tech.xmlというパスに「サンプルテック速報」というフィード名を対応させています。
デモ用に用意された2本目のフィードです。LINE 035: デモフィード辞書の終了。DEMO_FEEDS辞書の定義を閉じる行です。
この後、この辞書を参照して各種処理が行われます。LINE 036: デモの状態管理用辞書を定義。サイクルごとに増える記事数を管理するためのDEMO_STATE辞書を定義しています。
extraキーの値が増えることでデモの記事件数が変化します。LINE 037: デモ記事の基準日時を定義。デモ記事の公開日時を計算する際の基準となる日時DEMO_BASE_TIMEを定義しています。
この日時に時間を加算して各記事の公開日時を作ります。LINE 040: デモRSS生成関数の定義開始。指定したパスに対応するデモRSSのXML文字列を生成する関数build_demo_rssを定義しています。
この関数がデモサーバーの応答本文を作る役割を持ちます。LINE 041: 関数の説明docstring。この関数がRSS 2.0形式のXMLを作り、サイクルごとに記事が1件増える仕組みであることを説明しています。
関数の挙動を理解する手がかりになります。LINE 042: フィードタイトルの取得。DEMO_FEEDS辞書からパスに対応するフィードタイトルを取得しています。
この値がXML内のtitleタグに使われます。LINE 043: パスからスラッグ文字列を生成。パス文字列の前後スラッシュと拡張子を取り除いて、記事のリンクやguidに使う短い識別子slugを作っています。
URLやIDの一部として使われます。LINE 044: 記事リストの初期化。各記事のXML断片を格納していく空のリストitemsを用意しています。
この後のループでここに記事が追加されていきます。LINE 045: 記事数を決めるループ開始。1からDEMO_STATEのextra分だけ増えた件数までの記事を作るループを開始しています。
サイクルが進むごとにextraが増えるため記事数も増加します。LINE 046: 記事の公開日時を計算。基準日時に記事番号分の時間を加算し、RSSのpubDate形式に変換した公開日時を作っています。
記事ごとに異なる公開日時を持たせる役割です。LINE 047: 記事XML断片の追加開始。1件分の記事をitemタグとして組み立て、itemsリストに追加する処理を開始しています。
この後の複数行でタグの内容が構築されます。LINE 048: itemタグの開始。RSSのitem要素の開始タグを文字列として追加しています。
この記事に関する各種タグをこの中に含めていきます。LINE 049: タイトルタグの生成。フィードタイトルと記事番号を組み合わせて記事タイトルのtitleタグを作っています。
記事ごとに一意なタイトルになります。LINE 050: リンクタグの生成。スラッグと記事番号を使ってダミーのURLを持つlinkタグを作っています。
実在しないダミーURLですが実際のRSSと同じ形式です。LINE 051: guidタグの生成。記事を一意に識別するguidタグを作っています。
この値が後段でuidとして重複判定に使われる元になります。LINE 052: 公開日時タグの生成。先に計算した公開日時を使ってpubDateタグを作っています。
記事の公開日時情報としてfeedparser側で解析されます。LINE 053: 本文概要タグの生成。フィードタイトルと記事番号を使ってダミーの本文を持つdescriptionタグを作っています。
記事の概要として扱われる部分です。LINE 054: itemタグの終了。RSSのitem要素の終了タグを文字列として追加しています。
1件分の記事の組み立てがここで完了します。LINE 055: items.appendの呼び出し終了。組み立てた記事のXML文字列をitemsリストへ追加する呼び出しを終える行です。
ループが繰り返されるたびにこの処理が行われます。LINE 056: 戻り値の組み立て開始。完成したRSS全体のXML文字列を返すためのreturn文を開始しています。
この後複数の文字列を連結して最終的なXMLを作ります。LINE 057: XML宣言行。XML文書の先頭に必要なXML宣言を文字列として追加しています。
エンコーディングをUTF-8と明示しています。LINE 058: rss要素の開始タグ。RSS 2.0形式であることを示すrss要素の開始タグを追加しています。
この中にchannel要素が含まれます。LINE 059: channel要素の開始タグ。フィード全体の情報を格納するchannel要素の開始タグを追加しています。
この中にタイトルや記事一覧が含まれます。LINE 060: チャンネルタイトルタグ。フィード全体のタイトルを示すtitleタグを追加しています。
DEMO_FEEDSから取得したフィード名が使われます。LINE 061: チャンネルリンクタグ。フィード全体のリンク先を示すlinkタグを追加しています。
スラッグを使ったダミーのURLが設定されます。LINE 062: チャンネル説明タグ。フィードの説明文を示すdescriptionタグを追加しています。
デモ用であることを明示する固定文が入っています。LINE 063: 記事一覧の連結。先ほど作成したitemsリストの各記事XMLを改行で連結して1つの文字列にしています。
全記事分のitem要素をまとめて挿入する役割です。LINE 064: channel要素とrss要素の終了タグ。channel要素とrss要素の終了タグを追加してXML全体を閉じています。
これでRSS 2.0形式のXML文字列が完成します。LINE 065: return文の終了。組み立てた文字列全体を関数の戻り値として返す処理を終える行です。
この関数の呼び出し元でXML文字列として利用されます。RUN 2/9: デモ用RSSのXMLを組み立てる。build_demo_rssが完成した段階です。
生成されたXMLのitem数とタイトル行を表示して、RSS 2.0の形になっているか確認します。CHECK 2/9: 途中実行に成功。
item数5 行数43 ['<title>サンプルテック速報</title>', '<title>サンプルテック速報 記事01</title>', '<title>サンプルテック速報 記事02</title>'] RETURN 02: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 068: デモ用HTTPハンドラクラスの定義。デモRSSを配信するためのHTTPリクエストハンドラクラスDemoFeedHandlerを定義しています。BaseHTTPRequestHandlerを継承して独自の応答処理を実装します。
LINE 069: クラスの説明docstring。このクラスがデモRSSのXMLを返すだけの簡易ハンドラであることを説明しています。クラスの役割を明確にする役割です。
LINE 071: GETリクエスト処理メソッドの定義。HTTPのGETリクエストを受け取ったときに呼ばれるdo_GETメソッドを定義しています。この中でデモRSSの配信処理を実装します。
LINE 072: クエリを除いたパスの取得。リクエストされたパス文字列からクエリ部分を取り除いて、純粋なパスのみを取得しています。DEMO_FEEDSのキーと比較するための準備です。
LINE 073: 対象パスの存在チェック。リクエストされたパスがDEMO_FEEDSに登録されているかどうかを確認しています。存在しない場合は後続処理でエラー応答を返す準備です。
LINE 074: 404エラー応答の送信。対象パスが見つからない場合に404エラーとメッセージをクライアントへ送信しています。存在しないフィードへのリクエストに対する応答です。
LINE 075: 処理の中断。404エラーを返した後、これ以上の処理を行わずメソッドを終了しています。正常な配信処理へ進まないようにする役割です。
LINE 076: デモRSS本文の生成とバイト変換。build_demo_rss関数でXML文字列を生成し、UTF-8のバイト列に変換しています。この後のHTTP応答本文として使われます。
LINE 077: HTTP 200応答の開始。リクエストが成功したことを示すHTTPステータスコード200を送信しています。正常にフィードを配信することをクライアントに伝えます。
LINE 078: Content-Typeヘッダーの送信。応答本文がRSS XMLであることを示すContent-Typeヘッダーを送信しています。文字エンコーディングもUTF-8と明示しています。
LINE 079: Content-Lengthヘッダーの送信。応答本文のバイト長を示すContent-Lengthヘッダーを送信しています。クライアントが正しく本文を受信するために必要な情報です。
LINE 080: レスポンスヘッダーの送信完了。ここまでに設定したステータスコードやヘッダーを確定し、クライアントへ送信します。この後は本文（ボディ）の送信に進みます。
LINE 081: レスポンス本文の書き込み。生成したデモRSSのXMLデータをレスポンスボディとしてクライアントへ書き込みます。これによりHTTPリクエストへの応答が完了します。
LINE 083: ログ出力メソッドの定義。BaseHTTPRequestHandlerが標準で行うアクセスログ出力を上書きするためのメソッドです。次の行で何も出力しない処理を実装します。
LINE 084: アクセスログの抑制。何もせずにそのまま処理を終えることで、コンソールへアクセスログが表示されないようにしています。これによりデモ実行時の出力がすっきりします。
LINE 087: デモサーバー起動関数の定義。localhostでデモRSSを配信するサーバーを起動し、サーバー本体とアクセス用URLを返す関数を定義します。この関数は外部通信を行わずに動作確認できるようにするためのものです。
LINE 088: 関数の説明ドキュメント。この関数がlocalhostの空いているポートを使ってデモRSSを配信し、外部への通信は行わないことを説明しています。LINE 089: HTTPサーバーインスタンスの作成。
127.0.0.1のポート0（自動割り当て）でリクエストを受け付けるサーバーを作成します。リクエスト処理には先ほど定義したDemoFeedHandlerを使います。LINE 090: サーバーをバックグラウンドで起動。
サーバーの待ち受け処理を別スレッドで実行し、メインの処理をブロックしないようにします。daemon=Trueによりメインプログラム終了時にこのスレッドも自動的に終わります。LINE 091: 割り当てられたホストとポートの取得。
OSが自動的に割り当てたホスト名とポート番号をサーバーのアドレス情報から取得します。この情報を使ってアクセス用のURLを組み立てます。LINE 092: サーバーとURLの返却。
起動したサーバーオブジェクトと、そこへアクセスするためのURL文字列をまとめて呼び出し元に返します。呼び出し側はこのURLを使ってデモフィードを取得できます。LINE 096: DB初期化関数の定義。
SQLiteデータベースへ接続し、記事テーブルを準備する関数を定義します。この関数はcollectコマンドとlistコマンドの両方から使われます。LINE 097: SQLite接続の作成。
指定されたパスのSQLiteデータベースファイルへ接続します。ファイルが存在しない場合は新規に作成されます。LINE 098: 行の取得形式を辞書風に設定。
クエリ結果の各行をカラム名でアクセスできるsqlite3.Row形式に設定します。これにより後の処理で行データを辞書のように扱えるようになります。LINE 099: テーブル作成SQLの実行開始。
articlesテーブルを作成するSQL文の実行を開始します。次の行から続くSQL文でテーブルの構造を定義しています。LINE 100: SQL文の開始（複数行文字列）。
テーブル作成用のSQL文を複数行の文字列として記述しています。この形式により見やすくカラム定義を並べられます。LINE 101: テーブル作成文とテーブルが無い場合のみ作成。
articlesという名前のテーブルが存在しない場合のみ作成する宣言です。既にテーブルがある場合は何もしないため、繰り返し実行しても安全です。LINE 102: id列の定義。
各記事に自動で連番を振るid列を定義しています。この列はテーブルの主キーとして使われます。LINE 103: uid列の定義とユニーク制約。
記事を一意に識別するuid列を定義し、UNIQUE制約を付けています。この制約が後段の重複保存を防ぐ仕組みの土台になります。LINE 104: source列の定義。
記事の取得元となるフィード名を保存するsource列を定義しています。空の値は許可されない設定です。LINE 105: title列の定義。
記事のタイトルを保存するtitle列を定義しています。この列も必ず値が必要な設定になっています。LINE 106: link列の定義。
記事本文へのリンクURLを保存するlink列を定義しています。この列も空値を許さない設定です。LINE 107: published列の定義。
記事の公開日時を保存するpublished列を定義しています。この列は後で新着順の並び替えに使われます。LINE 108: summary列の定義。
記事の概要文を保存するsummary列を定義しています。空文字であっても値自体は必ず設定される列です。LINE 109: fetched_at列の定義。
記事を取得・保存した日時を記録するfetched_at列を定義しています。取得タイミングの記録に使われます。LINE 110: カラム定義の終わり。
ここまでのカラム定義が終わり、テーブル定義の括弧を閉じています。LINE 111: SQL文の終端。複数行にわたるテーブル作成SQL文字列がここで終わります。
LINE 112: テーブル作成の実行完了。ここまで定義してきたSQL文の実行呼び出しを閉じ、テーブル作成処理が完了します。LINE 113: インデックス作成SQLの実行。
published列を対象にした検索用インデックスを作成するSQL文の実行を開始します。存在しない場合のみ作成するよう指定しています。LINE 114: 公開日時インデックスの定義。
articlesテーブルのpublished列に対して降順のインデックスを作成しています。これにより新しい記事の一覧表示が高速になります。LINE 115: インデックス作成の実行完了。
インデックス作成SQLの呼び出しを閉じ、実行を完了させています。LINE 116: 変更の確定。テーブルとインデックスの作成内容をデータベースへ確定（コミット）します。
これにより変更が実際にファイルへ反映されます。LINE 117: 接続オブジェクトの返却。初期化済みのSQLite接続オブジェクトを呼び出し元へ返します。
以降この接続を使ってデータの保存や読み込みを行います。RUN 3/9: articlesテーブルの列構成を確認する。init_dbを書き終えた段階です。
テーブルを作成して列名の一覧と現在の件数を表示し、uid列やpublished列が用意されたか確かめます。CHECK 3/9: 途中実行に成功。
列一覧 ['id', 'uid', 'source', 'title', 'link', 'published', 'summary', 'fetched_at'] インデックス ['idx_articles_published', 'sqlite_autoindex_articles_1'] 保存件数0 RETURN 03: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 122: フィード取得関数の定義。requestsのセッションを使ってRSSフィードをHTTP取得する関数を定義します。
取得できたバイナリデータと結果メッセージのタプルを返します。LINE 123: 例外処理の開始。通信エラーが起きても処理全体が止まらないよう、GETリクエストをtry文の中で実行します。
LINE 124: GETリクエストの送信。指定したURLへタイムアウト付きでGETリクエストを送信します。session経由で送ることで、共通のヘッダー設定が自動的に適用されます。
LINE 125: 通信例外の捕捉。接続エラーやタイムアウトなどのリクエスト例外が発生した場合にここで捕まえます。プログラム全体を停止させずに次の処理へ進めるためです。
LINE 126: 取得失敗メッセージの返却。通信例外が発生した場合に、失敗した旨と例外の種類名を含むメッセージを返します。値の取得結果はNoneとして呼び出し元に伝えます。
LINE 127: HTTPステータスの確認。レスポンスが正常なステータスコードであるかを確認しています。異常な場合は次の行で失敗として扱います。
LINE 128: HTTPエラー時のメッセージ返却。ステータスコードが異常だった場合に、そのステータスコードを含む失敗メッセージを返します。取得データはNoneとして扱われます。
LINE 129: 取得成功時の戻り値開始。通信が正常に完了した場合の戻り値を作る処理の開始です。次の行で本文データとメッセージをまとめます。
LINE 130: 取得した本文データ。レスポンスのバイナリ本文（RSSのXMLデータ）を戻り値の一部として渡します。この後の解析処理で利用されます。
LINE 131: 取得成功メッセージの作成。URLとステータスコード、取得したバイト数を含む成功メッセージを作成しています。この文字列はログ表示に使われます。
LINE 132: 戻り値タプルの終了。取得データと成功メッセージのタプルを完成させ、呼び出し元へ返す準備を終えています。RUN 4/9: デモRSSをrequestsで取得する。
fetch_feedが完成した段階です。デモ配信サーバを立ててGETし、返ってきたメッセージとバイト数から取得が成功したか確認します。CHECK 4/9: 途中実行に成功。
取得成功http://127.0.0.1:46583/biz.xml -> HTTP 200 / 1880バイト 受信バイト数1880 RETURN 04: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 137: 記事解析関数の定義。
取得したXMLデータからRSS記事の情報を抜き出す関数を定義します。戻り値は記事情報を持つ辞書のリストです。LINE 138: XMLデータの解析。
feedparserライブラリを使って生のXMLデータをパースし、フィード情報や記事一覧を持つオブジェクトを得ます。LINE 139: フィード名（source）の決定。フィード自体のタイトルを取得し、それを記事の取得元名として使います。
タイトルが取得できない場合はフィードのURLを代わりに使います。LINE 140: 記事リストの初期化。解析済みの記事情報を格納していく空のリストを用意します。
この後のループで記事ごとの辞書を追加していきます。LINE 141: 記事エントリーのループ開始。解析済みフィード内の各記事エントリーを一つずつ処理するループを開始します。
LINE 142: 記事リンクの取得。記事本文へのリンクを取得します。値が無い場合は空文字にして、前後の余分な空白は取り除きます。
LINE 143: 記事タイトルの取得。記事のタイトルを取得します。タイトルが存在しない場合は「（タイトルなし）」という文言を代わりに使います。
LINE 144: 記事の一意識別子（uid）の決定。記事を重複なく識別するためのuidを決定します。フィードが提供するidを優先し、無ければリンク、それも無ければソース名とタイトルの組み合わせを使います。
LINE 145: uidが空の記事のチェック開始。uidがどうしても空になってしまう記事があるかを確認しています。LINE 146: 無効な記事のスキップ。
uidが空の記事は保存に使えないため、この記事の処理をスキップして次のエントリーへ進みます。LINE 147: 公開日時の元データ取得。記事の公開日時情報を取得します。
公開日時が無い場合は更新日時のデータを代わりに使います。LINE 148: 公開日時の文字列変換。取得した日時の構造体データを「年-月-日 時:分:秒」形式の文字列に変換します。
日時情報が全く無い場合は空文字にします。LINE 149: 記事概要（summary）の整形。記事の概要文を取得し、改行を空白に置き換えたうえで先頭200文字までに切り取ります。
値が無い場合は空文字として扱われます。LINE 150: 記事情報の辞書への追加開始。これまでに取得した記事情報をひとまとめの辞書にして、articlesリストへ追加する処理を開始します。
LINE 151: 記事情報の辞書の開始。1件分の記事情報をまとめる辞書リテラルの開始部分です。LINE 152: uidキーの格納。
決定した記事の一意識別子uidを辞書のuidキーに格納します。LINE 153: sourceキーの格納。フィードから取得したソース名を辞書のsourceキーに格納します。
LINE 154: titleキーの格納。記事のタイトルを辞書のtitleキーに格納します。LINE 155: linkキーの格納。
記事へのリンクURLを辞書のlinkキーに格納します。LINE 156: publishedキーの格納。整形済みの公開日時文字列を辞書のpublishedキーに格納します。
LINE 157: summaryキーの格納。整形済みの記事概要を辞書のsummaryキーに格納します。この行で1件分の記事情報の作成が完了します。
LINE 158: 記事情報の辞書を閉じる。1件分の記事情報をまとめた辞書リテラルを閉じています。ここまでの各キーの値が1つの記事データとして完成します。
LINE 159: 記事をリストへ追加する処理を閉じる。articles.appendの呼び出しを閉じ、作成した記事辞書をarticlesリストへ追加します。これによりループのたびに記事が積み重なっていきます。
LINE 160: 解析結果のリストを返す。すべてのエントリを処理し終えたarticlesリストを呼び出し元へ返します。この戻り値が後続の保存処理に使われます。
RUN 5/9: XMLから記事情報を取り出す。parse_entriesを書き終えた段階です。デモRSSを解析し、件数と先頭記事のタイトルやuid、公開日時を表示して辞書の中身を確かめます。
CHECK 5/9: 途中実行に成功。解析件数5 フィード名 サンプルビジネス通信 先頭タイトル サンプルビジネス通信 記事01 先頭uid urn:demo:biz:001 / 公開日時2026-08-01 10:00:00 RETURN 05: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 165: 新着記事を保存する関数の定義。接続済みのDBと記事リストを受け取り、新しく保存できた件数を返す関数を定義しています。ここが重複なし保存の中心処理になります。
LINE 166: 保存時刻を文字列で取得する。現在時刻を取得し、DBに保存しやすい文字列形式に変換しています。この値は各記事の取得時刻としてまとめて記録されます。
LINE 167: 新着件数の初期化。新しく保存できた記事の件数を数えるための変数を0で初期化しています。この後のループで実際にカウントしていきます。
LINE 168: トランザクションを開始する。with connによってデータベース操作をまとめて1つのトランザクションとして扱います。処理が正常に終われば自動でコミットされます。
LINE 169: 記事を1件ずつ処理するループ。受け取った記事リストから1件ずつ取り出し、以降の処理でデータベースへ保存していきます。LINE 170: INSERT文の実行を開始する。
1件分の記事データをデータベースへ書き込むためのSQL実行を開始しています。実行結果はcursorとして受け取ります。LINE 171: 挿入対象テーブルとINSERT OR IGNOREの指定。
articlesテーブルへの挿入命令であり、uidが重複した場合はエラーにせず無視するINSERT OR IGNOREを使っています。これが重複保存を防ぐ仕組みです。LINE 172: 挿入する列名の指定。
どの列に値を入れるかを列名で指定しています。uidから始まりfetched_atまでの7つの列が対象です。LINE 173: プレースホルダの指定。
実際の値を安全に埋め込むためのプレースホルダを7つ用意しています。SQLインジェクションを避けるための書き方です。LINE 174: バインドする値のタプル開始。
プレースホルダに対応する値をまとめるタプルを開始しています。ここから順番に7つの値を渡します。LINE 175: uidの値を渡す。
記事の一意な識別子であるuidをSQLへ渡しています。この値がUNIQUE制約と組み合わさり重複判定に使われます。LINE 176: sourceの値を渡す。
記事が属するフィード名（配信元）をSQLへ渡しています。LINE 177: titleの値を渡す。記事のタイトルをSQLへ渡しています。
LINE 178: linkの値を渡す。記事のリンクURLをSQLへ渡しています。LINE 179: publishedの値を渡す。
記事の公開日時をSQLへ渡しています。この値は一覧表示時の並び順に使われます。LINE 180: summaryの値を渡す。
記事の概要文をSQLへ渡しています。LINE 181: fetched_atの値を渡す。直前に取得した保存時刻をSQLへ渡しています。
これで全ての値がそろいます。LINE 182: バインド値のタプルを閉じる。7つの値をまとめたタプルを閉じ、SQL実行へ渡す準備を完了させています。
LINE 183: conn.executeの呼び出しを閉じる。conn.executeの呼び出し全体を閉じ、SQL文と値を確定させて実際にデータベースへ命令を送ります。LINE 184: 新規挿入されたかを判定する。
cursor.rowcountが0より大きいかどうかで、この記事が新しく挿入されたかを判定しています。重複していた場合は0になります。LINE 185: 新着件数を加算する。
新規挿入された場合にnew_countへ加算し、実際に保存できた記事の数を積み上げていきます。LINE 186: 新着件数を返す。ループが終わった時点の新着件数を呼び出し元へ返し、この関数の結果として利用されます。
RUN 6/9: 同じ記事を2回保存して重複を試す。save_entriesが完成した段階です。同じ記事リストを続けて保存し、1回目と2回目の新着件数の違いからUNIQUE制約の効果を確かめます。
CHECK 6/9: 途中実行に成功。解析件数5 1回目の新着5件 2回目の新着0件 保存総数5件 RETURN 06: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 191: 収集処理の関数定義開始。複数サイクルにわたって取得・解析・保存を行うcollect関数の定義を開始しています。ここからCLI全体の中心となる処理が始まります。
LINE 192: DBファイルパスの引数。保存先となるSQLiteのファイルパスを受け取る引数です。LINE 193: 対象フィードURLの引数。
取得対象となるRSSフィードのURLをリストで受け取る引数です。LINE 194: サイクル数の引数。何回繰り返して取得するかを指定する引数で、既定値はDEFAULT_CYCLESです。
LINE 195: 待機秒数の引数。サイクル間の待機時間を指定する引数で、既定値はDEFAULT_INTERVALです。LINE 196: デモ用状態の引数。
デモモード時に記事数を増やすための状態を受け取る引数で、通常はNoneです。LINE 197: 戻り値の型指定。この関数が最終的に文字列の実行結果を返すことを示しています。
LINE 198: 出力行の初期化。実行結果として表示するメッセージ行のリストを、対象フィード数・サイクル数・間隔の情報から初期化しています。LINE 199: データベース接続の初期化。
init_dbを呼び出してSQLiteへ接続し、テーブルが用意された接続オブジェクトを取得しています。LINE 200: HTTPセッションの作成。requests.Sessionを作成し、この後のフィード取得で共通のヘッダー設定を使い回せるようにしています。
LINE 201: リクエストヘッダーの更新開始。セッションに付与するヘッダーの設定を開始しています。LINE 202: ヘッダー辞書の開始。
User-AgentやAcceptなどのヘッダーをまとめる辞書を開始しています。LINE 203: User-Agentの設定。アクセス元を識別するためのUser-Agent文字列を設定しています。
LINE 204: Acceptヘッダーの設定。RSSやXML形式を優先的に受け取るようAcceptヘッダーを設定しています。LINE 205: ヘッダー辞書を閉じる。
ヘッダー設定用の辞書リテラルを閉じています。LINE 206: ヘッダー更新の呼び出しを閉じる。session.headers.updateの呼び出しを閉じ、以降のリクエストにこのヘッダーが使われるようにします。
LINE 207: 累計新着件数の初期化。全サイクルを通じた新着記事の合計件数を数えるための変数を0で初期化しています。LINE 208: 後始末を保証するtry開始。
取得処理中に例外が起きてもセッションと接続を確実に閉じられるよう、try文を開始しています。LINE 209: サイクルを繰り返すループ。指定されたcycles回だけ、取得・解析・保存の一連の処理を繰り返すループです。
LINE 210: サイクル開始メッセージの追加開始。何サイクル目の処理を開始したかを表すメッセージ行を追加する処理を開始しています。LINE 211: サイクル開始時刻の組み立て。
現在の時刻を含めた「サイクル開始」のメッセージ文字列を組み立てています。LINE 212: サイクル開始メッセージの追加を閉じる。lines.appendの呼び出しを閉じ、組み立てたメッセージを出力行リストへ追加します。
LINE 213: サイクル内の解析件数の初期化。このサイクルで解析した記事の総数を数えるための変数を0で初期化しています。LINE 214: サイクル内の新着件数の初期化。
このサイクルで新しく保存できた記事の総数を数えるための変数を0で初期化しています。LINE 215: フィードを1件ずつ処理するループ。対象となるフィードURLのリストから1つずつ取り出し、取得から保存までの処理を行うループです。
LINE 217: フィードのHTTP取得を実行する。fetch_feed関数を呼び出し、対象URLからRSSのXMLデータと結果メッセージを取得しています。LINE 218: 取得結果メッセージの追加。
取得成功か失敗かを示すメッセージを出力行リストへ追加し、後で表示できるようにしています。LINE 219: 取得失敗の判定。raw_xmlがNoneかどうかを確認し、取得に失敗したフィードかどうかを判定しています。
LINE 220: 失敗したフィードのスキップ。取得に失敗した場合は解析や保存を行わずに次のフィードへ処理を進めます。LINE 222: RSSの解析を実行する。
parse_entries関数を呼び出し、取得したXMLから記事情報のリストを作成しています。LINE 224: 記事のDB保存を実行する。save_entries関数を呼び出し、解析した記事をデータベースへ重複なく保存し、新着件数を受け取っています。
LINE 225: サイクル内解析件数の加算。今回のフィードで解析できた記事数をサイクル全体の解析件数へ加算しています。LINE 226: サイクル内新着件数の加算。
今回のフィードで新しく保存できた記事数をサイクル全体の新着件数へ加算しています。LINE 227: フィード結果メッセージの追加開始。1つのフィードにおける解析件数・新着件数・重複件数を表すメッセージを追加する処理を開始しています。
LINE 228: 解析件数と新着件数の文字列組み立て。解析した記事数と新着として保存できた記事数を含む文字列の前半部分を組み立てています。LINE 229: 重複スキップ件数の文字列組み立て。
解析件数から新着件数を引いた重複スキップ件数を含む文字列の後半部分を組み立てています。LINE 230: フィード結果メッセージの追加を閉じる。lines.appendの呼び出しを閉じ、組み立てたメッセージを出力行リストへ追加します。
LINE 231: 累計新着件数の加算。このサイクルで得られた新着件数を、全サイクルを通じた累計新着件数へ加算しています。LINE 232: サイクル合計メッセージの追加。
このサイクル全体の解析件数と新着件数をまとめたメッセージを出力行リストへ追加しています。LINE 233: デモ状態の有無を確認。demo_state引数がNoneでないかを確認しています。
デモモードで実行されたときだけ、次の処理でデモ記事の件数を増やす仕組みです。LINE 234: デモ記事数を1件増やす。demo_state辞書の"extra"の値を1増やしています。
これにより次のサイクルで取得するデモRSSに新しい記事が1件追加されます。LINE 235: 最終サイクルかどうかを判定。現在のサイクル番号cycleが指定回数cyclesより小さいかを確認しています。
まだサイクルが残っている場合だけ、待機の処理へ進みます。LINE 236: 待機時間の案内を追加。次の取得までの待機秒数を知らせる文言をlinesリストへ追加しています。
ユーザーが処理の進行状況を確認できるようにしています。LINE 237: 指定秒数だけ処理を停止。time.sleepを使ってinterval秒だけプログラムの実行を一時停止しています。
サイクル間隔を空けて次の取得を行うための処理です。LINE 238: 累計保存件数を取得。articlesテーブル全体の件数をSQLで数え、変数storedに格納しています。
全サイクル終了後にDB内の総記事数を確認するための処理です。LINE 239: 後処理ブロックの開始。tryブロックに対応するfinallyブロックを開始しています。
エラーが起きても必ず実行される後始末処理をここから記述します。LINE 240: HTTPセッションを終了。requestsのSessionオブジェクトを閉じてリソースを解放しています。
通信に使った接続を確実にクリーンアップするための処理です。LINE 241: DB接続を閉じる。SQLiteの接続connをクローズしています。
処理が終わったデータベースへの接続を確実に解放するための後始末です。LINE 242: 全体結果の要約を追加。保存先のDBパス、今回の新着件数total_new、累計件数storedをまとめた文言をlinesへ追加しています。
処理結果全体をユーザーへ伝える最後の行です。LINE 243: 結果文字列を組み立てて返す。linesリストの各要素を改行でつなぎ、1つの文字列にして返しています。
これまで積み上げた実行ログをまとめて呼び出し元へ渡します。RUN 7/9: 1サイクル分の収集レポートを見る。collectを書き終えた段階です。
デモ配信の2本を対象に1サイクルだけ実行し、取得から保存までのログがそろっているか確認します。CHECK 7/9: 途中実行に成功。
対象フィード2本 / サイクル1回 / 間隔0秒 [サイクル1/1] 15:49:49開始 取得成功http://127.0.0.1:37519/biz.xml -> HTTP 200 / 1880バイト 解析5件 / 新着5件 / 重複スキップ0件 取得成功http://127.0.0.1:37519/tech.xml -> HTTP 200 / 1858バイト 解析5件 / 新着5件 / 重複スキップ0件 サイクル計: 解析10件 / 新着10件 保存先 /tmp/tmpby6rkmjw/collect.db: 今回の新着10件 / 累計10件 RETURN 07: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 248: 記事一覧取得関数の定義。DBパスと取得件数limitを受け取り、総件数と記事一覧のタプルを返す関数を定義しています。
listサブコマンドの中心となる処理です。LINE 249: DB接続を初期化。init_db関数を呼び出してSQLiteへの接続を確立しています。
テーブルが存在しなければ作成される仕組みも含んでいます。LINE 250: 後処理付き処理の開始。tryブロックを開始しています。
この中でデータ取得を行い、後段のfinallyで確実に接続を閉じる構成になっています。LINE 251: 保存済み記事の総数を取得。articlesテーブルの全件数をCOUNT関数で数え、変数totalに格納しています。
保存されているニュースの総数を確認するための処理です。LINE 252: 記事データの取得を開始。SQL文を実行して記事情報を取得する処理を開始しています。
次の行で指定する列と並び順、件数制限に従ってデータを取り出します。LINE 253: 取得する列と並び順を指定。source、title、link、publishedの列を選び、published列とid列の降順で並べるSQL文です。
新しい記事から順に取得するための指定です。LINE 254: 取得件数の上限を指定。LIMIT句にプレースホルダーを使い、取得する記事数の上限を後から渡せるようにしています。
表示件数を制御するための仕組みです。LINE 255: 件数上限の値を渡す。limit変数をタプルとして渡し、SQL文中のプレースホルダーへ実際の値を割り当てています。
SQLインジェクションを防ぐ安全な書き方です。LINE 256: 検索結果を全件取得。fetchallメソッドを呼び出し、実行したSQLの結果を全て取得してrows変数へ格納しています。
ここで実際の記事データがメモリに読み込まれます。LINE 257: 後処理ブロックの開始。tryブロックに対応するfinallyブロックを開始しています。
データ取得の成否にかかわらず、DB接続を閉じる処理を確実に行うためです。LINE 258: DB接続を閉じる。SQLite接続connをクローズしています。
データ取得が終わったので接続リソースを解放しています。LINE 259: 取得結果を返す。総件数totalと、各行を辞書へ変換したリストをタプルとして返しています。
sqlite3.Rowオブジェクトを扱いやすい辞書型に変換している点がポイントです。LINE 262: 表示用文字列生成関数の定義。DBパス、総件数、記事一覧を受け取り、表示用の文字列を組み立てる関数を定義しています。
listコマンドの出力内容を作る処理です。LINE 263: 保存件数が0件かを判定。totalが0であるかどうかを確認しています。
保存されている記事が1件もない場合は、別の案内メッセージを返す分岐です。LINE 264: 記事なし時の案内を返す。保存済みニュースが0件であることと、collectコマンドの実行を促す案内文を返しています。
ユーザーへ次に取るべき操作をわかりやすく示しています。LINE 265: 表示リストの見出しを作成。保存件数totalと表示件数len(rows)を含む見出し文をリストの最初の要素として作成しています。
一覧表示の冒頭に表示される情報です。LINE 266: 各記事を順番に処理。enumerate関数を使い、rowsの各記事に1から始まる番号indexを付けながらループしています。
表示順に番号を振るための処理です。LINE 267: 記事情報の1行目を追加。番号、配信元source、公開日時publishedをまとめた行をlinesへ追加しています。
公開日時が空の場合は「公開日時なし」と表示する工夫がされています。LINE 268: 記事タイトル行を追加。記事のタイトルをインデント付きの行としてlinesへ追加しています。
見出しの下に記事タイトルを表示するための処理です。LINE 269: 記事リンク行を追加。記事のURLをインデント付きの行としてlinesへ追加しています。
ユーザーが元記事へアクセスできるようにリンクを表示しています。LINE 270: 一覧表示文字列を組み立てる。linesリストの各行を改行でつなぎ、1つの文字列として返しています。
ここまで組み立てた一覧表示の内容がまとめて出力されます。RUN 8/9: 保存済みニュースの一覧を整形する。load_articlesとformat_articlesが完成した段階です。
収集した内容を読み出して整形し、番号付きの一覧として表示されるか確かめます。CHECK 8/9: 途中実行に成功。
総件数5 / 取得行2 /tmp/tmpjqpgapli/list.db: 保存済み5件 / 公開日時の新しい2件を表示 1. [サンプルテック速報] 2026-08-01 14:00:00 サンプルテック速報 記事05 http://news.invalid/tech/005 2. [サンプルテック速報] 2026-08-01 13:00:00 サンプルテック速報 記事04 http://news.invalid/tech/004 RETURN 08: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 274: CLIパーサー構築関数の定義。argparseを用いたコマンドライン引数パーサーを組み立てる関数を定義しています。
この関数がCLI全体の入力仕様を決めています。LINE 275: パーサーオブジェクトの生成。ArgumentParserクラスのインスタンスを作成しています。
この後の行でプログラム名や説明文などの設定を行っていきます。LINE 276: プログラム名の設定。コマンドライン上で表示されるプログラム名をrss_collector.pyに設定しています。
ヘルプ表示などで利用される名前です。LINE 277: プログラムの説明文を設定。このCLIがrequestsとfeedparser、SQLiteを組み合わせたRSS収集ツールであることを説明する文言を設定しています。
ヘルプ表示に使われます。LINE 278: パーサー生成の呼び出し閉じ。ArgumentParserの引数指定を閉じています。
ここまでの設定でパーサーの基本情報が確定します。LINE 279: サブコマンド機能の追加。collectとlistのようなサブコマンドを扱うための仕組みをパーサーへ追加しています。
requiredをTrueにすることで、必ずどちらかのコマンドを指定する必要があります。LINE 281: collectサブコマンドの追加。collectという名前のサブコマンドをパーサーへ追加しています。
RSS取得と保存を行う機能をこのサブコマンドで実行します。LINE 282: collectのヘルプ文言を設定。collectサブコマンドの説明文として、RSSを定期取得し重複なく保存するという内容を設定しています。
ヘルプ表示で確認できる説明です。LINE 283: collect定義の呼び出し閉じ。add_parserメソッドの引数指定を閉じています。
ここでcollectサブコマンドの基本設定が確定します。LINE 284: feedオプションの追加開始。collect_parserに対して--feedという引数を追加する処理を開始しています。
取得対象のRSS URLを指定するための引数です。LINE 285: feed引数の動作指定。--feedをaction=appendとして複数回指定できるようにし、デフォルトを空リストに設定しています。
複数のRSS URLをリストへ積み重ねるための指定です。LINE 286: feed引数のヘルプ文言。--feedオプションの説明として、取得するRSSのURLを複数回指定できることを示す文言を設定しています。
LINE 287: feed引数定義の呼び出し閉じ。add_argumentメソッドの引数指定を閉じています。--feedオプションの設定がここで完結します。
LINE 288: demoオプションの追加開始。collect_parserに対して--demoという引数を追加する処理を開始しています。デモ用RSSを使うかどうかを切り替えるオプションです。
LINE 289: demo引数のフラグ指定。action=store_trueとすることで、--demoを指定するとTrueになるフラグ形式の引数にしています。値を伴わない切り替えオプションの定義です。
LINE 290: demo引数のヘルプ文言。--demoオプションの説明として、外部通信を行わずlocalhost配信のデモRSSを対象にする旨を設定しています。LINE 291: demo引数定義の呼び出し閉じ。
add_argumentメソッドの引数指定を閉じています。--demoオプションの設定がここで完結します。LINE 292: dbオプションの追加。
collect_parserに--dbオプションを追加し、デフォルト値としてDEFAULT_DBを設定しています。使用するSQLiteファイルのパスを指定するための引数です。LINE 293: cyclesオプションの追加開始。
collect_parserに対して--cyclesという引数を追加する処理を開始しています。取得サイクルの回数を指定するための引数です。LINE 294: cycles引数の型とデフォルト値。
--cyclesの値をint型として受け取り、デフォルト値にDEFAULT_CYCLESを設定し、説明文も付けています。取得を何回繰り返すかを制御する引数です。LINE 295: cycles引数定義の呼び出し閉じ。
add_argumentメソッドの引数指定を閉じています。--cyclesオプションの設定がここで完結します。LINE 296: intervalオプションの追加開始。
collect_parserに対して--intervalという引数を追加する処理を開始しています。サイクル間の待機秒数を指定するための引数です。LINE 297: interval引数の型とデフォルト値。
--intervalの値をfloat型として受け取り、デフォルト値にDEFAULT_INTERVALを設定し、説明文も付けています。サイクル間の待ち時間を制御する引数です。LINE 298: interval引数定義の呼び出し閉じ。
add_argumentメソッドの引数指定を閉じています。--intervalオプションの設定がここで完結します。LINE 300: listサブコマンドの追加。
listという名前のサブコマンドをパーサーへ追加し、保存済みニュースを一覧表示するという説明文を設定しています。LINE 301: listのdbオプション追加。list_parserに--dbオプションを追加し、デフォルト値としてDEFAULT_DBを設定しています。
一覧表示対象のSQLiteファイルを指定するための引数です。LINE 302: limitオプションの追加開始。list_parserに対して--limitという引数を追加する処理を開始しています。
表示する記事の件数を指定するための引数です。LINE 303: limit引数の型とデフォルト値。--limitの値をint型として受け取り、デフォルト値にDEFAULT_LIMITを設定し、表示件数を指定する旨の説明文を付けています。
LINE 304: limit引数定義の呼び出し閉じ。add_argumentメソッドの引数指定を閉じています。--limitオプションの設定がここで完結します。
LINE 305: 構築したパーサーを返す。ここまで設定したparserオブジェクトを呼び出し元へ返しています。build_parser関数の最終的な戻り値として、CLI全体の解析仕様が確定します。
RUN 9/9: サブコマンドの解析結果を確かめる。build_parserを書き終えた段階です。collectとlistの引数を解析させ、既定値やオプションが期待どおりに割り当てられるか確認します。
CHECK 9/9: 途中実行に成功。command collect / demo True / cycles 3 db news.db / interval 0.0 / feed [] command list / db news.db / limit 5 RETURN 09: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 308: メイン関数の定義。コマンドライン引数のリストargvを受け取り、実行結果の文字列を返すmain関数を定義しています。プログラム全体の入り口となる処理です。
LINE 309: コマンドライン引数の解析。build_parser関数で作成したパーサーを使ってargvを解析し、結果をargs変数に格納しています。ここでcollectやlistなどの指定内容が取り出されます。
LINE 311: collectコマンドかどうかの分岐。コマンドラインで指定されたサブコマンドがcollectかどうかを判定しています。collectの場合だけ、この後に続くRSS取得処理へ進みます。
LINE 312: サイクル回数の下限チェック。max関数を使って、args.cyclesが1未満にならないようにしています。これによりサイクル数に0やマイナスの値が指定されても、必ず1回以上は実行されるようにしています。
LINE 313: 待機秒数の下限チェック。max関数を使って、args.intervalが0.0未満にならないようにしています。これによりマイナスの待機秒数が指定されても0秒として扱われ、エラーを防ぎます。
LINE 314: デモモード指定の判定。--demoオプションが指定されたかどうかを判定しています。指定されている場合は、外部サイトへ接続せずローカルのデモRSSを使う処理へ進みます。
LINE 315: デモ配信サーバーの起動。start_demo_server関数を呼び出して、ローカルにデモ用RSSを配信するサーバーを起動しています。戻り値としてサーバー本体と接続先のURLを受け取っています。
LINE 316: デモフィードURL一覧の作成。DEMO_FEEDSのパスを名前順に並べ替え、先ほど取得したbase_urlと連結してフィードのURL一覧を作成しています。この一覧が後の取得処理に渡されます。
LINE 317: 例外処理の開始。この後の処理でエラーが発生してもサーバーを確実に終了できるように、try文で処理を囲んでいます。LINE 318: デモRSSの収集処理を実行。
collect関数を呼び出し、デモフィードのURL一覧やサイクル数、待機秒数、デモ状態を渡してRSSの取得・保存を実行しています。実行結果のレポート文字列をreportに格納しています。LINE 319: 後片付け処理の開始。
try節の処理が成功しても失敗しても必ず実行される、finally節の開始です。この後でデモサーバーの終了処理を行います。LINE 320: デモサーバーの停止指示。
server.shutdownを呼び出して、起動していたデモ配信サーバーのループ処理を止めています。LINE 321: デモサーバーのソケット解放。server.server_closeを呼び出して、サーバーが使用していたソケットを解放し、後片付けを完了させています。
LINE 322: デモ実行結果メッセージの返却。デモRSSの配信先URLと、収集処理の結果reportをまとめた文字列を作成し、main関数の戻り値として返しています。LINE 323: フィード未指定のチェック。
デモモードではない場合に、--feedオプションが一つも指定されていないかどうかを判定しています。LINE 324: フィード未指定時の案内メッセージ。取得対象のフィードが指定されていない場合に、--feedまたは--demoの指定を促す案内文字列を返しています。
LINE 325: 指定フィードでの収集処理を実行。デモではなく実際に指定されたargs.feedのURL一覧を使ってcollect関数を呼び出し、その結果をmain関数の戻り値として返しています。LINE 327: 保存済みニュースの取得。
load_articles関数を呼び出し、SQLiteに保存されている記事の総件数と、表示件数の上限を1件以上に補正したlimitで取得した記事一覧を受け取っています。LINE 328: ニュース一覧の整形と返却。format_articles関数を使って、保存件数と記事一覧を読みやすい文字列に整形し、main関数の戻り値として返しています。
LINE 331: スクリプト直接実行時の判定。このファイルがモジュールとしてインポートされたのではなく、直接実行された場合にのみ、以降の処理を行うための判定です。LINE 332: main関数の実行と結果表示。
main関数を呼び出して処理を実行し、その戻り値の文字列をprintで画面に出力しています。実行1/4: デモRSSを収集してSQLiteへ保存する。collectサブコマンドに--demoを付けて実行します。
localhostで配信されるRSS 2本を取得し、解析件数と新着件数、保存先の累計件数が表示されます。確認1/4: デモRSSを収集してSQLiteへ保存する。collectサブコマンドに--demoを付けて実行します。
localhostで配信されるRSS 2本を取得し、解析件数と新着件数、保存先の累計件数が表示されます。RETURN 10: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
実行2/4: 2サイクル実行して重複スキップを確かめる。--cycles 2と--interval 0を指定して連続で収集します。2サイクル目は同じ記事が重複スキップとして数えられ、増えた1件だけが新着になります。
確認2/4: 2サイクル実行して重複スキップを確かめる。--cycles 2と--interval 0を指定して連続で収集します。2サイクル目は同じ記事が重複スキップとして数えられ、増えた1件だけが新着になります。
RETURN 11: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。実行3/4: 保存済みニュースを新しい順に一覧する。
収集したデータベースに対してlistサブコマンドを実行します。公開日時の新しい順に指定件数だけ、フィード名と日時、タイトル、URLが並びます。確認3/4: 保存済みニュースを新しい順に一覧する。
収集したデータベースに対してlistサブコマンドを実行します。公開日時の新しい順に指定件数だけ、フィード名と日時、タイトル、URLが並びます。RETURN 12: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。実行4/4: 任意のフィードURLを指定して収集する。--feedへRSSのURLを渡すと、そのフィードだけを取得します。
取得成功の行にHTTPステータスと受信バイト数が出て、最後に保存先の件数が分かります。確認4/4: 任意のフィードURLを指定して収集する。--feedへRSSのURLを渡すと、そのフィードだけを取得します。
取得成功の行にHTTPステータスと受信バイト数が出て、最後に保存先の件数が分かります。RETURN 13: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
学習内容のまとめ。
session.headers.updateでUser-Agentを共通化するfeedparser.parseはバイト列のXMLをそのまま受け取るINSERT OR IGNOREでUNIQUE違反の行を無視REQUEST_TIMEOUTで待ち時間に10秒の上限を設ける 小さく実行確認しながら完成状態まで段階的に組み立てる エンディング。
Python研修はCodeCampでご確認ください。
requests・feedparser・SQLiteとは
今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。
HTTPライブラリ『requests』の役割
requestsは、URLを渡すだけでWebページやRSSのXMLを受け取れるHTTP通信のライブラリです。標準のurllibより記述が短く、Sessionを使うとヘッダーや接続を複数のリクエストで共有できます。
今回のCLIでは、User-Agentとタイムアウトを設定したSessionでRSSを取得し、通信例外やHTTPエラーが起きたフィードだけを飛ばす形にしました。
fetch_feed関数の中でrequestsをどう扱っているかの内訳です
- session.headers.updateでUser-Agentを共通化する
- timeout=REQUEST_TIMEOUTで待ち時間を10秒に区切る
- requests.RequestExceptionで接続失敗と名前解決失敗をまとめて捕まえる
- response.okは400未満で真になる判定
- response.contentでバイト列のままXMLを受け取る
- session.close()で通信の後始末をする
RSS解析のfeedparserと保存先のSQLite
feedparserは、RSSやAtomのXMLを辞書のように読み取れる解析ライブラリです。項目名の揺れを吸収してくれるため、entry.getという統一した書き方で記事情報を取り出せます。保存側のsqlite3はPythonに同梱されたデータベースモジュールで、ファイル1つにテーブルと制約を持てました。
今回はarticlesテーブルのuid列へUNIQUE制約を付け、重複の判定をデータベース側へ任せています。
parse_entriesとinit_dbが扱うRSS項目とテーブル定義の要点です
- feedparser.parseはバイト列のXMLをそのまま受け取る
- parsed.feed.get("title")をフィード名として採用
- entryのidが無い場合はlinkをuidへ流用
- published_parsedをtime.strftimeで文字列化
- 概要は改行を空白へ置換して200文字で切る
- conn.row_factoryへsqlite3.Rowを設定して列名アクセスにする
Python・requestsで開発する場合の環境構築
この記事のセットアップ手順と掲載コードは、Windows 11 Pro、PowerShell 5.1、Python 3.13.3で動作確認しています。仮想環境を有効化せず、その中のPythonを直接指定するため、以下のコマンドはPowerShellとコマンドプロンプト（cmd）の両方で使えます。
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install --upgrade pip
.\.venv\Scripts\python.exe -m pip install requests feedparser
python -c "import sqlite3; print('SQLite:', sqlite3.sqlite_version)"macOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。
python3 -m venv .venv
./.venv/bin/python -m pip install --upgrade pip
./.venv/bin/python -m pip install requests feedparser
python -c "import sqlite3; print('SQLite:', sqlite3.sqlite_version)"- SQLiteは標準ライブラリのsqlite3から使うため、追加インストールは不要です（入れるのはrequestsとfeedparserだけです）。
- feedparserは依存パッケージのsgmllib3kも一緒に導入されます。pipのログに出ても問題ありません。
- --demoはlocalhostの空きポートでHTTPサーバを起動します。ファイアウォールの確認が出た場合はローカル接続を許可してください。
- news.dbはコマンドを実行したフォルダに作られます。やり直したいときはこのファイルを削除します。
RSSニュース収集CLIの要件定義
目的は、requestsでRSSを取得しfeedparserで解析した記事をSQLiteへ重複なく保存し、collectとlistの2コマンドで新着件数と保存済み一覧を確認できるCLIを作ることです。
対象者として、Pythonの基本文法を学び終え、HTTP取得とRSS解析、SQLiteへの保存を1本のコマンドへまとめる手順を学びたい人を想定しています。
完成物は、requests・feedparser・sqlite3で作るRSSニュース収集CLIで、collectでの定期取得とlistでの新着一覧、uidのUNIQUE制約による重複排除を確認できる教材です。
実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。
機能要件
- requests.SessionへUser-Agentを設定してRSSをGETする
- タイムアウト10秒を指定してフィードを取得する
- 取得に失敗したフィードだけをスキップして次へ進む
- feedparser.parseでタイトルとリンクと公開日時と概要を取り出す
- entryのidかlinkからuidを決める
- INSERT OR IGNOREで新着記事だけをSQLiteへ保存する
- cursor.rowcountで新着件数を数える
- 解析件数と新着件数と重複スキップ件数をサイクルごとに報告する
- --cyclesの回数だけ取得を繰り返す
- --interval秒だけサイクル間で待機する
- listサブコマンドで公開日時の新しい順に一覧表示する
- --limitで表示件数を絞る
- --demoでlocalhost配信のデモRSSを対象にする
- --dbで保存先のSQLiteファイルを切り替える
非機能要件
- REQUEST_TIMEOUTで待ち時間に10秒の上限を設ける
- User-Agentへrss-news-collector/1.0を送る
- requests.RequestExceptionを捕まえて処理を続行する
- response.okが偽なら本文を使わずスキップする
- uid列のUNIQUE制約で同じ記事の二重登録を防ぐ
- CREATE TABLE IF NOT EXISTSで再実行に耐える
- published列の降順インデックスで並べ替えを支える
- with connで1サイクル分の挿入をまとめる
- finallyでセッションとコネクションを閉じる
- --demoでは127.0.0.1のローカル配信だけを使う
- 概要は先頭200文字へ切り詰める
- デモ配信サーバをdaemonスレッドで動かす
実装方針
今回はrequests・feedparser・SQLiteの基本動作を追いやすくするため、RSSニュース収集CLI本体を1つのPythonファイルへまとめます。
入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。
RSSニュース収集CLIを安全に組み立てるための実装方針は次のとおりです。
- REQUEST_TIMEOUTで待ち時間に10秒の上限を設ける
- User-Agentへrss-news-collector/1.0を送る
- requests.RequestExceptionを捕まえて処理を続行する
- response.okが偽なら本文を使わずスキップする
- uid列のUNIQUE制約で同じ記事の二重登録を防ぐ
- CREATE TABLE IF NOT EXISTSで再実行に耐える
- published列の降順インデックスで並べ替えを支える
- with connで1サイクル分の挿入をまとめる
- finallyでセッションとコネクションを閉じる
- --demoでは127.0.0.1のローカル配信だけを使う
- 概要は先頭200文字へ切り詰める
- デモ配信サーバをdaemonスレッドで動かす
完成と判断する条件
- collect --demoで2本のフィードから解析5件と新着5件を表示する
- 2サイクル目に解析6件と新着1件と重複スキップ5件を表示する
- 最終行へ保存先と今回の新着件数と累計件数を出す
- listで公開日時の新しい順に指定件数を表示する
- 保存が空のときはcollectでの取得を促す案内を返す
- --feedも--demoも無い場合に指定方法を案内する
- 取得に失敗したフィードを取得失敗として記録し次へ進む
requestsでRSSニュース収集CLIを作る際の重要ポイント
このCLIの中心は、取得と解析と保存を1サイクルとして繰り返し、新着だけを数える流れです。save_entriesはINSERT OR IGNOREを使い、uidが衝突した行を静かに読み飛ばします。挿入できた行数はcursor.rowcountで分かるので、解析件数から引くだけで重複スキップ件数も報告できました。
collectは--cyclesの回数だけこの流れを回し、次の取得まで--interval秒だけ待機します。
collectとlistが結果を組み立てるときの判断と条件の一覧です
- INSERT OR IGNOREでUNIQUE違反の行を無視
- rowcountが正の値のときだけ新着として加算
- 解析件数から新着件数を引いて重複スキップ件数を算出
- with connで1サイクル分の挿入をまとめてコミット
- 最終サイクルの後は待機せずに終了
- listは--limitの件数だけ取り出して番号付きで表示
collectサブコマンドが1サイクルで行う3工程
collectはフィードのURLを順に処理し、fetch_feedで取得、parse_entriesで解析、save_entriesで保存という3工程を回します。各フィードの結果はその場で行として積み上げ、サイクルの終わりに解析件数と新着件数の小計を出しました。
最終行では保存先のパスと今回の新着、累計件数をまとめて示すため、実行ログを読むだけで成果が分かります。
1サイクルの実行で出力される行の並びです
- 対象フィード数とサイクル数と間隔の見出し行
- フィードごとの取得成功または取得失敗の行
- 解析件数と新着件数と重複スキップ件数の行
- サイクル計の小計行
- 保存先と今回の新着と累計の最終行
listサブコマンドが一覧を組み立てる手順
listはload_articlesでSQLiteから総件数と行を取り出し、format_articlesで表示用の文字列へ整えます。取り出す列はsourceとtitleとlinkとpublishedの4つに絞り、公開日時とidの降順で並べました。保存が1件も無いときは、collectでの取得を促す1行だけを返す分岐も用意しています。
listの実行時に確認できる出力の構成です
- 先頭行に保存済み件数と表示件数
- 2桁そろえの連番とフィード名
- 次の行に記事タイトル
- その次の行に記事URL
- 保存が空のときは取得を促す案内
--demoで外部通信なしに確かめる仕組み
--demoを付けると、start_demo_serverが127.0.0.1の空きポートでThreadingHTTPServerを立ち上げ、そのURLをフィード一覧として使います。配信されるXMLはbuild_demo_rssが生成し、サイクルごとにDEMO_STATEのextraが増えて記事が1件追加されました。
おかげで外部サイトへつながないまま、新着と重複スキップの両方を再現できます。収集後のサーバはfinallyでshutdownしてから閉じます。
デモ実行時に用意される配信内容と後始末の流れです
- /biz.xmlと/tech.xmlの2本を配信
- 初回は各フィード5件の記事
- サイクルごとに記事が1件増える
- 収集後にserver.shutdownで停止
- 配信スレッドはdaemonとして起動
PythonでRSSニュース収集CLIの完成コード
実装は1ファイルにまとめ、冒頭のdocstringで取得・解析・保存・表示という4つの責務を宣言してから、関数を順に並べています。
取得側はrequests.Sessionを1つ作って使い回し、User-Agentとtimeoutを必ず添えます。成否はResponse.okで判定し、ステータスコードが400未満でなければ取得失敗として次のフィードへ進みました。
timeoutは接続と読み取りの待機時間に対する上限で、処理全体の合計時間を厳密に区切る指定ではありません。それでも省略すると、応答が来ないまま待ち続ける危険があります。
解析はfeedparser.parseに本文を渡すだけで、entriesからタイトルやリンク、公開日時を取り出せます。保存ではINSERT OR IGNOREの実行後にcursor.rowcountを見て、1なら新着、0なら重複として数えました。
collectは--cycles回のループで全フィードを回し、サイクルの間を--interval秒スリープします。ポイントとしては、取得と重複排除で外せない設定をまとめます。
このセクションの用語
- requests.Session
- ヘッダや接続を共有しながら複数のリクエストを送れる仕組みです。同じサイトへ続けてアクセスする処理に向きます。
- User-Agent
- 送信元のソフト名を相手へ伝えるHTTPヘッダです。誰からのアクセスかを明示する礼儀として付けます。
- Response.ok
- requestsの応答オブジェクトが持つ真偽値で、ステータスコードが400未満のときTrueになります。
- feedparser.parse
- RSSやAtomの中身を解析し、記事一覧を
entriesとして扱えるようにする関数です。 - rowcount
- 直前のSQLで影響を受けた行数です。
INSERT OR IGNOREでは1なら追加、0なら重複と判断できます。
"""RSSニュース収集CLI。
requestsでRSSフィードをHTTP取得する処理、feedparserでRSS記事を解析する処理、
SQLiteへ新着記事を重複なく保存する処理、そして複数サイクルで定期取得し保存済み
ニュースを一覧表示する処理を、それぞれ関数へ分けて実装している。
"""
from __future__ import annotations
import argparse
import sqlite3
import threading
import time
from datetime import datetime, timedelta, timezone
from email.utils import format_datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import feedparser
import requests
# 基本設定
DEFAULT_DB = "news.db"
DEFAULT_CYCLES = 1
DEFAULT_INTERVAL = 60.0
DEFAULT_LIMIT = 20
REQUEST_TIMEOUT = 10.0
USER_AGENT = "rss-news-collector/1.0"
# デモ用RSSのローカル配信（外部サイトへ接続せずに動作を確認する）
DEMO_FEEDS = {
"/biz.xml": "サンプルビジネス通信",
"/tech.xml": "サンプルテック速報",
}
DEMO_STATE = {"extra": 0}
DEMO_BASE_TIME = datetime(2026, 8, 1, 9, 0, tzinfo=timezone.utc)
def build_demo_rss(path: str) -> str:
"""デモフィードのRSS 2.0 XMLを作る（サイクルごとに記事が1件増える）。"""
feed_title = DEMO_FEEDS[path]
slug = path.strip("/").split(".")[0]
items = []
for number in range(1, 5 + DEMO_STATE["extra"] + 1):
published = format_datetime(DEMO_BASE_TIME + timedelta(hours=number))
items.append(
" <item>\n"
f" <title>{feed_title} 記事{number:02d}</title>\n"
f" <link>http://news.invalid/{slug}/{number:03d}</link>\n"
f" <guid>urn:demo:{slug}:{number:03d}</guid>\n"
f" <pubDate>{published}</pubDate>\n"
f" <description>{feed_title}の記事{number:02d}のダミー本文です。</description>\n"
" </item>"
)
return (
'<?xml version="1.0" encoding="UTF-8"?>\n'
'<rss version="2.0">\n'
" <channel>\n"
f" <title>{feed_title}</title>\n"
f" <link>http://news.invalid/{slug}</link>\n"
" <description>デモ用のRSSフィードです。</description>\n"
+ "\n".join(items)
+ "\n </channel>\n</rss>\n"
)
class DemoFeedHandler(BaseHTTPRequestHandler):
"""デモRSSのXMLを返すだけの簡易ハンドラ。"""
def do_GET(self) -> None: # noqa: N802
path = self.path.split("?")[0]
if path not in DEMO_FEEDS:
self.send_error(404, "feed not found")
return
body = build_demo_rss(path).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/rss+xml; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args) -> None:
return
def start_demo_server() -> tuple[ThreadingHTTPServer, str]:
"""localhostの空きポートでデモRSSを配信する（外部通信なし）。"""
server = ThreadingHTTPServer(("127.0.0.1", 0), DemoFeedHandler)
threading.Thread(target=server.serve_forever, daemon=True).start()
host, port = server.server_address[0], server.server_address[1]
return server, f"http://{host}:{port}"
# SQLiteの初期化（uidのUNIQUE制約が、後段の重複なし保存を支える土台になる）
def init_db(db_path: str) -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
conn.execute(
"""
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
uid TEXT NOT NULL UNIQUE,
source TEXT NOT NULL,
title TEXT NOT NULL,
link TEXT NOT NULL,
published TEXT NOT NULL,
summary TEXT NOT NULL,
fetched_at TEXT NOT NULL
)
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_articles_published ON articles(published DESC)"
)
conn.commit()
return conn
# 1. requestsでRSSフィードをHTTP取得する
# requests.Sessionへ User-Agent と timeout を設定してGETし、失敗時はそのフィードだけをスキップする。
def fetch_feed(session: requests.Session, url: str) -> tuple[bytes | None, str]:
try:
response = session.get(url, timeout=REQUEST_TIMEOUT)
except requests.RequestException as exc:
return None, f"取得失敗 {url} -> {exc.__class__.__name__}"
if not response.ok:
return None, f"取得失敗 {url} -> HTTP {response.status_code}"
return (
response.content,
f"取得成功 {url} -> HTTP {response.status_code} / {len(response.content)}バイト",
)
# 2. feedparserでRSS記事を解析する
# 取得したXMLをfeedparser.parseへ渡し、フィード名・タイトル・リンク・公開日時・概要を取り出す。
def parse_entries(raw_xml: bytes, feed_url: str) -> list[dict]:
parsed = feedparser.parse(raw_xml)
source = (parsed.feed.get("title") or feed_url).strip()
articles: list[dict] = []
for entry in parsed.entries:
link = (entry.get("link") or "").strip()
title = (entry.get("title") or "（タイトルなし）").strip()
uid = (entry.get("id") or link or f"{source}:{title}").strip()
if not uid:
continue
stamp = entry.get("published_parsed") or entry.get("updated_parsed")
published = time.strftime("%Y-%m-%d %H:%M:%S", stamp) if stamp else ""
summary = (entry.get("summary") or "").strip().replace("\n", " ")[:200]
articles.append(
{
"uid": uid,
"source": source,
"title": title,
"link": link,
"published": published,
"summary": summary,
}
)
return articles
# 3. SQLiteへ新着記事を重複なく保存する
# uid列のUNIQUE制約を前提に INSERT OR IGNORE を実行し、cursor.rowcountで新着件数を数える。
def save_entries(conn: sqlite3.Connection, articles: list[dict]) -> int:
fetched_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
new_count = 0
with conn:
for article in articles:
cursor = conn.execute(
"INSERT OR IGNORE INTO articles"
" (uid, source, title, link, published, summary, fetched_at)"
" VALUES (?, ?, ?, ?, ?, ?, ?)",
(
article["uid"],
article["source"],
article["title"],
article["link"],
article["published"],
article["summary"],
fetched_at,
),
)
if cursor.rowcount > 0:
new_count += cursor.rowcount
return new_count
# 4. 複数サイクルで定期取得し保存済みニュースを一覧表示する（その前半: 定期取得部分）
# --cycles回のループで全フィードを取得・解析・保存し、間を--interval秒スリープする。
def collect(
db_path: str,
feed_urls: list[str],
cycles: int = DEFAULT_CYCLES,
interval: float = DEFAULT_INTERVAL,
demo_state: dict | None = None,
) -> str:
lines = [f"対象フィード {len(feed_urls)}本 / サイクル {cycles}回 / 間隔 {interval:g}秒"]
conn = init_db(db_path)
session = requests.Session()
session.headers.update(
{
"User-Agent": USER_AGENT,
"Accept": "application/rss+xml, application/xml;q=0.9, */*;q=0.8",
}
)
total_new = 0
try:
for cycle in range(1, cycles + 1):
lines.append(
f"[サイクル {cycle}/{cycles}] {datetime.now().strftime('%H:%M:%S')} 開始"
)
cycle_parsed = 0
cycle_new = 0
for url in feed_urls:
# 1. requestsでRSSフィードをHTTP取得する
raw_xml, message = fetch_feed(session, url)
lines.append(f" {message}")
if raw_xml is None:
continue
# 2. feedparserでRSS記事を解析する
articles = parse_entries(raw_xml, url)
# 3. SQLiteへ新着記事を重複なく保存する
new_count = save_entries(conn, articles)
cycle_parsed += len(articles)
cycle_new += new_count
lines.append(
f" 解析 {len(articles)}件 / 新着 {new_count}件"
f" / 重複スキップ {len(articles) - new_count}件"
)
total_new += cycle_new
lines.append(f" サイクル計: 解析 {cycle_parsed}件 / 新着 {cycle_new}件")
if demo_state is not None:
demo_state["extra"] += 1
if cycle < cycles:
lines.append(f" 次の取得まで {interval:g}秒 待機")
time.sleep(interval)
stored = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
finally:
session.close()
conn.close()
lines.append(f"保存先 {db_path}: 今回の新着 {total_new}件 / 累計 {stored}件")
return "\n".join(lines)
# 4. 複数サイクルで定期取得し保存済みニュースを一覧表示する（その後半: 一覧表示部分）
# listサブコマンドから呼ばれ、公開日時の新しい順に保存済みニュースを取り出す。
def load_articles(db_path: str, limit: int) -> tuple[int, list[dict]]:
conn = init_db(db_path)
try:
total = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
rows = conn.execute(
"SELECT source, title, link, published FROM articles"
" ORDER BY published DESC, id DESC LIMIT ?",
(limit,),
).fetchall()
finally:
conn.close()
return total, [dict(row) for row in rows]
def format_articles(db_path: str, total: int, rows: list[dict]) -> str:
if total == 0:
return f"{db_path}: 保存済みニュースは0件です（collect コマンドで取得してください）"
lines = [f"{db_path}: 保存済み {total}件 / 公開日時の新しい {len(rows)}件を表示"]
for index, row in enumerate(rows, 1):
lines.append(f"{index:>2}. [{row['source']}] {row['published'] or '公開日時なし'}")
lines.append(f" {row['title']}")
lines.append(f" {row['link']}")
return "\n".join(lines)
# CLIの定義
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="rss_collector.py",
description="RSSニュース収集CLI（requests + feedparser + SQLite）",
)
subparsers = parser.add_subparsers(dest="command", required=True)
collect_parser = subparsers.add_parser(
"collect", help="RSSを定期取得してSQLiteへ重複なく保存する"
)
collect_parser.add_argument(
"--feed", action="append", default=[], metavar="URL",
help="取得するRSSのURL（複数回指定できます）",
)
collect_parser.add_argument(
"--demo", action="store_true",
help="localhost配信のデモRSSを対象にする（外部通信なし）",
)
collect_parser.add_argument("--db", default=DEFAULT_DB, help="SQLiteのDBファイル")
collect_parser.add_argument(
"--cycles", type=int, default=DEFAULT_CYCLES, help="取得サイクルの回数"
)
collect_parser.add_argument(
"--interval", type=float, default=DEFAULT_INTERVAL, help="サイクル間の待機秒数"
)
list_parser = subparsers.add_parser("list", help="保存済みニュースを一覧表示する")
list_parser.add_argument("--db", default=DEFAULT_DB, help="SQLiteのDBファイル")
list_parser.add_argument(
"--limit", type=int, default=DEFAULT_LIMIT, help="表示する件数"
)
return parser
def main(argv: list[str] | None = None) -> str:
args = build_parser().parse_args(argv)
if args.command == "collect":
cycles = max(1, args.cycles)
interval = max(0.0, args.interval)
if args.demo:
server, base_url = start_demo_server()
feed_urls = [base_url + path for path in sorted(DEMO_FEEDS)]
try:
report = collect(args.db, feed_urls, cycles, interval, DEMO_STATE)
finally:
server.shutdown()
server.server_close()
return f"デモRSSを {base_url} で配信（外部通信なし）\n{report}"
if not args.feed:
return "取得対象がありません（--feed URL または --demo を指定してください）"
return collect(args.db, args.feed, cycles, interval)
total, rows = load_articles(args.db, max(1, args.limit))
return format_articles(args.db, total, rows)
if __name__ == "__main__":
print(main())
コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。
docstringで4つの責務を宣言
"""RSSニュース収集CLI。
requestsでRSSフィードをHTTP取得する処理、feedparserでRSS記事を解析する処理、
SQLiteへ新着記事を重複なく保存する処理、そして複数サイクルで定期取得し保存済み
ニュースを一覧表示する処理を、それぞれ関数へ分けて実装している。
"""ファイルの先頭で、このスクリプトが担う4つの処理をはっきり書いています。後から読み返すときに、どの関数を探せばよいか迷いにくくなります。
取得・解析・保存に必要なimport
import argparse
import sqlite3
import threading
import time
from datetime import datetime, timedelta, timezone
from email.utils import format_datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import feedparser
import requests外部ライブラリはfeedparserとrequestsの2つだけで、残りは標準ライブラリです。threadingとhttp.serverはデモ用のローカル配信、sqlite3は保存に使います。
REQUEST_TIMEOUTなどの既定値
DEFAULT_DB = "news.db"
DEFAULT_CYCLES = 1
DEFAULT_INTERVAL = 60.0
DEFAULT_LIMIT = 20
REQUEST_TIMEOUT = 10.0
USER_AGENT = "rss-news-collector/1.0"既定のDB名やサイクル数、待機時間を先頭にまとめ、コードの奥に数値が散らばらないようにしました。REQUEST_TIMEOUTとUSER_AGENTをここへ置くと、通信の作法を1か所で調整できます。
DEMO_FEEDSとDEMO_STATEの役割
DEMO_FEEDS = {
"/biz.xml": "サンプルビジネス通信",
"/tech.xml": "サンプルテック速報",
}
DEMO_STATE = {"extra": 0}
DEMO_BASE_TIME = datetime(2026, 8, 1, 9, 0, tzinfo=timezone.utc)デモで配信する2本のフィードのパスとタイトルを辞書で持っています。DEMO_STATEのextraはサイクルごとに増える記事数のカウンタで、新着が出る状況を作るための仕掛けです。
build_demo_rssの記事生成ループ
for number in range(1, 5 + DEMO_STATE["extra"] + 1):
published = format_datetime(DEMO_BASE_TIME + timedelta(hours=number))記事は5件から始まり、extraが増えるほど末尾に1件ずつ足されます。公開日時は基準時刻に時間を足し、format_datetimeでRSSが使う日時表記へ変換しています。
itemタグにguidを埋め込む
f" <link>http://news.invalid/{slug}/{number:03d}</link>\n"
f" <guid>urn:demo:{slug}:{number:03d}</guid>\n"
f" <pubDate>{published}</pubDate>\n"guidにはurn:demo:tech:001のような一意な文字列を入れて、重複判定の鍵をはっきりさせました。linkとpubDateも揃えると、解析結果をそのまま保存へ流せます。
do_GETでRSSを応答する
def do_GET(self) -> None: # noqa: N802
path = self.path.split("?")[0]
if path not in DEMO_FEEDS:
self.send_error(404, "feed not found")
return
body = build_demo_rss(path).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/rss+xml; charset=utf-8")知らないパスには404を返し、登録済みのパスならXMLを組み立ててバイト列で応答します。Content-Typeにapplication/rss+xmlを指定しておくと、フィードとして素直に解析されます。
参考：
©Requests公式ドキュメントQuickstartYou can tell Requests to stop waiting for a response after a given number of seconds with the timeout parameter. Nearly all production code should use this parameter in nearly all requests.
タイムアウト：requestsに必ず指定
取得判定：okは400未満でTrue
重複排除：uidのUNIQUE制約に任せる
--demoでローカルRSSを2本配信する仕組み
外部のニュースサイトへ検証のたびにアクセスすると、相手先に負荷をかけますし、結果も毎回変わって確認しづらくなります。そこで--demoではlocalhostに小さなHTTPサーバを立て、RSSを2本配信しました。
配信内容はbuild_demo_rssが組み立てるRSS 2.0のXMLです。DEMO_STATEのextraが増えるとサイクルごとに記事が1件足されるので、新着保存と重複スキップの両方を1つのコマンドで再現できます。
サーバはThreadingHTTPServerを別スレッドで動かし、collectの処理と並行してリクエストへ応答させます。外部通信なしで、localhost内のHTTP通信だけで完結する構成です。
このセクションの用語
- http.server
- Python標準ライブラリの簡易HTTPサーバです。追加インストールなしで手元にテスト用のサーバを立てられます。
- ThreadingHTTPServer
- リクエストごとにスレッドを割り当てるHTTPサーバのクラスです。取得が重なっても応答を返せます。
- guid
- RSSのitemに入れる記事の識別子です。サイト内で一意な値が入るため、重複判定の鍵として使えます。
- Content-Type
- 応答の中身の種類を伝えるHTTPヘッダです。RSSでは
application/rss+xmlを指定します。
デモモードで意図的に用意した仕掛けを整理します。
- 配信するフィードは
/biz.xmlと/tech.xmlの2本 - 1サイクルごとに各フィードの記事が1件ずつ増える
-
log_messageを空実装にしてアクセスログを抑制 - 接続先はlocalhostだけで、外部サイトへは接続しない
PythonでRSSニュース収集CLIのエラー対処
収集ツールは相手のサーバやネットワークの状態に左右されるため、エラーは「起きる前提」で作った方が気楽になります。
今回の4回の実行はすべて終了コード0で終わりましたが、フィードを外部のURLへ差し替えれば通信まわりの失敗は普通に起こります。一般につまずきやすいのは、ライブラリの入れ忘れ、URLの書き方、DBの初期化漏れの3系統です。
このセクションの用語
- 例外
- 実行中の異常をPythonが投げて知らせる仕組みです。捕まえないとプログラムはそこで止まります。
- 仮想環境
- プロジェクトごとにライブラリを分けて入れる仕組みです。別環境で実行すると未インストール扱いになります。
- スキーム
- URLの先頭にある
httpやhttpsの部分です。requestsではこれが無いとURLとして解釈できません。
| エラー例 | 原因 | 対処 |
|---|---|---|
ModuleNotFoundError: No module named 'feedparser' |
ライブラリが未インストール、または別環境のPythonで実行している | 仮想環境を有効化してpip install feedparser requestsを実行し、python -c "import feedparser"で確認する |
requests.exceptions.ConnectTimeout |
相手ホストが応答せず、timeoutの待機時間を超えた |
REQUEST_TIMEOUTの値を見直し、失敗したフィードは飛ばして次へ進む作りにする |
requests.exceptions.MissingSchema: Invalid URL |
URLにhttp://やhttps://のスキームが付いていない |
フィードURLをスキームから始まる完全な形で渡す |
sqlite3.OperationalError: no such table: articles |
テーブル作成前にSELECTが走っている |
起動時にCREATE TABLE IF NOT EXISTSを通す初期化関数を必ず呼ぶ |
sqlite3.IntegrityError: UNIQUE constraint failed: articles.uid |
素のINSERTで既存のuidを書き込もうとした |
INSERT OR IGNOREに変えるか、例外を捕まえて重複として数える |
RSSニュース収集CLIで注意したい点
一番つまずくのはuidの決め方です。同じ記事でもURLに広告用のクエリパラメータが付いて変わることがあり、linkだけを鍵にすると同じ記事が二重に入りかねません。
guidがあればそれを優先し、無いときだけlinkで補うと安定します。何が入っているかはフィード次第なので、まず1回取得してentriesの中身を覗くのが早道でした。
取得間隔の設定でつまずくケースも多く見られます。
公開日時の扱いも落とし穴になりやすいです。RSSのpubDateはタイムゾーン付きの文字列なので、並び替えや期間の比較をする前にUTCなど1つの基準へ揃えます。
ポイントとしては、重複排除と定期取得でつまずきやすい設定をまとめます。
識別子：guid優先でlink補完
取得間隔：検証以外は分単位
公開日時：UTC基準に揃える
RSSニュース収集CLIの動作確認
実行はすべて--demo前提で、外部サイトへは接続していません。コマンドは4回で、いずれも終了コード0で完了しました。
最初にpython subject.py list --db news.dbを実行し、保存前の状態でも一覧処理が落ちないことを確認しています。
次にpython subject.py collect --demo --db news.db --cycles 1で1サイクルだけ収集しました。デモの2フィードを取得して解析し、DBへ書き込むまでの流れが1コマンドで通ります。
続けて--cycles 3 --interval 1で3サイクル回しました。サイクルごとにデモ側の記事が1件増える設計なので、既存記事のスキップと新着の追加が混ざった動きを追えます。
最後にpython subject.py list --db news.db --limit 5で表示件数を絞って一覧しました。各コマンドの標準出力はキャプチャにそのまま写っています。
このセクションの用語
- 終了コード
- コマンドが終わるときに返す番号です。0は正常終了を意味し、成否の判定に使えます。
- 標準出力
- プログラムが画面へ文字を出す既定の出口です。CLIの実行結果はここへ表示されます。
実行した4つのコマンドと、それぞれで確かめたかった点を並べます。
-
list --db news.db：保存前の状態でも一覧が成立するか -
collect --demo --cycles 1：取得から保存までが1周するか -
collect --demo --cycles 3 --interval 1：間隔スリープと重複スキップが動くか -
list --db news.db --limit 5：--limitで表示件数を絞れるか
SQLite保存を活かせる業務シーンと使い方
作ったものは小さいですが、uidで重複を弾きながら溜め続けられる性質は、日々の情報収集の下地として使えます。
collectだけをスケジューラへ登録し、読むときはlistを叩く形にすると運用が単純になります。保存先がファイル1つなので、後からSQLで絞り込む拡張もしやすいでしょう。
このセクションの用語
- cron
- Linuxなどで決まった時刻にコマンドを自動実行する仕組みです。Windowsではタスクスケジューラが同じ役割を担います。
- SQL
- データベースへ問い合わせるための言語です。溜めた記事の絞り込みや件数の集計に使えます。
| 使える場面 | 具体的な使い方 |
|---|---|
| 業界ニュースの朝の巡回 | 業界メディアのRSSを登録し、始業前にcollectをcronやタスクスケジューラで走らせ、list --limit 20で新着だけ読む |
| 競合のリリース追跡 | 競合のリリースノートRSSを集め、溜めたタイトルを月末にSQLで期間指定して抽出し、変更履歴の棚卸しに使う |
| 社内共有ダイジェストの下書き | 当日分のレコードをSELECTしてタイトルとURLをMarkdownへ整形し、チャットへ貼るサマリのたたき台にする |
| ステータス情報のウォッチ | 利用中のサービスのステータスRSSを登録し、collectが数えた新着件数が0以外のときだけ通知処理へ渡す |
| 学習用データの蓄積 | 同じフィードを長期間ためて、タイトルに出る語句の傾向を後から集計する土台にする |
RSSニュース収集CLI開発のまとめ
requestsで取得し、feedparserで解析し、sqlite3へINSERT OR IGNOREで入れる3段構えにすると、RSS収集CLIは短いコードでも形になりました。
重複判定をアプリ側の条件分岐ではなくUNIQUE制約に寄せた点が効いています。新着か重複かはcursor.rowcountを見るだけで分かるので、集計処理も素直に書けます。
--demoでlocalhostにRSSを2本配信する仕組みを入れたので、外部通信なしで新着保存と重複スキップの両方を試せる形になりました。実行したコマンドは4回で、いずれも終了コード0で完了しています。
次に手を入れたい拡張の候補を挙げます。
- フィードURLをコード内から設定ファイルへ移す
- 本文の要約列を足して
listの表示を読みやすくする - 取得に失敗したフィードを記録し、次回の再試行対象にする
参考にした一次情報
- ^ Requests: Quickstart（公式ドキュメント）. https://requests.readthedocs.io/en/latest/user/quickstart/, (参照26-08-14).
- ^ feedparser documentation. https://feedparser.readthedocs.io/en/latest/, (参照26-08-14).
- ^ sqlite3 — SQLiteデータベースのDB-API 2.0インターフェース（Python公式）. https://docs.python.org/ja/3/library/sqlite3.html, (参照26-08-14).
- ^ SQLite: ON CONFLICT clause（INSERT OR IGNOREの挙動）. https://www.sqlite.org/lang_conflict.html, (参照26-08-14).
※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。
