text: # -*- coding: utf-8 -*-
  """
  Created on 2017-01-09 10:38
  ---------
  @summary: 组装parser、 parser_control 和 collector
  ---------
  @author: Boris
  @email: boris_liu@foxmail.com
  """
  import threading
  import time
  from collections.abc import Iterable
  
  import feapder.setting as setting
  import feapder.utils.tools as tools
  from feapder.buffer.item_buffer import ItemBuffer
  from feapder.buffer.request_buffer import RequestBuffer
  from feapder.core.base_parser import BaseParser
  from feapder.core.collector import Collector
  from feapder.core.handle_failed_requests import HandleFailedRequests
  from feapder.core.parser_control import PaserControl
  from feapder.db.redisdb import RedisDB
  from feapder.network.item import Item
  from feapder.network.request import Request
  from feapder.utils.log import log
  from feapder.utils.redis_lock import RedisLock
  from feapder.utils import metrics
  
  SPIDER_START_TIME_KEY = "spider_start_time"
  SPIDER_END_TIME_KEY = "spider_end_time"
  SPIDER_LAST_TASK_COUNT_RECORD_TIME_KEY = "last_task_count_record_time"
  
  
  class Scheduler(threading.Thread):
      __custom_setting__ = {}
  
      def __init__(
          self,
          redis_key=None,
          thread_count=None,
          begin_callback=None,
          end_callback=None,
          delete_keys=(),
          keep_alive=None,
          auto_start_requests=None,
          batch_interval=0,
          wait_lock=True,
          task_table=None,
          **kwargs
      ):
          """
          @summary: 调度器
          ---------
          @param redis_key: 爬虫request及item存放redis中的文件夹
          @param thread_count: 线程数，默认为配置文件中的线程数
          @param begin_callback: 爬虫开始回调函数
          @param end_callback: 爬虫结束回调函数
          @param delete_keys: 爬虫启动时删除的key，类型: 元组/bool/string。 支持正则
          @param keep_alive: 爬虫是否常驻，默认否
          @param auto_start_requests: 爬虫是否自动添加任务
          @param batch_interval: 抓取时间间隔 默认为0 天为单位 多次启动时，只有当前时间与第一次抓取结束的时间间隔大于指定的时间间隔时，爬虫才启动
          @param wait_lock: 下发任务时否等待锁，若不等待锁，可能会存在多进程同时在下发一样的任务，因此分布式环境下请将该值设置True
          @param task_table: 任务表， 批次爬虫传递
          ---------
          @result:
          """
  
          super(Scheduler, self).__init__()
  
          for key, value in self.__class__.__custom_setting__.items():
              if key == "AUTO_STOP_WHEN_SPIDER_DONE":  # 兼容老版本的配置
                  setattr(setting, "KEEP_ALIVE", not value)
              else:
                  setattr(setting, key, value)
  
          self._redis_key = redis_key or setting.REDIS_KEY
          if not self._redis_key:
              raise Exception(
                  """
                  redis_key 为redis中存放request与item的目录。不能为空，
                  可在setting中配置，如 REDIS_KEY = 'test'
                  或spider初始化时传参, 如 TestSpider(redis_key='test')
                  """
              )
  
          self._request_buffer = RequestBuffer(redis_key)
          self._item_buffer = ItemBuffer(redis_key, task_table)
  
          self._collector = Collector(redis_key)
          self._parsers = []
          self._parser_controls = []
          self._parser_control_obj = PaserControl
  
          # 兼容老版本的参数
          if "auto_stop_when_spider_done" in kwargs:
              self._keep_alive = not kwargs.get("auto_stop_when_spider_done")
          else:
              self._keep_alive = (
                  keep_alive if keep_alive is not None else setting.KEEP_ALIVE
              )
          self._auto_start_requests = (
              auto_start_requests
              if auto_start_requests is not None
              else setting.SPIDER_AUTO_START_REQUESTS
          )
          self._batch_interval = batch_interval
  
          self._begin_callback = (
              begin_callback
              if begin_callback
              else lambda: log.info("\n********** feapder begin **********")
          )
          self._end_callback = (
              end_callback
              if end_callback
              else lambda: log.info("\n********** feapder end **********")
          )
  
          self._thread_count = (
              setting.SPIDER_THREAD_COUNT if not thread_count else thread_count
          )
  
          self._spider_name = redis_key
          self._project_name = redis_key.split(":")[0]
  
          self._tab_spider_time = setting.TAB_SPIDER_TIME.format(redis_key=redis_key)
          self._tab_spider_status = setting.TAB_SPIDER_STATUS.format(redis_key=redis_key)
          self._tab_requests = setting.TAB_REQUSETS.format(redis_key=redis_key)
          self._tab_failed_requests = setting.TAB_FAILED_REQUSETS.format(
              redis_key=redis_key
          )
  
          self._is_notify_end = False  # 是否已经通知结束
          self._last_task_count = 0  # 最近一次任务数量
          self._redisdb = RedisDB()
  
          self._project_total_state_table = "{}_total_state".format(self._project_name)
          self._is_exist_project_total_state_table = False
  
          # Request 缓存设置
          Request.cached_redis_key = redis_key
          Request.cached_expire_time = setting.RESPONSE_CACHED_EXPIRE_TIME
  
          delete_keys = delete_keys or setting.DELETE_KEYS
          if delete_keys:
              self.delete_tables(delete_keys)
  
          self._last_check_task_status_time = 0
          self.wait_lock = wait_lock
  
          self.init_metrics()
  
      def init_metrics(self):
          """
          初始化打点系统
          """
          metrics.init(**setting.METRICS_OTHER_ARGS)
  
      def add_parser(self, parser, **kwargs):
          parser = parser(**kwargs)  # parser 实例化
          if isinstance(parser, BaseParser):
              self._parsers.append(parser)
          else:
              raise ValueError("类型错误，爬虫需继承feapder.BaseParser或feapder.BatchParser")
  
      def run(self):
          if not self.is_reach_next_spider_time():
              return
  
          self._start()
  
          while True:
              try:
                  if self.all_thread_is_done():
                      if not self._is_notify_end:
                          self.spider_end()  # 跑完一轮
                          self.record_spider_state(
                              spider_type=1,
                              state=1,
                              spider_end_time=tools.get_current_date(),
                              batch_interval=self._batch_interval,
                          )
  
                          self._is_notify_end = True
  
                      if not self._keep_alive:
                          self._stop_all_thread()
                          break
  
                  else:
                      self._is_notify_end = False
  
                  self.check_task_status()
  
              except Exception as e:
                  log.exception(e)
  
              tools.delay_time(1)  # 1秒钟检查一次爬虫状态
  
      def __add_task(self):
          # 启动parser 的 start_requests
          self.spider_begin()  # 不自动结束的爬虫此处只能执行一遍
          self.record_spider_state(
              spider_type=1,
              state=0,
              batch_date=tools.get_current_date(),
              spider_start_time=tools.get_current_date(),
              batch_interval=self._batch_interval,
          )
  
          # 判断任务池中属否还有任务，若有接着抓取
          todo_task_count = self._collector.get_requests_count()
          if todo_task_count:
              log.info("检查到有待做任务 %s 条，不重下发新任务，将接着上回异常终止处继续抓取" % todo_task_count)
          else:
              for parser in self._parsers:
                  results = parser.start_requests()
                  # 添加request到请求队列，由请求队列统一入库
                  if results and not isinstance(results, Iterable):
                      raise Exception("%s.%s返回值必须可迭代" % (parser.name, "start_requests"))
  
                  result_type = 1
                  for result in results or []:
                      if isinstance(result, Request):
                          result.parser_name = result.parser_name or parser.name
                          self._request_buffer.put_request(result)
                          result_type = 1
  
                      elif isinstance(result, Item):
                          self._item_buffer.put_item(result)
                          result_type = 2
  
                      elif callable(result):  # callbale的request可能是更新数据库操作的函数
                          if result_type == 1:
                              self._request_buffer.put_request(result)
                          else:
                              self._item_buffer.put_item(result)
                      else:
                          raise TypeError(
                              "start_requests yield result type error, expect Request、Item、callback func, bug get type: {}".format(
                                  type(result)
                              )
                          )
  
                  self._request_buffer.flush()
                  self._item_buffer.flush()
  
      def _start(self):
          # 启动request_buffer
          self._request_buffer.start()
          # 启动item_buffer
          self._item_buffer.start()
          # 启动collector
          self._collector.start()
  
          # 启动parser control
          for i in range(self._thread_count):
              parser_control = self._parser_control_obj(
                  self._collector,
                  self._redis_key,
                  self._request_buffer,
                  self._item_buffer,
              )
  
              for parser in self._parsers:
                  parser_control.add_parser(parser)
  
              parser_control.start()
              self._parser_controls.append(parser_control)
  
          # 下发任务 因为时间可能比较长，放到最后面
          if setting.RETRY_FAILED_REQUESTS:
              # 重设失败的任务, 不用加锁，原子性操作
              handle_failed_requests = HandleFailedRequests(self._redis_key)
              handle_failed_requests.reput_failed_requests_to_requests()
  
          # 下发新任务
          if self._auto_start_requests:  # 自动下发
              if self.wait_lock:
                  # 将添加任务处加锁，防止多进程之间添加重复的任务
                  with RedisLock(key=self._spider_name) as lock:
                      if lock.locked:
                          self.__add_task()
              else:
                  self.__add_task()
  
      def all_thread_is_done(self):
          for i in range(3):  # 降低偶然性, 因为各个环节不是并发的，很有可能当时状态为假，但检测下一条时该状态为真。一次检测很有可能遇到这种偶然性
              # 检测 collector 状态
              if (
                  self._collector.is_collector_task()
                  or self._collector.get_requests_count() > 0
              ):
                  return False
  
              # 检测 parser_control 状态
              for parser_control in self._parser_controls:
                  if not parser_control.is_not_task():
                      return False
  
              # 检测 item_buffer 状态
              if (
                  self._item_buffer.get_items_count() > 0
                  or self._item_buffer.is_adding_to_db()
              ):
                  return False
  
              # 检测 request_buffer 状态
              if (
                  self._request_buffer.get_requests_count() > 0
                  or self._request_buffer.is_adding_to_db()
              ):
                  return False
  
              tools.delay_time(1)
  
          return True
  
      @tools.run_safe_model("check_task_status")
      def check_task_status(self):
          """
          检查任务状态 预警
          """
          # 每分钟检查一次
          now_time = time.time()
          if now_time - self._last_check_task_status_time > 60:
              self._last_check_task_status_time = now_time
          else:
              return
  
          # 检查redis中任务状态，若连续20分钟内任务数量未发生变化（parser可能卡死），则发出报警信息
          task_count = self._redisdb.zget_count(self._tab_requests)
  
          if task_count:
              if task_count != self._last_task_count:
                  self._last_task_count = task_count
                  self._redisdb.hset(
                      self._tab_spider_time,
                      SPIDER_LAST_TASK_COUNT_RECORD_TIME_KEY,
                      tools.get_current_timestamp(),
                  )  # 多进程会重复发消息， 使用reids记录上次统计时间
              else:
                  # 判断时间间隔是否超过20分钟
                  lua = """
                      -- local key = KEYS[1]
                      local field = ARGV[1]
                      local current_timestamp = ARGV[2]
  
                      -- 取值
                      local last_timestamp = redis.call('hget', KEYS[1], field)
                      if last_timestamp and current_timestamp - last_timestamp >= 1200 then
                          return current_timestamp - last_timestamp -- 返回任务停滞时间 秒
                      end
  
                      if not last_timestamp then
                          redis.call('hset', KEYS[1], field, current_timestamp)
                      end
  
                      return 0
  
                  """
                  redis_obj = self._redisdb.get_redis_obj()
                  cmd = redis_obj.register_script(lua)
                  overtime = cmd(
                      keys=[self._tab_spider_time],
                      args=[
                          SPIDER_LAST_TASK_COUNT_RECORD_TIME_KEY,
                          tools.get_current_timestamp(),
                      ],
                  )
  
                  if overtime:
                      # 发送报警
                      msg = "《{}》爬虫任务停滞 {}，请检查爬虫是否正常".format(
                          self._spider_name, tools.format_seconds(overtime)
                      )
                      log.error(msg)
                      self.send_msg(
                          msg,
                          level="error",
                          message_prefix="《{}》爬虫任务停滞".format(self._spider_name),
                      )
  
          else:
              self._last_task_count = 0
  
          # 检查失败任务数量 超过1000 报警，
          failed_count = self._redisdb.zget_count(self._tab_failed_requests)
          if failed_count > setting.WARNING_FAILED_COUNT:
              # 发送报警
              msg = "《%s》爬虫当前失败任务 %s, 请检查爬虫是否正常" % (self._spider_name, failed_count)
              log.error(msg)
              self.send_msg(
                  msg,
                  level="error",
                  message_prefix="《%s》爬虫当前失败任务数报警" % (self._spider_name),
              )
  
          # parser_control实时统计已做任务数及失败任务数，若成功率<0.5 则报警
          failed_task_count, success_task_count = PaserControl.get_task_status_count()
          total_count = success_task_count + failed_task_count
          if total_count > 0:
              task_success_rate = success_task_count / total_count
              if task_success_rate < 0.5:
                  # 发送报警
                  msg = "《%s》爬虫当前任务成功数%s, 失败数%s, 成功率 %.2f, 请检查爬虫是否正常" % (
                      self._spider_name,
                      success_task_count,
                      failed_task_count,
                      task_success_rate,
                  )
                  log.error(msg)
                  self.send_msg(
                      msg,
                      level="error",
                      message_prefix="《%s》爬虫当前任务成功率报警" % (self._spider_name),
                  )
  
          # 检查入库失败次数
          if self._item_buffer.export_falied_times > setting.EXPORT_DATA_MAX_FAILED_TIMES:
              msg = "《{}》爬虫导出数据失败，失败次数：{}， 请检查爬虫是否正常".format(
                  self._spider_name, self._item_buffer.export_falied_times
              )
              log.error(msg)
              self.send_msg(
                  msg, level="error", message_prefix="《%s》爬虫导出数据失败" % (self._spider_name)
              )
  
      def delete_tables(self, delete_tables_list):
          if isinstance(delete_tables_list, bool):
              delete_tables_list = [self._redis_key + "*"]
          elif not isinstance(delete_tables_list, (list, tuple)):
              delete_tables_list = [delete_tables_list]
  
          redis = RedisDB()
          for delete_tab in delete_tables_list:
              if not delete_tab.startswith(self._redis_key):
                  delete_tab = self._redis_key + delete_tab
              tables = redis.getkeys(delete_tab)
              for table in tables:
                  if table != self._tab_spider_time:
                      log.info("正在删除key %s" % table)
                      redis.clear(table)
  
      def _stop_all_thread(self):
          self._request_buffer.stop()
          self._item_buffer.stop()
          # 停止 collector
          self._collector.stop()
          # 停止 parser_controls
          for parser_control in self._parser_controls:
              parser_control.stop()
  
          self._started.clear()
  
      def send_msg(self, msg, level="debug", message_prefix=""):
          # log.debug("发送报警 level:{} msg{}".format(level, msg))
          tools.send_msg(msg=msg, level=level, message_prefix=message_prefix)
  
      def spider_begin(self):
          """
          @summary: start_monitor_task 方式启动，此函数与spider_end不在同一进程内，变量不可共享
          ---------
          ---------
          @result:
          """
  
          if self._begin_callback:
              self._begin_callback()
  
          for parser in self._parsers:
              parser.start_callback()
  
          # 记录开始时间
          if not self._redisdb.hexists(self._tab_spider_time, SPIDER_START_TIME_KEY):
              current_timestamp = tools.get_current_timestamp()
              self._redisdb.hset(
                  self._tab_spider_time, SPIDER_START_TIME_KEY, current_timestamp
              )
  
              # 发送消息
              self.send_msg("《%s》爬虫开始" % self._spider_name)
  
      def spider_end(self):
          self.record_end_time()
  
          if self._end_callback:
              self._end_callback()
  
          for parser in self._parsers:
              if not self._keep_alive:
                  parser.close()
              parser.end_callback()
  
          if not self._keep_alive:
              # 关闭webdirver
              if Request.webdriver_pool:
                  Request.webdriver_pool.close()
  
              # 关闭打点
              metrics.close()
          else:
              metrics.flush()
  
          # 计算抓取时长
          data = self._redisdb.hget(
              self._tab_spider_time, SPIDER_START_TIME_KEY, is_pop=True
          )
          if data:
              begin_timestamp = int(data)
  
              spand_time = tools.get_current_timestamp() - begin_timestamp
  
              msg = "《%s》爬虫结束，耗时 %s" % (
                  self._spider_name,
                  tools.format_seconds(spand_time),
              )
              log.info(msg)
  
              self.send_msg(msg)
  
          if self._keep_alive:
              log.info("爬虫不自动结束， 等待下一轮任务...")
          else:
              self.delete_tables(self._tab_spider_status)
  
      def record_end_time(self):
          # 记录结束时间
          if self._batch_interval:
              current_timestamp = tools.get_current_timestamp()
              self._redisdb.hset(
                  self._tab_spider_time, SPIDER_END_TIME_KEY, current_timestamp
              )
  
      def is_reach_next_spider_time(self):
          if not self._batch_interval:
              return True
  
          last_spider_end_time = self._redisdb.hget(
              self._tab_spider_time, SPIDER_END_TIME_KEY
          )
          if last_spider_end_time:
              last_spider_end_time = int(last_spider_end_time)
              current_timestamp = tools.get_current_timestamp()
              time_interval = current_timestamp - last_spider_end_time
  
              if time_interval < self._batch_interval * 86400:
                  log.info(
                      "上次运行结束时间为 {} 与当前时间间隔 为 {}, 小于规定的抓取时间间隔 {}。爬虫不执行，退出～".format(
                          tools.timestamp_to_date(last_spider_end_time),
                          tools.format_seconds(time_interval),
                          tools.format_seconds(self._batch_interval * 86400),
                      )
                  )
                  return False
  
          return True
  
      def record_spider_state(
          self,
          spider_type,
          state,
          batch_date=None,
          spider_start_time=None,
          spider_end_time=None,
          batch_interval=None,
      ):
          pass
  
      def join(self, timeout=None):
          """
          重写线程的join
          """
          if not self._started.is_set():
              return
  
          super().join()
  