import base64 import os import re from datetime import datetime from difflib import SequenceMatcher from operator import itemgetter from typing import Any, Dict, List, Tuple import requests from django.core.cache import cache from django.utils.timezone import make_aware from rest_framework import exceptions from config.utils.datetime import get_now from .models import Issue, Reciever, Status data_acq_url = os.environ.get('DATA_ACQUISITION_URL') redmine_params = '?limit=200&offset=0' def usernames_match(u1: str, u2: str, min_ratio: float = 0.84) -> Tuple[bool, float]: def normalize_username(u: str) -> str: return u.replace('.', ' ').lower() ratio = SequenceMatcher(None, normalize_username(u1), normalize_username(u2)).quick_ratio() if ratio >= min_ratio: return (True, ratio) return (False, ratio) class RedmineStatus: NEW = 1 IN_PROGRESS = 2 CLOSED = 5 DELETED = -1 def get_redmine_user_by_username( config: Dict[str, Any], username: str, project_identifier: str, ) -> Dict[str, Any] | None: api_url = config['api-url'] url = f'{api_url}projects/{project_identifier}/memberships.json{redmine_params}' headers = {'X-Redmine-API-Key': config['api-key']} response = requests.get(url, headers=headers) if (response.status_code == 200 and response.json()): memberships = response.json()['memberships'] matched_users: List[Tuple[float, Dict[str, Any]]] = [] for i in memberships: user = i.get('user', None) if user: redmine_username = user.get('name', '') # append matched user to list to get highest match later match = usernames_match(username, redmine_username) if match[0]: matched_users.append((match[1], i['user'])) # return user with highest match if matched_users: return max(matched_users, key=itemgetter(0))[1] return None def get_user(config: Dict[str, Any], user_id: int, project_identifier: str) -> List[Dict[str, Any]]: url = f'{data_acq_url}sebn-data-acquisition/api/sebn-taskbar-manager/employee-info/' response = requests.post(url, json={'id': user_id}) if (response.status_code == 200 and response.json()): employee = response.json()[0] api_url = config['api-url'] url = f'{api_url}projects/{project_identifier}/memberships.json{redmine_params}' headers = {'X-Redmine-API-Key': config['api-key']} response = requests.get(url, json={'id': user_id}, headers=headers) if (response.status_code == 200 and response.json()): memberships = response.json()['memberships'] user = [ i for i in memberships if employee['full_name_en'] in i['user']['name'] ] return user raise exceptions.APIException( code='503', detail="Can't fetch redmine memberships.") raise exceptions.APIException(code='503', detail="Can't fetch employee.") def is_redmine(reciever: Reciever) -> int: redmine_match = re.match(r'redmine', getattr(reciever, 'name', ''), re.IGNORECASE) return True if redmine_match else False def create_redmine_issue(reciever: Reciever, data: Dict[str, Any]) -> Tuple[bool, Dict[str, Any]]: def generate_json(data: Dict[str, Any]) -> dict[str, Any]: issue = {} # { # issue: { # project_id # tracker_id # status_id # priority_id # subject # description # category_id # fixed_version_id # assigned_to_id # parent_issue_id # custom_fields # watcher_user_ids # is_private # estimated_hours # } # } category = data.get('category', 'No category') department = data.get('department', None) if not department: raise exceptions.ValidationError( "Can't create redmine issue: department is None") project_id = getattr( category, 'configuration', {}).get('redmine', {}).get( 'project_id', None) or department.configuration.get( 'redmine', {}).get('identifier', None) if not project_id: raise exceptions.ValidationError( "Can't create redmine issue: project_id is None") issue['project_id'] = project_id issue['subject'] = data['title'] issue['description'] = data['description'] issue['priority_id'] = data['priority'] issue['uploads'] = data.get('uploads', []) return {"issue": issue} api_url = reciever.backend_configuration['api-url'] url = f'{api_url}issues.json' headers = { 'Content-Type': 'application/json; charset=utf-8', 'X-Redmine-API-Key': reciever.backend_configuration['api-key'] } json_data = generate_json(data) response = requests.post(url, json=json_data, headers=headers) if (response.status_code == 201): json_body = response.json() uploads = json_data['issue']['uploads'] issue_id = int(json_body['issue']['id']) # Dont forget change tgbot.handlers.helpdesk.handlers.parse_issue regex on text change href = f'{api_url}issues/{issue_id}' priority = reciever.configuration.get('priority', {}).get('data', {}).get( str(data['priority']), '') priority_emoji = "" match data['priority']: case 1: priority_emoji = '⚪' case 2: priority_emoji = '🔵' case 3: priority_emoji = '🟡' case 4: priority_emoji = '🟠' case 5: priority_emoji = '🔴' priority_text = f'<i>{priority_emoji} {priority.lower()}</i>' if priority else '' text = f'<b>Redmine issue</b> <a href="{href}">#{issue_id}</a> {"🖼️" if uploads else ""} {priority_text}' return (True, {'issue_id': issue_id, 'text': text}) return (False, { 'status_code': response.status_code, 'text': response.text }) def fetch_redmine_issue(config: Dict[str, Any], id: int, query_params: str | None = None) -> Any: api_url = config['api-url'] url = f'{api_url}issues/{id}.json?{query_params}' headers = {'X-Redmine-API-Key': config['api-key']} response = requests.get(url, headers=headers) if (response.status_code == 200): return response.json() if (response.status_code == 404): values: Dict[str, Any] = { 'issue': { 'status': { 'id': -1, 'name': 'Deleted' }, 'closed_on': get_now().isoformat() } } return values return None def format_datetime(string: str) -> datetime: dt = string.replace("Z", "") return datetime.fromisoformat(dt) def parse_notes(journals: List[Dict[str, Any]]) -> str: notes = '' if journals: for i, val in enumerate(journals): if (val['notes'] and not val['private_notes']): note = f"{val['user']['name']} ({format_datetime(val['created_on'])}):\n {val['notes']}" notes = f'{note}\n\n{notes}' return notes def save_redmine_fields(issue: Issue, fields: Dict[str, Any]) -> Issue: issue_status_id = fields['issue']['status']['id'] cached_existed_status_ids = cache.get('status_ids', []) if not cached_existed_status_ids: cached_existed_status_ids = Status.objects.all().values( 'id', 'redmine_code', 'sysaid_code') cache.set('status_ids', cached_existed_status_ids) for s in cached_existed_status_ids: if s['redmine_code'] == issue_status_id: issue.status_id = s['id'] # type: ignore ## not cached old code # status = Status.objects.filter( # redmine_code=fields['issue']['status']['id']).first() # if status: issue.status = status if (fields['issue'].get('assigned_to')): issue.responsibility = fields['issue']['assigned_to']['name'] issue.comments = parse_notes(fields['issue'].get('journals')) if (fields['issue'].get('updated_on')): issue.update_datetime = make_aware( format_datetime(fields['issue']['updated_on'])) if (fields['issue'].get('closed_on')): issue.close_datetime = make_aware( format_datetime(fields['issue']['closed_on'])) issue.save() return issue def update_redmine_issue(config: Dict[str, Any], data: Dict[str, Any]) -> Tuple[bool, int]: api_url = config['api-url'] url = f'{api_url}issues/{data["issue"]["issue_id"]}.json' headers = {'X-Redmine-API-Key': config['api-key']} response = requests.put(url, json=data, headers=headers) if (response.status_code == 204): return (True, 204) return (False, response.status_code) def upload_file(config: Dict[str, Any], data_uri: Any, filename: str = '') -> Tuple[bool, str]: header, encoded = data_uri.split(",", 1) ext = header.split('/')[1] filename = filename or 'attachment' api_url = config['api-url'] url = f'{api_url}uploads.json?filename={filename}.{ext}' headers = { 'X-Redmine-API-Key': config['api-key'], 'Content-Type': 'application/octet-stream', } response = requests.post(url, data=base64.b64decode(encoded), headers=headers) if (response.status_code == 201): return (True, response.text) return (False, f'{response.status_code}: {response.text}')