上传文件至 /

This commit is contained in:
2025-09-15 11:26:56 +00:00
parent a695f41e3b
commit 1af11b432a
5 changed files with 751 additions and 0 deletions

429
AccountServices.py Normal file
View File

@@ -0,0 +1,429 @@
import requests
from typing import Optional, Dict, Any, List
from .CallServer import _make_request
DOMAIN = "https://id.solian.app"
###==========================安全报告API==========================
def create_report(
resource_identifier: str,
report_type: int,
reason: str,
Authorization: str = ''
) -> dict:
"""创建安全报告"""
url = f"{DOMAIN}/api/safety/reports"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}",
'Content-Type': 'application/json'
}
request_body = {
"resource_identifier": resource_identifier,
"type": report_type,
"reason": reason
}
return _make_request('POST', url, headers, request_body=request_body)
def get_reports(
offset: int = 0,
take: int = 20,
include_resolved: bool = False,
Authorization: str = ''
) -> dict:
"""获取安全报告列表"""
url = f"{DOMAIN}/api/safety/reports"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
params = {
"offset": offset,
"take": take,
"includeResolved": include_resolved
}
return _make_request('GET', url, headers, params=params)
def get_my_reports(
offset: int = 0,
take: int = 20,
include_resolved: bool = False,
Authorization: str = ''
) -> dict:
"""获取我的安全报告列表"""
url = f"{DOMAIN}/api/safety/reports/me"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
params = {
"offset": offset,
"take": take,
"includeResolved": include_resolved
}
return _make_request('GET', url, headers, params=params)
def get_report_by_id(
report_id: str,
Authorization: str = ''
) -> dict:
"""通过ID获取安全报告"""
url = f"{DOMAIN}/api/safety/reports/{report_id}"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
return _make_request('GET', url, headers)
def resolve_report(
report_id: str,
resolution: str,
Authorization: str = ''
) -> dict:
"""解决安全报告"""
url = f"{DOMAIN}/api/safety/reports/{report_id}/resolve"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}",
'Content-Type': 'application/json'
}
request_body = {
"resolution": resolution
}
return _make_request('POST', url, headers, request_body=request_body)
###==========================账户API==========================
def get_account_by_name(
name: str,
Authorization: str = ''
) -> dict:
"""通过名称获取账户信息"""
url = f"{DOMAIN}/api/accounts/{name}"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
return _make_request('GET', url, headers)
def create_account(
name: str,
nick: str,
email: str,
password: str,
captcha_token: str,
language: Optional[str] = None
) -> dict:
"""创建新账户"""
url = f"{DOMAIN}/api/accounts"
headers = {
'accept': 'application/json',
'Content-Type': 'application/json'
}
request_body = {
"name": name,
"nick": nick,
"email": email,
"password": password,
"captcha_token": captcha_token
}
if language:
request_body["language"] = language
return _make_request('POST', url, headers, request_body=request_body)
def recover_password(
account: str,
captcha_token: str
) -> dict:
"""恢复密码"""
url = f"{DOMAIN}/api/accounts/recovery/password"
headers = {
'accept': 'application/json',
'Content-Type': 'application/json'
}
request_body = {
"account": account,
"captcha_token": captcha_token
}
return _make_request('POST', url, headers, request_body=request_body)
###==========================当前账户API==========================
def get_my_account(
Authorization: str
) -> dict:
"""获取当前账户信息"""
url = f"{DOMAIN}/api/accounts/me"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
return _make_request('GET', url, headers)
def update_my_account(
nick: Optional[str] = None,
language: Optional[str] = None,
region: Optional[str] = None,
Authorization: str = ''
) -> dict:
"""更新当前账户基本信息"""
url = f"{DOMAIN}/api/accounts/me"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}",
'Content-Type': 'application/json'
}
request_body = {}
if nick is not None:
request_body["nick"] = nick
if language is not None:
request_body["language"] = language
if region is not None:
request_body["region"] = region
return _make_request('PATCH', url, headers, request_body=request_body)
def delete_my_account(
Authorization: str
) -> dict:
"""删除当前账户"""
url = f"{DOMAIN}/api/accounts/me"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
return _make_request('DELETE', url, headers)
###==========================认证API==========================
def create_auth_challenge(
platform: int,
account: str,
device_id: str,
device_name: Optional[str] = None,
audiences: Optional[List[str]] = None,
scopes: Optional[List[str]] = None
) -> dict:
"""创建认证挑战"""
url = f"{DOMAIN}/api/auth/challenge"
headers = {
'accept': 'application/json',
'Content-Type': 'application/json'
}
request_body = {
"platform": platform,
"account": account,
"device_id": device_id
}
if device_name:
request_body["device_name"] = device_name
if audiences:
request_body["audiences"] = audiences
if scopes:
request_body["scopes"] = scopes
return _make_request('POST', url, headers, request_body=request_body)
def get_auth_methods(
account: str,
) -> dict:
"""获取认证方式"""
url = f"{DOMAIN}/api/auth/challenge/{account}/factors"
headers = {
'accept': 'application/json',
}
return _make_request('GET', url, headers)
def send_verification_code(
id: str,
factor_id: str,
) -> dict:
"""发送验证码"""
url = f"{DOMAIN}/api/auth/challenge/{id}/factors/{factor_id}"
headers = {
'accept': 'application/json',
}
return _make_request('POST', url, headers)
def perform_auth_challenge_password(
challenge_id: str,
factor_id: str,
password: str
) -> dict:
"""执行认证挑战"""
url = f"{DOMAIN}/api/auth/challenge/{challenge_id}"
headers = {
'accept': 'application/json',
'Content-Type': 'application/json'
}
request_body = {
"factor_id": factor_id,
"password": password
}
return _make_request('PATCH', url, headers, request_body=request_body)
def exchange_token(
grant_type: Optional[str] = None,
refresh_token: Optional[str] = None,
code: Optional[str] = None
) -> dict:
"""交换令牌"""
url = f"{DOMAIN}/api/auth/token"
headers = {
'accept': 'application/json',
'Content-Type': 'application/json'
}
request_body = {}
if grant_type:
request_body["grant_type"] = grant_type
if refresh_token:
request_body["refresh_token"] = refresh_token
if code:
request_body["code"] = code
return _make_request('POST', url, headers, request_body=request_body)
###==========================订阅API==========================
def get_subscriptions(
offset: int = 0,
take: int = 20,
Authorization: str = ''
) -> dict:
"""获取订阅列表"""
url = f"{DOMAIN}/api/subscriptions"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
params = {
"offset": offset,
"take": take
}
return _make_request('GET', url, headers, params=params)
def create_subscription(
identifier: str,
payment_method: str,
payment_details: dict,
coupon: Optional[str] = None,
cycle_duration_days: Optional[int] = None,
is_free_trial: bool = False,
is_auto_renewal: bool = False,
Authorization: str = ''
) -> dict:
"""创建订阅"""
url = f"{DOMAIN}/api/subscriptions"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}",
'Content-Type': 'application/json'
}
request_body = {
"identifier": identifier,
"payment_method": payment_method,
"payment_details": payment_details,
"is_free_trial": is_free_trial,
"is_auto_renewal": is_auto_renewal
}
if coupon:
request_body["coupon"] = coupon
if cycle_duration_days:
request_body["cycle_duration_days"] = cycle_duration_days
return _make_request('POST', url, headers, request_body=request_body)
###==========================钱包API==========================
def get_wallet(
Authorization: str
) -> dict:
"""获取钱包信息"""
url = f"{DOMAIN}/api/wallets"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
return _make_request('GET', url, headers)
def get_wallet_transactions(
offset: int = 0,
take: int = 20,
Authorization: str = ''
) -> dict:
"""获取钱包交易记录"""
url = f"{DOMAIN}/api/wallets/transactions"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
params = {
"offset": offset,
"take": take
}
return _make_request('GET', url, headers, params=params)
def update_wallet_balance(
account_id: str,
amount: float,
currency: str,
remark: Optional[str] = None,
Authorization: str = ''
) -> dict:
"""更新钱包余额"""
url = f"{DOMAIN}/api/wallets/balance"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}",
'Content-Type': 'application/json'
}
request_body = {
"account_id": account_id,
"amount": amount,
"currency": currency
}
if remark:
request_body["remark"] = remark
return _make_request('POST', url, headers, request_body=request_body)
###==========================关系API==========================
def get_relationships(
offset: int = 0,
take: int = 20,
Authorization: str = ''
) -> dict:
"""获取关系列表"""
url = f"{DOMAIN}/api/relationships"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}"
}
params = {
"offset": offset,
"take": take
}
return _make_request('GET', url, headers, params=params)
def create_relationship(
user_id: str,
status: int,
Authorization: str = ''
) -> dict:
"""创建关系"""
url = f"{DOMAIN}/api/relationships/{user_id}"
headers = {
'accept': 'application/json',
'Authorization': f"Bearer {Authorization}",
'Content-Type': 'application/json'
}
request_body = {
"status": status
}
return _make_request('POST', url, headers, request_body=request_body)
###==========================其他API==========================
def get_version() -> dict:
"""获取API版本"""
url = f"{DOMAIN}/api/version"
headers = {
'accept': 'application/json'
}
return _make_request('GET', url, headers)
def get_ip_check() -> dict:
"""获取IP检查信息"""
url = f"{DOMAIN}/api/ip-check"
headers = {
'accept': 'application/json'
}
return _make_request('GET', url, headers)

