piaoquan_crawler/scheduling/scheduling_main/crawler_scheduling.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/3/2
import os
import sys
import time

sys.path.append(os.getcwd())
from common.common import Common
from common.scheduling_db import MysqlHelper, RedisHelper


class Scheduling:
    # 读取任务表
    @classmethod
    def get_task(cls, log_type, crawler, env):
        get_sql = """ select * from crawler_task """
        all_task_list = MysqlHelper.get_values(log_type=log_type, crawler=crawler, sql=get_sql, env=env)
        pre_task_list = []
        for task in all_task_list:
            if int(time.time()) >= task["next_time"]:
                pre_task_list.append(task)
        return pre_task_list

    # 更新下次启动时间，调用时机：调度该 task_id 的任务时
    @classmethod
    def update_task(cls, log_type, crawler, task_id, next_time, interval_piaoquan, env):
        if interval_piaoquan > 0:
            new_next_time = next_time + interval_piaoquan
            update_sql = f""" UPDATE crawler_task SET next_time={new_next_time} WHERE task_id={task_id} """
            MysqlHelper.update_values(log_type, crawler, update_sql, env)

    # 资源分配 / 组装
    @classmethod
    def write_redis(cls, log_type, crawler, env):
        pre_task_list = cls.get_task(log_type=log_type, crawler=crawler, env=env)
        if len(pre_task_list) == 0:
            Common.logger(log_type, crawler).info("暂无新任务\n")
        else:
            for pre_task in pre_task_list:
                # machine字段是用来区分海外爬虫和国内爬虫使用的，不涉及任何其他含义
                machine = pre_task.get('machine', 'dev')

                if machine == "hk":
                    # 写入 redis
                    task_key = 'crawler_config_task_queue:hk'
                    RedisHelper.redis_push(env, task_key, str(pre_task))
                elif machine == "aliyun":
                    # 写入 redis
                    task_key = 'crawler_config_task_queue:aliyun'
                    RedisHelper.redis_push(env, task_key, str(pre_task))
                else:
                    # 写入 redis
                    task_key = 'crawler_config_task_queue:dev'
                    RedisHelper.redis_push(env, task_key, str(pre_task))

    @classmethod
    def get_redis(cls, log_type, crawler, env):
        while True:
            if env == 'hk':
                task_key = 'crawler_config_task_queue:hk'
            elif env == 'prod':
                task_key = 'crawler_config_task_queue:aliyun'
            else:
                task_key = 'crawler_config_task_queue:dev'

            redis_data = RedisHelper.redis_pop(env, task_key)
            if redis_data is None or len(redis_data) == 0:
                Common.logger(log_type, crawler).info("Redis为空，等待1秒")
                time.sleep(1)
            else:
                task = eval(str(redis_data, encoding="utf8"))
                return task

    @classmethod
    def scheduling_task(cls, log_type, crawler, env):
        task = cls.get_redis(log_type, crawler, env)
        Common.logger(log_type, crawler).info(f"task: {task}")
        Common.logger(log_type, crawler).info(f"已获取调度任务:{task}")
        task_id = task['task_id']
        source = task['source']
        next_time = task['next_time']
        interval_piaoquan = task['interval_piaoquan']
        spider_name = task['spider_name']
        if env == "aliyun":
            oss_endpoint = "inner"
        elif env == "hk":
            oss_endpoint = "hk"
        else:
            oss_endpoint = "out"

        if int(time.time()) >= next_time:
            cls.update_task(log_type, crawler, task_id, next_time, interval_piaoquan, env)
            # 正式环境，调度任务
            Common.logger(log_type, crawler).info(f"开始调度任务:{task}\n")
            task_str = [('task_id', str(task_id)), ('task_name', str(task['task_name'])),
                        ('source', str(task['source'])), ('next_time', str(task['next_time'])),
                        ('interval_piaoquan', str(task['interval_piaoquan'])),
                        ('play_cnt', eval(task['spider_rule'])['play_cnt']),
                        ('video_width', eval(task['spider_rule'])['video_width']),
                        ('video_height', eval(task['spider_rule'])['video_height']),
                        ('video_like', eval(task['spider_rule'])['video_like']),
                        ('share_cnt', eval(task['spider_rule'])['share_cnt']),
                        ('duration_min', eval(task['spider_rule'])['duration']['min']),
                        ('duration_max', eval(task['spider_rule'])['duration']['max']),
                        ('task_type', task['task_type']), ('spider_link', eval(task['spider_link'])),
                        ('spider_name', str(task['spider_name'])), ('min_publish_time', str(task['min_publish_time'])),
                        ('min_publish_day', str(task['min_publish_day'])), ('media_id', str(task['media_id'])),
                        ('applets_status', str(task['applets_status'])), ('app_status', str(task['app_status'])),
                        ('user_tag', str(task['user_tag'])), ('user_content_tag', str(task['user_content_tag'])),
                        ('machine', str(task['machine']))]
            task_str = str(task_str).replace(' ', '')
            cmd = f"""sh scheduling/scheduling_main/scheduling.sh {source}/{source}_main/{spider_name}_scheduling.py --log_type="{spider_name}" --crawler="{source}" --task="{str(task_str)}" --oss_endpoint="{oss_endpoint}" --env="{env}" {source}/{source}-nohup.log """
            Common.logger(log_type, crawler).info(f"cmd:{cmd}\n")
            os.system(cmd)


if __name__ == "__main__":
    # print(Scheduling.get_task("scheduling", "scheduling", "dev", "local"))
    # print(Scheduling.get_redis("scheduling", "scheduling", "dev", "local"))
    # Scheduling.write_redis("scheduling", "scheduling", "dev", "local")
    Scheduling.scheduling_task("scheduling", "scheduling", "dev")

    pass