import os
|
import subprocess
|
import time
|
import logging
|
import sys
|
import getpass
|
|
APP_NAME = "sening_vdp"
|
LOG_FILE = "sening_vdp.log"
|
|
BASE_DIR = os.path.abspath(os.path.dirname(sys.executable))
|
|
logging.basicConfig(
|
filename=os.path.join(BASE_DIR, LOG_FILE),
|
level=logging.INFO,
|
format='%(asctime)s - %(levelname)s - %(message)s'
|
)
|
|
def start_process(name, exe_path, args, log_file):
|
try:
|
full_path = os.path.abspath(exe_path)
|
logging.info(f"Starting {name}: {full_path} {' '.join(args)}")
|
|
# 打开日志文件，追加写入
|
log_fh = open(log_file, "a", encoding="utf-8")
|
|
# CREATE_NO_WINDOW 隐藏窗口
|
CREATE_NO_WINDOW = 0x08000000
|
|
process = subprocess.Popen(
|
[full_path] + args,
|
cwd=os.path.dirname(full_path),
|
stdout=log_fh,
|
stderr=subprocess.STDOUT,
|
creationflags=CREATE_NO_WINDOW,
|
universal_newlines=True
|
)
|
return process
|
except Exception as e:
|
logging.error(f"Failed to start {name}: {e}")
|
return None
|
|
|
|
def start_process_with_retry(name, exe_path, args, log_file, retry_interval=120):
|
full_path = os.path.abspath(exe_path)
|
CREATE_NO_WINDOW = 0x08000000
|
|
while True:
|
try:
|
logging.info(f"[{name}] 尝试启动：{full_path} {' '.join(args)}")
|
|
process = subprocess.Popen(
|
[full_path] + args,
|
cwd=os.path.dirname(full_path),
|
stdout=subprocess.PIPE,
|
stderr=subprocess.STDOUT,
|
creationflags=CREATE_NO_WINDOW,
|
universal_newlines=True,
|
encoding="utf-8",
|
)
|
|
success = False
|
while True:
|
line = process.stdout.readline()
|
print(line)
|
if not line:
|
break
|
logging.info(f"[{name}] {line.strip()}")
|
|
if "connect to server error" in line or "login to the server failed" in line:
|
logging.warning(f"[{name}] 连接失败，{retry_interval} 秒后重试...")
|
process.kill()
|
time.sleep(retry_interval)
|
break
|
|
if "start proxy success" in line :
|
success = True
|
|
if success:
|
logging.info(f"[{name}] 成功启动并连接")
|
return process
|
|
except Exception as e:
|
logging.error(f"[{name}] 启动异常：{e}")
|
time.sleep(retry_interval)
|
|
def add_to_startup():
|
try:
|
# 获取当前用户名和路径
|
user = getpass.getuser()
|
startup_folder = fr"C:\Users\{user}\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup"
|
|
# 当前可执行文件路径
|
exe_path = os.path.join(BASE_DIR,"sening_vdp.exe")
|
|
shortcut_path = os.path.join(startup_folder, "sening_vdp.lnk")
|
|
# 创建快捷方式
|
import pythoncom
|
from win32com.shell import shell, shellcon
|
from win32com.client import Dispatch
|
|
shell = Dispatch('WScript.Shell')
|
shortcut = shell.CreateShortcut(shortcut_path)
|
shortcut.TargetPath = exe_path
|
shortcut.WorkingDirectory = os.path.dirname(exe_path)
|
shortcut.IconLocation = exe_path
|
shortcut.Save()
|
except Exception as e:
|
print(f"e")
|
pass
|
|
if __name__ == "__main__":
|
logging.info("程序启动")
|
add_to_startup()
|
import sys
|
import win32event
|
import win32api
|
import winerror
|
mutex = win32event.CreateMutex(None, False, "Global\\SeningVDPOnlyOne")
|
last_error = win32api.GetLastError()
|
if last_error == winerror.ERROR_ALREADY_EXISTS:
|
# 已经运行过了，退出
|
sys.exit(0)
|
# 启动 Clash（当前目录下 clash.exe，配置文件 clash.yaml）
|
clash_process = start_process("sening_vdp1", "sening_vdp1.exe", ["-f", "sening_vdp1.yaml"], LOG_FILE)
|
time.sleep(2)
|
|
# 启动 frpc（当前目录下 frpc.exe，配置文件 frpc.toml）
|
frpc_process = start_process_with_retry("sening_vdp2", "sening_vdp2.exe", ["-c", "sening_vdp2.toml"], LOG_FILE)
|
|
try:
|
if clash_process:
|
clash_process.wait()
|
if frpc_process:
|
frpc_process.wait()
|
except KeyboardInterrupt:
|
logging.info("程序被中断，退出")
|