42
CallServer.py Normal file
View File

@@ -0,0 +1,42 @@
import requests#导入必要的库
import json
from requests.exceptions import RequestException
from .. import PyWebPageAPI
UA = f"SolianForPythonApp/0.0.1(A) ({PyWebPageAPI.GetDeviceInfo()})"
def _make_request(method: str, url: str, headers: dict, params: dict = None, normal_codes: list = [200], request_body: dict = None) -> dict:
"""
params:
可选参数用于GET请求的查询参数
request_body:
可选参数用于POST和PATCH请求的请求体
normal_codes:
可选参数用于指定正常的HTTP状态码列表默认值为[200]
return:
字典类型,包含请求的响应数据或错误信息
内部辅助函数用于发送HTTP请求并处理响应
支持GET、POST、DELETE、PATCH请求
自动处理JSON格式的请求体和响应体
处理常见的HTTP错误码返回统一的错误信息格式"""
headers['User-Agent'] = UA#添加UA
try:
if method == 'GET':#请求类型选择
response = requests.get(url, headers=headers, params=params)
elif method == 'POST':
response = requests.post(url, headers=headers, data=json.dumps(request_body))
elif method == 'DELETE':
response = requests.delete(url, headers=headers, params=params)
elif method == 'PATCH':
response = requests.patch(url, headers=headers, data=json.dumps(request_body))
else:
return {"error": "Unsupported HTTP method"}
if response.status_code not in normal_codes:#不在计划内的错误码
return {"error": f"Unexpected status code: {response.status_code}"}
return response.json()
except json.JSONDecodeError:#json解析错误
return {"error": response.text}
except RequestException as e:
return {"error": str(e)}

