import json from datetime import datetime from typing import Any, Dict, Tuple from channels.db import database_sync_to_async from helpdesk.models import Issue, Reciever from helpdesk.redmine import (RedmineStatus, fetch_redmine_issue, get_redmine_user_by_username, is_redmine, update_redmine_issue, upload_file, usernames_match) from telegram import Bot from tgbot.models import TelegramBot, TelegramUser from django.db.models import Q from django.utils.dateparse import parse_datetime from django.utils.timezone import make_aware, utc from django.utils.translation import gettext as _ def get_now() -> datetime: return make_aware(datetime.now()) class IssueStatus: OPEN = 1 CLOSED = 2 @database_sync_to_async def get_or_create_tg_user(user: Any, tg_user_id: int, username: str) -> str: try: tg_user, created = TelegramUser.objects.get_or_create( user=user, tg_user_id=tg_user_id, username=username, ) return str(tg_user) except Exception: filter_query = Q(tg_user_id=tg_user_id) | Q(user=user) TelegramUser.objects.filter(filter_query).delete() tg_user = TelegramUser.objects.create( user=user, tg_user_id=tg_user_id, username=username, ) return str(tg_user) @database_sync_to_async def get_tg_user_username(tg_user_id: int) -> str | None: tg_user = TelegramUser.objects.filter(tg_user_id=tg_user_id).first() user = getattr(tg_user, 'user', None) if tg_user and user: return str(tg_user.user) return None @database_sync_to_async def del_tg_user(user_id: int) -> Tuple[int, Dict[str, int]]: return TelegramUser.objects.filter(tg_user_id=user_id).delete() @database_sync_to_async def update_issue(issue_id: int, tg_user_id: int, status_id: int) -> Tuple[bool, str]: issue = Issue.objects.filter(issue_id=issue_id).first() department = getattr(issue, 'department', None) reciever = getattr(department, 'reciever', None) err_msg = _('Error on issue #%(issue_id)s update') % {'issue_id': issue_id} current_status = None current_assigned_name = None updated_on = None done_ratio = 0 tg_user = TelegramUser.objects.filter(tg_user_id=tg_user_id).first() user = getattr(tg_user, 'user', None) if not user: err = _('You are not logged in. Use /login') return (False, err) username = user.username if issue and reciever: if is_redmine(reciever): project_identifier = issue.department.configuration.get( 'redmine', None).get('identifier', None) if project_identifier: redmine_user = get_redmine_user_by_username( config=reciever.backend_configuration, username=username, project_identifier=project_identifier) if redmine_user: status = None if status_id == IssueStatus.OPEN: status = RedmineStatus.IN_PROGRESS msg = _('Opened by %(username)s at %(datetime)s.') % { 'username': user.username, 'datetime': datetime.now().strftime( "%d.%m.%Y %H:%M:%S") } elif status_id == IssueStatus.CLOSED: status = RedmineStatus.CLOSED done_ratio = 100 msg = _('Closed by %(username)s at %(datetime)s.') % { 'username': user.username, 'datetime': datetime.now().strftime( "%d.%m.%Y %H:%M:%S") } # get current redmine issue status redmine_result = fetch_redmine_issue( config=reciever.backend_configuration, id=issue_id) if redmine_result: current_status = redmine_result['issue']['status'][ 'id'] assigned_to = redmine_result['issue'].get( 'assigned_to', None) if assigned_to: current_assigned_name = assigned_to.get( 'name', None) updated_on = parse_datetime( # type: ignore redmine_result['issue']['updated_on'] ).astimezone(utc).strftime("%d.%m.%Y %H:%M:%S") # Deny close for different than open user if current_assigned_name and current_status == RedmineStatus.IN_PROGRESS: match = usernames_match(current_assigned_name, user.username) if not match[0]: err = _( 'Issue opened by %(username)s via redmine at %(updated_on)s' ) % { 'username': current_assigned_name, 'updated_on': updated_on } return (True, err) # Already closed in redmine if current_status == RedmineStatus.CLOSED: err = _( 'Issue closed by %(username)s via redmine at %(updated_on)s' ) % { 'username': current_assigned_name, 'updated_on': updated_on } return (True, err) data = { 'issue': { 'issue_id': issue_id, 'status_id': status, 'assigned_to_id': redmine_user['id'], 'done_ratio': done_ratio } } result = update_redmine_issue( reciever.backend_configuration, data, ) if result[0]: return (True, msg) else: err = _('Status code: %(status_code)s') % { 'status_code': result[1] } return (False, f'{err_msg}: {err}') err = _('Redmine user not found for this project') return (False, f'{err_msg}. {err}') err = _('Project_identifier not configured') return (False, f'{err_msg}. {err}') err = _('Update allowed for redmine issues only') return (False, f'{err_msg}. {err}') err = _('Issue or reciever error') return (False, f'{err_msg}. {err}') @database_sync_to_async def attach_file_redmine(issue_id: int, user_id: int, caption: str, data_uri: str) -> Tuple[bool, str]: config = None redmine_user = None issue = Issue.objects.filter(issue_id=issue_id).first() tg_user = TelegramUser.objects.filter(tg_user_id=user_id).first() user = getattr(tg_user, 'user', None) err_msg = _('Error on attaching file to redmine issue #%(issue_id)s') % { 'issue_id': issue_id } if not user: return (False, _('You are not logged in. Use /login')) # bad solution but how is it to attach without reference to helpdesk issue if issue: config = issue.department.reciever.backend_configuration else: reciever = Reciever.objects.filter(name__icontains='redmine').first() config = getattr(reciever, 'backend_configuration', None) if config: # Get issue project membership for access rights solution redmine_issue = fetch_redmine_issue(config=config, id=issue_id) if redmine_issue: redmine_user = get_redmine_user_by_username( config=config, username=user.username, project_identifier=redmine_issue['issue']['project']['id']) if not redmine_user: err = _('Access denied.') return (False, f'{err_msg}. {err}') filename = f'photo_{user}_{datetime.now()}'.replace(' ', '_') upload_result = upload_file( config, data_uri, filename, ) if upload_result[0]: token = json.loads(upload_result[1])['upload']['token'] user_upload_str = _('%(user)s upload a file') % {'user': user} notes = f"_{user_upload_str}_" if caption: notes += f': *{caption}*' data = { 'issue': { 'issue_id': issue_id, 'notes': notes, 'uploads': [{ 'token': token }] } } update_result = update_redmine_issue(config, data) if update_result[0]: return ( True, _('File successfully attached to an issue #%(issue_id)s') % { 'issue_id': issue_id }) return (False, f'{err_msg}. Code {update_result[1]}') err = _('Upload failed: %(result)s') % {'result': upload_result[1]} return (False, f'{err_msg}. {err}') err = _('Issue or reciever error') return (False, f'{err_msg}. {err}') def update_bot_info(app_bot: Bot, is_polling: bool = False) -> Any: bot_link = f"https://t.me/{app_bot.username}" bot, created = TelegramBot.objects.get_or_create( id=app_bot.id, username=app_bot.username, first_name=app_bot.first_name, link=bot_link, ) bot.is_polling = is_polling bot.polling_at = get_now() bot.save() return bot def update_bots_is_polling(is_polling: bool = False) -> None: TelegramBot.objects.all().update(is_polling=is_polling)