やってみる

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

📌 ？
トップ > Webスクレイピング > PythonでRSSからHTMLの本文を抽出してSQLite3に挿入する（未登録のみ挿入する）

PythonでRSSからHTMLの本文を抽出してSQLite3に挿入する（未登録のみ挿入する）

Webスクレイピング

　RSSデータがDB最新より新しいかチェックする。

成果物

主な変更コード

NewsDb.py

class NewsDb:
    def __is_get_latest(self):
        cur = self.conn.cursor()
        return cur.execute(self.__get_latest_sql()).fetchone()

    def __get_newer_news(self):
        self.news = sorted(self.news, key=operator.itemgetter(1)) # 第2キー: URL昇順
        self.news = sorted(self.news, key=operator.itemgetter(0), reverse=True) # 第1キー: 公開日時降順
        latest_published, latest_url = self.__is_get_latest() # DB最新
        if latest_published is None: return self.news # DBが0件なら全件挿入する

        # JSON最古がDB最新より新しければ全件挿入する
        if self.news[len(self.news)-1][0] > latest_published: return self.news
        # JSON最新がDB最新と同じか古ければ挿入しない
        if (self.news[0][0] < latest_published or
           (self.news[0][0] == latest_published and 
            self.news[0][1] == latest_url)): return None
        # JSON内にDB最新が存在する
        try: 
            # 先頭からDB最新+1までを挿入する
            idx = self.news.index((latest_published, latest_url))
            return self.news[0:idx]
        # JSON内にDB最新が存在しない
        except ValueError:
            # JSON先頭がDB最新より新しいなら全件挿入する
            if self.news[0][0] > latest_published: return self.news
            # JSON先頭がDB最新かそれより古いなら挿入しない
            else: return None
    def insert(self):
        if 0 == len(self.news): return
        try:
            ins_news = self.__get_newer_news()
            if ins_news is None: return
            cur = self.conn.cursor()
            cur.executemany(self.__insert_sql(), ins_news)
            self.conn.commit()
            self.news.clear()
        except: 
            import traceback
            traceback.print_exc()
            self.conn.rollback()

所感

　テストできてない。

　あと、最新チェックのところって表制約で簡略化できないか？　Pythonで実装するの超大変。

対象環境

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

前回まで