cashflow_analyse/cashflow_analyser.py

import asyncio
import kanboard
from time import sleep

kb = kanboard.Client(url="add", username="add", api_token="add")

async def fetch_tasks_from_column(project_id, column_name):
    all_tasks = await kb.get_all_tasks_async(project_id=project_id, column_name=column_name)
    return all_tasks

async def create_new_task(project_id, column_name, title, description):
    task_id = await kb.create_task_async(
        project_id=project_id,
        title=title,
        description=description,
        column_name=column_name
    )
    return task_id

async def sum_costs_from_tasks(project_id, column_names):
    total_cost = 0
    for column in column_names:
        tasks = await fetch_tasks_from_column(project_id, column)
        for task in tasks:
            if "Стоимость: " in task['description']:
                try:
                    cost = float(task['description'].split("Стоимость: ")[1])
                    total_cost += cost
                except ValueError as e:
                    print(f"Failed to extract cost from task {task['id']}. Error: {e}")
    return total_cost

async def main_loop(project_id):
    while True:
        try:
            total_cost = await sum_costs_from_tasks(project_id, ["Отгрузка", "Оплата"])
            await create_new_task(project_id, "Cтатистика", "Стоимость договоров", f"Total: {total_cost}")
            sleep(1200)  # Wait for 20 minutes
        except KeyboardInterrupt:
            print("Script interrupted by user.")
            break
        except Exception as e:
            print(f"Unexpected error: {e}")


PROJECT_ID = "2"

asyncio.run(main_loop(PROJECT_ID))