From 2bc979c9b405950fa235c669936b7b5df05f17db Mon Sep 17 00:00:00 2001 From: "Nikita Tyukalov, ASUS, Linux" Date: Sun, 19 Jul 2026 22:02:06 +0300 Subject: [PATCH] Initial commit --- .gitignore | 6 + README.md | 14 +++ ai.py | 150 ++++++++++++++++++++++ bot.py | 318 +++++++++++++++++++++++++++++++++++++++++++++++ config.py.proto | 15 +++ main.py | 27 ++++ requirements.txt | 48 +++++++ tools/builtin.py | 25 ++++ tools/slavery.py | 208 +++++++++++++++++++++++++++++++ 9 files changed, 811 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 ai.py create mode 100644 bot.py create mode 100644 config.py.proto create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 tools/builtin.py create mode 100644 tools/slavery.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0e4b1a9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__ +venv +*.swp +*.swo +*.json +config.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..2e80190 --- /dev/null +++ b/README.md @@ -0,0 +1,14 @@ +```bash +python -m venv venv +source venv/bin/activate +pip install -r requirements.txt +cp config.py.proto config.py +``` + +Отредактируйте config.py + +```bash +python main.py +``` + +Авторизуйтесь diff --git a/ai.py b/ai.py new file mode 100644 index 0000000..8b36a35 --- /dev/null +++ b/ai.py @@ -0,0 +1,150 @@ +"""This module interfaces with OpenWebUI using OpenAI-compatible /chat/completions API""" + +import asyncio +import json +import traceback +import os + +import importlib.util +import inspect +import sys +import re +from typing import Callable + +from dataclasses import dataclass +from enum import Enum + +from ollama import AsyncClient + +import config + +class Role(Enum): + SYSTEM = "system" + USER = "user" + ASSISTANT = "assistant" + +@dataclass +class Message: + role: Role + content: str + +# +# Data +# +_tools = None + +# +# Private +# +def _load_tools() -> list[Callable]: + """Returns a list of all callables in tools.*""" + tools = [] + tools_path = os.path.abspath("./tools/") + for tool_path in os.listdir(tools_path): + if not tool_path.endswith(".py"): + continue + module_name = tool_path[:-3] + file_path = os.path.join(tools_path, tool_path) + try: + spec = importlib.util.spec_from_file_location(module_name, file_path) + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + + for name, obj in inspect.getmembers(module): + if inspect.isfunction(obj) and obj.__module__ == module.__name__ and not name.startswith("_"): + tools.append(obj) + except: + traceback.print_exc() + continue + return tools + +# +# Public +# +async def chat(model: str, messages: list[Message], tools_whitelist: list[str] = [], tools_blacklist: list[str] = [], **kwargs) -> str | None: + """Chat with ollama""" + # load tools if they are not loaded + global _tools + if _tools is None: + _tools = _load_tools() + try: + client = AsyncClient( + host=config.OLLAMA_URL, + headers=config.OLLAMA_HEADERS + ) + # get tools list allowed for this chat + allowed_tools = [] + # add all whitelisted tools (if whitelist is enabled) + if tools_whitelist: + for tool in _tools: + for pattern in tools_whitelist: + if re.match(pattern, tool.__name__): + print(f"[I] Allowing tool {tool.__name__} (matched by `{pattern}`)") + allowed_tools.append(tool) + break + # add all tools if whitelist is missing + else: + allowed_tools = list(_tools) + # remove blacklisted tools if blacklist is present + if tools_blacklist: + for tool in list(allowed_tools): + for pattern in tools_blacklist: + if not pattern: + continue + if re.match(pattern, tool.__name__): + print(f"[I] Removing tool {tool.__name__} (matched by `{pattern}`)") + allowed_tools.remove(tool) + break + # convert messages to valid format + messages = list({"role": m.role.value, "content": m.content} for m in messages) + # execute until done + while True: + response = await client.chat( + model=model, + messages=messages, + tools=allowed_tools, + **kwargs + ) + if not response.message.tool_calls: + return response.message.content + messages.append(response.message) + calls = response.message.tool_calls + for call in calls: + tool_name = call.function.name + try: + print(f"[I] Calling tool `{tool_name}`...") + args = call.function.arguments + for a in args: + print(f"{a} = {args[a]}") + for t in allowed_tools: + if t.__name__ != tool_name: + continue + if inspect.iscoroutinefunction(t): + result = await t(**args) + else: + result = t(**args) + messages.append({ + "role": "tool", + "tool_name": tool_name, + "content": f"{result}" + }) + break + else: + print(f"[!] Tool `{tool_name}` is not found") + messages.append({ + "role": "tool", + "tool_name": tool_name, + "content": f"Tool `{tool_name}` does not exist" + }) + except Exception as e: + traceback.print_exc() + messages.append({ + "role": "tool", + "tool_name": tool_name, + "content": f"Exception occured: {e}" + }) + return response.message.content + except: + traceback.print_exc() + return None diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..6c6acab --- /dev/null +++ b/bot.py @@ -0,0 +1,318 @@ +"""This module implements matrix bot""" + +import asyncio +import json +import traceback +import time +import html +from datetime import datetime + +from nio import ClientConfig, AsyncClient +from nio import LoginResponse, SyncResponse, RoomMessagesResponse +from nio import MatrixRoom, RoomMessageText +from nio import RoomGetStateEventResponse, RoomSendResponse + +import config +import ai + +# +# Data +# +_client: AsyncClient = None +_client_task: asyncio.Task = None +_next_batch: str = None +_message_tasks = set() + +# +# Helpers +# +def _unix_time_to_readable_date(timestamp: int | float) -> str: + """Convert UNIX time to readable date""" + dt = datetime.fromtimestamp(timestamp) + return dt.strftime("%d.%m.%Y, %H:%M") + +async def _matrix_get_room_topic(room_id: str) -> str | None: + """Returns room topic as string""" + try: + response = await _client.room_get_state_event(room_id, "m.room.topic") + if not isinstance(response, RoomGetStateEventResponse): + print("[!] Failed to get room topic: {response}") + return None + return response.content["m.topic"]["m.text"][0]["body"].replace("
", "\n").replace("
", "\n") + except: + return None + +async def _matrix_send_text_to_room(room_id: str, text: str) -> str | None: + try: + response = await _client.room_send( + room_id=room_id, + message_type="m.room.message", + content={ + "msgtype": "m.text", + "body": str(text) + } + ) + if not isinstance(response, RoomSendResponse): + return None + return response.event_id + except: + traceback.print_exc() + return None + +async def _matrix_get_last_messages(room_id: str, n: int = 200) -> list[tuple[str, str]]: + response = await _client.room_messages( + room_id=room_id, + start=_next_batch, + direction="b", + limit=n + ) + if not isinstance(response, RoomMessagesResponse): + return [] + messages = [] + for e in response.chunk: + if not isinstance(e, RoomMessageText): + continue + if not e.body: + continue + if e.body.startswith("IGNORE"): + continue + messages.append((e.sender, e.body, _unix_time_to_readable_date(e.server_timestamp / 1000))) + return list(reversed(messages)) + +# +# Private +# +def _load_next_batch() -> str | None: + """Read state from disk""" + try: + with open(config.MATRIX_STATE_JSON, "r") as f: + j = json.load(f) + if j["access_token"] != _client.access_token: + raise RuntimeError() + return j["next_batch"] + except: + return None + +def _dump_next_batch(next_batch: str): + try: + j = { + "access_token": _client.access_token, + "next_batch": next_batch + } + with open(config.MATRIX_STATE_JSON, "w") as f: + json.dump(j, f) + return True + except: + return False + +def _get_login_data() -> dict: + """Returns data to use during login. This function decides whether to login or use existing access token.""" + try: + with open(config.MATRIX_SESSION_JSON, "r") as f: + existing_data = json.load(f) + # check if required fields are present + if not isinstance(existing_data["homeserver"], str): + raise RuntimeError() + if not isinstance(existing_data["username"], str): + raise RuntimeError() + if not isinstance(existing_data["device_id"], str): + raise RuntimeError() + if not isinstance(existing_data["access_token"], str): + raise RuntimeError() + # check if data has changed + if existing_data["homeserver"] != config.MATRIX_HOMESERVER: + raise RuntimeError() + if existing_data["username"] != config.MATRIX_USERNAME: + raise RuntimeError() + return existing_data + except: + traceback.print_exc() + return { + "homeserver": config.MATRIX_HOMESERVER, + "username": config.MATRIX_USERNAME + } + +def _write_login_data(login_response: LoginResponse) -> bool: + """Writes login data to disk. It will be used on the next start.""" + try: + j = { + "homeserver": config.MATRIX_HOMESERVER, + "username": login_response.user_id, + "device_id": login_response.device_id, + "access_token": login_response.access_token + } + with open(config.MATRIX_SESSION_JSON, "w") as f: + json.dump(j, f, indent=4) + return True + except: + traceback.print_exc() + return False + +async def _on_message_handle(room_id: str, model: str, messages: list, whitelist: list[str], blacklist: list[str]): + temp_id = await _matrix_send_text_to_room(room_id, "IGNORE Calling ollama... (btw send `CUT` to reset chat memory)") + try: + ollama_response = await ai.chat( + model, + messages, + whitelist, + blacklist + ) + await _client.room_redact( + room_id=room_id, + event_id=temp_id + ) + await _matrix_send_text_to_room(room_id, str(ollama_response)) + except Exception as e: + traceback.print_exc() + await _matrix_send_text_to_room(room_id, f"IGNORE Failed to call ollama: {e}") + + + +async def _on_message(room: MatrixRoom, event: RoomMessageText) -> None: + """This event is called when there is a new message.""" + # ignore message by ourselves + if event.sender == config.MATRIX_USERNAME: + return + # cut + if event.body.lower() == "cut": + await _matrix_send_text_to_room( + room.room_id, + "IGNORE Chat memory is erased" + ) + return + room_topic = await _matrix_get_room_topic(room.room_id) + if not room_topic: + await _matrix_send_text_to_room( + room.room_id, + "Missing room topic" + ) + return + # check if required fields are present + try: + j = json.loads(room_topic) + if "ollama" not in j: + raise RuntimeError("Missing `root[\"ollama\"]`") + if "model" not in j["ollama"]: + raise RuntimeError("Missing `root[\"ollama\"][\"model\"]`") + if "tools" not in j: + raise RuntimeError("Missing `root[\"tools\"]`") + if "whitelist" not in j["tools"]: + raise RuntimeError("Missing `root[\"tools\"][\"whitelist\"]`") + if "blacklist" not in j["tools"]: + raise RuntimeError("Missing `root[\"tools\"][\"blacklist\"]`") + if "system" not in j: + raise RuntimeError("Missing `root[\"system\"]`") + except Exception as e: + traceback.print_exc() + await _matrix_send_text_to_room( + room.room_id, + f"Exception has occured: {e}" + ) + return + # last messages + last_messages = (await _matrix_get_last_messages(room.room_id))[-20:] + for i in range(len(last_messages) - 1, -1, -1): + if last_messages[i][1].lower() == "cut": + last_messages = last_messages[i+1:] + break + # message text with timestamp + timestamp = _unix_time_to_readable_date(event.server_timestamp / 1000) + message_text = f"[Сообщение от {timestamp}]\n\n{event.body}" + # build messages history for ollama + messages = [] + messages.append(ai.Message(ai.Role.SYSTEM, j["system"])) + for m in last_messages: + role = ai.Role.SYSTEM if m[0] == config.MATRIX_USERNAME else ai.Role.USER + body = f"[Сообщение от {m[2]}]\n\n{m[1]}" + messages.append(ai.Message(role, body)) + messages.append(ai.Message(ai.Role.USER, message_text)) + # handle in background + t = asyncio.create_task(_on_message_handle( + room.room_id, + j["ollama"]["model"], + messages, + j["tools"]["whitelist"], + j["tools"]["blacklist"] + )) + _message_tasks.add(t) + t.add_done_callback(_message_tasks.discard) + +async def _sync_loop() -> None: + """This loop syncs forever.""" + global _next_batch + last_save_time = 0 + while True: + try: + response = await _client.sync(timeout=30000, since=_next_batch) + if isinstance(response, SyncResponse): + _next_batch = response.next_batch + d = time.time() - last_save_time + if d >= config.MATRIX_NEXT_BATCH_DUMP_PERIOD: + _dump_next_batch(_next_batch) + last_save_time = time.time() + else: + print(f"[!] Sync error: {response}") + await asyncio.sleep(5) + except asyncio.CancelledError: + raise + except: + traceback.print_exc() + await asyncio.sleep(1) + +# +# Public +# +async def start() -> bool: + """Start the bot using data in config.py or stored data""" + global _client, _client_task, _next_batch + # do not continue if the client already exists + if _client is not None: + return False + # get data to use for login + login_data = _get_login_data() + if "access_token" not in login_data: + print(f"[I] Authenticating as {login_data['username']}...") + login_data["password"] = input(" Password: ") + # create the client + _client = AsyncClient( + config.MATRIX_HOMESERVER, + config.MATRIX_USERNAME + ) + # add callbacks + _client.add_event_callback(_on_message, RoomMessageText) + # authenticate + if "password" in login_data: + login_response = await _client.login(login_data["password"]) + if not isinstance(login_response, LoginResponse): + print(f"[!] Failed to authenticate: {login_response}") + _client = None + return False + _write_login_data(login_response) + print("[I] Authenticated using password") + # already authenticated + else: + _client.access_token = login_data["access_token"] + _client.user_id = login_data["username"] + print("[I] Authenticated using access token") + # work + _next_batch = _load_next_batch() + _client_task = asyncio.create_task(_sync_loop()) + return True + +async def stop() -> bool: + """Stop the bot""" + global _client, _client_task + if _client is None: + return False + try: + _client_task.cancel() + await _client_task + except: + pass + _client_task = None + try: + await _client.close() + except: + traceback.print_exc() + _dump_next_batch(_next_batch) + return True diff --git a/config.py.proto b/config.py.proto new file mode 100644 index 0000000..ba91ea8 --- /dev/null +++ b/config.py.proto @@ -0,0 +1,15 @@ +"""Configuration for the application""" +MATRIX_HOMESERVER = "https://matrix.server.su" +MATRIX_USERNAME = "@bot-username:matrix.server.su" +MATRIX_SESSION_JSON = "session.json" +MATRIX_STATE_JSON = "state.json" +MATRIX_NEXT_BATCH_DUMP_PERIOD = 60 + +OLLAMA_URL = "https://ai.server.su/ollama" +OLLAMA_HEADERS = { + "Authorization": "Bearer sk-gdsfgsfdgsfd95623dchdfhdfhfd" +} + +SHEETS_SERVICE_ACCOUNT_FILE = "sheets_service_account.json" +SHEETS_URL = "https://docs.google.com/spreadsheets/d/gdfgdsfgdsfkbdigdsfgsfdgfsdgdfsgdsfgdsfzMA/edit" +SHEETS_CONFIG_SHEET_NAME = "AI" diff --git a/main.py b/main.py new file mode 100644 index 0000000..7e4081a --- /dev/null +++ b/main.py @@ -0,0 +1,27 @@ +import asyncio +import os +import traceback +import signal +from typing import Callable + +import bot +import config +import ai +from ai import Message, Role + +async def main() -> None: + """Application entry point""" + stop_event = asyncio.Event() + loop = asyncio.get_running_loop() + def shutdown_handler(): + stop_event.set() + + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, shutdown_handler) + # get available tools + await bot.start() + await stop_event.wait() + await bot.stop() + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2ab7fbb --- /dev/null +++ b/requirements.txt @@ -0,0 +1,48 @@ +aiofiles==24.1.0 +aiohappyeyeballs==2.7.1 +aiohttp==3.14.1 +aiohttp_socks==0.11.0 +aiosignal==1.4.0 +annotated-types==0.7.0 +anyio==4.14.2 +attrs==26.1.0 +certifi==2026.6.17 +cffi==2.1.0 +charset-normalizer==3.4.9 +cryptography==49.0.0 +frozenlist==1.8.0 +google-auth==2.56.0 +google-auth-oauthlib==1.4.0 +gspread==6.0.2 +gspread_asyncio==2.0.0 +h11==0.16.0 +h2==4.3.0 +hpack==4.2.0 +httpcore==1.0.9 +httpx==0.28.1 +hyperframe==6.1.0 +idna==3.18 +jsonschema==4.26.0 +jsonschema-specifications==2025.9.1 +matrix-nio==0.25.2 +multidict==6.7.1 +oauthlib==3.3.1 +ollama==0.6.2 +propcache==0.5.2 +pyasn1==0.6.4 +pyasn1_modules==0.4.2 +pycparser==3.0 +pycryptodome==3.23.0 +pydantic==2.13.4 +pydantic_core==2.46.4 +python-socks==2.8.2 +referencing==0.37.0 +requests==2.34.2 +requests-oauthlib==2.0.0 +rpds-py==2026.6.3 +StrEnum==0.4.15 +typing-inspection==0.4.2 +typing_extensions==4.16.0 +unpaddedbase64==2.1.0 +urllib3==2.7.0 +yarl==1.24.2 diff --git a/tools/builtin.py b/tools/builtin.py new file mode 100644 index 0000000..4ec18ff --- /dev/null +++ b/tools/builtin.py @@ -0,0 +1,25 @@ +""" +This module implements some builtin tools. +""" + +import asyncio +import sys +from typing import Literal + +async def builtin_network_ping(host: str) -> str: + """Ping `host` to check if it is reachable.""" + try: + param = '-n' if sys.platform.lower() == 'windows' else '-c' + cmd = ['ping', param, '1', host] + process = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await process.communicate() + if process.returncode == 0: + return f"{host} is reachable" + else: + return f"{host} is unreachable" + except Exception as e: + return f"Exception happened: {e}" diff --git a/tools/slavery.py b/tools/slavery.py new file mode 100644 index 0000000..7feaf7b --- /dev/null +++ b/tools/slavery.py @@ -0,0 +1,208 @@ +""" +This module implements work hours keeping. +""" + +import asyncio +import sys +import random +import time +from datetime import datetime + +import gspread_asyncio +from google.oauth2.service_account import Credentials + +import config + + +# +# Internal functions +# +def _unix_time_to_readable_date(timestamp: int | float) -> str: + """Convert UNIX time to readable date""" + dt = datetime.fromtimestamp(timestamp) + return dt.strftime("%d.%m.%Y, %H:%M") + +def _g_get_creds(): + creds = Credentials.from_service_account_file(config.SHEETS_SERVICE_ACCOUNT_FILE) + scoped = creds.with_scopes([ + "https://www.googleapis.com/auth/spreadsheets", + ]) + return scoped + +_g_manager = gspread_asyncio.AsyncioGspreadClientManager(_g_get_creds) + +async def _g_get_ai_config() -> dict: + client = await _g_manager.authorize() + spreadsheet = await client.open_by_url(config.SHEETS_URL) + worksheet = await spreadsheet.worksheet(config.SHEETS_CONFIG_SHEET_NAME) + values = await worksheet.batch_get(["A1:B20"]) + res = {} + for pair in values[0]: + res[pair[0]] = pair[1] + return res + +async def _g_set_ai_config(conf: dict): + client = await _g_manager.authorize() + spreadsheet = await client.open_by_url(config.SHEETS_URL) + worksheet = await spreadsheet.worksheet(config.SHEETS_CONFIG_SHEET_NAME) + values = [] + for c in conf: + values.append([c, conf[c]]) + if len(values) < 20: + values += [None, None] * (20 - len(values)) + await worksheet.batch_update([{ + "range": "A1:B20", + "values": values + }]) + +# +# Tools for ollama +# +async def slavery_create_entry() -> str: + """Create new `entry` in the database. + + Returns: + Human-readable result of the operation. + """ + # get the AI config + ai_config = await _g_get_ai_config() + # allocate entry ID + new_entry_id = ai_config["entry_free_id"] + # increment free ID + ai_config["entry_free_id"] = str(int(ai_config["entry_free_id"]) + 1) + await _g_set_ai_config(ai_config) + # return the ID + return f"Unique ID of new entry is `{new_entry_id}`. You MUST set its properties using other tools!" + +async def slavery_get_entry_data(entry_id: str) -> str: + """Get information about the entry. + + Args: + entry_id: Unique ID of the entry. + + Returns: + Human-readable text containing all entry information (or error explanation). + """ + # try to get entry_id as number + try: + entry_id = int(entry_id) + except: + return f"`entry_id` must be an integer, `{entry_id}` is invalid value" + # get the AI config + ai_config = await _g_get_ai_config() + min_entry_id = int(ai_config["entry_first_id"]) + max_entry_id = int(ai_config["entry_free_id"]) - 1 + if entry_id < min_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + if entry_id > max_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + # get description + client = await _g_manager.authorize() + spreadsheet = await client.open_by_url(config.SHEETS_URL) + worksheet = await spreadsheet.worksheet(ai_config["hours_sheet_name"]) + cells = [ + ai_config["entry_start_col"] + str(entry_id), + ai_config["entry_end_col"] + str(entry_id), + ai_config["entry_description_col"] + str(entry_id) + ] + response = await worksheet.batch_get(cells) + response = [(c if c else ["(not set)"]) for c in response] + response_text = f"Entry #`{entry_id}`" + response_text += f"\nStart: {response[0][0]}" + response_text += f"\nEnd: {response[1][0]}" + response_text += f"\n--- remaining text is description ---\n\n" + response_text += f"{response[2][0]}" + return response_text + +async def slavery_set_entry_start(entry_id: str, date_and_time: str) -> str: + """Set entry `start` date and time. + + Args: + entry_id: Unique ID of the entry. + date_and_time: Date and time in format `%d.%m.%Y, %H:%M` (format is VERY important) + + Returns: + Human-readable result of the operation. + """ + # try to get entry_id as number + try: + entry_id = int(entry_id) + except: + return f"`entry_id` must be an integer, `{entry_id}` is invalid value" + # get the AI config + ai_config = await _g_get_ai_config() + min_entry_id = int(ai_config["entry_first_id"]) + max_entry_id = int(ai_config["entry_free_id"]) - 1 + if entry_id < min_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + if entry_id > max_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + # set entry start + client = await _g_manager.authorize() + spreadsheet = await client.open_by_url(config.SHEETS_URL) + worksheet = await spreadsheet.worksheet(ai_config["hours_sheet_name"]) + cell_id = ai_config["entry_start_col"] + str(entry_id) + await worksheet.update([[date_and_time]], cell_id, raw=False) + return f"Successfully set `start` datetime to {date_and_time} for entry #{entry_id}" + +async def slavery_set_entry_end(entry_id: str, date_and_time: str) -> str: + """Set entry `end` date and time. + + Args: + entry_id: Unique ID of the entry. + date_and_time: Date and time in format `%d.%m.%Y, %H:%M` (format is VERY important) + + Returns: + Human-readable result of the operation. + """ + # try to get entry_id as number + try: + entry_id = int(entry_id) + except: + return f"`entry_id` must be an integer, `{entry_id}` is invalid value" + # get the AI config + ai_config = await _g_get_ai_config() + min_entry_id = int(ai_config["entry_first_id"]) + max_entry_id = int(ai_config["entry_free_id"]) - 1 + if entry_id < min_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + if entry_id > max_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + # set entry start + client = await _g_manager.authorize() + spreadsheet = await client.open_by_url(config.SHEETS_URL) + worksheet = await spreadsheet.worksheet(ai_config["hours_sheet_name"]) + cell_id = ai_config["entry_end_col"] + str(entry_id) + await worksheet.update([[date_and_time]], cell_id, raw=False) + return f"Successfully set `end` datetime to {date_and_time} for entry #{entry_id}" + +async def slavery_set_entry_description(entry_id: str, description: str) -> str: + """Set entry description. + + Args: + entry_id: Unique ID of the entry. + description: Entry description in Russian language + + Returns: + Human-readable result of the operation. + """ + # try to get entry_id as number + try: + entry_id = int(entry_id) + except: + return f"`entry_id` must be an integer, `{entry_id}` is invalid value" + # get the AI config + ai_config = await _g_get_ai_config() + min_entry_id = int(ai_config["entry_first_id"]) + max_entry_id = int(ai_config["entry_free_id"]) - 1 + if entry_id < min_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + if entry_id > max_entry_id: + return f"`entry_id` of `{entry_id}` is invalid" + # set entry start + client = await _g_manager.authorize() + spreadsheet = await client.open_by_url(config.SHEETS_URL) + worksheet = await spreadsheet.worksheet(ai_config["hours_sheet_name"]) + cell_id = ai_config["entry_description_col"] + str(entry_id) + await worksheet.update([[description]], cell_id, raw=True) + return f"Successfully updated description of the `{entry_id}`"