AutoScraperX/application/config/common/redis/redis_helper.py

import redis
import os
import sys

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


class RedisHelper:
    @classmethod
    def connect_redis(cls, env):
        if env == 'hk':
            redis_pool = redis.ConnectionPool(
                # host='r-bp154bpw97gptefiqk.redis.rds.aliyuncs.com',  # 内网地址
                # host='r-bp154bpw97gptefiqkpd.redis.rds.aliyuncs.com',  # 测试地址
                host='r-bp1mb0v08fqi4hjffupd.redis.rds.aliyuncs.com',  # 外网地址
                port=6379,
                db=2,
                password='Wqsd@2019'
            )
            redis_conn = redis.Redis(connection_pool=redis_pool)
        elif env == 'prod':
            redis_pool = redis.ConnectionPool(
                host='r-bp1mb0v08fqi4hjffu.redis.rds.aliyuncs.com',  # 内网地址
                # host='r-bp1mb0v08fqi4hjffupd.redis.rds.aliyuncs.com',  # 外网地址
                port=6379,
                db=2,
                password='Wqsd@2019'
            )
            redis_conn = redis.Redis(connection_pool=redis_pool)
        else:
            redis_pool = redis.ConnectionPool(
                # host='r-bp154bpw97gptefiqk.redis.rds.aliyuncs.com',  # 内网地址
                host='r-bp154bpw97gptefiqkpd.redis.rds.aliyuncs.com',  # 外网地址
                port=6379,
                db=2,
                password='Qingqu2019'
            )
            redis_conn = redis.Redis(connection_pool=redis_pool)
        return redis_conn

    @classmethod
    def redis_push(cls, env, task_key, data):
        redis_conn = cls.connect_redis(env)
        # print("开始写入数据")
        redis_conn.lpush(task_key, data)
        # print("数据写入完成")

    @classmethod
    def redis_pop(cls, env, task_key):
        redis_conn = cls.connect_redis(env)
        if redis_conn.llen(task_key) == 0:
            return None
        else:
            return redis_conn.rpop(task_key)