119 lines
4.6 KiB
Python
119 lines
4.6 KiB
Python
import logging
|
||
|
||
from colorama import Fore
|
||
import requests
|
||
from tqdm import tqdm
|
||
|
||
from config import RedmineConfig, config
|
||
from classes import Activity, ActivityType
|
||
from enums import ActivityTypes
|
||
from exceptions import InvalidToken
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
WORKSPACE_ID = config.WORKSPACE_ID
|
||
USER_ID = config.USER_ID
|
||
|
||
|
||
class RedmineAPI(RedmineConfig):
|
||
"""API клиент для взаимодействия с Redmine.
|
||
|
||
Позволяет заносить трудочасы и получать виды деятельности.
|
||
"""
|
||
def __init__(self, token):
|
||
super().__init__()
|
||
self.token = token
|
||
self._check_token()
|
||
|
||
def track_activities(self, activities: list[Activity]) -> list[Activity]:
|
||
"""Заносит трудочасы на сервер Redmine"""
|
||
is_beginner = config.is_beginner
|
||
if is_beginner:
|
||
logging.info(f"{Fore.YELLOW}Делим трудочасы пополам{Fore.RESET}")
|
||
tracked = []
|
||
for activity in tqdm(activities, desc="Занесение трудочасов"):
|
||
if not(not activity.is_tracked and activity.date_end):
|
||
continue
|
||
try:
|
||
resp = requests.post(
|
||
self.base_url + self.time_entries_url,
|
||
headers={"X-Redmine-Api-Key": self.token},
|
||
json={
|
||
"time_entry": {
|
||
"issue_id": activity.task_id,
|
||
"hours": activity.time_spent / (2 if is_beginner else 1),
|
||
"activity_id": ActivityTypes.DEVELOPMENT,
|
||
"comments": activity.description,
|
||
"spent_on": activity.date_start.strftime(self.datetime_format)
|
||
}
|
||
}
|
||
)
|
||
if resp.ok:
|
||
activity.is_tracked = True
|
||
tracked.append(activity)
|
||
logger.debug(activity)
|
||
else:
|
||
logger.error(Fore.RED + f"{activity} Не была затрекана" + Fore.RED)
|
||
logger.error(f"{resp.json()['errors']} {resp.status_code}")
|
||
except Exception as e:
|
||
logger.error(e)
|
||
continue
|
||
if len(activities) == len(tracked):
|
||
logger.info(Fore.GREEN + "Все задачи были успешно занесены в Redmine" + Fore.RESET)
|
||
else:
|
||
logger.info(f"{Fore.YELLOW}Не все задачи были занесены в Redmine. Занесено {len(tracked)}{Fore.RESET}")
|
||
return tracked
|
||
|
||
def get_time_entry_activities(self) -> list[ActivityTypes]:
|
||
"""Получает виды Деятельности с сервера Redmine"""
|
||
resp = requests.get(self.base_url + self.time_entry_activities, headers={"X-Redmine-API-KEY": self.token})
|
||
logging.debug(resp.json())
|
||
if resp.ok:
|
||
return [
|
||
ActivityType(id=act.get("id"), title=act.get("name"))
|
||
for act in resp.json().get("time_entry_activities")
|
||
if act.get("active")
|
||
]
|
||
|
||
def get_user(self):
|
||
"""Проверяет токен и возвращает данные юзера"""
|
||
resp = requests.get(self.base_url + self.my_account_url, headers={"X-Redmine-Api-Key": self.token})
|
||
if not resp.ok:
|
||
logger.error("Invalid Token for RedmineAPI")
|
||
raise InvalidToken
|
||
json = resp.json()["user"]
|
||
return {
|
||
"id": json["id"],
|
||
"first_name": json["firstname"],
|
||
"last_name": json["lastname"],
|
||
"email": json["mail"],
|
||
"username": json["login"]
|
||
}
|
||
|
||
def get_issue(self, issue_id: int) -> dict:
|
||
"""Получает задачу по её id с сервера Redmine"""
|
||
resp = requests.get(
|
||
self.base_url + self.issues.format(**{"id": issue_id}),
|
||
headers={"X-Redmine-Api-Key": self.token},
|
||
timeout=None
|
||
)
|
||
if not resp.ok:
|
||
logger.error("Ошибка RedmineAPI")
|
||
return {}
|
||
issue = resp.json()["issue"]
|
||
return {
|
||
"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"]
|
||
}
|
||
|
||
def _check_token(self):
|
||
user = self.get_user()
|
||
logging.debug(user)
|