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

Много чего сделал

Browse Source
This commit is contained in:
Власов Эдуард
2025-07-02 14:23:00 +03:00
parent a3fd037008
commit ca20bd8620
5 changed files with 129 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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(15):
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}"

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"

48
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__)
@@ -228,7 +229,7 @@ class Database:
except Exception as e:
logging.error(e)
def insert_tasks(self, tasks: list[dict]):
def insert_tasks(self, tasks: list[Task]):
"""Сохраняет в БД Задачи из Redmine"""
for task in tasks:
try:
@@ -244,7 +245,16 @@ class Database:
"project"
)
VALUES(?, ?, ?, ?, ?, ?, ?, ?)""",
[v for k, v in task.items()])
[
task.id,
task.subject,
task.description,
task.status,
task.priority,
task.author,
task.tracker,
task.project,
])
except sq.IntegrityError:
self.cursor.execute("""
UPDATE task
@@ -257,15 +267,33 @@ class Database:
"project" = ?
WHERE "id" = ?""",
(
task["subject"],
task["description"],
task["status"],
task["priority"],
task["author"],
task["tracker"],
task["project"],
task["id"]
task.subject,
task.description,
task.status,
task.priority,
task.author,
task.tracker,
task.project,
task.id
)
)
finally:
self.con.commit()
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)

68
track.py
View File

@@ -59,7 +59,7 @@ def main():
)
if not WORKSPACE_ID or not USER_ID:
if not os.path.exists(config.ENV_TXT):
logging.info(f"{Fore.YELLOW}Необходимые переменные не были обнаружены. Попытка получения...{Fore.RESET}")
logger.info(f"{Fore.YELLOW}Необходимые переменные не были обнаружены. Попытка получения...{Fore.RESET}")
clock.getenvs()
else:
logger.info(f'{Fore.YELLOW}Проверьте файл "{config.ENV_TXT}". Если необходимых переменных в нём нет, то просто удалите его.{Fore.RESET}')
@@ -82,34 +82,56 @@ 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})
# Обновление Тегов
logger.info(f"{Fore.YELLOW}Обновление Тегов...\n{Fore.RESET}")
logging.info(f"{Fore.YELLOW}Получение Задач с Redmine{Fore.RESET}")
issues = [red.get_issue(parse_task_id(tag.title)) for tag in tags]
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}")
logging.info(f"{Fore.YELLOW}Получение Проектов Ветро{Fore.RESET}")
vetro_projects = clock.get_vetro_projects()
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}")
logging.info(f"{Fore.YELLOW}Получение Типов деятельности{Fore.RESET}")
activity_types = red.get_time_entry_activities()
# # Обновление Проектов Ветро
# logger.info(f"{Fore.YELLOW}Получение Проектов Ветро{Fore.RESET}")
# vetro_projects = clock.get_vetro_projects()
# logger.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 vetro_projects:
# logger.info(f"{Fore.YELLOW}Сохранение Проектов Ветро{Fore.RESET}")
# db.insert_vetro_projects(vetro_projects)
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)
# if activity_types:
# logger.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: