e.vlasov/redmine_tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Большой рефактор. Обновление тегов, задач и типов деятельности при сохранении активностей в БД' (#11) from dev into master

Browse Source 
Reviewed-on: #11
This commit was merged in pull request #11.
This commit is contained in:
Власов Эдуард
2025-07-03 15:51:59 +03:00
parent 572c123432 f627f318a6
commit d3c0a6a8c2
8 changed files with 299 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
api/clockify.py
View File

@@ -1,5 +1,6 @@
from datetime import datetime
import logging
from typing import Union
from colorama import Fore
import requests
@@ -52,17 +53,28 @@ class ClockifyAPI(ClockifyConfig):
start = datetime.strptime(activity["timeInterval"]["start"], config.DATETIME_FORMAT)
end = datetime.strptime(activity["timeInterval"]["end"], config.DATETIME_FORMAT) if activity["timeInterval"]["end"] else None
time_worked_out_total += time_to_hour((end if end else datetime.utcnow()) - start)
tag = [tag for tag in tags if tag.id == tag_id]
if not tag:
logging.info(f"{Fore.YELLOW}В базе данных нет нужного тега {Fore.GREEN}{tag_id}{Fore.YELLOW}! Обновите БД при помощи флага -i [--init-db]{Fore.RESET}")
break
tag = self.get_tags(True, tag_id=tag_id)
if tag:
db.insert_tags([tag])
from config import red
try:
task_id = parse_task_id(tag.title)
except Exception as e:
logger.error(e)
continue
task = red.get_issue(task_id)
if task:
db.insert_tasks([task])
else:
logger.error(f"{Fore.RED} Тег не найден: {tag_id}{Fore.RESET}")
else:
tag = tag[0]
try:
task_id = parse_task_id(tag.title)
except ValueError:
logger.info(f'{activity["timeInterval"]["start"]}: {activity["description"]} - Неправильный формат тега {tag}')
continue
author = db.get_user_by("clockify_user_id", activity["userId"])
@@ -86,7 +98,7 @@ class ClockifyAPI(ClockifyConfig):
return result
def get_tags(self, is_updating=False, **kwargs) -> list[Tag]:
def get_tags(self, is_updating=False, tag_id="", **kwargs) -> Union[list[Tag], Tag]:
"""
Получает с сервера Clockify теги
name: str - optional
@@ -95,7 +107,14 @@ class ClockifyAPI(ClockifyConfig):
"""
if is_updating:
tags = requests.get(self.base_url + self.tags_url, params={"sort-column": "name", "sort-order": "descending", **kwargs}, headers={"X-API-KEY": self.token}).json()
logging.debug(f"{Fore.YELLOW}Получено тегов: {len(tags)}.{Fore.RESET}")
if tag_id:
for tag in tags:
if tag_id == tag["id"]:
return Tag(
id=tag["id"],
title=tag["name"]
)
logger.debug(f"{Fore.YELLOW}Получено тегов: {len(tags)}.{Fore.RESET}")
return [Tag(
id=tag["id"],
title=tag["name"]

34
api/redmine.py
View File

@@ -4,6 +4,7 @@ from colorama import Fore
import requests
from tqdm import tqdm
from classes.classes import Task
from config import RedmineConfig, config
from classes import Activity, ActivityType
from enums import ActivityTypes
@@ -91,10 +92,10 @@ class RedmineAPI(RedmineConfig):
"username": json["login"]
}
def get_issue(self, issue_id: int) -> dict:
def get_issue(self, issue_id: int) -> Task:
"""Получает задачу по её id с сервера Redmine"""
resp = requests.get(
self.base_url + self.issues.format(**{"id": issue_id}),
self.base_url + self.issue.format(**{"id": issue_id}),
headers={"X-Redmine-Api-Key": self.token},
timeout=None
)
@@ -102,7 +103,7 @@ class RedmineAPI(RedmineConfig):
logger.error("Ошибка RedmineAPI")
return {}
issue = resp.json()["issue"]
return {
return Task(**{
"id": issue["id"],
"subject": issue["subject"],
"description": issue["description"],
@@ -111,7 +112,32 @@ class RedmineAPI(RedmineConfig):
"author": issue["author"]["name"],
"tracker": issue["tracker"]["name"],
"project": issue["project"]["name"]
}
})
def get_issues(self) -> list[Task]:
"""Получает задачи с сервера Redmine"""
issues = []
for i in range(25):
resp = requests.get(
self.base_url + self.issues,
headers={"X-Redmine-Api-Key": self.token},
params={"limit": 100, "offset": 100 * i, "status_id": "*"},
timeout=None
)
if not resp.ok or not resp.json():
logger.error("Ошибка RedmineAPI")
return {}
issues.extend(resp.json()["issues"])
return [Task(**{
"id": issue["id"],
"subject": issue["subject"],
"description": issue["description"],
"status": issue["status"]["name"],
"priority": issue["priority"]["name"],
"author": issue["author"]["name"],
"tracker": issue["tracker"]["name"],
"project": issue["project"]["name"]
}) for issue in issues]
def _check_token(self):
user = self.get_user()

14
classes/classes.py
View File

@@ -46,3 +46,17 @@ class VetroProject:
def __str__(self):
return self.title
@dataclass
class Task:
id: int
subject: str
description: str
status: str
priority: str
author: str
tracker: str
project: str
def __str__(self):
return f"#{self.id}: {self.subject}"

10
config/__init__.py
View File

@@ -80,3 +80,13 @@ def configure_argument_parser():
arg_parser = configure_argument_parser()
config = Config()
db = Database(config.DB_NAME, config.SQL_PATH)
from api.clockify import ClockifyAPI
from api.redmine import RedmineAPI
clock = ClockifyAPI(
token=config.CLOCKIFY_TOKEN,
workspace_id=config.WORKSPACE_ID,
user_id=config.USER_ID
)
red = RedmineAPI(config.REDMINE_TOKEN)

3
config/redmine.py
View File

@@ -5,4 +5,5 @@ class RedmineConfig:
self.my_account_url = "/my/account.json"
self.datetime_format = "%Y-%m-%d"
self.time_entry_activities = "/enumerations/time_entry_activities.json"
self.issues = "/issues/{id}.json"
self.issue = "/issues/{id}.json"
self.issues = "/issues.json"

150
db/db.py
View File

@@ -4,6 +4,7 @@ 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__)
@@ -31,16 +32,7 @@ class Database:
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(
@@ -90,8 +82,6 @@ class Database:
)
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"""
@@ -194,7 +184,58 @@ class Database:
)
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:
@@ -217,55 +258,54 @@ class Database:
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
for tag in self.cursor.execute("SELECT * FROM tag").fetchall()
]
except Exception as e:
logging.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:
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()
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:
logging.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:
logging.error(e)

79
track.py
View File

@@ -1,5 +1,4 @@
import logging
import os.path
import sys
logging.basicConfig(
@@ -14,13 +13,8 @@ logging.basicConfig(
from colorama import Fore # noqa: E402
from api import ClockifyAPI, RedmineAPI # noqa: E402
from config import arg_parser, config, db # noqa: E402
from utils import format_date, str_to_date, today_start, today_end, parse_task_id # noqa: E402
WORKSPACE_ID = config.WORKSPACE_ID
USER_ID = config.USER_ID
from config import arg_parser, clock, db, red # noqa: E402
from utils import check_envs, db_update, format_date, get_start_end_dates # noqa: E402
def main():
@@ -31,40 +25,11 @@ def main():
logger = logging.getLogger(__name__)
start, end = None, None
if args.start:
start = str_to_date(args.start)
if args.end:
end = str_to_date(args.end)
if args.start is None and args.end is None:
logger.info("Парсим задачи за сегодняшний день")
start = today_start()
end = today_end()
else:
if start:
from_ = f"{start.day} {config.MONTHS.get(start.month)} {start.year} г."
start, end = get_start_end_dates(args)
if end:
to_ = f"{end.day} {config.MONTHS.get(end.month)} {end.year} г."
if start == end:
logger.info(f"{Fore.YELLOW}Парсим задачи за {from_}{Fore.RESET}")
else:
logger.info(Fore.YELLOW + "Парсим задачи" + (f" с {from_} " if start else " ") + (f"до {to_}" if end else "") + Fore.RESET)
clock = ClockifyAPI(
token=config.CLOCKIFY_TOKEN,
workspace_id=WORKSPACE_ID,
user_id=USER_ID
)
if not WORKSPACE_ID or not USER_ID:
if not os.path.exists(config.ENV_TXT):
logging.info(f"{Fore.YELLOW}Необходимые переменные не были обнаружены. Попытка получения...{Fore.RESET}")
clock.getenvs()
else:
logger.info(f'{Fore.YELLOW}Проверьте файл "{config.ENV_TXT}". Если необходимых переменных в нём нет, то просто удалите его.{Fore.RESET}')
if not check_envs():
return
red = RedmineAPI(config.REDMINE_TOKEN)
user = clock.get_user()
clock_user = {
"clockify_user_id": user["id"],
@@ -81,39 +46,7 @@ def main():
}
db.insert_user(**clock_user, **red_user)
if args.init_db:
logger.info(Fore.YELLOW + "Обновление базы данных..." + Fore.RESET)
try:
logging.info(f"{Fore.YELLOW}Получение Тегов{Fore.RESET}")
tags = clock.get_tags(is_updating=True, **{"page-size": clock.tags_count})
logging.info(f"{Fore.YELLOW}Получение Задач с Redmine{Fore.RESET}")
issues = [red.get_issue(parse_task_id(tag.title)) for tag in tags]
logging.info(f"{Fore.YELLOW}Получение Проектов Ветро{Fore.RESET}")
vetro_projects = clock.get_vetro_projects()
logging.info(f"{Fore.YELLOW}Получение Типов деятельности{Fore.RESET}")
activity_types = red.get_time_entry_activities()
if tags:
logging.info(f"{Fore.YELLOW}Сохранение Тегов{Fore.RESET}")
db.insert_tags(tags)
if issues:
logging.info(f"{Fore.YELLOW}Сохранение Задач Redmine{Fore.RESET}")
db.insert_tasks(issues)
if vetro_projects:
logging.info(f"{Fore.YELLOW}Сохранение Проектов Ветро{Fore.RESET}")
db.insert_vetro_projects(vetro_projects)
if activity_types:
logging.info(f"{Fore.YELLOW}Сохранение Типов деятелности{Fore.RESET}")
db.insert_activity_types(activity_types)
except Exception as e:
logger.error(Fore.RED + f"Ошибка при обновлении базы данных: {e}" + Fore.RESET)
else:
logger.info(Fore.GREEN + "База данных теперь актуальна." + Fore.RESET)
db_update()
activities = sorted(clock.get_activities(
**{

115
utils.py
View File

@@ -1,9 +1,16 @@
from argparse import Namespace
from datetime import datetime, timedelta
import logging
import os
import re
from typing import Optional
from colorama import Fore
from config import config
logger = logging.getLogger(__name__)
def time_to_hour(time: timedelta) -> float:
return time.total_seconds() / 3600
@@ -32,3 +39,111 @@ def format_date(date: datetime) -> datetime:
def parse_task_id(tag: str) -> int:
return int(re.search(r"^(?P<task_id>\d{5})", tag).group("task_id"))
def db_update():
from config import clock, db, red
logger.info(Fore.YELLOW + "Обновление базы данных..." + Fore.RESET)
try:
# Обновление Тегов
logger.info(f"{Fore.YELLOW}Обновление Тегов...\n{Fore.RESET}")
clock_tags = clock.get_tags(is_updating=True, **{"page-size": clock.tags_count})
db_tags = db.get_tags()
new_tags = [tag for tag in clock_tags if tag not in db_tags]
if new_tags:
try:
db.insert_tags(new_tags)
logger.info(f"{Fore.GREEN}Обновлено тегов: {len(new_tags)}{Fore.RESET}")
except Exception as e:
logger.error(f"{Fore.RED}Ошибка при сохранении тегов!{Fore.RESET}")
else:
logger.info(f"{Fore.YELLOW}Обновление тегов не требуется.\n{Fore.RESET}")
# Обновление Задач
logger.info(f"{Fore.YELLOW}Обновление Задач...\n{Fore.RESET}")
db_tasks = db.get_tasks()
red_tasks = red.get_issues()
clock_tasks = [task for task in red_tasks if task.id in [parse_task_id(tag.title) for tag in [*db_tags, *new_tags]]]
new_tasks = [task for task in clock_tasks if task not in db_tasks]
if new_tasks:
try:
db.insert_tasks(new_tasks)
logger.info(f"{Fore.GREEN}Обновлено задач: {len(new_tasks)}{Fore.RESET}")
except Exception as e:
logger.error(f"{Fore.RED}Ошибка при сохранении задач!{Fore.RESET}")
else:
logger.info(f"{Fore.YELLOW}Обновление задач не требуется\n{Fore.RESET}")
# Обновление Проектов Ветро
logger.info(f"{Fore.YELLOW}Обновление Проектов Ветро{Fore.RESET}")
clock_projects = clock.get_vetro_projects()
db_projects = db.get_vetro_projects()
new_projects = [project for project in clock_projects if project not in db_projects]
if new_projects:
try:
db.insert_vetro_projects(new_projects)
logger.info(f"{Fore.GREEN}Обновлено Проектов Ветро: {len(new_projects)}{Fore.RESET}")
except Exception as e:
logger.error(f"{Fore.RED}Ошибка при сохранении Проектов Ветро!{Fore.RESET}")
else:
logger.info(f"{Fore.YELLOW}Обновление Проектов Ветро не требуется\n{Fore.RESET}")
# Обновление Типов деятельности
logger.info(f"{Fore.YELLOW}Обновление Типов деятельности{Fore.RESET}")
red_types = red.get_time_entry_activities()
db_types = db.get_activity_types()
new_types = [type_ for type_ in red_types if type_ not in db_types]
if new_types:
try:
db.insert_activity_types(new_types)
logger.info(f"{Fore.GREEN}Обновлено Типов деятельности: {len(new_types)}{Fore.RESET}")
except Exception as e:
logger.error(f"{Fore.RED}Ошибка при сохранении Типов деятельности!{Fore.RESET}")
else:
logger.info(f"{Fore.YELLOW}Обновление Типов деятельности не требуется\n{Fore.RESET}")
except Exception as e:
logger.error(Fore.RED + f"Ошибка при обновлении базы данных: {e}" + Fore.RESET)
else:
logger.info(Fore.GREEN + "База данных теперь актуальна." + Fore.RESET)
def get_start_end_dates(args: Namespace):
if args.start:
start = str_to_date(args.start)
if args.end:
end = str_to_date(args.end)
if args.start is None and args.end is None:
logger.info("Парсим задачи за сегодняшний день")
start = today_start()
end = today_end()
else:
if start:
from_ = f"{start.day} {config.MONTHS.get(start.month)} {start.year} г."
if end:
to_ = f"{end.day} {config.MONTHS.get(end.month)} {end.year} г."
if start == end:
logger.info(f"{Fore.YELLOW}Парсим задачи за {from_}{Fore.RESET}")
else:
logger.info(Fore.YELLOW + "Парсим задачи" + (f" с {from_} " if start else " ") + (f"до {to_}" if end else "") + Fore.RESET)
return start, end
def check_envs():
from config import config, clock
if not config.WORKSPACE_ID or not config.USER_ID:
if not os.path.exists(config.ENV_TXT):
logger.info(f"{Fore.YELLOW}Необходимые переменные не были обнаружены. Попытка получения...{Fore.RESET}")
clock.getenvs()
else:
logger.info(f'{Fore.YELLOW}Проверьте файл "{config.ENV_TXT}". Если необходимых переменных в нём нет, то просто удалите его.{Fore.RESET}')
return False
return True