Todesk自动安装上线系统!

cls822001 · 发表于昨天 06:10

直接上代码自行修改编译安装代码install_todesk.py

import os
import sys
import subprocess
import shutil
import requests
import time
import winreg
import psutil
from pathlib import Path
# 配置变量
INSTALL_DIR = os.path.join(os.environ['ProgramFiles'], 'Todesk')
DOWNLOAD_URL = 'http://down.jywangluo.cn:88/ToDesk/Todesk.7z'
TOOLS_URL = 'http://down.jywangluo.cn:88/Tools/todesk.7z'
# 日志文件路径
LOG_FILE = os.path.join(os.environ['TEMP'], 'todesk_install.log')
# 确保中文显示正常
if sys.stdout is not None:
try:
sys.stdout.reconfigure(encoding='utf-8')
except AttributeError:
# 对于不支持reconfigure的Python版本或环境，忽略此错误
pass
def log(message):
log_time = time.strftime('%Y-%m-%d %H:%M:%S')
log_message = f"[{log_time}] {message}"
# 输出到控制台
print(log_message)
# 写入到日志文件
try:
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write(log_message + '\n')
f.flush()
except Exception as e:
print(f"无法写入日志文件: {e}")
def stop_todesk_processes():
try:
log("开始停止ToDesk相关进程...")
# 获取当前进程PID，避免终止自身
current_pid = os.getpid()
# 使用psutil查找并终止ToDesk相关进程
processes_stopped = 0
for proc in psutil.process_iter(['pid', 'name']):
try:
# 跳过当前进程
if proc.info['pid'] == current_pid:
continue
proc_name = proc.info['name'].lower()
# 只终止真正的ToDesk进程，排除安装程序自身
if 'todesk' in proc_name and 'install' not in proc_name:
log(f"发现ToDesk进程: {proc.info['name']} (PID: {proc.info['pid']})")
proc.terminate()
proc.wait(timeout=5) # 等待进程终止
log(f"已终止进程: {proc.info['name']} (PID: {proc.info['pid']})")
processes_stopped += 1
except (psutil.NoSuchProcess, psutil.AccessDenied):
# 进程已终止或无权限访问，继续处理其他进程
continue
except psutil.TimeoutExpired:
# 进程未在指定时间内终止，强制杀死
log(f"进程 {proc.info['name']} 未正常终止，强制杀死")
proc.kill()
processes_stopped += 1
# 停止ToDesk服务
try:
subprocess.run(['sc', 'stop', 'ToDesk_Service'], shell=True, capture_output=True, text=True)
log("已停止ToDesk服务")
except Exception as e:
log(f"停止ToDesk服务时出错: {e}")
# 使用taskkill作为备用方案
try:
result = subprocess.run(['taskkill', '/f', '/im', 'ToDesk.exe', '/T'],
shell=True, capture_output=True, text=True)
if result.returncode == 0:
log("taskkill命令成功执行")
except Exception as e:
log(f"taskkill执行出错: {e}")
log(f"停止进程完成，共处理了 {processes_stopped} 个ToDesk相关进程")
except Exception as e:
log(f"停止进程时出错: {e}")
def get_todesk_client_id():
log("获取ToDesk机器码...")
max_attempts = 5
attempt = 0
while attempt < max_attempts:
attempt += 1
try:
config_path = os.path.join(INSTALL_DIR, 'config.ini')
if os.path.exists(config_path):
with open(config_path, 'r', encoding='utf-8') as f:
content = f.read()
for line in content.splitlines():
if line.startswith('clientId='):
client_id = line.split('=', 1)[1].strip()
if client_id:
log(f"成功获取机器码: {client_id}")
return client_id
log(f"尝试 {attempt}/{max_attempts}: 未在配置文件中找到有效的clientId")
else:
log(f"尝试 {attempt}/{max_attempts}: 配置文件不存在: {config_path}")
except Exception as e:
log(f"尝试 {attempt}/{max_attempts}: 读取机器码时出错: {e}")
if attempt < max_attempts:
log(f"等待3秒后重试...")
time.sleep(3)
log("达到最大尝试次数，未能获取机器码")
return None
def get_server_url():
"""从配置文件读取服务器IP/域名，然后拼接成完整URL"""
# 配置文件路径 - 兼容开发和编译后的环境
if getattr(sys, 'frozen', False):
# 编译后的可执行文件环境
base_dir = os.path.dirname(os.path.abspath(sys.executable))
else:
# 开发环境
base_dir = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(base_dir, 'config.ini')
default_server = 'localhost:7888' # 默认服务器地址和端口
api_path = '/api/register' # API路径部分
# 如果配置文件不存在，创建默认配置文件
if not os.path.exists(config_path):
log(f"配置文件不存在，创建默认配置文件: {config_path}")
try:
with open(config_path, 'w', encoding='utf-8') as f:
f.write('[ServerConfig]\n')
f.write(f'server_host={default_server}\n')
log(f"默认配置文件创建成功")
server_host = default_server
except Exception as e:
log(f"创建配置文件失败: {e}")
server_host = default_server
else:
# 读取配置文件
try:
server_host = None
with open(config_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line.startswith('server_host='):
server_host = line.split('=', 1)[1].strip()
break
# 向后兼容：如果配置文件仍使用旧的server_url格式
elif line.startswith('server_url='):
server_url = line.split('=', 1)[1].strip()
# 从完整URL中提取host和port部分
import re
match = re.match(r'https?://([^/]+)', server_url)
if match:
server_host = match.group(1)
log(f"检测到旧格式配置，提取服务器地址: {server_host}")
break
if not server_host:
log("配置文件中未找到server_host，使用默认值")
server_host = default_server
except Exception as e:
log(f"读取配置文件失败: {e}")
server_host = default_server
# 构建完整URL
# 确保URL格式正确（添加http://前缀如果没有）
if not server_host.startswith(('http://', 'https://')):
server_url = f'http://{server_host}{api_path}'
else:
server_url = f'{server_host}{api_path}'
log(f"服务器地址: {server_host}")
log(f"完整服务器URL: {server_url}")
return server_url
def upload_client_id(client_id):
log(f"开始上传机器码到服务器: {client_id}")
max_retries = 3
retry_count = 0
while retry_count < max_retries:
retry_count += 1
try:
# 从配置文件读取服务器URL
SERVER_URL = get_server_url()
# 上传数据
data = {'client_id': client_id}
# 发送请求
log(f"尝试 {retry_count}/{max_retries}: 发送请求到 {SERVER_URL}")
response = requests.post(
SERVER_URL,
json=data,
timeout=30
)
# 检查响应
if response.status_code == 200:
try:
result = response.json()
if result.get('status') == 'success':
log("机器码上传成功！")
device_info = result.get('device_info', {})
log(f"设备ID: {device_info.get('device_id')}")
log(f"首次上线: {device_info.get('first_online')}")
log(f"最后上线: {device_info.get('last_online')}")
return True
else:
log(f"服务器返回错误: {result.get('message')}")
except ValueError:
log(f"无法解析服务器响应: {response.text}")
else:
log(f"上传失败，HTTP状态码: {response.status_code}")
log(f"响应内容: {response.text}")
except requests.exceptions.ConnectionError:
log("连接服务器失败，请检查服务器是否运行")
except requests.exceptions.Timeout:
log("连接服务器超时")
except Exception as e:
log(f"上传机器码时出错: {e}")
if retry_count < max_retries:
log(f"等待5秒后重试...")
time.sleep(5)
log("达到最大重试次数，上传失败")
return False
def download_file(url, save_path):
"""下载文件"""
log(f"下载文件: {url} -> {save_path}")
try:
response = requests.get(url, stream=True)
response.raise_for_status()
with open(save_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
log(f"文件下载成功: {save_path}")
return True
except Exception as e:
log(f"下载文件时出错: {e}")
return False
def extract_7z(archive_path, extract_dir):
"""使用py7zr库解压7z文件"""
log(f"解压文件: {archive_path} -> {extract_dir}")
# 确保解压目录存在
os.makedirs(extract_dir, exist_ok=True)
try:
# 优先使用py7zr库解压
try:
import py7zr
log("使用py7zr库解压...")
with py7zr.SevenZipFile(archive_path, mode='r') as z:
z.extractall(extract_dir)
# 验证解压结果
if os.listdir(extract_dir):
log("文件解压成功")
return True
else:
log("解压目录为空，解压可能未完全成功")
return False
except ImportError:
log("py7zr库未安装，请先安装: pip install py7zr --only-binary :all:")
return False
except Exception as e:
log(f"使用py7zr解压时出错: {e}")
return False
except Exception as e:
log(f"解压过程发生未预期错误: {e}")
return False
def get_file_version(file_path):
"""获取Windows可执行文件的版本信息"""
try:
import ctypes
from ctypes import wintypes
# 定义所需的结构体
class VS_FIXEDFILEINFO(ctypes.Structure):
_fields_ = [
('dwSignature', wintypes.DWORD),
('dwStrucVersion', wintypes.DWORD),
('dwFileVersionMS', wintypes.DWORD),
('dwFileVersionLS', wintypes.DWORD),
('dwProductVersionMS', wintypes.DWORD),
('dwProductVersionLS', wintypes.DWORD),
('dwFileFlagsMask', wintypes.DWORD),
('dwFileFlags', wintypes.DWORD),
('dwFileOS', wintypes.DWORD),
('dwFileType', wintypes.DWORD),
('dwFileSubtype', wintypes.DWORD),
('dwFileDateMS', wintypes.DWORD),
('dwFileDateLS', wintypes.DWORD),
]
# 获取文件版本信息
size = ctypes.windll.version.GetFileVersionInfoSizeW(file_path, None)
if size == 0:
log(f"无法获取文件版本信息: {file_path}")
return "1.0.0.0"
buffer = ctypes.create_string_buffer(size)
if not ctypes.windll.version.GetFileVersionInfoW(file_path, None, size, buffer):
log(f"获取文件版本信息失败: {file_path}")
return "1.0.0.0"
# 解析版本信息
dwLen = wintypes.UINT()
lpData = ctypes.c_void_p()
if not ctypes.windll.version.VerQueryValueW(buffer, '\\', ctypes.byref(lpData), ctypes.byref(dwLen)):
log(f"解析版本信息失败: {file_path}")
return "1.0.0.0"
vsinfo = VS_FIXEDFILEINFO.from_address(lpData.value)
# 提取版本号
major = (vsinfo.dwFileVersionMS >> 16) & 0xFFFF
minor = vsinfo.dwFileVersionMS & 0xFFFF
build = (vsinfo.dwFileVersionLS >> 16) & 0xFFFF
revision = vsinfo.dwFileVersionLS & 0xFFFF
return f"{major}.{minor}.{build}.{revision}"
except Exception as e:
log(f"获取文件版本时出错: {e}")
return "1.0.0.0"
def configure_todesk():
"""配置ToDesk"""
log("配置ToDesk...")
try:
# 下载配置文件（注意：虽然URL是.7z，但文件内容本身是INI格式）
temp_config = os.path.join(os.environ['TEMP'], 'config.ini')
# 初始化auth_pass_ex为空
auth_pass_ex = ''
if download_file(TOOLS_URL, temp_config):
# 直接读取下载的文件作为INI（无需解压）
try:
with open(temp_config, 'r', encoding='utf-8') as f:
in_config_info = False
for line in f:
line = line.strip()
# 检查是否进入ConfigInfo小节
if line == '[ConfigInfo]':
in_config_info = True
continue
# 检查是否离开ConfigInfo小节（进入其他小节）
if line.startswith('[') and line.endswith(']') and line != '[ConfigInfo]':
in_config_info = False
continue
# 只在ConfigInfo小节内查找authPassEx
if in_config_info and line.startswith('authPassEx='):
auth_pass_ex = line.split('=', 1)[1].strip()
log(f"从配置文件ConfigInfo小节读取到authPassEx: {auth_pass_ex}")
break
if not auth_pass_ex:
log("配置文件ConfigInfo小节中未找到authPassEx项")
except Exception as e:
log(f"读取临时配置文件失败: {str(e)}")
# 获取ToDesk.exe的版本信息
todesk_exe_path = os.path.join(INSTALL_DIR, 'ToDesk.exe')
version = "1.0.0.0" # 默认版本号
if os.path.exists(todesk_exe_path):
version = get_file_version(todesk_exe_path)
log(f"获取到ToDesk.exe版本信息: {version}")
else:
log("ToDesk.exe不存在，使用默认版本号")
# 创建配置文件
config_path = os.path.join(INSTALL_DIR, 'config.ini')
with open(config_path, 'w', encoding='utf-8') as f:
f.write('[ConfigInfo]\n')
if auth_pass_ex:
f.write(f"authPassEx={auth_pass_ex}\n")
log("已将从下载的配置文件中读取的authPassEx写入配置")
else:
f.write(f"authPassEx=default_auth_key\n")
log("未获取到有效的authPassEx，使用默认值")
f.write("autoStart=1\n")
f.write("autoupdate=1\n")
f.write("AuthMode=1\n")
f.write(f"Version={version}\n")
f.write("autoLockScreen=0\n")
f.write("filetranstip=0\n")
f.write("PrivateScreenLockScreen=0\n")
f.write("DisableHardCode=1\n")
f.write("ResolutionConfig=0\n")
f.write("isupdate=0\n")
# 清理临时文件
if os.path.exists(temp_config):
os.remove(temp_config)
except Exception as e:
log(f"配置ToDesk时出错: {e}")
def write_registry():
"""写入注册表信息"""
log("写入注册表信息...")
try:
# 创建或打开注册表项
key = winreg.CreateKeyEx(winreg.HKEY_LOCAL_MACHINE, r'SOFTWARE\ToDesk',
0, winreg.KEY_WRITE | winreg.KEY_WOW64_64KEY)
# 写入安装路径
winreg.SetValueEx(key, 'InstPath', 0, winreg.REG_SZ, INSTALL_DIR)
winreg.CloseKey(key)
# 创建卸载信息
uninstall_key = winreg.CreateKeyEx(winreg.HKEY_LOCAL_MACHINE,
r'SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\ToDesk',
0, winreg.KEY_WRITE | winreg.KEY_WOW64_64KEY)
winreg.SetValueEx(uninstall_key, 'DisplayName', 0, winreg.REG_SZ, 'ToDesk')
winreg.SetValueEx(uninstall_key, 'UninstallString', 0, winreg.REG_SZ,
os.path.join(INSTALL_DIR, 'uninst.exe'))
winreg.SetValueEx(uninstall_key, 'DisplayIcon', 0, winreg.REG_SZ,
os.path.join(INSTALL_DIR, 'ToDesk.exe'))
winreg.SetValueEx(uninstall_key, 'Publisher', 0, winreg.REG_SZ,
'Hainan YouQu Technology Co., Ltd')
winreg.SetValueEx(uninstall_key, 'DisplayVersion', 0, winreg.REG_SZ, '1.0.0.0')
winreg.SetValueEx(uninstall_key, 'todeskpath', 0, winreg.REG_SZ, INSTALL_DIR)
winreg.SetValueEx(uninstall_key, 'developer', 0, winreg.REG_SZ, 'Wanli@todesk.com')
winreg.SetValueEx(uninstall_key, 'website', 0, winreg.REG_SZ, 'https://www.todesk.com')
winreg.CloseKey(uninstall_key)
except Exception as e:
log(f"写入注册表时出错: {e}")
def create_shortcut():
"""在公共桌面目录创建快捷方式"""
log("在公共桌面目录创建快捷方式...")
try:
# 为所有用户创建快捷方式
all_users_desktop = os.path.join(os.environ['PUBLIC'], 'Desktop')
all_users_shortcut = os.path.join(all_users_desktop, 'Todesk.lnk')
if os.path.exists(all_users_shortcut):
os.remove(all_users_shortcut)
# 使用WScript.Shell创建快捷方式
import win32com.client
shell = win32com.client.Dispatch("WScript.Shell")
all_users_sc = shell.CreateShortCut(all_users_shortcut)
all_users_sc.Targetpath = os.path.join(INSTALL_DIR, "ToDesk.exe")
all_users_sc.WorkingDirectory = INSTALL_DIR
all_users_sc.IconLocation = os.path.join(INSTALL_DIR, "ToDesk.exe")
all_users_sc.save()
except ImportError:
log("未安装pywin32，跳过创建快捷方式")
except Exception as e:
log(f"创建快捷方式时出错: {e}")
def delete_privatedata_registry():
# 定义注册表路径、键名和访问权限
REG_PATH = r'SOFTWARE\ToDesk'
VALUE_NAME = 'PrivateData'
KEY_64BIT = winreg.KEY_WOW64_64KEY
WRITE_ACCESS = winreg.KEY_SET_VALUE | KEY_64BIT
# 检查并删除PrivateData值
try:
with winreg.ConnectRegistry(None, winreg.HKEY_LOCAL_MACHINE) as hklm:
try:
# 尝试删除值
with winreg.OpenKey(hklm, REG_PATH, 0, WRITE_ACCESS) as write_key:
winreg.DeleteValue(write_key, VALUE_NAME)
return True
except FileNotFoundError:
# 值不存在，视为删除成功
return True
except Exception:
return False
def start_todesk():
"""启动ToDesk"""
log("启动ToDesk...")
# 在启动ToDesk前删除PrivateData注册表项
delete_privatedata_registry()
try:
todesk_exe = os.path.join(INSTALL_DIR, 'ToDesk.exe')
if os.path.exists(todesk_exe):
# 使用管理员权限启动ToDesk
import ctypes
result = ctypes.windll.shell32.ShellExecuteW(
None, "runas", todesk_exe, "", INSTALL_DIR, 1
)
if result > 32:
log("ToDesk启动成功，正在等待配置文件生成...")
# 等待一段时间确保ToDesk完全启动
time.sleep(8)
return True
else:
log(f"ToDesk启动失败，返回代码: {result}")
# 尝试普通方式启动
log("尝试以普通方式启动ToDesk...")
process = subprocess.Popen([todesk_exe], cwd=INSTALL_DIR)
log("ToDesk启动成功，正在等待配置文件生成...")
time.sleep(8)
return True
else:
log(f"ToDesk可执行文件不存在: {todesk_exe}")
except Exception as e:
log(f"启动ToDesk时出错: {e}")
return False
def main():
"""主函数"""
# 清除旧的日志文件
if os.path.exists(LOG_FILE):
try:
os.remove(LOG_FILE)
except Exception as e:
print(f"无法删除旧日志文件: {e}")
log("开始ToDesk自动安装...")
log(f"日志文件已创建在: {LOG_FILE}")
# 导入ctypes模块用于管理员权限检查
import ctypes
# 以管理员权限运行检查
if not ctypes.windll.shell32.IsUserAnAdmin():
log("请以管理员权限运行此脚本")
# 尝试以管理员权限重启
ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, " ".join(sys.argv), None, 1)
sys.exit(1)
# 停止现有进程
stop_todesk_processes()
# 跳过备份配置步骤
# 创建安装目录
if os.path.exists(INSTALL_DIR):
log(f"删除现有安装目录: {INSTALL_DIR}")
shutil.rmtree(INSTALL_DIR, ignore_errors=True)
os.makedirs(INSTALL_DIR, exist_ok=True)
# 下载安装包
temp_7z = os.path.join(os.environ['TEMP'], 'Todesk.7z')
if download_file(DOWNLOAD_URL, temp_7z):
# 解压安装包
if extract_7z(temp_7z, INSTALL_DIR):
# 清理下载的安装包
if os.path.exists(temp_7z):
os.remove(temp_7z)
# 配置ToDesk
configure_todesk()
# 写入注册表
write_registry()
# 创建快捷方式
create_shortcut()
# 启动ToDesk并等待配置文件生成
todesk_started = start_todesk()
# 如果ToDesk启动失败，尝试手动启动
if not todesk_started:
log("再次尝试启动ToDesk服务...")
try:
subprocess.run(['sc', 'start', 'ToDesk_Service'], shell=True, capture_output=True, text=True)
log("ToDesk服务启动中，等待配置文件生成...")
time.sleep(10)
except Exception as e:
log(f"启动ToDesk服务失败: {e}")
# 获取机器码
client_id = get_todesk_client_id()
if client_id:
# 尝试上传机器码到服务器
upload_client_id(client_id)
else:
log("未能获取有效的机器码")
log("ToDesk自动安装完成！")
else:
log("安装失败：解压文件失败")
else:
log("安装失败：下载文件失败")
if __name__ == "__main__":
# 导入ctypes模块用于管理员权限检查
import ctypes
main()

复制代码

服务端代码
server.py

from flask import Flask, request, jsonify
import datetime
import json
import os
import sys
import uuid
import tkinter as tk
from tkinter import ttk
import threading
import time
import tempfile
app = Flask(__name__)
# 存储数据的文件路径
DATA_FILE = 'todesk_devices.json'
# 全局变量用于UI刷新通知
gui_refresh_event = None
# 确保数据文件存在
if not os.path.exists(DATA_FILE):
with open(DATA_FILE, 'w', encoding='utf-8') as f:
json.dump({}, f, ensure_ascii=False, indent=2)
def load_devices():
"""从文件加载设备数据"""
try:
with open(DATA_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"加载设备数据失败: {e}")
return {}
def save_devices(devices):
"""保存设备数据到文件"""
try:
with open(DATA_FILE, 'w', encoding='utf-8') as f:
json.dump(devices, f, ensure_ascii=False, indent=2)
return True
except Exception as e:
print(f"保存设备数据失败: {e}")
return False
@app.route('/api/register', methods=['POST'])
def register_device():
"""注册设备，接收机器码并记录上线时间"""
try:
data = request.json
if not data or 'client_id' not in data:
return jsonify({'status': 'error', 'message': '缺少client_id参数'}), 400
client_id = data['client_id']
devices = load_devices()
# 获取当前时间
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 如果设备已存在，更新上线时间；否则创建新记录
if client_id in devices:
devices[client_id]['last_online'] = now
print(f"设备 {client_id} 再次上线，时间: {now}")
else:
# 生成精简的设备ID（使用时间戳+随机数）
import random
timestamp = int(time.time() * 1000)
random_num = random.randint(1000, 9999)
device_id = f"{timestamp}{random_num}"
devices[client_id] = {
'device_id': device_id,
'client_id': client_id,
'first_online': now,
'last_online': now,
'ip_address': request.remote_addr
}
print(f"新设备注册: {client_id}, 时间: {now}, IP: {request.remote_addr}")
# 有新设备注册时触发UI刷新
if gui_refresh_event:
gui_refresh_event.set()
# 保存数据
if save_devices(devices):
return jsonify({
'status': 'success',
'message': '设备注册成功',
'device_info': devices[client_id]
}), 200
else:
return jsonify({'status': 'error', 'message': '保存数据失败'}), 500
except Exception as e:
print(f"处理注册请求时出错: {e}")
return jsonify({'status': 'error', 'message': f'服务器内部错误: {str(e)}'}), 500
@app.route('/api/device/<client_id>', methods=['DELETE'])
def delete_device(client_id):
"""删除设备记录"""
try:
devices = load_devices()
if client_id in devices:
del devices[client_id]
if save_devices(devices):
print(f"设备 {client_id} 已删除")
return jsonify({'status': 'success', 'message': '设备已删除'}), 200
else:
return jsonify({'status': 'error', 'message': '保存数据失败'}), 500
else:
return jsonify({'status': 'error', 'message': '设备不存在'}), 404
except Exception as e:
print(f"删除设备时出错: {e}")
return jsonify({'status': 'error', 'message': f'服务器内部错误: {str(e)}'}), 500
@app.route('/', methods=['GET'])
def index():
"""首页"""
return jsonify({
'message': 'ToDesk 自动上线系统服务器',
'version': '1.0.0',
'endpoints': {
'register': '/api/register (POST)',
'devices': '/api/devices (GET)',
'device': '/api/device/<client_id> (GET, DELETE)'
}
})
class DeviceGUI:
def __init__(self, root):
self.root = root
self.root.title("ToDesk设备管理系统")
self.root.geometry("800x600")
# 获取屏幕尺寸
screen_width = root.winfo_screenwidth()
screen_height = root.winfo_screenheight()
# 计算窗口居中位置，并向上偏移80像素
x = (screen_width - 800) // 2 # 800是窗口宽度
y = (screen_height - 600) // 2 - 80 # 600是窗口高度，向上偏移80像素
# 确保Y坐标不为负数
if y < 0:
y = 0
# 设置窗口位置
self.root.geometry(f"800x600+{x}+{y}")
# 创建事件用于UI刷新通知
self.refresh_event = threading.Event()
global gui_refresh_event
gui_refresh_event = self.refresh_event
# 启动事件监听线程
self.start_event_listener()
# 创建标题
title_label = tk.Label(root, text="ToDesk设备管理系统", font=("微软雅黑", 16, "bold"))
title_label.pack(pady=10)
# 创建操作按钮框架
button_frame = tk.Frame(root)
button_frame.pack(pady=5)
# 创建刷新按钮
refresh_button = tk.Button(button_frame, text="刷新设备列表", command=self.refresh_devices)
refresh_button.pack(side=tk.LEFT, padx=10)
# 创建清理按钮
clean_button = tk.Button(button_frame, text="清理离线设备", command=self.clean_offline_devices, bg="#ffcccc")
clean_button.pack(side=tk.LEFT, padx=10)
# 创建统计信息框架
stats_frame = tk.Frame(root)
stats_frame.pack(fill=tk.X, padx=10, pady=5)
self.total_label = tk.Label(stats_frame, text="总设备数: 0", font=("微软雅黑", 10))
self.total_label.pack(side=tk.LEFT, padx=10)
self.online_label = tk.Label(stats_frame, text="在线设备数: 0", font=("微软雅黑", 10))
self.online_label.pack(side=tk.LEFT, padx=10)
# 创建设备列表表格
columns = ("device_id", "client_id", "first_online", "last_online", "ip_address", "status")
self.tree = ttk.Treeview(root, columns=columns, show="headings")
# 设置列标题和宽度
self.tree.heading("device_id", text="设备ID")
self.tree.heading("client_id", text="机器码")
self.tree.heading("first_online", text="首次上线时间")
self.tree.heading("last_online", text="最后上线时间")
self.tree.heading("ip_address", text="IP地址")
self.tree.heading("status", text="状态")
self.tree.column("device_id", width=100)
self.tree.column("client_id", width=150)
self.tree.column("first_online", width=120)
self.tree.column("last_online", width=120)
self.tree.column("ip_address", width=100)
self.tree.column("status", width=80)
# 添加滚动条
scrollbar = ttk.Scrollbar(root, orient=tk.VERTICAL, command=self.tree.yview)
self.tree.configure(yscroll=scrollbar.set)
# 放置表格和滚动条
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.tree.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
# 创建日志文本框
log_label = tk.Label(root, text="系统日志:", font=("微软雅黑", 10, "bold"))
log_label.pack(anchor=tk.W, padx=10, pady=5)
self.log_text = tk.Text(root, height=10, wrap=tk.WORD)
self.log_text.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
self.log_text.config(state=tk.DISABLED)
# 初始加载设备
self.refresh_devices()
# 定期刷新设备列表（每30秒）
self.root.after(30000, self.auto_refresh)
def start_event_listener(self):
"""启动事件监听线程"""
def event_listener():
while True:
# 等待刷新事件
self.refresh_event.wait()
# 清除事件状态
self.refresh_event.clear()
# 在UI线程中执行刷新
self.root.after(0, self.refresh_devices)
self.log_message("检测到新设备注册，自动刷新设备列表")
# 启动后台线程
listener_thread = threading.Thread(target=event_listener, daemon=True)
listener_thread.start()
def log_message(self, message):
"""添加日志消息到文本框"""
self.log_text.config(state=tk.NORMAL)
timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
self.log_text.insert(tk.END, f"[{timestamp}] {message}\n")
self.log_text.see(tk.END)
self.log_text.config(state=tk.DISABLED)
def is_device_online(self, last_online_str):
"""判断设备是否在线（2小时内有活动）"""
try:
last_online = datetime.datetime.strptime(last_online_str, '%Y-%m-%d %H:%M:%S')
now = datetime.datetime.now()
# 计算时间差
delta = now - last_online
# 如果时间差小于2小时，则认为在线
return delta.total_seconds() < 7200 # 7200秒 = 2小时
except Exception as e:
self.log_message(f"检查设备在线状态时出错: {e}")
return False
def refresh_devices(self):
"""刷新设备列表"""
try:
# 清空现有数据
for item in self.tree.get_children():
self.tree.delete(item)
# 加载设备数据
devices = load_devices()
total_count = len(devices)
online_count = 0
# 更新统计信息
self.total_label.config(text=f"总设备数: {total_count}")
# 添加设备到表格
for client_id, device_info in devices.items():
is_online = self.is_device_online(device_info['last_online'])
if is_online:
online_count += 1
status_text = "在线"
status_color = "green"
else:
status_text = "离线"
status_color = "red"
# 添加行数据
self.tree.insert("", tk.END, values=(
device_info['device_id'],
client_id,
device_info['first_online'],
device_info['last_online'],
device_info['ip_address'],
status_text
))
# 设置状态单元格颜色
item_id = self.tree.get_children()[-1]
self.tree.tag_configure(status_color, foreground=status_color)
self.tree.item(item_id, tags=(status_color,))
# 更新在线设备数量
self.online_label.config(text=f"在线设备数: {online_count}")
self.log_message("设备列表已刷新")
except Exception as e:
self.log_message(f"刷新设备列表时出错: {e}")
def auto_refresh(self):
"""自动刷新设备列表"""
self.refresh_devices()
# 继续定时刷新
self.root.after(30000, self.auto_refresh)
def clean_offline_devices(self):
"""手动清理离线设备"""
try:
self.log_message("开始清理2小时未上线的设备...")
deleted_count, deleted_devices = clean_inactive_devices()
if deleted_count > 0:
self.log_message(f"成功清理 {deleted_count} 个离线设备")
# 刷新设备列表以显示最新状态
self.refresh_devices()
else:
self.log_message("没有需要清理的离线设备")
except Exception as e:
self.log_message(f"清理离线设备时出错: {e}")
def run_server():
"""在单独线程中运行Flask服务器"""
print("ToDesk自动上线系统服务器启动中...")
print(f"数据将保存到: {os.path.abspath(DATA_FILE)}")
print("API接口:")
print(" POST /api/register - 注册设备，提交client_id")
print(" GET /api/devices - 获取所有设备列表")
print(" GET /api/device/<client_id> - 获取特定设备信息")
print(" DELETE /api/device/<client_id> - 删除设备记录")
print(" POST /api/clean - 清理离线设备")
print("\n服务器启动在 http://0.0.0.0:7888")
# 在生产环境中，应该使用更安全的方式启动Flask应用
# 这里为了方便测试，不使用debug模式
app.run(host='0.0.0.0', port=7888, debug=False, use_reloader=False)
# 定时清理任务
def scheduled_cleanup():
"""定时执行清理任务的线程函数"""
while True:
print("执行定时清理任务...")
deleted_count, _ = clean_inactive_devices()
if deleted_count > 0:
print(f"定时清理完成，共清理 {deleted_count} 个离线设备")
# 每小时执行一次清理
time.sleep(3600) # 3600秒 = 1小时
# 添加设备在线状态检查的帮助函数，供API使用
def get_device_status(device_info):
"""获取设备状态"""
try:
last_online = datetime.datetime.strptime(device_info['last_online'], '%Y-%m-%d %H:%M:%S')
now = datetime.datetime.now()
delta = now - last_online
return 'online' if delta.total_seconds() < 7200 else 'offline'
except:
return 'unknown'
def clean_inactive_devices():
"""清理超过2小时未上线的设备数据"""
try:
devices = load_devices()
now = datetime.datetime.now()
devices_to_delete = []
# 找出超过2小时未上线的设备
for client_id, device_info in devices.items():
try:
last_online = datetime.datetime.strptime(device_info['last_online'], '%Y-%m-%d %H:%M:%S')
delta = now - last_online
# 如果时间差大于等于2小时，则标记为删除
if delta.total_seconds() >= 7200: # 7200秒 = 2小时
devices_to_delete.append(client_id)
except Exception as e:
print(f"检查设备 {client_id} 时间时出错: {e}")
# 删除标记的设备
deleted_count = 0
for client_id in devices_to_delete:
if client_id in devices:
del devices[client_id]
deleted_count += 1
print(f"已清理离线设备: {client_id}")
# 保存更新后的数据
if devices_to_delete:
save_devices(devices)
return deleted_count, devices_to_delete
except Exception as e:
print(f"清理离线设备时出错: {e}")
return 0, []
# 更新API返回，添加设备状态信息
@app.route('/api/devices', methods=['GET'])
def get_devices():
"""获取所有已注册设备列表"""
try:
devices = load_devices()
devices_with_status = []
for device_info in devices.values():
device_with_status = device_info.copy()
device_with_status['status'] = get_device_status(device_info)
devices_with_status.append(device_with_status)
return jsonify({
'status': 'success',
'total': len(devices),
'online': len([d for d in devices_with_status if d['status'] == 'online']),
'devices': devices_with_status
}), 200
except Exception as e:
print(f"获取设备列表时出错: {e}")
return jsonify({'status': 'error', 'message': f'服务器内部错误: {str(e)}'}), 500
# 添加清理设备的API端点
@app.route('/api/clean', methods=['POST'])
def clean_devices():
"""通过API清理离线设备"""
try:
deleted_count, deleted_devices = clean_inactive_devices()
return jsonify({
'status': 'success',
'message': f'已清理 {deleted_count} 个离线设备',
'deleted_devices': deleted_devices
}), 200
except Exception as e:
print(f"通过API清理设备时出错: {e}")
return jsonify({'status': 'error', 'message': f'服务器内部错误: {str(e)}'}), 500
# 更新单个设备API，添加状态信息
@app.route('/api/device/<client_id>', methods=['GET'])
def get_device(client_id):
"""获取特定设备的信息"""
try:
devices = load_devices()
if client_id in devices:
device_info = devices[client_id].copy()
device_info['status'] = get_device_status(devices[client_id])
return jsonify({
'status': 'success',
'device_info': device_info
}), 200
else:
return jsonify({'status': 'error', 'message': '设备不存在'}), 404
except Exception as e:
print(f"获取设备信息时出错: {e}")
return jsonify({'status': 'error', 'message': f'服务器内部错误: {str(e)}'}), 500
if __name__ == '__main__':
# 在单独的线程中运行Flask服务器
server_thread = threading.Thread(target=run_server, daemon=True)
server_thread.start()
# 启动定时清理线程
cleanup_thread = threading.Thread(target=scheduled_cleanup, daemon=True)
cleanup_thread.start()
print("定时清理任务已启动，每小时执行一次")
# 启动GUI
root = tk.Tk()
# 先隐藏窗口，避免定位过程中的闪现
root.withdraw()
# 设置窗口图标
try:
# 处理打包后的情况和开发环境
if hasattr(sys, '_MEIPASS'):
# 当使用PyInstaller打包后，图标文件会在临时目录中
icon_path = os.path.join(sys._MEIPASS, 'Server.ico')
else:
# 开发环境中，图标文件在当前目录
icon_path = 'Server.ico'
# 检查图标文件是否存在并设置
if os.path.exists(icon_path):
root.iconbitmap(icon_path)
print(f"成功加载图标: {icon_path}")
else:
print(f"警告: 图标文件不存在: {icon_path}")
except Exception as e:
print(f"设置窗口图标时出错: {e}")
app_gui = DeviceGUI(root)
# 所有设置完成后显示窗口
root.deiconify()
# 设置窗口关闭时的处理
def on_closing():
print("关闭服务器和GUI...")
root.destroy()
root.protocol("WM_DELETE_WINDOW", on_closing)
# 启动GUI主循环
root.mainloop()

复制代码

smile_z · 发表于昨天 06:36

感谢分享

lazhi2025 · 发表于昨天 06:54

感谢分享

it323 · 发表于昨天 07:26

感谢分享！

燕飞龙 · 发表于昨天 07:49

谢谢分享

ylac · 发表于昨天 07:51

这个TOXX现在各种难用

ebaqiang · 发表于昨天 07:57

谢谢楼主分享！

a66 · 发表于昨天 08:05

纯支持～

a272121742 · 发表于昨天 08:08

感谢分享！

wn168cn@163.com · 发表于昨天 08:20

支持原创

lxy60017482 · 发表于昨天 08:21

这个不错，支持

john138 · 发表于昨天 08:24

谢谢分享

dksoft · 发表于昨天 08:31

感谢分享

win82 · 发表于昨天 08:32

感谢分享

cncecpcy · 发表于昨天 08:34

好。。。。。。。。。。。。。。

cncecpcy · 发表于昨天 08:34

好。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。

cncecpcy · 发表于昨天 08:34

好。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。

Darewa · 发表于昨天 08:35

感谢分享，支持

wang1126 · 发表于昨天 08:36

谢谢楼主分享

好好学些 · 发表于昨天 08:40

感谢分享

dxy796 · 发表于昨天 08:41

自动上线，自动安装？

zjf199249 · 发表于昨天 08:51

感谢

yyz2191958 · 发表于昨天 09:04

cncecpcy 发表于 2025-11-25 08:34
好。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。 ...

三连冠

剑转流云 · 发表于昨天 09:05

真长啊这代码1

kgdwfn · 发表于昨天 09:20

完全看不懂这是干嘛的

tanglf · 发表于昨天 09:20

谢谢分享

erdos47 · 发表于昨天 09:24

感谢分享！支持原创

cls822001 · 发表于昨天 09:32

cncecpcy 发表于 2025-11-25 08:34
好。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。 ...

恶意刷帖吗

stz123456 · 发表于昨天 09:35

感谢分享！

锌锰铜铁 · 发表于昨天 09:42

具体怎么用？

		自动登录	找回密码
密码			注册

[原创] Todesk自动安装上线系统!

点评

浏览过的版块