定期的にを叩くのではなく、更新そのものを追いたい。あるサイトの更新を追いたい、という要件はよくあります。新着のお知らせ、公開された資料、価格の改定。画面を見れば一覧があるので、で取ろうと考えます。
その前に確認してほしいのが、の有無です。あれば、決まった形式で新着だけが返ってきます。HTML を解析する必要がなく、サイトの見た目が変わっても壊れません。
見落とされがちですが、公的機関や報道機関のサイトには今でも用意されていることが多いです。画面上にリンクが無くても、決まった場所に置かれている場合があります。
フィードを探す
見つかれば、のために書くはずだった処理のほとんどが不要になります。探し方は2つあります。HTML の中に自動検出用の指定があるか、よくある場所を直接試すか。前者が確実なので、まずはページの先頭部分を確認します。
import refrom urllib.parse import urljoin
import requests
COMMON_PATHS = ("/feed", "/rss", "/rss.xml", "/atom.xml", "/index.xml", "/feed.xml")
def discover_feeds(page_url: str, timeout=(5, 30)) -> list[str]: """まず HTML の指定を見て、無ければ定番の場所を叩いて確かめる。""" resp = requests.get(page_url, timeout=timeout) resp.raise_for_status()
found = [] for m in re.finditer(r"<link[^>]+>", resp.text, re.I): tag = m.group(0) if "alternate" in tag and re.search(r"application/(rss|atom)\+xml", tag, re.I): href = re.search(r"""href=["']([^"']+)["']""", tag) if href: found.append(urljoin(page_url, href.group(1)))
if found: return found
for path in COMMON_PATHS: url = urljoin(page_url, path) try: head = requests.head(url, timeout=timeout, allow_redirects=True) except requests.RequestException: continue ctype = head.headers.get("Content-Type", "") if head.status_code == 200 and ("xml" in ctype or "rss" in ctype): found.append(url) return foundフィードが見つかったら、まず1回取って中身を見てください。必要な情報が本当に含まれているかを確かめるためです。タイトルとリンクしか無く、欲しい本文が入っていないことはよくあります。その場合は、リンク先を別途取りに行くことになります。
配信されている件数も確認しておきます。10件しか入っていないフィードを1日1回しか見に行かないと、更新が多い日に取りこぼします。件数と更新頻度から、確認する間隔を決めてください。
新着だけを取り出す
フィードには最新の数十件しか入っていません。毎回全部を処理してもよいのですが、すでに処理したものを覚えておくと無駄が減ります。各項目には一意の識別子が付いているので、これを使います。
識別子は、リンクの URL と同じであることも、別に振られていることもあります。URL を鍵にすると、同じ記事の URL が変わったときに二重に処理されます。識別子があるならそちらを優先してください。
import jsonfrom pathlib import Path
# 外部から取ってきた XML を標準の xml.etree で読むと、# 外部実体参照(XXE)や展開爆弾の対象になりうる。# 取得したフィードは自分で作ったものではないので、防御済みの実装を使う。from defusedxml import ElementTree as ET
def parse_items(xml_text: str) -> list[dict]: """RSS 2.0 の item を取り出す。guid が無ければ link で代用する。""" root = ET.fromstring(xml_text) out = [] for item in root.findall(".//item"): link = (item.findtext("link") or "").strip() guid = (item.findtext("guid") or link).strip() out.append( { "id": guid, "title": (item.findtext("title") or "").strip(), "link": link, "published": (item.findtext("pubDate") or "").strip(), } ) return out
def only_new(items: list[dict], seen_path: Path) -> list[dict]: seen = set(json.loads(seen_path.read_text())) if seen_path.exists() else set() fresh = [i for i in items if i["id"] not in seen] seen.update(i["id"] for i in items) seen_path.write_text(json.dumps(sorted(seen), ensure_ascii=False)) return freshRSS と Atom は目的が同じで、要素の名前が違います。片方だけを前提に書くと、もう片方で何も取れません。上のコードは RSS 用なので、Atom も扱うなら分岐が要ります。専用のライブラリを使えば両方まとめて扱えます。
Python 標準の xml.etree は、外部実体参照(XXE)や展開爆弾に対して無防備です。フィードは自分が書いたものではないので、防御済みの実装(defusedxml)を使ってください。手元で試すだけなら気になりませんが、定期実行に載せるなら必須です。
取得の間隔を空ける
取得したフィードはに残しておくと、後から解釈を変えたくなったときにやり直せます。フィードは軽いので、頻繁に取りに行きたくなります。ただし更新が1日1回のサイトを1分ごとに叩いても、得られるものは何もありません。更新の頻度に合わせた間隔にしてください。
無駄を減らすなら が使えます。前回の版の印を添えて要求すると、変わっていなければ本文が返りません。相手の負荷もこちらの転送量も減るので、頻繁に確認したい場合はとくに有効です。
フィードが無い場合に初めて、画面の解析を検討します。その判断については EP.16 を参照してください。次回は、機械が読むことを想定していない形式を扱います。
更新を追う仕組みは、一度作れば長く動きます。形式が決まっているので、サイトの改装で壊れることも少ない。手をかける価値のある部類です。
配信が止まることもあります。サイトの改装でフィードだけ廃止される、という例は実際にあります。取得件数がゼロの日が続いたら知らせる仕組みは、ここでも必要です。
フィードが用意されているかどうかで、必要な工数は何倍も変わります。実装に入る前の数分を、この確認に使う価値があります。
なお、フィードで取れるのは更新の事実と概要までです。中身をすべて扱いたいなら、リンク先を別途取りにいくことになります。その場合は、前回までに扱った作法がそのまま必要になります。
この記事の感想を教えてください
あなたの 1 クリックで、本当にこの記事は更新されます。「もっと詳しく」「続編希望」が一定数集まった記事は、 ふくふくが 実際に内容を拡充したり続編記事を公開 します。 送信したリアクションはお使いのブラウザに記録され、再カウントされません。