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