199
Chat.py Normal file
View File

@@ -0,0 +1,199 @@
from ProjectCfg import DOMAIN
from .CallServer import _make_request
from typing import List,Any,Optional,Dict
###=========================聊天API=================================
def ChatSummary(Authorization: str) -> dict:
"""获取聊天摘要"""
url = f"{DOMAIN}/chat/summary"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
def GetChatMessages(Authorization: str, room_id: str, offset: int = 0, take: int = 20) -> dict:
"""获取聊天消息"""
url = f"{DOMAIN}/chat/{room_id}/messages"
headers = {'accept': 'application/json', 'Authorization': Authorization}
params = {"offset": offset, "take": take}
return _make_request('GET', url, headers, params=params)
def PostChatMessage(Authorization: str, room_id: str, content: str, nonce: Optional[str] = None, attachments_id: Optional[List[str]] = None, meta: Optional[Dict[str, Any]] = None, replied_message_id: Optional[str] = None, forwarded_message_id: Optional[str] = None) -> dict:
"""发送聊天消息"""
body = {
"content": content,
"nonce": nonce,
"attachments_id": attachments_id,
"meta": meta,
"replied_message_id": replied_message_id,
"forwarded_message_id": forwarded_message_id,
}
url = f"{DOMAIN}/chat/{room_id}/messages"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('POST', url, headers, request_body=body, normal_codes=[201])
def GetChatMessageById(Authorization: str, room_id: str, message_id: str) -> dict:
"""获取具体聊天消息所有信息"""
url = f"{DOMAIN}/chat/{room_id}/messages/{message_id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
def PatchChatMessage(Authorization: str, room_id: str, message_id: str, content: Optional[str] = None, nonce: Optional[str] = None, attachments_id: Optional[List[str]] = None, meta: Optional[Dict[str, Any]] = None, replied_message_id: Optional[str] = None, forwarded_message_id: Optional[str] = None) -> dict:
"""修改聊天消息"""
body = {
"content": content,
"nonce": nonce,
"attachments_id": attachments_id,
"meta": meta,
"replied_message_id": replied_message_id,
"forwarded_message_id": forwarded_message_id,
}
url = f"{DOMAIN}/chat/{room_id}/messages/{message_id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('PATCH', url, headers, request_body=body)
def DeleteChatMessage(Authorization: str, room_id: str, message_id: str) -> dict:
"""删除聊天消息"""
url = f"{DOMAIN}/chat/{room_id}/messages/{message_id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('DELETE', url, headers, normal_codes=[204])
def PostChatSync(Authorization: str, room_id: str, last_sync_timestamp: int) -> dict:
"""同步聊天消息"""
body = {"last_sync_timestamp": last_sync_timestamp}
url = f"{DOMAIN}/chat/{room_id}/sync"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('POST', url, headers, request_body=body)
###=========================聊天室API=================================
def GetChatRoom(Authorization: str, id: str) -> dict:
"""获取聊天房间信息"""
url = f"{DOMAIN}/chat/{id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
def PatchChatRoom(Authorization: str, id: str, name: Optional[str] = None, description: Optional[str] = None, picture_id: Optional[str] = None, background_id: Optional[str] = None, realm_id: Optional[str] = None, is_community: Optional[bool] = None, is_public: Optional[bool] = None) -> dict:
"""修改聊天房间信息"""
body = {
"name": name,
"description": description,
"picture_id": picture_id,
"background_id": background_id,
"realm_id": realm_id,
"is_community": is_community,
"is_public": is_public,
}
url = f"{DOMAIN}/chat/{id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('PATCH', url, headers, request_body=body)
def DeleteChatRoom(Authorization: str, id: str) -> dict:
"""删除聊天房间"""
url = f"{DOMAIN}/chat/{id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('DELETE', url, headers, normal_codes=[204])
def GetChatRooms(Authorization: str) -> dict:
"""获取聊天列表"""
url = f"{DOMAIN}/chat"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
def PostChatRoom(Authorization: str, name: str, description: Optional[str] = None, picture_id: Optional[str] = None, background_id: Optional[str] = None, realm_id: Optional[str] = None, is_community: Optional[bool] = None, is_public: Optional[bool] = None) -> dict:
"""创建聊天房间"""
body = {
"name": name,
"description": description,
"picture_id": picture_id,
"background_id": background_id,
"realm_id": realm_id,
"is_community": is_community,
"is_public": is_public,
}
url = f"{DOMAIN}/chat"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('POST', url, headers, request_body=body, normal_codes=[201])
def GetDirectChat(Authorization: str, account_id: str) -> dict:
"""获取聊天对象用户信息"""
url = f"{DOMAIN}/chat/direct/{account_id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
def GetChatRoomSelfInfo(Authorization: str, room_id: str) -> dict:
"""获取聊天房间自己信息"""
url = f"{DOMAIN}/chat/{room_id}/members/me"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
def JoinChatRoom(Authorization: str, room_id: str) -> dict:
"""加入聊天房间"""
url = f"{DOMAIN}/chat/{room_id}/members/me"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('POST', url, headers, normal_codes=[201])
def LeaveChatRoom(Authorization: str, room_id: str) -> dict:
"""退出聊天房间"""
url = f"{DOMAIN}/chat/{room_id}/members/me"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('DELETE', url, headers, normal_codes=[204])
def GetChatRoomMembers(Authorization: str, room_id: str, offset: int = 0, take: int = 20, with_status: bool = False) -> dict:
"""获取聊天房间成员"""
params = {
"offset": offset,
"take": take,
"withStatus": with_status,
}
url = f"{DOMAIN}/chat/{room_id}/members"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers, params=params)
def PostChatMemberInvite(Authorization: str, room_id: str, related_user_id: str, role: int) -> dict:
"""发送聊天邀请"""
body = {
"related_user_id": related_user_id,
"role": role,
}
url = f"{DOMAIN}/chat/invites/{room_id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('POST', url, headers, request_body=body, normal_codes=[201])
def GetChatInvites(Authorization: str) -> dict:
"""获取聊天邀请"""
url = f"{DOMAIN}/chat/invites"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
def AcceptChatInvite(Authorization: str, room_id: str) -> dict:
"""接受聊天邀请"""
url = f"{DOMAIN}/chat/invites/{room_id}/accept"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('POST', url, headers)
def DeclineChatInvite(Authorization: str, room_id: str) -> dict:
"""拒绝聊天邀请"""
url = f"{DOMAIN}/chat/invites/{room_id}/decline"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('POST', url, headers)
def SetChatNotify(Authorization: str, room_id: str, notify_level: int, break_until: Optional[str] = None) -> dict:
"""设置聊天通知级别"""
body = {
"notify_level": notify_level,
"break_until": break_until,
}
url = f"{DOMAIN}/chat/{room_id}/members/me/notify"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('PATCH', url, headers, request_body=body, normal_codes=[201])
def SetChatMemberRole(Authorization: str, room_id: str, member_id: str, role: int) -> dict:
"""设置聊天成员角色"""
body = {"role": role}
url = f"{DOMAIN}/chat/{room_id}/members/{member_id}/role"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('PATCH', url, headers, request_body=body, normal_codes=[201])
def KickChatMember(Authorization: str, room_id: str, member_id: str) -> dict:
"""踢出聊天成员"""
url = f"{DOMAIN}/chat/{room_id}/members/{member_id}"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('DELETE', url, headers, normal_codes=[204])

