321 lines
12 KiB
Python
321 lines
12 KiB
Python
import logging
|
||
import sqlite3 as sq
|
||
|
||
from colorama import Fore
|
||
|
||
from classes import Activity, ActivityType, Tag, VetroProject
|
||
from classes.classes import Task
|
||
from enums import ActivityTypes
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Database:
|
||
"""Класс для работы с SQLite базой данных приложения.
|
||
|
||
Обеспечивает сохранение и обновление данных из Clockify и Redmine.
|
||
"""
|
||
def __init__(self, db_name: str, sql_path: str):
|
||
self.con = sq.connect(db_name)
|
||
self.cursor = self.con.cursor()
|
||
|
||
try:
|
||
with open(sql_path, 'r', encoding='utf-8') as f:
|
||
sql_script = f.read()
|
||
|
||
self.cursor.executescript(sql_script)
|
||
self.con.commit()
|
||
logger.info("База данных успешно инициализирована!")
|
||
except Exception as e:
|
||
logger.error("Ошибка при инициализации БД: ", e)
|
||
self.con.rollback()
|
||
|
||
def insert_activities(self, activities: list[Activity]):
|
||
"""Сохраняет в базу данных активности Clockify"""
|
||
for act in activities:
|
||
try:
|
||
self.cursor.execute(
|
||
"""INSERT INTO activity(
|
||
id,
|
||
description,
|
||
time_spent,
|
||
date_start,
|
||
date_end,
|
||
is_tracked,
|
||
vetro_project_id,
|
||
tag_id,
|
||
task_id,
|
||
project_id,
|
||
author_id
|
||
) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
(
|
||
act.id,
|
||
act.description,
|
||
act.time_spent,
|
||
act.date_start,
|
||
act.date_end,
|
||
act.is_tracked,
|
||
act.vetro_project_id,
|
||
act.tag_id,
|
||
act.task_id,
|
||
act.project_id,
|
||
act.author_id,
|
||
)
|
||
)
|
||
except sq.IntegrityError:
|
||
self.cursor.execute(
|
||
"""UPDATE activity
|
||
SET description = ?, time_spent = ?, date_start = ?, date_end = ?, is_tracked = ?, vetro_project_id = ?, tag_id = ?, task_id = ?, project_id = ?
|
||
WHERE id = ?""",
|
||
(
|
||
act.description,
|
||
act.time_spent,
|
||
act.date_start,
|
||
act.date_end,
|
||
act.is_tracked,
|
||
act.vetro_project_id,
|
||
act.tag_id,
|
||
act.task_id,
|
||
act.project_id,
|
||
act.id,
|
||
)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
def insert_time_entries(self, activities: list[Activity]):
|
||
"""Сохраняет в базу данных трудочасы Redmine"""
|
||
for act in activities:
|
||
try:
|
||
self.cursor.execute(
|
||
"""INSERT INTO time_entry
|
||
(date, description, activity_id, time_spent, task_id, activity_type_id, author_id)
|
||
VALUES(?,?,?,?,?,?,?)""",
|
||
(
|
||
act.date_start.date(),
|
||
act.description,
|
||
act.id,
|
||
act.time_spent,
|
||
act.task_id,
|
||
ActivityTypes.DEVELOPMENT,
|
||
act.author_id
|
||
)
|
||
)
|
||
self.con.commit()
|
||
except sq.IntegrityError:
|
||
pass
|
||
|
||
|
||
def insert_tags(self, tags: list[Tag]):
|
||
"""Сохраняет в базу данных теги Clockify"""
|
||
for tag in tags:
|
||
try:
|
||
self.cursor.execute("INSERT INTO tag VALUES(?, ?)",(tag.id, tag.title))
|
||
except sq.IntegrityError:
|
||
self.cursor.execute(
|
||
"""UPDATE tag
|
||
SET title = ?
|
||
WHERE id = ?""",
|
||
(tag.title, tag.id)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
|
||
def insert_vetro_projects(self, vetro_projects: list[VetroProject]):
|
||
"""Сохраняет в базу данных Проекты (ДО) Ветро Clockify"""
|
||
if not vetro_projects:
|
||
return
|
||
for vetro_project in vetro_projects:
|
||
try:
|
||
self.cursor.execute("INSERT INTO vetro_project VALUES(?, ?)", (vetro_project.id, vetro_project.title))
|
||
except sq.IntegrityError:
|
||
self.cursor.execute("""
|
||
UPDATE vetro_project
|
||
SET title = ?
|
||
WHERE id = ?""",
|
||
(vetro_project.title, vetro_project.id)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
def insert_activity_types(self, activity_types: list[ActivityType]):
|
||
"""Сохраняет в базу данных Деятельности Redmine"""
|
||
for activity_type in activity_types:
|
||
try:
|
||
self.cursor.execute("INSERT INTO activity_type VALUES(?, ?)", (activity_type.id, activity_type.title))
|
||
except sq.IntegrityError:
|
||
self.cursor.execute("""
|
||
UPDATE activity_type
|
||
SET title = ?
|
||
WHERE id = ?""",
|
||
(activity_type.title, activity_type.id)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
def insert_user(self, **kwargs):
|
||
"""Сохраняет в базу данных пользователя"""
|
||
try:
|
||
self.cursor.execute("""INSERT INTO user(
|
||
clockify_user_id, clockify_email, clockify_username,
|
||
redmine_user_id, redmine_email, first_name, last_name, redmine_username)
|
||
VALUES(?, ?, ?, ?, ?, ?, ?, ?)""", [v for k, v in kwargs.items()])
|
||
except sq.IntegrityError:
|
||
self.cursor.execute("""
|
||
UPDATE user
|
||
SET clockify_email = ?,
|
||
clockify_username = ?,
|
||
redmine_email = ?,
|
||
first_name = ?,
|
||
last_name = ?,
|
||
redmine_username = ?
|
||
WHERE redmine_user_id = ? AND clockify_user_id = ?""",
|
||
(
|
||
kwargs["clockify_email"],
|
||
kwargs["clockify_username"],
|
||
kwargs["redmine_email"],
|
||
kwargs["first_name"],
|
||
kwargs["last_name"],
|
||
kwargs["redmine_username"],
|
||
kwargs["redmine_user_id"],
|
||
kwargs["clockify_user_id"]
|
||
)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
def insert_tasks(self, tasks: list[Task]):
|
||
"""Сохраняет в БД Задачи из Redmine"""
|
||
for task in tasks:
|
||
try:
|
||
self.cursor.execute("""
|
||
INSERT INTO task(
|
||
"id",
|
||
"subject",
|
||
"description",
|
||
"status",
|
||
"priority",
|
||
"author",
|
||
"tracker",
|
||
"project"
|
||
)
|
||
VALUES(?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
[
|
||
task.id,
|
||
task.subject,
|
||
task.description,
|
||
task.status,
|
||
task.priority,
|
||
task.author,
|
||
task.tracker,
|
||
task.project,
|
||
])
|
||
except sq.IntegrityError:
|
||
self.cursor.execute("""
|
||
UPDATE task
|
||
SET "subject" = ?,
|
||
"description" = ?,
|
||
"status" = ?,
|
||
"priority" = ?,
|
||
"author" = ?,
|
||
"tracker" = ?,
|
||
"project" = ?
|
||
WHERE "id" = ?""",
|
||
(
|
||
task.subject,
|
||
task.description,
|
||
task.status,
|
||
task.priority,
|
||
task.author,
|
||
task.tracker,
|
||
task.project,
|
||
task.id
|
||
)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
def get_user_by(self, by_field: str, value: str):
|
||
"""Получает пользователя из бд"""
|
||
try:
|
||
user = self.cursor.execute(f"SELECT * FROM user WHERE {by_field} = ?", (value,)).fetchone()
|
||
return {
|
||
"id": user[0],
|
||
"first_name": user[1],
|
||
"last_name": user[2],
|
||
"clockify_user_id": user[3],
|
||
"redmine_user_id": user[4],
|
||
"clockify_email": user[5],
|
||
"clockify_username": user[6],
|
||
"redmine_email": user[7],
|
||
"redmine_username": user[8]
|
||
}
|
||
except Exception as e:
|
||
logger.error(e)
|
||
|
||
|
||
def get_tags(self) -> list[Tag]:
|
||
"""Получает теги из бд"""
|
||
try:
|
||
return [
|
||
Tag(
|
||
id=tag[0],
|
||
title=tag[1]
|
||
)
|
||
for tag in self.cursor.execute("SELECT * FROM tag").fetchall()
|
||
]
|
||
except Exception as e:
|
||
logger.error(e)
|
||
|
||
def get_tasks(self) -> list[Task]:
|
||
"""Получает задачи из бд"""
|
||
try:
|
||
return [
|
||
Task(**{
|
||
"id": task[0],
|
||
"subject": task[1],
|
||
"description": task[2],
|
||
"status": task[3],
|
||
"priority": task[4],
|
||
"author": task[5],
|
||
"tracker": task[6],
|
||
"project": task[7]
|
||
}) for task in self.cursor.execute("SELECT * FROM task").fetchall()
|
||
]
|
||
except Exception as e:
|
||
logger.error(e)
|
||
|
||
def get_vetro_projects(self) -> list[VetroProject]:
|
||
"""Получает Проекты Ветро из бд"""
|
||
try:
|
||
return [
|
||
VetroProject(**{
|
||
"id": vetro_project[0],
|
||
"title": vetro_project[1]
|
||
}) for vetro_project in self.cursor.execute("SELECT * FROM vetro_project").fetchall()
|
||
]
|
||
except Exception as e:
|
||
logger.error(e)
|
||
|
||
def get_activity_types(self) -> list[ActivityType]:
|
||
"""Получает Виды деятельности из бд"""
|
||
try:
|
||
return [
|
||
ActivityType(**{
|
||
"id": activity_type[0],
|
||
"title": activity_type[1]
|
||
}) for activity_type in self.cursor.execute("SELECT * FROM activity_type").fetchall()
|
||
]
|
||
except Exception as e:
|
||
logger.error(e)
|
||
|
||
def has_time_entry(self, activity_id: str) -> bool:
|
||
"""Проверяет, есть ли запись о времени для активности"""
|
||
try:
|
||
result = self.cursor.execute("SELECT * FROM time_entry WHERE activity_id = ?", (activity_id,)).fetchone()
|
||
return bool(result)
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return False
|