automatic_crawler/application/functions/zqkd_db_redis.py

import os
import sys
import threading
import traceback
from datetime import datetime, timedelta

import redis

from application.common import Local

sys.path.append(os.getcwd())

from application.common.mysql import MysqlHelper


class DatabaseOperations:
    def __init__(self, mode, platform):
        self.mysql = MysqlHelper(mode=mode, platform=platform)
        self.LocalLog = Local.logger(platform, mode)

    def check_user_id(self, uid):
        """
        检查指定用户ID是否存在于数据库的zqkd_uid表中。

        :param uid:要检查的用户ID
        :return:如果用户ID存在于表中返回True，否则返回False
        """
        try:
            query_sql = f""" SELECT uid FROM zqkd_uid WHERE uid = "{uid}"; """
            result = self.mysql.select(sql=query_sql)
            return bool(result)
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"检查用户ID失败: {e}\n{tb}")
            return False

    def update_user(self, uid, user_name, avatar_url):
        """
        更新数据库中指定用户的用户名和头像URL。

        :param uid:要更新信息的用户ID
        :param user_name:新的用户名
        :param avatar_url:新的头像URL
        :return:如果更新操作成功，返回更新操作的结果（通常是影响的行数），失败则返回None或抛出异常
        """
        try:
            update_sql = f""" UPDATE zqkd_uid SET avatar_url = "{avatar_url}", user_name = "{user_name}" WHERE uid = "{uid}"; """
            return self.mysql.update(sql=update_sql)
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"更新用户信息失败: {e}\n{tb}")
            return None

    def insert_user(self, uid, user_name, avatar_url):
        """
        向数据库的zqkd_uid表中插入新的用户信息，包含用户ID、用户名、头像URL和当前时间。

        :param uid:新用户的ID
        :param user_name:新用户的用户名
        :param avatar_url:新用户的头像URL
        :return:如果插入操作成功，返回插入操作的结果（通常是影响的行数），失败则返回None或抛出异常
        """
        try:
            current_time = datetime.now()
            formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
            insert_sql = f""" INSERT INTO zqkd_uid (uid, avatar_url, user_name, data_time) VALUES ('{uid}', '{avatar_url}', '{user_name}', '{formatted_time}'); """
            return self.mysql.update(sql=insert_sql)
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"插入用户信息失败: {e}\n{tb}")
            return None

    def select_user(self, last_scanned_id=0):
        """
        根据last_scanned_id分页查询用户数据
        :param last_scanned_id: 上次扫描的ID，0表示从头开始
        :return: 查询结果列表
        """
        try:
            # 构建查询（根据last_scanned_id过滤）
            query = "SELECT id, uid FROM zqkd_uid"
            if last_scanned_id > 0:
                query += f" WHERE id > {last_scanned_id}"
            query += " ORDER BY id ASC"

            return self.mysql.select(query)
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"查询用户列表失败: {e}\n{tb}")
            return []


class RedisOperations:
    _pool: redis.ConnectionPool = None
    _instance = None
    _lock = threading.Lock()  # 用于线程安全的单例创建

    @classmethod
    def get_instance(cls, mode="", platform=""):
        """线程安全的单例获取方法"""
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = cls(mode, platform)
        return cls._instance

    def __init__(self, mode, platform):
        # 私有构造函数，使用 get_instance() 获取实例
        self.mode = mode
        self.platform = platform
        self.LocalLog = Local.logger(self.platform, self.mode)
        if RedisOperations._instance is not None:
            raise Exception("请使用 get_instance() 获取实例")

        self._pool = self._get_pool()
        self.client = redis.Redis(connection_pool=self._pool, decode_responses=True)  # 复用同一个客户端

    def _get_pool(self) -> redis.ConnectionPool:
        if self._pool is None:
            try:
                self._pool = redis.ConnectionPool(
                    host="r-bp1mb0v08fqi4hjffu.redis.rds.aliyuncs.com",
                    port=6379,
                    db=0,
                    password="Wqsd@2019",
                    max_connections=50,  # 增加最大连接数
                    socket_timeout=10,
                    retry_on_timeout=True
                )
            except Exception as e:
                tb = traceback.format_exc()
                self.LocalLog.error(f"创建Redis连接池失败: {e}\n{tb}")
                raise
        return self._pool

    def close(self):
        """关闭连接池"""
        try:
            if self._pool:
                self._pool.disconnect(inuse_connections=True)
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"关闭Redis连接池失败: {e}\n{tb}")

    def get_recommend_video(self, task="task:zqkd_video_id"):
        """从Redis的指定列表中弹出并返回最左边的视频ID"""
        try:
            value_bytes = self.client.rpop(task)
            value_str = value_bytes.decode('utf-8')
            return value_str
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"获取推荐视频ID失败: {e}\n{tb}")
            return None

    def check_video_id_exists(self, videoID):
        """检查指定的视频ID是否已经存在于Redis中"""
        try:
            key = f"crawler:zqkd:{videoID}"
            return self.client.exists(key)
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"检查视频ID是否存在失败: {e}\n{tb}")
            return False

    def save_video_id(self, videoID):
        """将视频ID存储到Redis中，并为其设置7天的过期时间"""
        try:
            key = f"crawler:zqkd:{videoID}"
            expiration_time = int(timedelta(days=7).total_seconds())
            self.client.setex(key, expiration_time, "1")
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"保存视频ID失败: {e}\n{tb}")

    def save_recommend_video(self, videoID):
        """将推荐视频ID添加到Redis的指定列表中，并为该列表设置2天的过期时间"""
        try:
            task = "task:zqkd_video_id"
            pipe = self.client.pipeline()  # 使用管道执行多个命令
            pipe.rpush(task, videoID)
            pipe.expire(task, int(timedelta(days=2).total_seconds()))
            pipe.execute()

            # 检查数据是否写入成功
            list_length = self.client.llen(task)
            self.LocalLog.info(f"保存推荐视频ID成功，列表长度: {list_length}")
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"保存推荐视频ID失败: {e}\n{tb}")

    def get_last_scanned_id(self):
        """获取上次扫描的ID"""
        try:
            return self.client.get("zqkd_last_scanned_id").decode('utf-8')
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"获取上次扫描的ID失败: {e}\n{tb}")
            return None

    def set_last_scanned_id(self, last_scanned_id):
        """设置上次扫描的ID"""
        try:
            result = self.client.set("zqkd_last_scanned_id", last_scanned_id)
            if result:
                self.LocalLog.info(f"成功设置上次扫描的ID: {last_scanned_id}")
        except Exception as e:
            tb = traceback.format_exc()
            self.LocalLog.error(f"设置上次扫描的ID失败: {e}\n{tb}")
            return False


if __name__ == '__main__':
    db = DatabaseOperations("12", "123")
    user = db.select_user(10000)
    print(user)