# 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()
|
