From acbdbff9f4f02a3bce3d21ee1a4e2698c7d34e29 Mon Sep 17 00:00:00 2001 From: zhangweijian97 Date: Sat, 19 Feb 2022 00:40:06 +0800 Subject: [PATCH] =?UTF-8?q?chore.=E6=95=B4=E7=90=86python=E5=AE=9A?= =?UTF-8?q?=E6=97=B6demo?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/Python/timed_start/requirements.txt | 1 + .../timed_sample.py} | 31 +++-- .../定时启动MAA_v2.0.0_by_Black_Cat_Bon.py | 123 ------------------ 3 files changed, 21 insertions(+), 134 deletions(-) create mode 100644 src/Python/timed_start/requirements.txt rename src/Python/{定时启动MAA_v2.0.1_by_Black_Cat_Bon.py => timed_start/timed_sample.py} (87%) delete mode 100644 src/Python/定时启动MAA_v2.0.0_by_Black_Cat_Bon.py diff --git a/src/Python/timed_start/requirements.txt b/src/Python/timed_start/requirements.txt new file mode 100644 index 0000000000..b397e36222 --- /dev/null +++ b/src/Python/timed_start/requirements.txt @@ -0,0 +1 @@ +schedule diff --git a/src/Python/定时启动MAA_v2.0.1_by_Black_Cat_Bon.py b/src/Python/timed_start/timed_sample.py similarity index 87% rename from src/Python/定时启动MAA_v2.0.1_by_Black_Cat_Bon.py rename to src/Python/timed_start/timed_sample.py index e2c69296dd..aae364e62c 100644 --- a/src/Python/定时启动MAA_v2.0.1_by_Black_Cat_Bon.py +++ b/src/Python/timed_start/timed_sample.py @@ -1,10 +1,14 @@ -from operator import ne +""" +function: 定时启动 maa +author: Black Cat Bon +version: v2.0.1 +""" + import threading import time import schedule import datetime -from threading import Timer import json import pathlib import sys @@ -12,9 +16,11 @@ import re from interface import Asst, Message + def get_now_time(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + def job(): # print("I'm running on thread %s" % threading.current_thread()) print("现在时间：" + datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) @@ -26,10 +32,11 @@ def job(): asst.append_visit() asst.append_mall(True) asst.append_award() - # asst.set_penguin_id('18688523') + # asst.set_penguin_id('1234567') asst.start() + def run_threaded(job_func): job_thread = threading.Thread(target=job_func) job_thread.start() @@ -41,10 +48,11 @@ def get_target_time(input_time): target_time = input_time else: while True: - input_str = input('Invalid time format, please input again!\n For example, 20:00 \n> ') + input_str = input( + 'Invalid time format, please input again!\n For example, 20:00 \n> ') if (input_str == 'exit()'): sys.exit() - + matchObj = re.match('\d\d:\d\d', input_str) if matchObj: @@ -56,16 +64,18 @@ def get_target_time(input_time): return target_time + def get_target_port(input_port): matchObj = re.match('\d\d\d\d', input_port) if matchObj: target_time = input_port else: while True: - input_str = input('Invalid port format, please input again!\n For example, 5565 \n> ') + input_str = input( + 'Invalid port format, please input again!\n For example, 5565 \n> ') if (input_str == 'exit()'): sys.exit() - + matchObj = re.match('\d\d\d\d', input_str) if matchObj: @@ -88,11 +98,11 @@ if __name__ == "__main__": elif len(sys.argv) == 2: target_time = get_target_time(sys.argv[1]) port = '5555' - elif len(sys.argv) == 3: + elif len(sys.argv) == 3: target_time = get_target_time(sys.argv[1]) port = get_target_port(sys.argv[2]) - # 声明asst + # asst 的回调 @Asst.CallBackType def my_callback(msg, details, arg): m = Message(msg) @@ -100,7 +110,6 @@ if __name__ == "__main__": print(m, d, arg) - dirname: str = (pathlib.Path.cwd()).__str__() asst = Asst(dirname=dirname, callback=my_callback) @@ -118,6 +127,6 @@ if __name__ == "__main__": # print('下次定时任务开始于：' + target_time) - while 1: + while True: schedule.run_pending() time.sleep(1) diff --git a/src/Python/定时启动MAA_v2.0.0_by_Black_Cat_Bon.py b/src/Python/定时启动MAA_v2.0.0_by_Black_Cat_Bon.py deleted file mode 100644 index 3f6d81ee17..0000000000 --- a/src/Python/定时启动MAA_v2.0.0_by_Black_Cat_Bon.py +++ /dev/null @@ -1,123 +0,0 @@ -from operator import ne -import threading -import time -import schedule - -import datetime -from threading import Timer -import json -import pathlib -import sys -import re - -from interface import Asst, Message - -def get_now_time(): - return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - -def job(): - # print("I'm running on thread %s" % threading.current_thread()) - print("现在时间：" + datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) - - # asst.append_fight("LastBattle", 0, 0, 999) - # asst.append_recruit(3, [4, 5], [3, 4, 5], True, False) - # asst.append_infrast(1, ["Mfg", "Trade", "Control", - # "Power", "Reception", "Office", "Dorm"], "Money", 0.1) - # asst.append_visit() - # asst.append_mall(True) - asst.append_award() - asst.set_penguin_id('18688523') - - asst.start() - -def run_threaded(job_func): - job_thread = threading.Thread(target=job_func) - job_thread.start() - - -def get_target_time(input_time): - matchObj = re.match('\d\d:\d\d', input_time) - if matchObj: - target_time = input_time - else: - while True: - input_str = input('Invalid time format, please input again!\n For example, 20:00 \n> ') - if (input_str == 'exit()'): - sys.exit() - - matchObj = re.match('\d\d:\d\d', input_str) - - if matchObj: - break - else: - print('Please input a valid time!') - - target_time = input_str - - return target_time - -def get_target_port(input_port): - matchObj = re.match('\d\d\d\d', input_port) - if matchObj: - target_time = input_port - else: - while True: - input_str = input('Invalid port format, please input again!\n For example, 5565 \n> ') - if (input_str == 'exit()'): - sys.exit() - - matchObj = re.match('\d\d\d\d', input_str) - - if matchObj: - break - else: - print('Please input a valid time!') - - target_time = input_str - - return target_time - - -if __name__ == "__main__": - global target_time - global asst - - if (len(sys.argv) == 1): - target_time = '20:00' - port = '5555' - elif len(sys.argv) == 2: - target_time = get_target_time(sys.argv[1]) - port = '5555' - elif len(sys.argv) == 3: - target_time = get_target_time(sys.argv[1]) - port = get_target_port(sys.argv[2]) - - # 声明asst - @Asst.CallBackType - def my_callback(msg, details, arg): - m = Message(msg) - d = json.loads(details.decode('utf-8')) - - print(m, d, arg) - - - dirname: str = (pathlib.Path.cwd()).__str__() - asst = Asst(dirname=dirname, callback=my_callback) - - print('version', asst.get_version()) - - if asst.catch_custom('127.0.0.1:' + port): - print('连接成功') - else: - print('连接失败') - exit() - - schedule.every().day.at(target_time).do(run_threaded, job) - # schedule.every(30).seconds.do(run_threaded, job) - print("定时任务已设置，将于每天 " + target_time + " 运行") - - # print('下次定时任务开始于：' + target_time) - - while 1: - schedule.run_pending() - time.sleep(1)