246 lines
7.8 KiB
Python
246 lines
7.8 KiB
Python
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')
|