import csv import json import os import sys import time import requests from logging_config import setup_logging from urllib.parse import urlencode logger = setup_logging(__name__) base_dir = os.path.dirname(os.path.realpath(__file__)) asana_data_json = os.path.join(base_dir, './data/data_asana.json') users_data = os.path.join(base_dir, './data/users.mapping.asana2ya.csv') sys.stdin.reconfigure(encoding='utf-8') sys.stdout.reconfigure(encoding='utf-8') QUEUE_NAME = 'TESTIMPORT' headers = { 'Host': 'api.tracker.yandex.net', 'Authorization': 'OAuth y0_AgAEA7qkB3KjAArkXwAAAADzMlP9oR1lwMzBS2e94jHzdnII8Laxi7k', 'X-Org-ID': '7095769', 'Content-Type': 'appication/json', } def get_assignee_data(): """ Получение данных для поля assignee """ mapping = {} with open(users_data, 'r', encoding="utf8") as csv_file: reader = csv.DictReader(csv_file) for row in reader: asana_id = row['ID'] ya_login = row['ya_login'] mapping[asana_id] = ya_login return mapping assignee = get_assignee_data() def get_task_status(task): """ Получение данных о статусе задачи """ if task.get('completed') is True: status = 'resolved' completed_at = task.get('completed_at', '') else: status = 'open' completed_at = '' return status, completed_at # здесь функция для дальнейшей работы с parent def get_parent_task_ids(task, ya_imported_task_response): """ Получение соответствий между идентификаторами подзадач в Асане и Яндекс Трекере """ parent_task_ids = {} if task.get('parent') and 'gid' in task['parent']: parent_gid = task['parent']['gid'] if parent_gid in ya_imported_task_response: parent_id = ya_imported_task_response[parent_gid] parent_task_ids[task['gid']] = parent_id return parent_task_ids def transform_data(data): """ Преобразование данных из Asana в ЯндексТрекер в совместимом формате """ transformed_data = [] for task in data: status, completed_at = get_task_status(task) transformed_task = { 'gid': task['gid'], 'summary': task['name'], 'description': task['notes'], 'createdAt': task['created_at'], 'deadline': task['due_on'], 'assignee': task['assignee'], 'status': status, 'completedAt': completed_at, 'parent': task['parent'], 'subtasks': transform_data(task['subtasks']) } transformed_data.append(transformed_task) return transformed_data def create_task(task): base_url = 'https://api.tracker.yandex.net/v2/issues/_import' print(task) if task.get('assignee') and 'gid' in task['assignee']: ya_assignee = assignee.get(task['assignee']['gid'], 'dr.cyrill') else: ya_assignee = 'dr.cyrill' payload = { 'queue': QUEUE_NAME, 'summary': task['summary'], 'description': task['description'], 'createdAt': task['createdAt'], 'createdBy': 'dr.cyrill', 'deadline': task['deadline'], 'assignee': ya_assignee, 'status': task['status'], } logger.debug('Request: %s', json.dumps(payload)) response = requests.post( base_url, headers=headers, data=json.dumps(payload), params=urlencode({'parent': task['parent']}) # добавила параметр для parent в url-запрос ) if response.status_code == 201: print('Задача успешно создана в ЯндексТрекер') # здесь добавлено про родительскую задачу return response.json()['id'] else: print( 'Ошибка при создании задачи в ЯндексТрекер:', response.content, ) logger.debug('Response: %s', response.content) def create_tasks_in_tracker(data, limit=10): """ Создание задач в ЯндексТрекер """ datalen = (len(data) if limit==-1 else min(limit, len(data))) logger.info('Import started. Task count: %d', datalen) # этот датасет тоже для parent добавлен ya_imported_task_response = {} for i in range(datalen): task = data[i] task_id = create_task(task) logger.info('Created task id = %s',task_id) for subtask in task['subtasks']: logger.info('Found subtask gid = %s for id = %s',subtask['gid'], task_id) subtask_id = create_task(subtask) assign_parent_task(subtask_id, task_id) # и эта функция для родительской задачи написана def assign_parent_task(task_id, parent_id): """ Назначение родительской задачи """ base_url = f'https://api.tracker.yandex.net/v2/issues/{task_id}' headers = { 'Host': 'api.tracker.yandex.net', 'Authorization': 'OAuth y0_AgAEA7qkB3KjAArkXwAAAADzMlP9oR1lwMzBS2e94jHzdnII8Laxi7k', 'X-Org-ID': '7095769', 'Content-Type': 'application/json', } payload = { 'parent': parent_id, } logger.debug('Request: %s', json.dumps(payload)) response = requests.patch( base_url, headers=headers, data=json.dumps(payload), ) if response.status_code == 200: print('Родительская задача успешно назначена') logger.info('Task %s successfully assigned a parent',task_id) else: print('Ошибка при назначении родительской задачи:', response.content) logger.debug('Response: %s', response.content) time.sleep(1) # добавляем задержку перед следующим запросом к API # здесь я закончила добавлять про родительскую задачу с большой надеждой # пробую закрыть задачи уже в Трекере def close_tasks_for_completedAt(queue_name): base_url = 'https://api.tracker.yandex.net/v2/issues/_search' payload = { "filter": { "queue": queue_name, "status": "resolved" # Фильтр для задач cо статусом closed } } response = requests.post( base_url, headers=headers, data=json.dumps(payload) ) if response.status_code == 200: logger.debug('Response: %s', response.content) tasks = response.json() for task in tasks: task_id = task['id'] close_payload = { 'resolution': 'fixed' } close_url = f'https://api.tracker.yandex.net/v2/issues/{task_id}/transitions/close/_execute' print(close_url) print(headers) close_response = requests.post( close_url, headers=headers, data=json.dumps(close_payload) ) if close_response.status_code == 200: print(f'Задача {task_id} закрыта успешно') else: print(f'Ошибка при закрытии задачи {task_id}: {close_response.content}') else: print(f'Ошибка при получении задач для очереди {queue_name}: {response.content}') file = open(asana_data_json, "r", encoding="utf8") json_data = json.loads(file.read()) yandex_tracker_data = transform_data(json_data['data']) create_tasks_in_tracker(yandex_tracker_data, limit=3) close_tasks_for_completedAt('TESTIMPORT')