feat: Add the basic functions

Author: Keniis0712

.gitignore

@@ -0,0 +1,3 @@
+.venv
+.idea
+cli.py

src/StealthIM/__init__.py

@@ -0,0 +1,15 @@
+import logging
+
+logger = logging.getLogger(__name__)
+console_handler = logging.StreamHandler()
+formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+
+logger.setLevel(logging.DEBUG)
+console_handler.setLevel(logging.DEBUG)
+console_handler.setFormatter(formatter)
+logger.addHandler(console_handler)
+
+from . import apis
+
+from .user import User
+from .server import Server

src/StealthIM/apis/__init__.py

@@ -0,0 +1,3 @@
+from . import common
+from . import user
+from . import util

src/StealthIM/apis/common.py

@@ -0,0 +1,54 @@
+import dataclasses
+from typing import Any
+
+import aiohttp
+
+
+@dataclasses.dataclass
+class Result:
+    code: int
+    msg: str
+
+@dataclasses.dataclass
+class NoValResult:
+    result: Result
+
+
+async def request(
+        url: str,
+        method: str,
+        data: dict = None,
+        body: dict = None,
+        headers: dict = None,
+        retry: int = 3
+) -> dict[str, Any]:
+    """
+    Make a request to the server.
+
+    Args:
+        url (str): The URL of the server.
+        method (str): The HTTP method to use (default is "GET").
+        data (dict): The data to send in the request (default is None).
+        body (dict): The body of the request (default is None).
+        headers (dict): The headers to include in the request (default is None).
+        retry (int): The number of times to retry the request in case of failure (default is 3).
+
+    Returns:
+        dict[str, Any]: The response data from the server.
+    Raises:
+        RuntimeError: If the server returns an internal error too many times.
+    """
+    current_retry = 0
+    while True:
+        async with aiohttp.ClientSession() as session:
+            async with session.request(method, url, data=data, headers=headers, json=body) as response:
+                if response.status != 200:
+                    raise RuntimeError(f"Request failed with status: {response.status}")
+                response_data = await response.json()
+
+                if 900 <= response_data["result"]["code"] <= 999:
+                    current_retry += 1
+                    continue
+
+                return response_data
+    raise RuntimeError(f"Request failed with code: {response_data['result']['code']}")

src/StealthIM/apis/group.py

