| 2838 | |
| 2839 | |
| 2840 | class _Script: |
| 2841 | __slots__ = ('name', 'cmd', 'interval', '_is_first', '_ident', 'congestion', 'prep_func') |
| 2842 | |
| 2843 | def __init__(self, name: str, cmd: str, interval: int, congestion=True, |
| 2844 | prep_func: Callable[[], bool]=None): |
| 2845 | self.name = name |
| 2846 | self.cmd = cmd |
| 2847 | self.interval = interval |
| 2848 | self._is_first = True |
| 2849 | self._ident = 0 |
| 2850 | self.congestion = congestion # 拥塞检查，为True时，如果下次执行前上次执行未结束，则触发拥塞处理 |
| 2851 | self.prep_func = prep_func or (lambda: True) |
| 2852 | |
| 2853 | def first(self) -> int: |
| 2854 | if self._is_first: |
| 2855 | self._is_first = False |
| 2856 | return int(time.time()) |
| 2857 | else: |
| 2858 | return 0 |
| 2859 | |
| 2860 | def get_next_time(self) -> int: |
| 2861 | """ |
| 2862 | 获取下一次执行时间 |
| 2863 | """ |
| 2864 | return int(time.time() + self.interval) |
| 2865 | |
| 2866 | def get_ident(self) -> int: |
| 2867 | """ |
| 2868 | 获取任务标识 |
| 2869 | """ |
| 2870 | return self._ident |
| 2871 | |
| 2872 | def set_ident(self, ident: int): |
| 2873 | """ |
| 2874 | 设置任务标识 |
| 2875 | """ |
| 2876 | self._ident = ident |
| 2877 | |
| 2878 | def run(self): |
| 2879 | """ |
| 2880 | 运行任务 |
| 2881 | """ |
| 2882 | if not self.prep_func(): |
| 2883 | write_log("任务: {} 无需启动，已跳过".format(self.name)) |
| 2884 | return |
| 2885 | write_log("执行任务: {}".format(self.name)) |
| 2886 | try: |
| 2887 | # write_log("cmd:", self.cmd, _level='debug') |
| 2888 | os.system(self.cmd) |
| 2889 | except Exception as e: |
| 2890 | write_log("执行{}任务失败: {}".format(self.name, str(e)), traceback.format_exc(), _level='error', color='red') |
| 2891 | |
| 2892 | |
| 2893 | class _ScriptService: |
no outgoing calls
no test coverage detected