上传文件至 core
This commit is contained in:
246
core/AccountServices.py
Normal file
246
core/AccountServices.py
Normal file
@@ -0,0 +1,246 @@
|
||||
import requests
|
||||
import json
|
||||
import uuid
|
||||
import platform
|
||||
import hashlib
|
||||
|
||||
class AuthClient:
|
||||
def __init__(self, base_url):
|
||||
self.base_url = base_url
|
||||
self.stage = 'find-account'
|
||||
self.error = None
|
||||
self.account_identifier = ''
|
||||
self.device_id = ''
|
||||
self.challenge = None
|
||||
self.factors = []
|
||||
self.selected_factor_id = None
|
||||
self.password = ''
|
||||
self.user_store = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3' # 您可以自定义此User-Agent字符串
|
||||
|
||||
def generate_device_id(self):
|
||||
"""生成设备指纹
|
||||
:return: 设备指纹ID（字符串）
|
||||
"""
|
||||
platform_info = platform.system() + platform.release() + platform.machine() # 修正了这里的拼接错误
|
||||
unique_id = uuid.getnode()
|
||||
device_info = f"{platform_info}-{unique_id}"
|
||||
# 使用SHA-256哈希生成设备ID
|
||||
self.device_id = hashlib.sha256(device_info.encode()).hexdigest()
|
||||
return self.device_id
|
||||
|
||||
def check_account_validity(self, account_identifier):
|
||||
"""检查账户有效性
|
||||
:param account_identifier: 用户的账户标识符（邮箱或用户名）
|
||||
:return: 挑战信息（字典）
|
||||
"""
|
||||
self.account_identifier = account_identifier
|
||||
self.error = None
|
||||
if not self.account_identifier:
|
||||
self.error = 'Please enter your email or username.'
|
||||
return None
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{self.base_url}/auth/challenge",
|
||||
headers={'Content-Type': 'application/json', 'User-Agent': self.user_store,'device_name': "A Python login app.By NRFF & Deepseek"}, # 添加自定义User-Agent
|
||||
data=json.dumps({
|
||||
'platform': 1,
|
||||
'account': self.account_identifier,
|
||||
'device_id': self.generate_device_id(),
|
||||
})
|
||||
)
|
||||
if not response.ok:
|
||||
raise Exception(response.text or 'Account not found.')
|
||||
self.challenge = response.json()
|
||||
self.stage = 'select-factor'
|
||||
return self.challenge
|
||||
except Exception as e:
|
||||
self.error = str(e)
|
||||
return None
|
||||
|
||||
def get_authentication_factors(self):
|
||||
"""获取可用的认证因素
|
||||
:return: 可用的认证因素列表（列表）
|
||||
"""
|
||||
self.error = None
|
||||
if not self.challenge:
|
||||
self.error = 'No challenge information available.'
|
||||
return None
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{self.base_url}/auth/challenge/{self.challenge['id']}/factors",
|
||||
headers={'User-Agent': self.user_store,'device_name': "A Python login app.By NRFF & Deepseek"} # 添加自定义User-Agent
|
||||
)
|
||||
if not response.ok:
|
||||
raise Exception('Could not fetch authentication factors.')
|
||||
available_factors = response.json()
|
||||
self.factors = [factor for factor in available_factors if factor['id'] not in self.challenge.get('blacklist_factors', [])]
|
||||
if len(self.factors) == 0:
|
||||
self.error = 'No available authentication factors.'
|
||||
return None
|
||||
else:
|
||||
self.stage = 'select-factor'
|
||||
return self.factors
|
||||
except Exception as e:
|
||||
self.error = str(e)
|
||||
return None
|
||||
|
||||
def request_verification_code(self, selected_factor_id):
|
||||
"""请求验证码
|
||||
:param selected_factor_id: 用户选择的认证因素ID（字符串类型）
|
||||
:return: None
|
||||
"""
|
||||
self.selected_factor_id = selected_factor_id
|
||||
self.error = None
|
||||
if not self.selected_factor_id:
|
||||
self.error = 'No authentication factor selected.'
|
||||
return None
|
||||
selected_factor = self.get_selected_factor()
|
||||
if not selected_factor:
|
||||
self.error = 'Selected factor not found.'
|
||||
return None
|
||||
hint = {'contact': selected_factor['contact']}
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{self.base_url}/auth/challenge/{self.challenge['id']}/factors/{self.selected_factor_id}",
|
||||
headers={'Content-Type': 'application/json', 'User-Agent': self.user_store, 'device_name': "A Python login app.By NRFF & Deepseek"}, # 添加自定义User-Agent
|
||||
data=json.dumps(hint)
|
||||
)
|
||||
if not response.ok:
|
||||
raise Exception(response.text or 'Failed to send code.')
|
||||
self.stage = 'enter-code'
|
||||
except Exception as e:
|
||||
self.error = str(e)
|
||||
self.stage = 'select-factor'
|
||||
return None
|
||||
|
||||
def get_selected_factor(self):
|
||||
"""获取选择的认证因素
|
||||
:return: 选择的认证因素（字典）或 None
|
||||
"""
|
||||
if not self.selected_factor_id:
|
||||
return None
|
||||
return next((factor for factor in self.factors if factor['id'] == self.selected_factor_id), None)
|
||||
|
||||
def verify_factor(self, selected_factor_id, password):
|
||||
"""验证所选的认证因素
|
||||
:param selected_factor_id: 用户选择的认证因素ID（字符串类型）
|
||||
:param password: 用户输入的密码或验证码（字符串类型）
|
||||
:return: None
|
||||
"""
|
||||
self.selected_factor_id = selected_factor_id
|
||||
self.password = password
|
||||
self.error = None
|
||||
if not self.selected_factor_id or not self.password:
|
||||
self.error = 'Please enter your password/code.'
|
||||
return None
|
||||
try:
|
||||
response = requests.patch(
|
||||
f"{self.base_url}/auth/challenge/{self.challenge['id']}",
|
||||
headers={'Content-Type': 'application/json', 'User-Agent': self.user_store,'device_name': "A Python login app.By NRFF & Deepseek"}, # 添加自定义User-Agent
|
||||
data=json.dumps({
|
||||
'factor_id': self.selected_factor_id,
|
||||
'password': self.password,
|
||||
})
|
||||
)
|
||||
if not response.ok:
|
||||
raise Exception(response.text or 'Verification failed.')
|
||||
self.challenge = response.json()
|
||||
self.password = ''
|
||||
if self.challenge['step_remain'] == 0:
|
||||
self.stage = 'token-exchange'
|
||||
else:
|
||||
self.stage = 'select-factor'
|
||||
except Exception as e:
|
||||
self.error = str(e)
|
||||
self.stage = 'select-factor'
|
||||
|
||||
def exchange_token(self):
|
||||
"""交换令牌以完成登录
|
||||
:return: None
|
||||
"""
|
||||
self.error = None
|
||||
if not self.challenge:
|
||||
self.error = 'No challenge information available.'
|
||||
return None
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{self.base_url}/auth/token",
|
||||
headers={'Content-Type': 'application/json', 'User-Agent': self.user_store,'device_name': "A Python login app.By NRFF & Deepseek"}, # 添加自定义User-Agent
|
||||
data=json.dumps({
|
||||
'grant_type': 'authorization_code',
|
||||
'code': self.challenge['id'],
|
||||
})
|
||||
)
|
||||
if not response.ok:
|
||||
raise Exception(response.text or 'Token exchange failed.')
|
||||
token_info = response.json()
|
||||
token = token_info['token']
|
||||
#self.user_store.fetchUser()
|
||||
redirect_uri = 'redirect_uri_from_query' # 这里需要根据实际情况获取
|
||||
if redirect_uri:
|
||||
print(f"Redirecting to: {redirect_uri}")
|
||||
else:
|
||||
print("Navigating to home page.")
|
||||
return token
|
||||
except Exception as e:
|
||||
self.error = str(e)
|
||||
self.stage = 'select-factor'
|
||||
|
||||
def get_factor_name(self, factor_type):
|
||||
"""根据认证因素类型返回相应的名称
|
||||
:param factor_type: 认证因素类型（整数）
|
||||
:return: 认证因素名称（字符串）
|
||||
"""
|
||||
factor_names = {
|
||||
0: 'Password',
|
||||
1: 'Email',
|
||||
2: 'Authenticator App',
|
||||
}
|
||||
return factor_names.get(factor_type, 'Unknown Factor')
|
||||
|
||||
# 示例调用
|
||||
if __name__ == "__main__":
|
||||
auth_client = AuthClient(base_url='https://api.solian.app/id')
|
||||
account_identifier = ''#你的账号
|
||||
selected_factor_id = "enter-code" # 用户需要选择一个因素
|
||||
password = ''#你的密码
|
||||
# 第一步：检查账户有效性
|
||||
challenge = auth_client.check_account_validity(account_identifier)
|
||||
if auth_client.error:
|
||||
print(f"Error in check account validity: {auth_client.error}")
|
||||
elif challenge:
|
||||
print(f"Challenge information: {challenge}")
|
||||
# 第二步：获取可用的认证因素
|
||||
factors = auth_client.get_authentication_factors()
|
||||
if auth_client.error:
|
||||
print(f"Error in get authentication factors: {auth_client.error}")
|
||||
elif factors:
|
||||
print(f"Available factors: {factors}")
|
||||
# 假设用户选择第一个因素
|
||||
selected_factor_id = factors[0]['id']
|
||||
print(f"Selected factor ID: {selected_factor_id}")
|
||||
# 第三步：请求验证码（如果需要）
|
||||
if factors[0]['type'] == 1:
|
||||
auth_client.request_verification_code(selected_factor_id)
|
||||
if auth_client.error:
|
||||
print(f"Error in request verification code: {auth_client.error}")
|
||||
elif auth_client.stage == 'enter-code':
|
||||
print("Verification code requested. Please enter the code.")
|
||||
# 第四步：验证所选的认证因素
|
||||
auth_client.verify_factor(selected_factor_id, password)
|
||||
if auth_client.error:
|
||||
print(f"Error in verify factor: {auth_client.error}")
|
||||
elif auth_client.stage == 'token-exchange':
|
||||
print("Factor verified. Ready to exchange token.")
|
||||
# 第五步：交换令牌以完成登录
|
||||
if auth_client.stage == 'token-exchange':
|
||||
token = auth_client.exchange_token()
|
||||
if auth_client.error:
|
||||
print(f"Error in exchange token: {auth_client.error}")
|
||||
else:
|
||||
print("Login successful!")
|
||||
print(f"Token: {token}")
|
||||
else:
|
||||
print("Login failed.")
|
||||
else:
|
||||
print("No available authentication factors.")
|
||||
92
core/CallServerAPIs.py
Normal file
92
core/CallServerAPIs.py
Normal file
@@ -0,0 +1,92 @@
|
||||
import requests
|
||||
from requests.exceptions import RequestException
|
||||
import json
|
||||
|
||||
DOMAIN = "https://solian.app/api"
|
||||
|
||||
def _make_request(method: str, url: str, headers: dict, params: dict = None, data: dict = None) -> dict:
|
||||
"""内部辅助函数，用于发送HTTP请求并处理响应"""
|
||||
try:
|
||||
if method == 'GET':
|
||||
response = requests.get(url, headers=headers, params=params)
|
||||
elif method == 'POST':
|
||||
response = requests.post(url, headers=headers, data=data)
|
||||
elif method == 'DELETE':
|
||||
response = requests.delete(url, headers=headers, params=params)
|
||||
elif method == 'PATCH':
|
||||
response = requests.patch(url, headers=headers, data=data)
|
||||
else:
|
||||
return {"error": "Unsupported HTTP method"}
|
||||
|
||||
if response.status_code != 200:
|
||||
return {"error": response.status_code}
|
||||
|
||||
return response.json()
|
||||
|
||||
except json.JSONDecodeError:
|
||||
return {"error": response.text}
|
||||
except RequestException as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
def ActivityAPIs(cursor: str = '', filter: str = '', take: int = 20, debuginclude: str = '', Authorization: str = '') -> dict:
|
||||
"""获取首页内容"""
|
||||
url = f"{DOMAIN}/activities"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
params = {"cursor": cursor, "filter": filter, "take": take, "debuginclude": debuginclude}
|
||||
return _make_request('GET', url, headers, params)
|
||||
|
||||
def ChatSummary(Authorization: str) -> dict:
|
||||
"""获取聊天摘要"""
|
||||
url = f"{DOMAIN}/chat/summary"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('GET', url, headers)
|
||||
|
||||
def GetChatMessageAllInfo(Authorization: str, roomid: str, offset: int = 0, take: int = 20) -> dict:
|
||||
"""获取聊天消息"""
|
||||
url = f"{DOMAIN}/chat/{roomid}/message"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
params = {"offset": offset, "take": take}
|
||||
return _make_request('GET', url, headers, params)
|
||||
|
||||
def GetChatMessageBaseInfo(Authorization: str, roomid: str) -> dict:
|
||||
"""获取聊天消息基础信息"""
|
||||
url = f"{DOMAIN}/chat/{roomid}/message/base"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('POST', url, headers)
|
||||
|
||||
def GetAMessageAllInfo(Authorization: str, roomid: str, messageid: str) -> dict:
|
||||
"""获取具体聊天消息所有信息"""
|
||||
url = f"{DOMAIN}/chat/{roomid}/message/{messageid}"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('GET', url, headers)
|
||||
|
||||
def GetAMessageInfo(Authorization: str, roomid: str, messageid: str) -> dict:
|
||||
"""获取聊天消息基础信息。暂时作为保留"""
|
||||
url = f"{DOMAIN}/chat/{roomid}/message/{messageid}/info"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('POST', url, headers)
|
||||
|
||||
def DelMessage(Authorization: str, roomid: str, messageid: str) -> dict:
|
||||
"""删除聊天消息"""
|
||||
url = f"{DOMAIN}/chat/{roomid}/message/{messageid}"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('DELETE', url, headers)
|
||||
|
||||
def GetChatRoomInfo(Authorization: str, id: str) -> dict:
|
||||
"""获取聊天房间信息"""
|
||||
url = f"{DOMAIN}/chat/{id}"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('GET', url, headers)
|
||||
|
||||
def ModifyChatRoomInfo(Authorization: str, id: str, data: dict) -> dict:
|
||||
"""修改聊天房间信息"""
|
||||
url = f"{DOMAIN}/chat/{id}"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('PATCH', url, headers, data=data)
|
||||
|
||||
def DeleteChatRoom(Authorization: str, id: str) -> dict:
|
||||
"""删除聊天房间"""
|
||||
url = f"{DOMAIN}/chat/{id}"
|
||||
headers = {'accept': 'application/json', 'Authorization': Authorization}
|
||||
return _make_request('DELETE', url, headers)
|
||||
|
||||
78
core/PyWebPageAPI.py
Normal file
78
core/PyWebPageAPI.py
Normal file
@@ -0,0 +1,78 @@
|
||||
import os
|
||||
import sys
|
||||
import importlib
|
||||
import platform
|
||||
import ProjectCfg
|
||||
def EnvironmentCheck() -> bool:
|
||||
"""
|
||||
检查环境是否符合要求
|
||||
:return:
|
||||
"""
|
||||
if sys.version_info < ProjectCfg.PYTHON_VERSION_MIN and ProjectCfg.PYTHON_VERSION_MIN!=(0,0):
|
||||
return False
|
||||
if sys.version_info > ProjectCfg.PYTHON_VERSION_MAX and ProjectCfg.PYTHON_VERSION_MAX!=(0,0):
|
||||
return False
|
||||
for lib in ProjectCfg.LIB_LIST:
|
||||
if importlib.util.find_spec(lib) is None:
|
||||
return False
|
||||
if platform.system() not in ProjectCfg.SUPPORTED_PLATFORMS and ProjectCfg.SUPPORTED_PLATFORMS!=['All']:
|
||||
return False
|
||||
return True
|
||||
|
||||
def CheckWebviewInstalled() -> tuple[bool,str]:
|
||||
"""
|
||||
检查WebView框架是否已安装
|
||||
:return:
|
||||
"""
|
||||
system_platform : str = platform.system()
|
||||
|
||||
if system_platform == "Windows":
|
||||
# 检查WebView2
|
||||
webview2_path : str = r'C:\Program Files (x86)\Microsoft\EdgeWebView2'
|
||||
webview2_dll : str = r'\WebView2Loader.dll'
|
||||
for root, dirs, files in os.walk(webview2_path):
|
||||
if webview2_dll in files:
|
||||
return (True,'')
|
||||
return (False,"No installed WebView2")
|
||||
|
||||
elif system_platform == "Darwin":
|
||||
# 检查WebKit框架
|
||||
webkit_path = '/System/Library/Frameworks/WebKit.framework'
|
||||
if os.path.exists(webkit_path):
|
||||
return (True,'')
|
||||
return (False,"Unsupported WKWebView")
|
||||
|
||||
elif system_platform == "Linux":
|
||||
# 检查WebKitGTK
|
||||
webkitgtk_paths = [
|
||||
'/usr/lib/libwebkit2gtk-4.0.so',
|
||||
'/usr/lib/x86_64-linux-gnu/libwebkit2gtk-4.0.so'
|
||||
]
|
||||
for path in webkitgtk_paths:
|
||||
if os.path.exists(path):
|
||||
return (True,'')
|
||||
|
||||
# 检查QtWebEngine
|
||||
qtwebengine_paths = [
|
||||
'/usr/lib/qt5/bin/qtwebengine_process',
|
||||
'/usr/lib/x86_64-linux-gnu/qt5/bin/qtwebengine_process'
|
||||
]
|
||||
for path in qtwebengine_paths:
|
||||
if os.path.exists(path):
|
||||
return "QtWebEngine已安装"
|
||||
|
||||
return (False,"Not Found Webview framework.")
|
||||
|
||||
else:
|
||||
return (False,"Unsupported OS.")
|
||||
|
||||
def CheckPortAvailable(port : int) -> bool:
|
||||
"""
|
||||
检查端口是否可用
|
||||
:param port:
|
||||
:return:
|
||||
"""
|
||||
import socket
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
return s.connect_ex(('localhost', port)) != 0
|
||||
return False
|
||||
29
core/WebApp.py
Normal file
29
core/WebApp.py
Normal file
@@ -0,0 +1,29 @@
|
||||
import flask
|
||||
|
||||
app = flask.Flask(__name__, template_folder='../webfile', static_folder='../webfile/static')
|
||||
|
||||
@app.errorhandler(500)
|
||||
def internal_server_error(error):
|
||||
# 返回500错误的HTML页面
|
||||
return flask.render_template("/error/500.html",error=error,static_url_path='/static')
|
||||
|
||||
@app.errorhandler(404)
|
||||
def not_found(error):
|
||||
# 返回404错误的HTML页面
|
||||
return flask.render_template("/error/404.html",error=error,static_url_path='/static'),404
|
||||
|
||||
@app.errorhandler(403)
|
||||
def forbidden(error):
|
||||
# 返回403错误的HTML页面
|
||||
return flask.render_template("/error/403.html",error=error,static_url_path='/static'),403
|
||||
|
||||
@app.route('/')
|
||||
def index():
|
||||
return flask.render_template('index.html')
|
||||
|
||||
def AppStart(host: str, port: int):
|
||||
"""启动Flask应用"""
|
||||
app.run(host=host, port=port, debug=True, use_reloader=False)
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(host="127.0.0.1", port=5000, debug=True)
|
||||
7
core/WebViewWIndow.py
Normal file
7
core/WebViewWIndow.py
Normal file
@@ -0,0 +1,7 @@
|
||||
import webview
|
||||
|
||||
def WebViewWIndow(url: str,title: str,width: int,height: int):#创建Webview窗口
|
||||
webview.create_window(title=title, url=url, width=width, height=height)
|
||||
webview.settings['ALLOW_DOWNLOADS'] = True
|
||||
webview.start(icon="./assets/img/icon.png")
|
||||
exit(0)
|
||||
Reference in New Issue
Block a user