fz-mirrors/MaaAssistantArknights
0
0
Fork 0
mirror of https://github.com/MaaAssistantArknights/MaaAssistantArknights.git synced 2026-07-16 01:40:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore.整理python定时demo

Browse Source
This commit is contained in:
zhangweijian97
2022-02-19 00:40:06 +08:00
committed by MistEO
parent 3d0362c7b9
commit acbdbff9f4
3 changed files with 21 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/Python/timed_start/requirements.txt Normal file
View File

@@ -0,0 +1 @@
schedule

31
src/Python/定时启动MAA_v2.0.1_by_Black_Cat_Bon.py → src/Python/timed_start/timed_sample.py
View File

@@ -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)

123
src/Python/定时启动MAA_v2.0.0_by_Black_Cat_Bon.py
View File

@@ -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)