import time
from typing import NamedTuple, Optional
class ReturnFromPostProcessor(NamedTuple):
is_operation_successful: bool
next_operation_id: Optional[str]
local_vars: dict
global_vars: dict
def post_processor(
response, operation, user_vars, listener, user_class, *args, **kwargs
) -> ReturnFromPostProcessor:
# Инициализируем дефолтные значения на случай полной деградации user_vars
current_iter = 0
total_iter = 1
# ОПТИМИЗАЦИЯ 1: Безопасное извлечение переменных за один проход
try:
if isinstance(user_vars, dict):
iter_obj = user_vars.get('iteration')
total_obj = user_vars.get('total_iterations')
# Извлекаем значения независимо от того, объект это UserVar или чистый тип
current_iter = int(iter_obj.value if hasattr(iter_obj, 'value') else iter_obj)
total_iter = int(total_obj.value if hasattr(total_obj, 'value') else total_obj)
except (ValueError, TypeError, AttributeError, KeyError):
pass # В случае любой аномалии безопасно работаем со значениями по умолчанию
# Вычисляем флаг продолжения итераций заранее, убирая дублирование функции
remaining = current_iter < (total_iter - 1)
# Определяем успешность HTTP-запроса
is_success = False
# ОПТИМИЗАЦИЯ 2: Защита от NoneType при проверке статус-кода
if hasattr(response, "status_code") and response.status_code == 200:
is_success = True
# Логика пэйсинга (вычисляется только при успехе и наличии оставшихся итераций)
if remaining:
try:
conf = f"{user_class.environment.config=}"
position = conf.find("UC29_prelogin_AUTH_pp.py")
if position != -1:
pacing_part = conf[position:position+130].split(",")[3]
pos_quote = pacing_part.find("'")
pacing_clean = pacing_part[pos_quote+1:-1]
pos_dash = pacing_clean.find("-")
numA = float(pacing_clean[0:pos_dash])
numB = float(pacing_clean[pos_dash+1:])
pacing_time = (((numB + numA) / 2) - 15) / total_iter
if pacing_time > 0:
time.sleep(pacing_time)
except (IndexError, ValueError, AttributeError, TypeError):
pass # Если конфиг изменен или недоступен — продолжаем без задержки
# ОПТИМИЗАЦИЯ 3: Единая матрица возврата (исключает дублирование веток return)
if remaining:
return ReturnFromPostProcessor(
is_success,
"UC29_GET_contacts_WEB",
{"iteration": str(current_iter + 1)},
{}
)
else:
# Вместо математического умножения строки на 0 — сразу пишем строковый ноль "0"
return ReturnFromPostProcessor(
is_success,
None,
{"iteration": "0"},
{}
)