79 lines
3.2 KiB
Python
79 lines
3.2 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.CLOCKIFY_WORKSPACE_ID
|
||
USER_ID = config.CLOCKIFY_USER_ID
|
||
|
||
|
||
class RedmineAPI(RedmineConfig):
|
||
def __init__(self, token):
|
||
super().__init__()
|
||
self.token = self._check_token(token)
|
||
|
||
def track_activities(self, activities: list[Activity]) -> list[Activity]:
|
||
if len(activities) == 0:
|
||
logger.info("Нет задач для занесения трудочасов")
|
||
return
|
||
tracked = []
|
||
for activity in tqdm(activities, desc="Занесение трудочасов"):
|
||
if not activity.is_tracked:
|
||
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,
|
||
"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("Все задачи были успешно занесены в Redmine")
|
||
else:
|
||
logger.info(f"Не все задачи были занесены в Redmine. Занесено {len(tracked)}")
|
||
return tracked
|
||
|
||
def get_time_entry_activities(self) -> list[ActivityTypes]:
|
||
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 _check_token(self, token):
|
||
resp = requests.get(self.base_url + self.my_account_url, headers={"X-Redmine-Api-Key": token})
|
||
if not resp.ok:
|
||
logger.error("Invalid Token for RedmineAPI")
|
||
raise InvalidToken
|
||
resp = resp.json()['user']
|
||
logger.debug({"login": resp["login"], "email": resp["mail"], "full_name": f'{resp["firstname"]} {resp["lastname"]}'})
|
||
return token
|