Initial commit

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}"

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}`"