やってみる

アウトプットすべく己を導くためのブログ。その試行錯誤すらたれ流す。

PythonでRSSからHTMLの本文を抽出してSQLite3に挿入する(未登録のみ。UNIQUE制約でチェック)

 DBにまかせてPythonコードを減らす。 

成果物

概要

 DBより新しいニュースだけを取り込む。

 その方法としてSQLite3の一意制約(UNIQUE)とコンフリクトを用いた。公開日時とURLの2つが重複したら同一ニュースと判断しUNIQUE制約違反となる。insert or fail ...によりUNIQUE制約違反のときは中断する。

  1. RSSから新しくニュースを取得する
  2. 1を公開日時の降順にソートする
  3. DB既存と比較して重複があれば中断する(それ以降のニュースもinsertしない)

前回との違い

 今回のコードは、RSSがDBより古くても取り込んでしまう。前回はPythonにて日付の比較をしていたが、今回はUNIQUE制約にまかせたから。

 だが、RSSは常に最新状態なので、古いニュースは入らないはず。仮に古いニュースだとしても、入って困るわけではない。よって、思い切って削除した。

課題

  • 「n件の新しいニュースを取得しました」みたいなメッセージかログが欲しい
    • 取得日時, 情報源RSS, も添えて

コード

 変更対象のファイルのみ抜粋。

NewsDb.py

import sqlite3
import os
import operator

class NewsDb:
    def __init__(self, root):
        path = os.path.join(root, 'news.db')
        self.conn = sqlite3.connect(path)
        self.create_table()
        self.news = []
    def __del__(self): self.conn.close()
    def create_table(self):    
        cur = self.conn.cursor()
        cur.executescript(self.__create_table_sql())
    def __create_table_sql(self):
        return '''
create table if not exists news(
  id         integer primary key,
  published  text, 
  url        text,
  title      text,
  body       text, -- URL先から本文だけを抽出したプレーンテキスト
  UNIQUE(published,url) -- 記事の一意確認
);
create index if not exists idx_news on 
  news(published desc, id desc, url, title);
create table if not exists sources(
  id       integer primary key,
  domain   text, -- URLのドメイン名
  name     text, -- 情報源名
  created  text  -- 登録日時(同一ドメイン名が複数あるとき新しいほうを表示する)
);
create index if not exists idx_sources on 
  sources(domain, created desc, id desc, name);
'''
    def __get_latest_sql(self): return '''
with 
  latest(max_published) as (
    select max(published) max_published from news
  )
select 
  published as latest_published, 
  max(id) as latest_id 
from news,latest
where news.published=latest.max_published;
'''
    def __insert_sql(self): 
        return 'insert or fail into news(published,url,title,body) values(?,?,?,?)'
    def append_news(self, published, url, title, body):
        self.news.append((published, url, title, body))
    def insert(self):
        if 0 == len(self.news): return
        try:
            self.news = sorted(self.news, key=operator.itemgetter(1)) # 第2キー: URL昇順
            self.news = sorted(self.news, key=operator.itemgetter(0), reverse=True) # 第1キー: 公開日時降順
            self.conn.cursor().executemany(self.__insert_sql(), self.news)
            self.conn.commit()
        except sqlite3.IntegrityError as err_sql_integ:
            import traceback
            import sys
            msg = str(err_sql_integ.with_traceback(sys.exc_info()[2])).lower() # UNIQUE constraint failed: news.published, news.url
            # DB既存と重複した時点で中断する
            if ('UNIQUE'.lower() in msg and 'published' in msg and 'url' in msg): pass
            # それ以外ならエラー表示&ロールバックする
            else: 
                traceback.print_exc()
                self.conn.rollback() 
        except: # それ以外
            import traceback
            traceback.print_exc()
            self.conn.rollback() # ロールバックする
        finally: self.news.clear()

対象環境

$ uname -a
Linux raspberrypi 4.19.42-v7+ #1218 SMP Tue May 14 00:48:17 BST 2019 armv7l GNU/Linux

前回まで