Всем привет, сегодня я хочу описать работу над задачей которую мы сделали в компании ради избавления от рутинных операций.
Я начинающий разработчик в команде MarkOnlineStudio, и хочу рассказать о своем опыте работы.
Дело в том что наш руководитель - Марк использует Adesk для учета финансов туда он заносит расходы, доходы и другие операции. А также у нас в компании используется Kaiten как внутренняя канбан доска, в которой мы ведем учет выполнения задач.
Для того чтобы понимать сколько времени(а значит и денег) было потрачено на клиента, каждую операцию расхода Марк вручную «разбивал» и вносил туда сумму по формуле *формула* которую он считал при помощи таблицы в Excel.
Каждый месяц Марк садился за выполнение этой задачи для более прозрачного учета времени и средств. На выполнение у него уходило от 3 до 5-ти часов рабочего времени.
Чтобы избавиться от этой рутинной операции, мы разработали простую интеграцию, которая автоматизирует процесс. Алгоритм работы скрипта следующий:
Скрипт анализирует данные из Kaiten, включая количество потраченного времени на клиента в минутах.
-
Затем скрипт:
Проверяет имя контрагента.
Получает из Adesk все операции типа "расход" по контрагенту за текущий месяц, где в статье расходов содержится слово "Зарплата".
Скрипт суммирует все операции по данному контрагенту и общее количество потраченных минут.
Далее, он делит сумму операций на общее количество минут, получая стоимость одной минуты работы.
Это значение округляется до двух знаков после запятой.
На основе полученной стоимости минуты скрипт вычисляет сумму для каждой задачи.
Производится проверка, чтобы сумма всех задач соответствовала общей сумме операций.
Остаток заносится в значение "Касса"
Разбивка операции
Также мы изучили наш Kaiten и поняли что человеческий фактор оказывает большое влияние на формирование выгрузки, поскольку при отсутствии метки(или наличие более 1-й метки) на задаче работа скрипта выполняется неправильно. А так как метки в Kaiten и проекты в Adesk у нас должны быть одинаковые то ошибок быть не должно. Для исправления этого мы подготовили телеграмм бот который проверят наличие меток у задачи, и в случае ее отсутствия, или наличия более одной метки он отправит сообщение администратору о том что этот тикет сделан неправильно. Его алгоритм тоже довольно прост:
Телеграм бот получает от администратора запрос который состоит из года и месяца например 2023-09
Подключится по API к нашему Kaiten и сделать запрос на получение первых 100 карточек
Проверить наличие меток у карточек
В случае отсутствия метки оправить сообщение в бот с текстом Нет метки
В случае наличия более 1-й метки отправить сообщение в бот
Повторить 25 раз
После того как мы получим проблемные карточки, мы исправляем их, автоматизировать это действие нельзя, поскольку только сотрудник создавший эту карточку знает к какому проекту она относится.
Зная что выгрузка из kaiten больше не имеет проблем мы можем приступать к работе с приложением.
Нам нужно просто выбрать файл выгрузки и указать месяц и год для разбивки операций.
Ради автономности этой интеграции я добавил к нему простой UI сделанный при помощи customTkinter. Программа выглядит так:
После проверки на тестовом портале Adesk мы приступили к тестированию на реальных данных, и обнаружили что все работает отлично, за исключением некоторых ошибок в отсутствующих старых проектов. На выполнение разбивки операций за 1 месяц у нас ушло 10 минут вместо 4-х часов.
Таким образом Марк сэкономил ~ 3 часа и 50 минут рабочего времени в месяц с помощью очень простой интеграции.
Вот так выглядит код UI:
import customtkinter as ctk
import tkinter.filedialog
import tkinter as tk
import main # Импортируйте ваш основной скрипт здесь
class App(ctk.CTk):
WIDTH = 800
HEIGHT = 600
def __init__(self):
super().__init__()
self.title("Adesk Operation Splitter")
self.geometry(f"{self.WIDTH}x{self.HEIGHT}")
# Левая часть интерфейса
left_frame = ctk.CTkFrame(self)
left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
self.label_file = ctk.CTkLabel(left_frame, text="Вас приветсвует Адеск Operation Splitter!\n\n\nВыберите файл CSV:")
self.label_file.pack(pady=10)
self.button_file = ctk.CTkButton(left_frame, text="Обзор...", command=self.open_file_dialog)
self.button_file.pack(pady=10)
self.entry_file = ctk.CTkEntry(left_frame)
self.entry_file.pack(pady=10)
self.label_year = ctk.CTkLabel(left_frame, text="Введите год:")
self.label_year.pack(pady=10)
self.entry_year = ctk.CTkEntry(left_frame)
self.entry_year.pack(pady=10)
self.label_month = ctk.CTkLabel(left_frame, text="Введите месяц:")
self.label_month.pack(pady=10)
self.entry_month = ctk.CTkEntry(left_frame)
self.entry_month.pack(pady=10)
self.run_button = ctk.CTkButton(left_frame, text="Запустить", command=self.run_script)
self.run_button.pack(pady=20)
# Правая часть интерфейса для вывода консоли
right_frame = ctk.CTkFrame(self)
right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)
self.console_output = tk.Text(right_frame, height=20, width=100)
self.console_output.pack(pady=10, padx=10)
def open_file_dialog(self):
file_path = tkinter.filedialog.askopenfilename(filetypes=[("CSV files", "*.csv")])
self.entry_file.delete(0, ctk.END)
self.entry_file.insert(0, file_path)
def run_script(self):
print('Running script...')
file_path = self.entry_file.get()
year = self.entry_year.get()
month = self.entry_month.get()
# Здесь вам нужно будет модифицировать функцию main в main_script для логирования в виджет Text
main.main(file_path, year, month, self.console_output) # Обновите вызов функции main
if __name__ == "__main__":
app = App()
app.mainloop()А вот так выглядит код интеграции:
import requests
import pandas as pd
from datetime import datetime
import json
import urllib.parse
import ui
def read_kaiten_data(file_path):
df = pd.read_csv(file_path, delimiter=';', encoding='utf-8')
return df
def get_adesk_operations(api_token):
api_url = f'https://api.adesk.ru/v1/transactions?api_token={api_token}'
response = requests.get(api_url)
response.raise_for_status()
return response.json()
def get_adesk_projects(api_token):
api_url = f'https://api.adesk.ru/v1/projects?api_token={api_token}'
response = requests.get(api_url)
response.raise_for_status()
return response.json().get('projects', [])
def check_project_exists(project_name, projects):
return any(project.get('name') == project_name for project in projects)
def create_project_in_adesk(name, api_token):
print('Creating project in Adesk...')
api_url = f'https://api.adesk.ru/v1/project?api_token={api_token}'
data = {'name': name}
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
response = requests.post(api_url, data=data, headers=headers)
response.raise_for_status()
print(response.json())
return response.json() # Предполагаем, что API возвращает информацию о созданном проекте
def read_kaiten_data(file_path):
df = pd.read_csv(file_path, delimiter=';', encoding='utf-8')
return df
def validate_kaiten_data(df):
for index, row in df.iterrows():
if pd.isna(row['Метки']):
print(f"Найдена строка с пустым значением 'Проект': {row}")
confirmation = input("Введите 'Y' для продолжения: ")
while confirmation.lower() != 'y':
confirmation = input("Введите 'Y' для продолжения: ")
def update_adesk_operation(operation_id, parts, api_token):
print(operation_id, 'update_adesk_operation')
api_url = f'https://api.adesk.ru/v1/transaction/{operation_id}?api_token={api_token}'
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
parts_json = json.dumps(parts)
data = {
'is_splitted': 'true',
'parts': parts_json
}
data_encoded = urllib.parse.urlencode(data)
response = requests.post(api_url, headers=headers, data=data_encoded)
return response
def main(kaiten_file_path, year, month, console_output):
print(kaiten_file_path, year, month, console_output)
api_token = “ВАШ_ТОКЕН_ТУТ”'
kaiten_data = read_kaiten_data(kaiten_file_path)
validate_kaiten_data(kaiten_data)
existing_projects = get_adesk_projects(api_token)
adesk_operations = get_adesk_operations(api_token)
selected_year = int(year)
selected_month = int(month)
kassa_project_id = next((p['id'] for p in existing_projects if p['name'] == "КАССА"), None)
for contractor_name, group in kaiten_data.groupby('Пользователь'):
total_time = group['Сумма (м)'].sum()
projects = group.groupby('Метки')['Сумма (м)'].sum()
for operation in adesk_operations['transactions']:
if operation['dateIso'].startswith(f'{selected_year}-{selected_month:02d}') and "Зарплата" in operation.get(
'category', {}).get('name', '') and operation.get('contractor', {}).get('name') == contractor_name:
print(contractor_name)
total_operation_amount = float(operation['amount'])
operation_date = operation['dateIso']
category_id = operation.get('category', {}).get('id')
contractor_id = operation.get('contractor', {}).get('id')
parts = []
for i, (project_name, project_time) in enumerate(projects.items()):
if project_name in ["Внутренняя задача", "Первый контакт"]:
project_id = kassa_project_id
else:
if not check_project_exists(project_name, existing_projects):
raise ValueError(f"Ошибка: Проект '{project_name}' не найден в Adesk.")
project_id = next((p['id'] for p in existing_projects if p['name'] == project_name), None)
project_cost = total_operation_amount * (project_time / total_time)
if i < len(projects) - 1:
project_cost = round(project_cost, 2)
else:
print(project_cost)
if project_id:
part = {
"project": project_id,
"contractor": contractor_id,
"category": category_id,
"amount": project_cost,
"related_date": operation['dateIso']
}
parts.append(part)
else:
raise ValueError(f"Ошибка: Проект '{project_name}' не найден в Adesk.")
if parts:
total_parts_sum = sum(part['amount'] for part in parts[:-1])
parts[-1]['amount'] = round(total_operation_amount - total_parts_sum, 2)
response = update_adesk_operation(operation['id'], parts, api_token)
console_output.insert(ui.tk.END,
f"Операция по \"{contractor_name}\" суммой \"{total_operation_amount}\" от \"{operation_date}\"\n")
console_output.see(ui.tk.END)
else:
print(f"Operation {operation['id']} has no parts to update.")
response = update_adesk_operation(operation['id'], parts, api_token)
print(f"Operation {operation['id']} response: {response.json()}")
else:
print(operation['dateIso'], selected_year, selected_month, (operation.get('category', {}).get('name', '')), (operation.get('contractor', {}).get('name')),contractor_name)
if __name__ == '__main__':
try:
main()
except ValueError as e:
print(e)Технически можно более серьёзно углубится в автоматизацию данного процесса, но на промежуточных этапах все таки необходим контроль. Может бот упадет, может проект будет неверно записан, и как раз сейчас будет возможность оперативно устранить ошибки.