auto/utils/utils_create_rss_record.py

import time

import psycopg2
from psycopg2 import Error


class CreateRssRecord(object):
    def __init__(self):
        self.hostname = 'erhe.top'
        self.port = 20788
        self.database = 'freshrss'
        self.user = 'freshrss'
        self.password = 'freshrss'
        self.conn = None

    def connect(self):
        """连接到 PostgreSQL 数据库"""
        try:
            self.conn = psycopg2.connect(
                dbname=self.database,
                user=self.user,
                password=self.password,
                host=self.hostname,
                port=self.port
            )
        except Error as e:
            print(f"Error connecting to the database: {e}")
        else:
            print("Connected to the database successfully.")

        if not self.conn:
            raise Exception("Database connection failed")

    def check_and_insert(self, data):
        """检查 URL 是否存在，如果不存在则插入整个数据字典"""
        try:
            with self.conn.cursor() as cursor:
                # 检查 URL 是否存在
                select_sql = "SELECT COUNT(*) FROM freshrsstoor_feed WHERE url = %s"
                cursor.execute(select_sql, (data['url'],))
                result = cursor.fetchone()
                if result[0] == 0:
                    # URL 不存在，插入新记录
                    columns = ', '.join(data.keys())
                    values = ', '.join(['%s'] * len(data))
                    insert_sql = """INSERT INTO freshrsstoor_feed 
                    (url, kind, category, "name", website, description, "lastUpdate", priority, "pathEntries", "httpAuth", "error", "ttl", attributes, "cache_nbEntries", "cache_nbUnreads") 
                    VALUES ('{}', {}, {}, '{}', '{}', '{}', {}, {}, '{}', '{}', {}, {}, '{}', {}, {});""".format(
                        data['url'],
                        data['kind'],
                        data['category'],
                        data['name'],
                        data['website'],
                        data['description'],
                        data['lastUpdate'],
                        data['priority'],
                        data['pathEntries'],
                        data['httpAuth'],
                        data['error'],
                        data['ttl'],
                        data['attributes'],
                        data['cache_nbEntries'],
                        data['cache_nbUnreads']
                    )
                    cursor.execute(insert_sql, tuple(data.values()))
                    self.conn.commit()
                    print("Data inserted successfully.")
                else:
                    print("URL already exists.")
        except Error as e:
            print(f"Error: {e}")
        finally:
            if self.conn is not None:
                self.conn.close()


# 使用示例
if __name__ == "__main__":
    cr = CreateRssRecord()
    cr.connect()
    insert_data = {
        'url': 'https://rsshub.app/jike/topic/556688fae4b00c57d9dd46ee',
        'category': 7,
        'name': '今日份的摄影 - 即刻圈子',
        'website': 'http://finance.sina.com.cn/china/',
        'description': '爱摄影的即友都在这里~分享原创摄影作品，感受照片背后的共鸣吧！ - Powered by RSSHub',
        'kind': 0,
        'lastUpdate': int(time.time()),
        'priority': 10,
        'pathEntries': '',
        'httpAuth': '',
        'error': 0,
        'ttl': 0,
        'attributes': '{"curl_params":null,"ssl_verify":null,"timeout":null}',
        'cache_nbEntries': 0,
        'cache_nbUnreads': 0
    }
    cr.check_and_insert(insert_data)