134 lines
5.0 KiB
Python
134 lines
5.0 KiB
Python
import logging
|
||
import sqlite3 as sq
|
||
|
||
from classes import Activity, ActivityType, Tag, VetroProject
|
||
from enums import ActivityTypes
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Database:
|
||
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]):
|
||
is_updated = True
|
||
for act in activities:
|
||
if is_updated:
|
||
tags = self.cursor.execute("SELECT * FROM tags WHERE id = ?", (act.tag_id, )).fetchall()
|
||
vetro_projects = self.cursor.execute("SELECT * FROM vetro_projects WHERE id = ?", (act.vetro_project_id, )).fetchall()
|
||
if not tags or not vetro_projects:
|
||
is_updated = False
|
||
try:
|
||
self.cursor.execute(
|
||
"INSERT INTO activities 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.project_id,
|
||
)
|
||
)
|
||
except sq.IntegrityError:
|
||
self.cursor.execute(
|
||
"""UPDATE activities
|
||
SET description = ?, time_spent = ?, date_start = ?, date_end = ?, is_tracked = ?, vetro_project_id = ?, tag_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.project_id,
|
||
act.id,
|
||
)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
if not is_updated:
|
||
logging.info("База данных устарела. Рекомендуем обновить с помощью флага -i [--init-db]")
|
||
|
||
def insert_time_entries(self, activities: list[Activity]):
|
||
for act in activities:
|
||
try:
|
||
self.cursor.execute(
|
||
"""INSERT INTO time_entry
|
||
(date, description, activity_id, time_spent, task_id, activity_type_id)
|
||
VALUES(?,?,?,?,?,?)""",
|
||
(
|
||
act.date_start.date(),
|
||
act.description,
|
||
act.id,
|
||
act.time_spent,
|
||
act.task_id,
|
||
ActivityTypes.DEVELOPMENT
|
||
)
|
||
)
|
||
self.con.commit()
|
||
except sq.IntegrityError:
|
||
pass
|
||
|
||
|
||
def insert_tags(self, tags: list[Tag]):
|
||
for tag in tags:
|
||
try:
|
||
self.cursor.execute("INSERT INTO tags VALUES(?, ?)",(tag.id,tag.title))
|
||
except sq.IntegrityError:
|
||
self.cursor.execute(
|
||
"""UPDATE tags
|
||
SET title = ?
|
||
WHERE id = ?""",
|
||
(tag.title, tag.id)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
|
||
def insert_vetro_projects(self, vetro_projects: list[VetroProject]):
|
||
for vetro_project in vetro_projects:
|
||
try:
|
||
self.cursor.execute("INSERT INTO vetro_projects VALUES(?, ?)", (vetro_project.id, vetro_project.title))
|
||
except sq.IntegrityError:
|
||
self.cursor.execute("""
|
||
UPDATE vetro_projects
|
||
SET title = ?
|
||
WHERE id = ?""",
|
||
(vetro_project.title, vetro_project.id)
|
||
)
|
||
finally:
|
||
self.con.commit()
|
||
|
||
def insert_activity_types(self, activity_types: list[ActivityType]):
|
||
for activity_type in activity_types:
|
||
try:
|
||
self.cursor.execute("INSERT INTO activity_types VALUES(?, ?)", (activity_type.id, activity_type.title))
|
||
except sq.IntegrityError:
|
||
self.cursor.execute("""
|
||
UPDATE activity_types
|
||
SET title = ?
|
||
WHERE id = ?""",
|
||
(activity_type.title, activity_type.id)
|
||
)
|
||
finally:
|
||
self.con.commit()
|