Mapping_ya_tracker/mapping.py

import json

import requests
from yandex_tracker_client import TrackerClient

client = TrackerClient(
    token='YANDEX_TRACKER_TOKEN',
    org_id='ORG_ID'
)


YANDEX_TRACKER_TOKEN = 'y0_AgAEA7qkfTFqAArTqQAAAADyduv806rRKh8MS6GOg01ZdwPE-F4ZHuQ'
ORG_ID = '35d164dcf81e48f2b6a532f4badddd33'


asana_data_json = './data/data_asana.json'


# ниже функция для случаев, когда не будет готового файла для импорта
# def get_data_from_asana():
#     """ Получение данных из Asana """

#     headers = {
#         'Authorization': 'ASANA_TOKEN',
#     }
#     response = requests.get('https://api.asana.com/projects/PROJECT_ID/tasks', headers=headers)
#     data = response.json()
#     return data


def transform_data(data):
    """ Преобразование данных из Asana в ЯндексТрекер в совместимом формате """

    transformed_data = []
    for task in data:
        transformed_task = {
            'summary': task['name'],
            'description': task['notes'],
            'createdAt': task['created_at'],
            'deadline': task['due_on'],
            'assignee': task['assignee'],
            'milestone': task['section'],
            # 'memberships': task['memberships'],
            # 'parent': task[''],
            'status': task['completed'],
        }
        transformed_data.append(transformed_task)
    return transformed_data


def create_tasks_in_tracker(data):
    """ Создание задач в ЯндексТрекер """

    base_url = 'https://api.tracker.yandex.net/v2/issues'
    headers = {
        'Authorization': 'YANDEX_TRACKER_TOKEN',
        'Content-Type': 'asana_data_json',
    }
    for task in data:
        payload = {
            'summary': task['summary'],
            'description': task['description'],
            'createdAt': task['createdAt'],
            'deadline': task['deadline'],
            'assignee': task['assignee'],
            'milestone': task['milestone'],
            # 'memberships': task['memberships'],
            # 'parent': task['parent'],
            'status': task['status'],
        }
        response = requests.post(
            base_url,
            headers=headers,
            data=json.dumps(payload),
        )
        if response.status_code == 201:
            print('Задача успешно создана в ЯндексТрекер')
        else:
            print(
                'Ошибка при создании задачи в ЯндексТрекер:',
                response.content,
            )


def main():
    """ Основная функция для переноса данных """
    # asana_data = get_data_from_asana()
    with open('asana_data_json', 'r') as file:
        asana_data = json.load(file)
            # return asana_data
    yandex_tracker_data = transform_data(asana_data)
    create_tasks_in_tracker(yandex_tracker_data)


# Запуск основной функции
main()