209 lines
7.3 KiB
Python
209 lines
7.3 KiB
Python
"""
|
|
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}`"
|