@@ -0,0 +1,289 @@
+import dataclasses
+import enum
+
+from .common import request, Result, NoValResult
+from .. import logger
+
+
+@dataclasses.dataclass
+class GetGroupsResult:
+    groups: list[int]
+    result: Result
+
+class GroupMemberType(enum.Enum):
+    Member = 0
+    Manager = 1
+    Owner = 2
+
+@dataclasses.dataclass
+class GroupMember:
+    name: str
+    type: GroupMemberType
+
+@dataclasses.dataclass
+class GroupInfoResult:
+    members: list[GroupMember]
+    result: Result
+
+@dataclasses.dataclass
+class GroupPublicInfoResult:
+    create_at: str
+    name: str
+    result: Result
+
+@dataclasses.dataclass
+class CreateGroupResult:
+    groupid: int
+    result: Result
+
+JoinGroupResult = NoValResult
+InviteGroupResult = NoValResult
+SetMemberRoleResult = NoValResult
+KickMemberResult = NoValResult
+ChangeGroupNameResult = NoValResult
+ChangeGroupPasswordResult = NoValResult
+
+
+async def get_groups(
+        url: str,
+        session: str,
+) -> GetGroupsResult:
+    api_address = f"{url}/api/v1/group/"
+    logger.debug(f"Called API get_groups with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+
+    response_data = await request(api_address, "GET", headers=header)
+    logger.debug(f"Response data: {response_data}")
+
+    return GetGroupsResult(
+        groups=response_data["groups"],
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def get_group_info(
+        url: str,
+        session: str,
+        groupid: int,
+):
+    api_address = f"{url}/api/v1/group/{groupid}"
+    logger.debug(f"Called API get_group_info with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+
+    response_data = await request(api_address, "GET", headers=header)
+    logger.debug(f"Response data: {response_data}")
+
+    return GroupInfoResult(
+        members=[
+            GroupMember(
+                name=member["name"],
+                type=GroupMemberType(member.get("type", 0)),
+            ) for member in response_data["members"]
+        ],
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def get_group_public_info(
+        url: str,
+        session: str,
+        groupid: int,
+) -> GroupPublicInfoResult:
+    api_address = f"{url}/api/v1/group/{groupid}/public"
+    logger.debug(f"Called API get_group_public_info with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+
+    response_data = await request(api_address, "GET", headers=header)
+    logger.debug(f"Response data: {response_data}")
+
+    return GroupPublicInfoResult(
+        create_at=response_data["create_at"],
+        name=response_data["name"],
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def create_group(
+        url: str,
+        session: str,
+        group_name: str,
+) -> CreateGroupResult:
+    api_address = f"{url}/api/v1/group"
+    logger.debug(f"Called API create_group with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+    body = {
+        "name": group_name,
+    }
+
+    response_data = await request(api_address, "POST", headers=header, body=body)
+    logger.debug(f"Response data: {response_data}")
+
+    return CreateGroupResult(
+        groupid=response_data["groupid"],
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def join_group(
+        url: str,
+        session: str,
+        groupid: int,
+        password: str,
+) -> JoinGroupResult:
+    api_address = f"{url}/api/v1/group/{groupid}/join"
+    logger.debug(f"Called API join_group with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+    body = {
+        "password": password,
+    }
+
+    response_data = await request(api_address, "POST", headers=header, body=body)
+    logger.debug(f"Response data: {response_data}")
+
+    return JoinGroupResult(
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def invite_group(
+        url: str,
+        session: str,
+        groupid: int,
+        username: str,
+) -> InviteGroupResult:
+    api_address = f"{url}/api/v1/group/{groupid}/invite"
+    logger.debug(f"Called API invite_group with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+    body = {
+        "username": username,
+    }
+
+    response_data = await request(api_address, "POST", headers=header, body=body)
+    logger.debug(f"Response data: {response_data}")
+
+    return InviteGroupResult(
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def set_user_role(
+        url: str,
+        session: str,
+        groupid: int,
+        username: str,
+        user_type: GroupMemberType,
+) -> SetMemberRoleResult:
+    api_address = f"{url}/api/v1/group/{groupid}/{username}"
+    logger.debug(f"Called API set_user_type with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+    body = {
+        "type": user_type.value,
+    }
+
+    response_data = await request(api_address, "PUT", headers=header, body=body)
+    logger.debug(f"Response data: {response_data}")
+
+    return SetMemberRoleResult(
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def kick_user(
+        url: str,
+        session: str,
+        groupid: int,
+        username: str,
+) -> KickMemberResult:
+    api_address = f"{url}/api/v1/group/{groupid}/{username}"
+    logger.debug(f"Called API kick_user with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+
+    response_data = await request(api_address, "DELETE", headers=header)
+    logger.debug(f"Response data: {response_data}")
+
+    return KickMemberResult(
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def change_group_name(
+        url: str,
+        session: str,
+        groupid: int,
+        name: str,
+) -> ChangeGroupNameResult:
+    api_address = f"{url}/api/v1/group/{groupid}/name"
+    logger.debug(f"Called API change_group_name with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+    body = {
+        "name": name,
+    }
+
+    response_data = await request(api_address, "PATCH", headers=header, body=body)
+    logger.debug(f"Response data: {response_data}")
+
+    return ChangeGroupNameResult(
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )
+
+async def change_group_password(
+        url: str,
+        session: str,
+        groupid: int,
+        password: str,
+) -> ChangeGroupPasswordResult:
+    api_address = f"{url}/api/v1/group/{groupid}/password"
+    logger.debug(f"Called API change_group_password with url: {api_address}")
+    header = {
+        "Authorization": f"Bearer {session}"
+    }
+    body = {
+        "password": password,
+    }
+
+    response_data = await request(api_address, "PATCH", headers=header, body=body)
+    logger.debug(f"Response data: {response_data}")
+
+    return ChangeGroupPasswordResult(
+        result=Result(
+            code=response_data["result"]["code"],
+            msg=response_data["result"]["msg"]
+        )
+    )