371 lines
15 KiB
Python
371 lines
15 KiB
Python
import logging
|
||
from importlib.util import module_from_spec, spec_from_file_location
|
||
from pathlib import Path
|
||
import sqlite3 as sq
|
||
|
||
from colorama import Fore
|
||
|
||
from classes import Activity, ActivityType, Tag, VetroProject
|
||
from classes.classes import Task
|
||
from db.constants import USER_LOOKUP_FIELDS
|
||
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 run_migrations(self, migrations_path: str):
|
||
"""Выполняет новые миграции и сохраняет время их применения."""
|
||
self.cursor.execute(
|
||
"""CREATE TABLE IF NOT EXISTS migration (
|
||
name TEXT PRIMARY KEY,
|
||
executed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
)"""
|
||
)
|
||
self.con.commit()
|
||
applied = {
|
||
row[0]
|
||
for row in self.cursor.execute("SELECT name FROM migration").fetchall()
|
||
}
|
||
for path in sorted(Path(migrations_path).glob("*.py")):
|
||
if path.name.startswith("_") or path.name in applied:
|
||
continue
|
||
try:
|
||
spec = spec_from_file_location(f"migration_{path.stem}", path)
|
||
if not spec or not spec.loader:
|
||
raise ImportError(f"Не удалось загрузить миграцию {path.name}")
|
||
module = module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
module.upgrade(self.con)
|
||
self.cursor.execute(
|
||
"INSERT INTO migration(name) VALUES(?)",
|
||
(path.name,),
|
||
)
|
||
self.con.commit()
|
||
except Exception:
|
||
self.con.rollback()
|
||
logger.exception("Не удалось выполнить миграцию %s", path.name)
|
||
raise
|
||
logger.info("Миграция %s успешно выполнена", path.name)
|
||
|
||
def insert_activities(self, activities: list[Activity]):
|
||
"""Сохраняет в базу данных активности Clockify"""
|
||
try:
|
||
for act in activities:
|
||
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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(id) DO UPDATE SET
|
||
description = excluded.description,
|
||
time_spent = excluded.time_spent,
|
||
date_start = excluded.date_start,
|
||
date_end = excluded.date_end,
|
||
is_tracked = excluded.is_tracked,
|
||
vetro_project_id = excluded.vetro_project_id,
|
||
tag_id = excluded.tag_id,
|
||
task_id = excluded.task_id,
|
||
project_id = excluded.project_id,
|
||
author_id = excluded.author_id""",
|
||
(
|
||
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.Error as e:
|
||
self.con.rollback()
|
||
logger.error(f"Не удалось сохранить активности: {e}")
|
||
else:
|
||
self.con.commit()
|
||
|
||
def insert_time_entry(
|
||
self,
|
||
activity: Activity,
|
||
source_activities: list[Activity],
|
||
redmine_time_entry_id: int,
|
||
):
|
||
"""Сохраняет запись Redmine и связывает с ней активности Clockify."""
|
||
try:
|
||
self.cursor.execute(
|
||
"""INSERT INTO time_entry
|
||
(id, date, description, time_spent, task_id, activity_type_id, author_id)
|
||
VALUES(?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(id) DO UPDATE SET
|
||
date = excluded.date,
|
||
description = excluded.description,
|
||
time_spent = excluded.time_spent,
|
||
task_id = excluded.task_id,
|
||
activity_type_id = excluded.activity_type_id,
|
||
author_id = excluded.author_id""",
|
||
(
|
||
redmine_time_entry_id,
|
||
activity.date_start.date(),
|
||
activity.description,
|
||
activity.time_spent,
|
||
activity.task_id,
|
||
ActivityTypes.DEVELOPMENT,
|
||
activity.author_id,
|
||
)
|
||
)
|
||
self.cursor.executemany(
|
||
"UPDATE activity SET time_entry_id = ? WHERE id = ?",
|
||
[
|
||
(redmine_time_entry_id, source_activity.id)
|
||
for source_activity in source_activities
|
||
],
|
||
)
|
||
except sq.IntegrityError as e:
|
||
self.con.rollback()
|
||
logger.error(f"Не удалось сохранить трудочасы: {e}")
|
||
else:
|
||
self.con.commit()
|
||
|
||
|
||
def insert_tags(self, tags: list[Tag]):
|
||
"""Сохраняет в базу данных теги Clockify"""
|
||
try:
|
||
for tag in tags:
|
||
self.cursor.execute(
|
||
"""INSERT INTO tag VALUES(?, ?)
|
||
ON CONFLICT(id) DO UPDATE SET title = excluded.title""",
|
||
(tag.id, tag.title)
|
||
)
|
||
except sq.Error as e:
|
||
self.con.rollback()
|
||
logger.error(f"Не удалось сохранить теги: {e}")
|
||
else:
|
||
self.con.commit()
|
||
|
||
|
||
def insert_vetro_projects(self, vetro_projects: list[VetroProject]):
|
||
"""Сохраняет в базу данных Проекты (ДО) Ветро Clockify"""
|
||
if not vetro_projects:
|
||
return
|
||
try:
|
||
for vetro_project in vetro_projects:
|
||
self.cursor.execute(
|
||
"""INSERT INTO vetro_project VALUES(?, ?)
|
||
ON CONFLICT(id) DO UPDATE SET title = excluded.title""",
|
||
(vetro_project.id, vetro_project.title)
|
||
)
|
||
except sq.Error as e:
|
||
self.con.rollback()
|
||
logger.error(f"Не удалось сохранить проекты Ветро: {e}")
|
||
else:
|
||
self.con.commit()
|
||
|
||
def insert_activity_types(self, activity_types: list[ActivityType]):
|
||
"""Сохраняет в базу данных Деятельности Redmine"""
|
||
try:
|
||
for activity_type in activity_types:
|
||
self.cursor.execute(
|
||
"""INSERT INTO activity_type VALUES(?, ?)
|
||
ON CONFLICT(id) DO UPDATE SET title = excluded.title""",
|
||
(activity_type.id, activity_type.title)
|
||
)
|
||
except sq.Error as e:
|
||
self.con.rollback()
|
||
logger.error(f"Не удалось сохранить типы деятельности: {e}")
|
||
else:
|
||
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(?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(clockify_user_id) DO UPDATE SET
|
||
clockify_email = excluded.clockify_email,
|
||
clockify_username = excluded.clockify_username,
|
||
redmine_email = excluded.redmine_email,
|
||
first_name = excluded.first_name,
|
||
last_name = excluded.last_name,
|
||
redmine_username = excluded.redmine_username""",
|
||
(
|
||
kwargs["clockify_user_id"],
|
||
kwargs["clockify_email"],
|
||
kwargs["clockify_username"],
|
||
kwargs["redmine_user_id"],
|
||
kwargs["redmine_email"],
|
||
kwargs["first_name"],
|
||
kwargs["last_name"],
|
||
kwargs["redmine_username"],
|
||
)
|
||
)
|
||
except sq.Error as e:
|
||
self.con.rollback()
|
||
logger.error(f"Не удалось сохранить пользователя: {e}")
|
||
else:
|
||
self.con.commit()
|
||
|
||
def insert_tasks(self, tasks: list[Task]):
|
||
"""Сохраняет в БД Задачи из Redmine"""
|
||
try:
|
||
for task in tasks:
|
||
self.cursor.execute("""
|
||
INSERT INTO task(
|
||
"id",
|
||
"subject",
|
||
"description",
|
||
"status",
|
||
"priority",
|
||
"author",
|
||
"tracker",
|
||
"project"
|
||
)
|
||
VALUES(?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(id) DO UPDATE SET
|
||
"subject" = excluded."subject",
|
||
"description" = excluded."description",
|
||
"status" = excluded."status",
|
||
"priority" = excluded."priority",
|
||
"author" = excluded."author",
|
||
"tracker" = excluded."tracker",
|
||
"project" = excluded."project" """,
|
||
[
|
||
task.id,
|
||
task.subject,
|
||
task.description,
|
||
task.status,
|
||
task.priority,
|
||
task.author,
|
||
task.tracker,
|
||
task.project,
|
||
])
|
||
except sq.Error as e:
|
||
self.con.rollback()
|
||
logger.error(f"Не удалось сохранить задачи: {e}")
|
||
else:
|
||
self.con.commit()
|
||
|
||
def get_user_by(self, by_field: str, value: str):
|
||
"""Получает пользователя из бд"""
|
||
if by_field not in USER_LOOKUP_FIELDS:
|
||
raise ValueError(f"Недопустимое поле поиска пользователя: {by_field}")
|
||
try:
|
||
user = self.cursor.execute(f"SELECT * FROM user WHERE {by_field} = ?", (value,)).fetchone()
|
||
if not user:
|
||
logger.error(f"Пользователь не найден: {by_field}={value}")
|
||
return None
|
||
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 time_entry_id FROM activity WHERE id = ? AND time_entry_id IS NOT NULL",
|
||
(activity_id,),
|
||
).fetchone()
|
||
return bool(result)
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return False
|