223 lines
9.3 KiB
Python
223 lines
9.3 KiB
Python
import logging
|
||
|
||
from colorama import Fore
|
||
import requests
|
||
|
||
from classes.classes import Task
|
||
from config import RedmineConfig, config
|
||
from classes import Activity, ActivityType
|
||
from exceptions import InvalidToken
|
||
from .constants import (
|
||
ACTIVITY_TYPE_BY_VETRO_PROJECT,
|
||
CLOCKIFY_ACTIVITY_COMMENT_PREFIX,
|
||
DEFAULT_ACTIVITY_TYPE,
|
||
REDMINE_ISSUES_LIMIT,
|
||
REDMINE_TIME_ENTRIES_LIMIT,
|
||
REQUEST_TIMEOUT,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class RedmineAPI(RedmineConfig):
|
||
"""API клиент для взаимодействия с Redmine.
|
||
|
||
Позволяет заносить трудочасы и получать виды деятельности.
|
||
"""
|
||
def __init__(self, token):
|
||
super().__init__()
|
||
self.token = token
|
||
self._check_token()
|
||
|
||
def track_activities(self, activities: list[Activity], on_success=None) -> list[Activity]:
|
||
"""Заносит трудочасы на сервер Redmine"""
|
||
is_beginner = config.is_beginner
|
||
if is_beginner:
|
||
logger.info(f"{Fore.YELLOW}Делим трудочасы пополам{Fore.RESET}")
|
||
tracked = []
|
||
failed_activities = []
|
||
activities_to_track = [
|
||
activity
|
||
for activity in activities
|
||
if not activity.is_tracked and activity.date_end
|
||
]
|
||
for activity in activities_to_track:
|
||
try:
|
||
if not config.is_debug and self.has_time_entry_for_activity(activity):
|
||
activity.is_tracked = True
|
||
if on_success:
|
||
on_success(activity)
|
||
tracked.append(activity)
|
||
logger.info(f"{Fore.GREEN}{activity}: уже есть в Redmine {Fore.RESET}".ljust(100) + "✅")
|
||
continue
|
||
if not config.is_debug:
|
||
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": round(activity.time_spent / (2 if is_beginner else 1), 3),
|
||
"activity_id": self.get_activity_type_id(activity),
|
||
"comments": self.get_comment(activity),
|
||
"spent_on": activity.date_start.strftime(self.datetime_format)
|
||
}
|
||
},
|
||
timeout=REQUEST_TIMEOUT,
|
||
)
|
||
if config.is_debug or resp.ok:
|
||
activity.is_tracked = True
|
||
if on_success:
|
||
on_success(activity)
|
||
tracked.append(activity)
|
||
logger.info(f"{Fore.GREEN}{activity}: успешно {Fore.RESET}".ljust(100) + "✅")
|
||
else:
|
||
try:
|
||
errors = resp.json()["errors"]
|
||
except Exception:
|
||
errors = resp.text
|
||
logger.debug(errors)
|
||
logger.error(f"{Fore.RED}{activity}: с ошибкой {Fore.RESET}".ljust(100) + "❌")
|
||
logger.error(f"{errors} {resp.status_code}")
|
||
failed_activities.append(activity)
|
||
except Exception as e:
|
||
logger.error(e)
|
||
failed_activities.append(activity)
|
||
continue
|
||
if len(activities_to_track) == len(tracked):
|
||
logger.info(Fore.GREEN + "Все задачи были успешно занесены в Redmine" + Fore.RESET)
|
||
else:
|
||
logger.info(f"{Fore.YELLOW}Не все задачи были занесены в Redmine. Занесено {len(tracked)}{Fore.RESET}")
|
||
logger.info(f"{Fore.YELLOW}Не занесены задачи:{Fore.RESET}")
|
||
for activity in failed_activities:
|
||
logger.info(f"{Fore.RED}{activity}{Fore.RESET}".ljust(60) + "❌")
|
||
return tracked
|
||
|
||
def get_comment(self, activity: Activity) -> str:
|
||
"""Формирует комментарий Redmine с внешним id активности Clockify."""
|
||
return f"{activity.description}\n{CLOCKIFY_ACTIVITY_COMMENT_PREFIX} {activity.id}"
|
||
|
||
def get_activity_type_id(self, activity: Activity) -> int:
|
||
"""Возвращает тип деятельности Redmine для активности Clockify."""
|
||
return ACTIVITY_TYPE_BY_VETRO_PROJECT.get(
|
||
activity.vetro_project_id,
|
||
DEFAULT_ACTIVITY_TYPE,
|
||
)
|
||
|
||
def has_time_entry_for_activity(self, activity: Activity) -> bool:
|
||
"""Проверяет, была ли активность Clockify уже перенесена в Redmine."""
|
||
resp = requests.get(
|
||
self.base_url + self.time_entries_url,
|
||
headers={"X-Redmine-Api-Key": self.token},
|
||
params={
|
||
"issue_id": activity.task_id,
|
||
"spent_on": activity.date_start.strftime(self.datetime_format),
|
||
"limit": REDMINE_TIME_ENTRIES_LIMIT,
|
||
},
|
||
timeout=REQUEST_TIMEOUT,
|
||
)
|
||
if not resp.ok:
|
||
logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}")
|
||
return False
|
||
marker = f"{CLOCKIFY_ACTIVITY_COMMENT_PREFIX} {activity.id}"
|
||
return any(
|
||
marker in time_entry.get("comments", "")
|
||
for time_entry in resp.json().get("time_entries", [])
|
||
)
|
||
|
||
def get_time_entry_activities(self) -> list[ActivityType]:
|
||
"""Получает виды Деятельности с сервера Redmine"""
|
||
resp = requests.get(
|
||
self.base_url + self.time_entry_activities,
|
||
headers={"X-Redmine-API-KEY": self.token},
|
||
timeout=REQUEST_TIMEOUT,
|
||
)
|
||
if not resp.ok:
|
||
logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}")
|
||
return []
|
||
data = resp.json()
|
||
logger.debug(data)
|
||
return [
|
||
ActivityType(id=act.get("id"), title=act.get("name"))
|
||
for act in data.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},
|
||
timeout=REQUEST_TIMEOUT,
|
||
)
|
||
if not resp.ok:
|
||
logger.error(f"Invalid Token for RedmineAPI: {resp.text}")
|
||
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) -> Task | None:
|
||
"""Получает задачу по её id с сервера Redmine"""
|
||
resp = requests.get(
|
||
self.base_url + self.issue.format(**{"id": issue_id}),
|
||
headers={"X-Redmine-Api-Key": self.token},
|
||
timeout=REQUEST_TIMEOUT,
|
||
)
|
||
if not resp.ok:
|
||
logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}")
|
||
return None
|
||
issue = resp.json()["issue"]
|
||
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"]
|
||
})
|
||
|
||
def get_issues(self) -> list[Task]:
|
||
"""Получает задачи с сервера Redmine"""
|
||
issues = []
|
||
offset = 0
|
||
total_count = None
|
||
while total_count is None or offset < total_count:
|
||
resp = requests.get(
|
||
self.base_url + self.issues,
|
||
headers={"X-Redmine-Api-Key": self.token},
|
||
params={"limit": REDMINE_ISSUES_LIMIT, "offset": offset, "status_id": "*"},
|
||
timeout=REQUEST_TIMEOUT,
|
||
)
|
||
if not resp.ok:
|
||
logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}")
|
||
return []
|
||
data = resp.json()
|
||
if not data:
|
||
logger.error("Ошибка RedmineAPI: пустой ответ")
|
||
return []
|
||
issues.extend(data["issues"])
|
||
total_count = data.get("total_count", len(issues))
|
||
offset += data.get("limit", REDMINE_ISSUES_LIMIT)
|
||
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()
|
||
logger.debug(user)
|