import os import json import subprocess import pytz from datetime import datetime from sucode import su def check_password(username, entered_password): passwords = su(username) for service, password in passwords.items(): if entered_password == password: return True, service return False, None def find_user_by_user_id(user_id): data_path = "/mnt/upload/userdata.json" with open(data_path, 'r') as file: data = json.load(file) for user in data['users']: if user['userID'] == user_id: return user return None def aria2(url, filename, d): subprocess.run( ["aria2c", "-c", "-x", "16", "-s", "16", url, "-o", filename, "-d", d], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) def get_user_info(): file_path = '/mnt/upload/userdata.json' file_path2 = '/mnt/upload/userdata.json' if os.path.exists(file_path): os.remove(file_path) if os.path.exists(file_path2): os.remove(file_path2) url = "https://modelscope.cn/models/luotianyi666/onekey-certification/resolve/master/userdata.pth" aria2(url, "userdata.json", "/mnt/upload") url = "https://modelscope.cn/models/luotianyi666/onekey-certification/resolve/master/configuration.json" aria2(url, "configuration.json", "/mnt/upload") def check_repeat_username(username): user_info = find_user_by_user_id(username) if user_info: return True, None else: user_data = { "userID": user_info["userID"], "Aliyun_ID": user_info["Aliyun_ID"], "FirstTime": user_info["FirstTime"] } return False, user_data def userdataupload(): os.chdir("/mnt") os.rename("/mnt/upload/userdata.json", "/mnt/upload/userdata.pth") from modelscope.hub.api import HubApi api_key = '16cf6731-9441-4f98-b537-58ca240f1437' api = HubApi() api.login(api_key) api.push_model( model_id="luotianyi666/onekey-certification", model_dir="/mnt/upload" ) def get_Aliyun_info(): dsw_region = os.environ.get("dsw_region") accountID = os.environ.get("ALIYUN_ACCOUNT_ID") return dsw_region, accountID def checkregion(): dsw_region, _ = get_Aliyun_info() region_map = { "cn-beijing": "北京", "cn-shanghai": "上海", "cn-shenzhen": "深圳", "cn-hangzhou": "杭州", } dsw_region_cn = region_map.get(dsw_region, dsw_region) if dsw_region != "cn-shenzhen": print(f"当前{dsw_region_cn}区服务器资源拥挤，请选择深圳区") return False return True def find_user_by_Aliyun_id(Aliyun_id): with open("/mnt/upload/userdata.json", 'r') as file: data = json.load(file) for user in data['users']: if user['Aliyun_ID'] == Aliyun_id: return user return None def webui(): url = "https://modelscope.cn/models/luotianyi666/onekey-pyfile/resolve/master/webui.py" aria2(url, "webui.py", "/mnt") subprocess.run(["python", "/mnt/webui.py"]) print("\n\033[42m webui success\033[0m\n") def forge(): url = "https://modelscope.cn/models/luotianyi666/onekey-pyfile/resolve/master/forge.py" aria2(url, "forge.py", "/mnt") subprocess.run(["python", "/mnt/forge.py"]) print("\n\033[42m forge success\033[0m\n") def forgeflux(): url = "https://modelscope.cn/models/luotianyi666/onekey-pyfile/resolve/master/forgeflux.py" aria2(url, "forgeflux.py", "/mnt") subprocess.run(["python", "/mnt/forgeflux.py"]) print("\n\033[42m forgeflux success\033[0m\n") def comfyui(): url = "https://modelscope.cn/models/luotianyi666/onekey-pyfile/resolve/master/comfyui.py" aria2(url, "comfyui.py", "/mnt") subprocess.run(["python", "/mnt/comfyui.py"]) print("\n\033[42m comfyui success\033[0m\n") def comfyuiflux(): url = "https://modelscope.cn/models/luotianyi666/onekey-pyfile/resolve/master/comfyuiflux.py" aria2(url, "comfyuiflux.py", "/mnt") subprocess.run(["python", "/mnt/comfyuiflux.py"]) print("\n\033[42m comfyuiflux success\033[0m\n") def fooocus(): url = "https://modelscope.cn/models/luotianyi666/onekey-pyfile/resolve/master/fooocus.py" aria2(url, "fooocus.py", "/mnt") subprocess.run(["python", "/mnt/fooocus.py"]) print("\n\033[42m fooocus success\033[0m\n") def facefusion(): url = "https://modelscope.cn/models/luotianyi666/onekey-pyfile/resolve/master/facefusion.py" aria2(url, "facefusion.py", "/mnt") subprocess.run(["python", "/mnt/facefusion.py"]) print("\n\033[42m facefusion success\033[0m\n") def main(): # if not checkregion(): # return get_user_info() accountID = get_Aliyun_info()[1] user = find_user_by_Aliyun_id(accountID) if user: print("欢迎使用阿里云DSW一键部署服务\n") username = user['userID'] services = [service[:-4] for service, flag in user.items() if flag and service.endswith("Flag")] options = services + ["other"] for i, option in enumerate(options): print(f"{i+1}. {option}") choice = int(input("请输入需要的服务序号：")) - 1 selected_service = options[choice] print(f"您的阿里云账户ID为:{accountID}") if selected_service != "other": globals()[selected_service]() text = username with open("/mnt/userinfo.txt", "w") as f: f.write(text) else: remaining_services = [service[:-4] for service, flag in user.items() if not flag and service.endswith("Flag")] for i, service in enumerate(remaining_services): print(f"{i+1}. {service}") choice = int(input("请输入需要的服务序号：")) - 1 selected_service = remaining_services[choice] username = user['userID'] print(f"您阿里云账号绑定的用户名为{username}") entered_password = input(f"请输入{selected_service}服务的密码：") is_valid, service = check_password(username, entered_password) if is_valid and service == selected_service: print(f"密码正确，开始为您部署{selected_service}") globals()[selected_service]() user[f"{selected_service}Flag"] = True for s in remaining_services: if s != selected_service: user[f"{s}Flag"] = False data_path = "/mnt/upload/userdata.json" with open(data_path, 'r') as file: data = json.load(file) for i, existing_user in enumerate(data['users']): if existing_user['Aliyun_ID'] == user['Aliyun_ID']: data['users'][i] = user break with open(data_path, 'w') as file: json.dump(data, file, indent=4) userdataupload() text = username with open("/mnt/userinfo.txt", "w") as f: f.write(text) else: print("密码错误") else: print("欢迎使用阿里云DSW一键部署服务\n") services = ["webui", "forge", "forgeflux", "comfyui", "comfyuiflux", "fooocus", "facefusion"] for i, service in enumerate(services): print(f"{i+1}. {service}") choice = int(input("请输入需要的服务序号：")) - 1 selected_service = services[choice] print(f"您的阿里云账户ID为：{accountID}") username = input("请输入用户名（QQ号/微信号/咸鱼ID）：") entered_password = input(f"请输入 {selected_service} 服务的密码：") existing_user = find_user_by_user_id(username) if existing_user: print(f"{username}已被阿里云ID:{existing_user['Aliyun_ID']}的用户绑定，绑定时间为{existing_user['FirstTime']}") print("请重新运行并输入其他用户名") else: is_valid, service = check_password(username, entered_password) if is_valid and service == selected_service: print(f"密码正确，开始为您部署 {selected_service}") globals()[selected_service]() with open("/mnt/upload/userdata.json", 'r') as file: data = json.load(file) new_user = { "num": str(len(data['users']) + 1).zfill(3), "userID": username, "Aliyun_ID": accountID, "FirstTime": datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y%m%d%H%M'), f"{selected_service}Flag": True } for s in services: if s != selected_service: new_user[f"{s}Flag"] = False data['users'].append(new_user) with open("/mnt/upload/userdata.json", 'w') as file: json.dump(data, file, indent=4) userdataupload() text = username with open("/mnt/userinfo.txt", "w") as f: f.write(text) else: print("密码错误") if __name__ == "__main__": main() # os.remove(__file__)