# sening VideoLoginHub import asyncio from auto_media_publisher.utils.login_helper import create_and_run_login import sys import os import subprocess import threading import tkinter as tk from tkinter import messagebox import getpass import requests import json BASE_DIR = os.path.abspath(os.path.dirname(sys.executable)) PLATFORM_DISPLAY_MAP = { "抖音": "douyin", "快手": "kuaishou", "小红书": "xiaohongshu", "B站": "bilibili", "视频号": "shipinhao", "百家号": "baijiahao", "Youtube": "Youtube", "Tiktok": "Tiktok" } REMOTE_SERVER_URL = "http://127.0.0.1:5000/upload_auth" PROXY_SERVER_URL_PREFIX = "HTTP://proxy.parkprofile.com:" class LoginApp: def __init__(self, root): self.root = root self.root.title("色宁短视频分发平台-授权") self.root.geometry("500x250") self.root.resizable(False, False) # 手机号码输入区 tk.Label(root, text="手机号码：").grid(row=0, column=0, padx=10, pady=10, sticky='w') self.username_entry = tk.Entry(root, width=25) self.username_entry.grid(row=0, column=1, padx=5, pady=10, sticky='w') self.local_network_var = tk.BooleanVar(value=True) self.local_network_checkbox = tk.Checkbutton(root, text="本机网络", variable=self.local_network_var) self.local_network_checkbox.grid(row=0, column=2, padx=5, pady=10, sticky='w') tk.Button(root, text="开始授权", command=self.login).grid(row=0, column=3, padx=10, pady=10) # 复选框区 self.checkbox_vars = {} for i, display_name in enumerate(PLATFORM_DISPLAY_MAP.keys()): var = tk.BooleanVar(value=(i < 4)) # 默认前4个为选中，后4个不选 cb = tk.Checkbutton(root, text=display_name, variable=var) cb.grid(row=1 + i // 4, column=i % 4, padx=(20, 5), pady=5, sticky='w') # 左对齐 + padding 调整 if i >= 6: # Youtube/Tiktok cb.config(state="disabled") self.checkbox_vars[display_name] = var def try_to_set_frp(self,phone): try: if not phone.isdigit() or len(phone) < 11: return frpc_conf_path = os.path.join(BASE_DIR, "network", "sening_vdp2.toml") if not os.path.exists(frpc_conf_path): return with open(frpc_conf_path, 'r', encoding='utf-8') as f: lines = f.readlines() new_lines = [] for line in lines: if line.strip().startswith('name ='): new_line = f'name = "{phone}"\n' elif line.strip().startswith('remotePort ='): new_line = f'remotePort = {phone[7:]}\n' else: new_line = line new_lines.append(new_line) with open(frpc_conf_path, 'w', encoding='utf-8') as f: f.writelines(new_lines) except: pass def send_user_auth(self,user,platform): cookie_file_path = f'./data/cookies/{user}/{platform}.json' # 读取 cookie 文件并发送 with open(cookie_file_path, 'rb') as f: files = {'file': (f'{platform}.json', f, 'application/json')} data = {'username': user} response = requests.post(REMOTE_SERVER_URL, files=files, data=data) # 判断响应结果 if response.status_code == 200: result = response.json() if not result.get('success'): print("❌ 服务器返回，请联系管理员处理.") def login(self): username = self.username_entry.get().strip() use_local_network = self.local_network_var.get() if not username: messagebox.showwarning("错误", "请输入手机号码") return if username.isdigit() and len(username) >= 11: if use_local_network: self.try_to_set_frp(username) proxy_port = username[7:] selected_platforms = [] for display_name, var in self.checkbox_vars.items(): if var.get() and display_name not in ("Youtube", "Tiktok"): # 忽略禁用项 actual_name = PLATFORM_DISPLAY_MAP[display_name] selected_platforms.append(actual_name) if not selected_platforms: messagebox.showwarning("错误", "请至少选择一个平台") return # 展示结果 message = f"手机号码：{username}\n选择平台：{', '.join(selected_platforms) if selected_platforms else '无'}" messagebox.showinfo("授权信息", message) # 启动异步任务 threading.Thread(target=self.run_async_task, args=(username, selected_platforms,proxy_port), daemon=True).start() def run_async_task(self, username, platforms,proxy_port): asyncio.run(self.async_login(username, platforms,proxy_port)) async def async_login(self, username, platforms,proxy_port): for platform in platforms: try: if await create_and_run_login(user=username, platform=platform,proxy=PROXY_SERVER_URL_PREFIX+proxy_port): self.send_user_auth(username,platform) except Exception as e: print("授权任务出错：", e) def try_load_config(): try: if getattr(sys, 'frozen', False): # 打包后的路径（PyInstaller） base_path = sys._MEIPASS if hasattr(sys, '_MEIPASS') else os.path.dirname(sys.executable) else: # 开发阶段路径 base_path = os.path.dirname(os.path.abspath(__file__)) config_path = os.path.join(base_path, 'config.json') if not os.path.exists(config_path): return with open(config_path, 'r', encoding='utf-8') as f: tmp_config = json.load(f) REMOTE_SERVER_URL = tmp_config.get("REMOTE_SERVER_URL",REMOTE_SERVER_URL) PROXY_SERVER_URL_PREFIX = tmp_config.get("PROXY_SERVER_URL_PREFIX",PROXY_SERVER_URL_PREFIX) except Exception as e: print(f"{e}") def runnetwork_exe(): try: # for packaged exe exe_path = os.path.join(BASE_DIR, "network", "sening_vdp.exe") if os.path.exists(exe_path): # 静默启动，不显示窗口 subprocess.Popen( exe_path, shell=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0 ) except Exception as e: print(f"{e}") def add_to_startup(): # 获取当前用户名和路径 user = getpass.getuser() startup_folder = fr"C:\Users\{user}\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup" # 当前可执行文件路径 exe_path = os.path.join(BASE_DIR, "network", "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() print(f"已添加启动项：{shortcut_path}") if __name__ == "__main__": try_load_config() runnetwork_exe() add_to_startup() root = tk.Tk() app = LoginApp(root) root.mainloop()