261 lines
9.9 KiB
Python
261 lines
9.9 KiB
Python
import logging
|
||
import sqlite3 as sq
|
||
|
||
from colorama import Fore
|
||
|
||
from classes import Activity, ActivityType, Tag, VetroProject
|
||
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"""
|
||
is_updated = True
|
||
for act in activities:
|
||
if is_updated:
|
||
tags = self.cursor.execute("SELECT * FROM tag WHERE id = ?", (act.tag_id, )).fetchall()
|
||
vetro_project_id = act.vetro_project_id
|
||
if vetro_project_id:
|
||
vetro_projects = self.cursor.execute("SELECT * FROM vetro_project WHERE id = ?", (vetro_project_id, )).fetchall()
|
||
is_updated = tags or vetro_projects
|
||
else:
|
||
is_updated = bool(tags)
|
||
try:
|
||
self.cursor.execute(
|
||
"INSERT INTO activity VALUES(?,?,?,?,?,?,?,?,?,?,?)",
|
||
(
|
||
act.id,
|
||
act.description,
|
||
act.author_id,
|
||
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,
|
||
)
|
||
)
|
||
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()
|
||
if not is_updated:
|
||
logging.info(Fore.YELLOW + "База данных устарела. Рекомендуем обновить с помощью флага -i [--init-db]" + Fore.RESET)
|
||
|
||
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 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:
|
||
logging.error(e)
|
||
|
||
|
||
def get_tags(self) -> list[Tag]:
|
||
"""Получает теги из бд"""
|
||
try:
|
||
tags = self.cursor.execute("SELECT * FROM tag").fetchall()
|
||
return [
|
||
Tag(
|
||
id=tag[0],
|
||
title=tag[1]
|
||
)
|
||
for tag in tags
|
||
]
|
||
except Exception as e:
|
||
logging.error(e)
|
||
|
||
def insert_tasks(self, tasks: list[dict]):
|
||
"""Сохраняет в БД Задачи из Redmine"""
|
||
for task in tasks:
|
||
try:
|
||
self.cursor.execute("""
|
||
INSERT INTO task(
|
||
"id",
|
||
"subject",
|
||
"description",
|
||
"status",
|
||
"priority",
|
||
"author",
|
||
"tracker",
|
||
"project"
|
||
)
|
||
VALUES(?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
[v for k, v in task.items()])
|
||
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()
|
||
|