246 lines
9.7 KiB
Python
246 lines
9.7 KiB
Python
import time
|
||
import os
|
||
import re
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from datetime import datetime
|
||
from main import DeviceAutomation
|
||
import globals.apis as apis
|
||
|
||
## --- 配置区 ---
|
||
API_URL = "http://your-api-server.com/api/tasks"
|
||
MAX_WORKERS = 3
|
||
TIME_FILE_PATH = r"D:\uploadInfo\time.txt"
|
||
POLLING_INTERVAL = 600 # 10分钟轮询一次
|
||
|
||
# 文件操作锁，防止多线程同时写入文件造成冲突
|
||
file_lock = threading.Lock()
|
||
|
||
def update_file_status(username, from_status, to_status):
|
||
"""
|
||
安全地更新 time.txt 中该用户的状态
|
||
例如: 将 'true' 改为 'running', 或将 'running' 改为 'done'
|
||
"""
|
||
if not os.path.exists(TIME_FILE_PATH):
|
||
return False
|
||
|
||
success = False
|
||
with file_lock:
|
||
try:
|
||
with open(TIME_FILE_PATH, 'r', encoding='utf-8') as f:
|
||
lines = f.readlines()
|
||
|
||
# new_lines = []
|
||
# for line in lines:
|
||
# clean_line = line.strip()
|
||
# # 匹配逻辑：包含用户名 且 以 from_status 结尾
|
||
# if f" {username} " in line and clean_line.endswith(from_status):
|
||
# line = line.replace(from_status, to_status)
|
||
# success = True
|
||
# new_lines.append(line)
|
||
new_lines = []
|
||
for line in lines:
|
||
# 使用正则确保精准匹配用户名和结尾状态
|
||
# 匹配规则：行内包含该用户名，且该行以 from_status 结尾
|
||
if re.search(rf'\b{username}\b', line) and line.strip().endswith(from_status):
|
||
# 只替换行尾的那个状态词
|
||
line = re.sub(rf'{from_status}$', to_status, line.rstrip()) + '\n'
|
||
success = True
|
||
new_lines.append(line)
|
||
|
||
with open(TIME_FILE_PATH, 'w', encoding='utf-8') as f:
|
||
f.writelines(new_lines)
|
||
|
||
if success:
|
||
print(f"📝 [文件更新] 用户 {username}: {from_status} -> {to_status}")
|
||
return success
|
||
except Exception as e:
|
||
print(f"❌ 更新文件状态失败 ({username}): {e}")
|
||
return False
|
||
|
||
def parse_time_config():
|
||
"""
|
||
解析 time.txt，只获取状态为 true 的任务
|
||
"""
|
||
time_map = {}
|
||
if not os.path.exists(TIME_FILE_PATH):
|
||
print(f"⚠️ 文件不存在: {TIME_FILE_PATH}")
|
||
return time_map
|
||
|
||
try:
|
||
with file_lock:
|
||
with open(TIME_FILE_PATH, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
# 匹配：用户名 时间 true (仅获取待处理任务)
|
||
match = re.search(r'(\w+)\s+(\d{4}-\d{1,2}-\d{1,2}\s+\d{1,2}:\d{2}:\d{2})\s+true$', line)
|
||
if match:
|
||
username, scheduled_time = match.group(1), match.group(2)
|
||
time_map[username] = scheduled_time
|
||
except Exception as e:
|
||
print(f"❌ 解析 time.txt 失败: {e}")
|
||
return time_map
|
||
|
||
def normalize_datetime(time_str):
|
||
"""
|
||
将时间字符串格式化为标准格式：YYYY-MM-DD HH:MM:SS
|
||
补全单数字的月、日、时
|
||
例如：2024-1-15 9:52:20 -> 2024-01-15 09:52:20
|
||
"""
|
||
try:
|
||
# 分割日期和时间部分
|
||
if ' ' in time_str:
|
||
date_part, time_part = time_str.split(' ', 1)
|
||
|
||
# 补全日期部分的单数字
|
||
date_parts = date_part.split('-')
|
||
if len(date_parts) == 3:
|
||
year = date_parts[0]
|
||
month = date_parts[1].zfill(2) # 月补零
|
||
day = date_parts[2].zfill(2) # 日补零
|
||
date_part = f"{year}-{month}-{day}"
|
||
|
||
# 补全时间部分的单数字小时
|
||
time_parts = time_part.split(':')
|
||
if len(time_parts) >= 1:
|
||
hour = time_parts[0].zfill(2) # 小时补零
|
||
time_part = f"{hour}:{':'.join(time_parts[1:])}"
|
||
|
||
return f"{date_part} {time_part}"
|
||
return time_str
|
||
except Exception as e:
|
||
print(f"⚠️ 时间格式标准化失败 ({time_str}): {e}")
|
||
return time_str
|
||
|
||
def get_combined_tasks():
|
||
"""
|
||
# 结合接口(is_ok==1)和本地文件(true)筛选任务
|
||
结合接口(is_ok==1)和获取解析云端轨迹文件筛选任务
|
||
"""
|
||
try:
|
||
|
||
|
||
|
||
# 调用你的 API 接口获取账号信息
|
||
accounts = apis.get_accounts_from_server("68ef0e02b0138d25e2ac9918")
|
||
if not accounts:
|
||
return {}
|
||
|
||
task_list = {}
|
||
today = datetime.now().strftime("%Y%m%d")
|
||
|
||
for account in accounts:
|
||
if account.get('is_ok') == 1:
|
||
user = account.get('username')
|
||
ip = account.get('device_ip')
|
||
port = account.get('device_port')
|
||
project_name = account.get('project_name')
|
||
|
||
# 获取轨迹文件
|
||
save_path = apis.fetch_and_save_track_file(f"https://database.yuxindazhineng.com/team-bucket/69378c5b4f42d83d9504560d/track/{today}/{today}_1_{user}.txt")
|
||
if save_path:
|
||
|
||
# 1. 解析原始文件
|
||
raw_records = apis.parse_track_file(save_path)
|
||
|
||
# 2. 按线路ID去重（重复则保留后面的记录）
|
||
deduplicated_records = apis.deduplicate_by_line_id(raw_records)
|
||
|
||
# 3. 从去重后的记录中找最晚结束时间
|
||
latest_time = apis.get_latest_end_time(deduplicated_records)
|
||
|
||
# 4. 写入end.txt
|
||
# write_success = apis.write_to_file(latest_time)
|
||
write_success = apis.writer_file_status(user, latest_time, "true")
|
||
time.sleep(2)
|
||
|
||
|
||
# 获取状态为true的账号
|
||
local_times = parse_time_config()
|
||
if not local_times:
|
||
return {}
|
||
|
||
# 只有在写入成功且 time.txt 中是 true 的账号才会被加入
|
||
if write_success and user in local_times and ip and port:
|
||
address = f"{ip}:{port}"
|
||
# 确保时间是两位数格式
|
||
raw_time = local_times[user]
|
||
full_time = normalize_datetime(raw_time)
|
||
task_list[address] = {"time": full_time, "user": user, "project_name": project_name}
|
||
elif not write_success:
|
||
print(f"❌ 写入end.txt失败，跳过任务: {user}")
|
||
|
||
return task_list
|
||
except Exception as e:
|
||
print(f"❌ 获取任务异常: {e}")
|
||
return {}
|
||
|
||
def run_task(address, target_time, username, project_name):
|
||
"""
|
||
单个执行线程：锁定状态 -> 等待 -> 执行 -> 完成
|
||
"""
|
||
# 1. 尝试将状态从 true 改为 running (锁定任务)
|
||
# 如果此时文件状态已被其他逻辑修改，则放弃执行，防止重复
|
||
if not update_file_status(username, "true", "running"):
|
||
return f"⏭️ {username} 状态已变更，跳过执行。"
|
||
|
||
print(f"🚀 [任务锁定] 设备: {address} | 用户: {username} | 计划时间: {target_time} | 项目: {project_name}")
|
||
|
||
try:
|
||
# 2. 计算并执行等待逻辑
|
||
# target_dt = datetime.strptime(target_time, "%Y-%m-%d %H:%M:%S")
|
||
# wait_secs = (target_dt - datetime.now()).total_seconds()
|
||
|
||
# if wait_secs > 0:
|
||
# print(f"⏳ {username} 距离执行还有 {int(wait_secs)} 秒...")
|
||
# time.sleep(wait_secs)
|
||
|
||
# 3. 调用 main.py 中的自动化逻辑
|
||
print(f"▶️ [正在执行] {username} 开始自动化操作...")
|
||
automation = DeviceAutomation(address, project_name)
|
||
result = automation.handle_app_state()
|
||
|
||
# 4. 执行完成后，将状态从 running 改为 done
|
||
# update_file_status(username, "running", "done")
|
||
return f"✅ {username} 执行成功: {result}"
|
||
|
||
except Exception as e:
|
||
# 如果中间报错，将状态改为 error 方便排查
|
||
update_file_status(username, "running", "error")
|
||
return f"❌ {username} 执行异常: {str(e)}"
|
||
|
||
def monitor_center():
|
||
"""调度中心：每半小时检查一次"""
|
||
while True:
|
||
now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
print(f"\n{'='*20} 周期性检查开始 ({now_str}) {'='*20}")
|
||
|
||
tasks = get_combined_tasks()
|
||
|
||
if tasks:
|
||
print(f"📡 发现 {len(tasks)} 个符合条件且未跑过的任务，准备启动线程池...")
|
||
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
|
||
# 提交任务，将 address, time, username, project_name 传入
|
||
future_to_user = {
|
||
executor.submit(run_task, addr, info['time'], info['user'], info.get('project_name', '')): info['user']
|
||
for addr, info in tasks.items()
|
||
}
|
||
|
||
for f in as_completed(future_to_user):
|
||
print(f.result())
|
||
else:
|
||
print("📭 暂无待处理任务 (所有账号已完成或 is_ok != 1)")
|
||
|
||
print(f"💤 本轮轮询结束，等待 {POLLING_INTERVAL//60} 分钟进行下次检查...")
|
||
time.sleep(POLLING_INTERVAL)
|
||
|
||
if __name__ == "__main__":
|
||
print("🌟 自动化调度服务已启动")
|
||
print(f"📂 配置文件: {TIME_FILE_PATH}")
|
||
print(f"⏱️ 轮询频率: {POLLING_INTERVAL}秒")
|
||
try:
|
||
monitor_center()
|
||
except KeyboardInterrupt:
|
||
print("\n👋 用户手动停止程序。")
|
||
|