文件时间进度扫描控制，可回溯，空闲扫描，系统时间调整不影响

1.文件扫描获取

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import time
import glob
import logging
from datetime import datetime
from datetime import timedelta

from utils import Utils
from netflowmeta_config import NOT_SCAN
from netflowmeta_config import FREE_TIMES
from netflowmeta_config import SLEEP_TIME
from netflowmeta_config import FINISH_SCAN
from netflowmeta_config import TCP_FLOW_APTH
from netflowmeta_config import BASE_SCAN_NODE

from progress_manage import DateManager

class TcpAvroScanner(object):

    def __init__(self, time_span=60, delay=20):
        """
        :param time_span:文件获取时间跨度，单位秒
        """
        self.today = None
        self.delay = delay
        self.offset = time_span
        self._ctrl_info = self.init_scan_control(time_span)
        self.progress_manager = DateManager(TCP_FLOW_APTH, self.offset)



    def get_new_files(self):
        """
        扫描获取新文件
        :return:
        """
        self.progress_manager.flush_progress()
        self.today = datetime.now()
        try:
            scan_date = self.schedule_date_ctl()
            status, index, offset = self.schedule_time_ctl(scan_date)
            if status:
                start_time, end_time = self._get_scan_time_range(index, offset)
                files_dict = self.select_file(scan_date, start_time, end_time)
                self.save_progress(scan_date, index)
            else:
                files_dict = {}
            return files_dict
        except Exception as e:
            logging.error('get new file list warn, msg: {}'.format(e.message))
            time.sleep(SLEEP_TIME)
            return {}

    def schedule_time_ctl(self, scan_date):
        """
        根据调度日期，去获取扫描点
        :param scan_date: 扫描日期
        :return: index 扫描点
        """
        today = self.today.strftime("%Y%m%d")
        if scan_date not in self.progress_manager.get():
            self._ctrl_info['counter'] = FREE_TIMES
            logging.info("There is no file in {}".format(scan_date))
            return False, 0, 0
        range_val = self.progress_manager.get(scan_date)["range"]
        cur_offset = self.progress_manager.get(scan_date)["offset"]
        if scan_date == today:
            scan_now = self.today - timedelta(seconds=self.delay)
            max_range = (scan_now.hour * 60 + scan_now.minute) * 60 // cur_offset
            for i in range(max_range):
                if Utils.get_scan_val(range_val, i) == NOT_SCAN:
                    self._ctrl_info['counter'] = 0
                    return True, i, cur_offset
            self._ctrl_info['counter'] += 1
            return False, max_range - 1, cur_offset
        else:
            scan_node = self.progress_manager.init_scan_node(cur_offset)
            if Utils.get_scan_val(range_val, scan_node) != FINISH_SCAN:
                for i in range(scan_node):
                    if Utils.get_scan_val(range_val, i) == NOT_SCAN:
                        return True, i, cur_offset
            self.set_scan_finished(scan_date)
            logging.info("Forbid scan date {}. There is no files to get.".format(scan_date))
            return False, scan_node, cur_offset

    def schedule_date_ctl(self):
        """
        日期调度，优先调度当天。
        空闲调度：当前系统时间连续调度 FREE_TIMES 次，回溯调度一次
        :return:date:%Y%m%d
        """
        if self._ctrl_info['counter'] >= FREE_TIMES:
            self._ctrl_info['counter'] = 0
            all_date = self.progress_manager.get().keys()
            if len(all_date) > 1:
                all_date.sort(reverse=True)
                today_format = self.today.strftime('%Y%m%d')
                backtrack_index = all_date.index(today_format) + 1 if today_format in all_date else 0
                for i in range(backtrack_index, len(all_date)):
                    scan_node = self.progress_manager.init_scan_node(
                        self.progress_manager.get(all_date[i])["offset"]
                    )
                    if Utils.get_scan_val(
                            self.progress_manager.get(all_date[i])["range"],
                            scan_node
                    ) != FINISH_SCAN:
                        # 不等于0 就可以扫描调度，否则排除当前日期
                        return all_date[i]
                #无文件休眠
                time.sleep(SLEEP_TIME)
        #非调度，时间为当天时
        # 此处与扫描时间保持一致 - delay秒，防止跨天进度保存bug
        return (self.today - timedelta(seconds=self.delay)).strftime("%Y%m%d")

    def _get_scan_time_range(self, index, offset):
        """
        获取扫描时间范围
        :return:
        """
        return self.change_index_to_time_range(index, offset)

    def change_index_to_time_range(self, index, offset):
        """
        根据扫描索引转化为扫描时间
        :param index: 扫描点
        :return: 起始时间，结束时间
        """
        start = index * offset
        end = (index + 1) * offset
        return [int(self.format_sec(start)), int(self.format_sec(end))]

    def format_sec(self, total):
        """
        格式化文件时间表达式
        :param total:
        :return:
        """
        total = total if total < BASE_SCAN_NODE else BASE_SCAN_NODE
        return "{}{}{}".format(
            str(total // 3600).zfill(2),
            str(total % 3600 // 60).zfill(2),
            str(total % 3600 % 60).zfill(2)
        )

    def select_file(self, scan_date, start_time, end_time):
        pattern = os.path.join(TCP_FLOW_APTH, scan_date, '*')
        file_list_of_dev = {}
        dev_path_list = [dp for dp in glob.glob(pattern) if os.path.isdir(dp)]
        for dev_path in dev_path_list:
            dev_name = "{}_{}".format(os.path.basename(dev_path),scan_date)
            file_list_of_dev[dev_name] = []
            file_pattern = os.path.join(dev_path, '*.avro')
            file_gen = (fl for fl in glob.glob(file_pattern))
            for fl in file_gen:
                fl_time = int(os.path.basename(fl).split("_")[0])
                if start_time <= fl_time < end_time:
                    file_list_of_dev[dev_name].append(fl)
        return file_list_of_dev

    def init_scan_control(self,span):
        progress = {}
        progress['counter'] = 0
        progress['offset'] = span
        return progress

    def save_progress(self, scan_date, index):
        """
        保存扫描进度
        :param scan_date: 日期
        :param index:扫描点
        :return:
        """
        self.progress_manager.save_scan_info(scan_date, index)

    def set_scan_finished(self, scan_date):
        """
        设定扫描完成
        :param scan_date:
        :return:
        """
        self.progress_manager.set_scan_finished(scan_date)


if __name__ == '__main__':
    ta = TcpAvroScanner(57)
    while True:
        ta.get_new_files()
                        

2.进度控制

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import re
import copy
from datetime import date

from cfgopr import CfgOpr

from utils import Utils
from netflowmeta_config import NOT_SCAN
from netflowmeta_config import FINISH_SCAN
from netflowmeta_config import BASE_SCAN_NODE
from netflowmeta_config import DATE_PROGRESS_FILE


class BaseManager(object):

    def __init__(self, path):
        self.path = path


class DateManager(BaseManager):


    def __init__(self, path, span=60):
        super(DateManager, self).__init__(path)
        self.offset = span
        self.__progress = None
        self.init_progress()

    def init_progress(self):
        """
        日期进度初始化
        扫描进度初始化
        :return:
        """
        if not self.__progress:
            self.get_all_progress()
        else:
            self.flush_progress()

    def get_all_progress(self, pattern='^d{8}$'):
        """
        获取所有key下的进度
        :param key: 日志类型.文件类型
        :param path: 日志类型目录
        :param span: 默认获取跨度
        :param pattern: 日期正则模式
        :return: True
        """
        all_progress = {}
        date_path = [
            (os.path.join(self.path, tar_date, DATE_PROGRESS_FILE),tar_date) 
            for tar_date in os.listdir(self.path) 
            if (re.match(pattern, tar_date) and os.path.isdir(os.path.join(self.path, tar_date)))
        ]
        for item in date_path:
            _, dp_data = CfgOpr.json_load(item[0])
            if dp_data:
                dp_data['range'] = Utils.base32to10(dp_data['range'])
                all_progress[item[1]] = dp_data
            else:
                dp_data['range'] = Utils.set_scan_val(
                    NOT_SCAN,
                    self.init_scan_node(self.offset)
                )
                dp_data['offset'] = self.offset
                all_progress[item[1]] = dp_data
        cur_date = date.today().strftime("%Y%m%d")
        if cur_date not in all_progress:
            all_progress[cur_date] = {}
            all_progress[cur_date]['range'] = Utils.set_scan_val(
                    NOT_SCAN,
                    self.init_scan_node(self.offset)
                )
            all_progress[cur_date]['offset'] = self.offset
        self.__progress = all_progress
        return True

    def flush_progress(self, pattern='^d{8}$'):
        """
        刷新进度，防止数据目录淘汰后异常，或跨天新增目录，未纳入管理
        :param key: 日志名.文件名
        :param path: 进度文件目录（日期目录同级）
        :param span: 文件获取时间跨度
        :param pattern: 文件夹正则表达式
        :return: True
        """
        all_exist_date = [
            tar_date for tar_date in os.listdir(self.path) 
            if (re.match(pattern, tar_date) and
                os.path.isdir(os.path.join(self.path, tar_date))
                )
        ]
        new_progress = {}
        for old_date in self.__progress.keys():
            if old_date in all_exist_date:
                new_progress[old_date] = self.__progress[old_date]
        for exist_date in all_exist_date:
            if exist_date not in new_progress:
                new_progress[exist_date] = {}
                new_progress[exist_date]['range'] = Utils.set_scan_val(
                    NOT_SCAN,
                    self.init_scan_node(self.offset)
                )
                new_progress[exist_date]['offset'] = self.offset
        self.__progress = new_progress
        return True

    def set_scan_finished(self, scan_date):
        """
        设定扫描完成
        :param key:日志类型.文件类型
        :param path: 保存路径
        :param scan_date: 扫描日期
        :return: True or False
        """
        range_val = self.__progress[scan_date]["range"]
        scan_node = self.init_scan_node(
            self.__progress[scan_date]["offset"]
        )
        self.__progress[scan_date]["range"] = 
            Utils.set_scan_val(range_val, scan_node, FINISH_SCAN)
        save_data = copy.deepcopy(self.__progress[scan_date])
        save_data["range"] = Utils.base10to32(save_data["range"])
        progress_path = os.path.join(self.path, scan_date, DATE_PROGRESS_FILE)
        return CfgOpr.json_dump(save_data, progress_path)

    def save_scan_info(self, scan_date, index):
        """
        保存扫描信息
        :param key:日志类型.文件类型
        :param path: 保存路径
        :param scan_date: 扫描日期
        :param index: 扫描点
        :return: True or False
        """
        self.__progress[scan_date]["range"] = 
            Utils.set_scan_val(
                self.__progress[scan_date]["range"],
                index
            )
        save_data = copy.deepcopy(self.__progress[scan_date])
        save_data["range"] = Utils.base10to32(save_data["range"])
        progress_path = os.path.join(self.path, scan_date, DATE_PROGRESS_FILE)
        return CfgOpr.json_dump(save_data, progress_path)

    def init_scan_node(self, span):
        """
        初始化扫描节点
        :param span:文件获取时间跨度
        :return: 当前扫描节点
        """
        return BASE_SCAN_NODE // int(span) + (1 if BASE_SCAN_NODE % int(span) else 0)
    
    def get(self, scan_date=None):
        return self.__progress if scan_date == None else self.__progress.get(scan_date, {})

3.配置

BASE_SCAN_NODE = 86400

FINISH_SCAN = 0
NOT_SCAN = 0


FREE_TIMES = 3

SLEEP_TIME = 1