SolianForPython/core/SNAPI/AccountServices.py

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