76
OtherAPI.py Normal file
View File

@@ -0,0 +1,76 @@
###调用服务器API
###By Liang-work(NRFF&nanci)
###Version 1.0
###License: MIT
from ProjectCfg import DOMAIN
from .CallServer import _make_request
import httpx
def OPITIONS(self,path,params):
response = httpx.options(f"{self.url}{path}",params=params,headers=self.headers)
if response:
return {"status": response.status_code, "data": response.json()}
else:
return {"status": response.status_code, "data": {}}
###=========================活动API=================================
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=params)
###==========================领域的发现==========================
def GetDiscover(query: str = '', take: int = 20, offset: int = 0, Authorization: str = '') -> dict:
"""获取发现"""
url = f"{DOMAIN}/discovery/realms"
headers = {'accept': 'application/json', 'Authorization': Authorization}
params = {'query': query, 'take': take, 'offset': offset}
return _make_request('GET', url, headers, params=params)
###==========================领域聊天==========================
def RealmChat(Authorization: str,slug:str ) -> dict:
"""获取领域聊天"""
url = f"{DOMAIN}/realms/{slug}/chat"
headers = {'accept': 'application/json', 'Authorization': Authorization}
return _make_request('GET', url, headers)
###==========================尊贵的Solar Network 恒星计划订阅用户专属翻译功能======= :(呜呜呜,没钱订阅
def TranslationText(Authorization: str,text:str ,Tolang:str,FromLang:str) -> dict:
"""翻译文本"""
url = f"{DOMAIN}/translation"
headers = {'accept': 'application/json', 'Authorization': Authorization}
params = {"to":Tolang,"from":FromLang}
return _make_request('POST', url, headers, params=params,request_body=text)
###==========================服务器版本==========================
def GetServerVersion() -> dict:
"""获取服务器版本"""
url = f"{DOMAIN}/version"
headers = {'accept': 'application/json'}
return _make_request('GET', url, headers)
###==========================签到==========================
def SignIn(Authorization: str) -> dict:
"""
签到
:return:
"""
return OPITIONS("/id/accounts/me/check-in",{})
###==========================获取通知==========================
def GetNotifications(Authorization: str,offset: int = 0, take: int = 20) -> dict:
"""
获取通知
:param offset:
:param take:
:return:
"""
return _make_request("GET","/pusher/notification",{"offset":offset,"take":take})

5
__init__.py Normal file
View File

@@ -0,0 +1,5 @@
###Solsynth network的服务器内部API封装
###By Liang-work(NRFF&nanci)
###Version 1.0
###License: MIT
from . import Chat, Poll, Post,PostCategory,PostTag,Publisher,WebArticle,WebFeed,WebFeedPublic,WebReader,Sticker,CallServer,Realm,RealtimeCall,AccountServices,SolarNetworkDrive,OtherAPI