From e1e5e725c747417a320c8c94e8d86c653520c265 Mon Sep 17 00:00:00 2001 From: gitadmin Date: Sun, 21 Dec 2025 13:51:55 +0300 Subject: [PATCH] =?UTF-8?q?[=D0=92=D0=9E=D0=9B=D0=9A]=20=D0=B4=D0=BE=D0=B1?= =?UTF-8?q?=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=BE=D1=82=D0=BB?= =?UTF-8?q?=D0=B0=D0=B4=D0=BE=D1=87=D0=BD=D0=BE=D0=B3=D0=BE=20=D1=81=D0=BA?= =?UTF-8?q?=D1=80=D0=B8=D0=BF=D1=82=D0=B0=20=D0=B4=D0=BB=D1=8F=20=D0=BB?= =?UTF-8?q?=D0=BE=D0=B2=D0=BB=D0=B8=20=D0=B1=D0=B0=D0=B3=D0=B0=20=D1=81=20?= =?UTF-8?q?volk26=5Fnri?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- volk_category_testing.py | 291 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 291 insertions(+) create mode 100644 volk_category_testing.py diff --git a/volk_category_testing.py b/volk_category_testing.py new file mode 100644 index 0000000..0eedfce --- /dev/null +++ b/volk_category_testing.py @@ -0,0 +1,291 @@ +#!/usr/bin/env python3 +""" +Тестовый модуль для проверки загрузки описания категории. +Использование: python volk_category_testing.py +""" +import os +import sys +import json +import time +import traceback +import socket +import argparse +from datetime import datetime +from urllib.request import urlopen, Request +from urllib.error import URLError, HTTPError +from dotenv import load_dotenv + +# Загрузка переменных окружения +load_dotenv() + +# Параметры из .env +VOLK_CATEGORY_DESC_PREFIX = os.getenv('VOLK_CATEGORY_DESC_PREFIX') +VOLK_API_VERSION = os.getenv('VOLK_API_VERSION') + +# Лог-файл для тестирования +TEST_LOG_FILE = 'volk_category_testing.log' + +def log_message(message, category="TEST"): + """Функция для записи в лог-файл с повторными попытками при блокировке""" + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + log_entry = f"[{timestamp}] [{category}] {message}\n" + + # До 5 попыток записи с увеличением задержки + for attempt in range(5): + try: + with open(TEST_LOG_FILE, 'a', encoding='utf-8') as f: + f.write(log_entry) + return True + except (IOError, OSError) as e: + if attempt < 4: # Не последняя попытка + time.sleep(0.4) # Задержка 400ms между попытками + else: + return False + return False + +def safe_log_direct(message, category="TEST"): + """Безопасная функция для прямой записи в лог-файл без обработки исключений + Используется в критических ситуациях, когда обычный log_message может не сработать""" + try: + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + log_entry = f"[{timestamp}] [{category}] {message}\n" + # Прямая запись без контекстного менеджера для максимальной надежности + f = open(TEST_LOG_FILE, 'a', encoding='utf-8') + try: + f.write(log_entry) + f.flush() # Принудительно сбрасываем буфер + finally: + f.close() + except: + # Игнорируем все исключения - эта функция должна быть максимально безопасной + pass + +def test_category_description_load(category_id): + """Тестирует загрузку описания категории""" + log_message("=" * 80) + log_message(f"НАЧАЛО ТЕСТИРОВАНИЯ ЗАГРУЗКИ ОПИСАНИЯ КАТЕГОРИИ: {category_id}") + log_message("=" * 80) + + # Проверка наличия префикса URL + if not VOLK_CATEGORY_DESC_PREFIX: + error_msg = "Переменная окружения VOLK_CATEGORY_DESC_PREFIX не установлена" + log_message(error_msg, "ERROR") + print(f"ОШИБКА: {error_msg}") + return False + + # Формирование URL + desc_url = VOLK_CATEGORY_DESC_PREFIX + category_id + log_message(f"URL для запроса: {desc_url}") + log_message(f"Префикс URL: {VOLK_CATEGORY_DESC_PREFIX}") + log_message(f"ID категории: {category_id}") + + desc_response = None + + # Шаг 1: Запрос через urlopen + log_message("ШАГ 1: Вызов urlopen()") + try: + req = Request(desc_url) + log_message(f"Request объект создан: {req}") + safe_log_direct("Выполняется urlopen с таймаутом 30 секунд", "DEBUG") + response = urlopen(req, timeout=30) + safe_log_direct("urlopen успешно выполнен", "DEBUG") + log_message("urlopen успешно выполнен") + except (URLError, HTTPError, OSError, socket.gaierror, socket.timeout, ConnectionError) as e: + # Используем безопасное логирование для критических ошибок + safe_log_direct("ПЕРЕХВАЧЕНА СЕТЕВАЯ ОШИБКА urlopen", "ERROR") + safe_log_direct(f"Категория: {category_id}, URL: {desc_url}", "ERROR") + safe_log_direct(f"Тип ошибки: {type(e).__name__}, Сообщение: {str(e)}", "ERROR") + try: + traceback_str = traceback.format_exc() + safe_log_direct(f"Трассировка:\n{traceback_str}", "ERROR") + log_message(f"Трассировка сетевой ошибки:\n{traceback_str}", "ERROR") + except: + pass + try: + error_msg = f"Ошибка сетевого подключения при запросе описания категории {category_id} из {desc_url}: {type(e).__name__}: {e}" + log_message(error_msg, "ERROR") + except: + pass + print(f"ОШИБКА: Сетевая ошибка при запросе - {type(e).__name__}: {e}") + return False + except Exception as e: + # Перехватываем любые другие исключения, которые могут возникнуть при urlopen + safe_log_direct("ПЕРЕХВАЧЕНА НЕОЖИДАННАЯ ОШИБКА urlopen", "ERROR") + safe_log_direct(f"Категория: {category_id}, URL: {desc_url}", "ERROR") + safe_log_direct(f"Тип ошибки: {type(e).__name__}, Сообщение: {str(e)}", "ERROR") + try: + traceback_str = traceback.format_exc() + safe_log_direct(f"Трассировка:\n{traceback_str}", "ERROR") + log_message(f"Трассировка неожиданной ошибки:\n{traceback_str}", "ERROR") + except: + pass + try: + error_msg = f"Неожиданная ошибка при запросе описания категории {category_id} из {desc_url}: {type(e).__name__}: {e}" + log_message(error_msg, "ERROR") + except: + pass + print(f"ОШИБКА: Неожиданная ошибка - {type(e).__name__}: {e}") + return False + + # Шаг 2: Чтение и декодирование данных + log_message("ШАГ 2: Чтение и декодирование данных") + try: + with response: + # Явно указываем кодировку UTF-8 при декодировании + raw_data = response.read() + log_message(f"raw data read, размер: {len(raw_data)} байт") + safe_log_direct(f"Размер raw_data: {len(raw_data)} байт", "DEBUG") + + try: + decoded_data = raw_data.decode('utf-8') + log_message("data decoded (UTF-8)") + safe_log_direct("Декодирование UTF-8 успешно", "DEBUG") + except UnicodeDecodeError as e: + error_msg = f"Ошибка декодирования ответа для категории {category_id}: {e}. Попытка декодирования как latin-1." + log_message(error_msg, "WARNING") + decoded_data = raw_data.decode('latin-1') + log_message("data decoded (latin-1 fallback)") + + log_message(f"Длина декодированной строки: {len(decoded_data)} символов") + log_message(f"Первые 200 символов ответа:\n{decoded_data[:200]}") + + # Шаг 3: Парсинг JSON + log_message("ШАГ 3: Парсинг JSON") + try: + desc_response = json.loads(decoded_data) + log_message("json loaded успешно") + safe_log_direct("JSON успешно распарсен", "DEBUG") + except json.JSONDecodeError as e: + error_msg = f"Ошибка парсинга JSON для категории {category_id}: {e}. Первые 500 символов ответа: {decoded_data[:500]}" + log_message(error_msg, "ERROR") + log_message(f"Полный ответ (первые 1000 символов):\n{decoded_data[:1000]}", "ERROR") + print(f"ОШИБКА: Не удалось распарсить JSON - {e}") + return False + except Exception as e: + error_msg = f"Ошибка при чтении ответа для категории {category_id}: {type(e).__name__}: {e}" + log_message(error_msg, "ERROR") + traceback_str = traceback.format_exc() + log_message(f"Трассировка ошибки чтения:\n{traceback_str}", "ERROR") + print(f"ОШИБКА: Ошибка чтения ответа - {type(e).__name__}: {e}") + return False + + # Проверяем, что desc_response был успешно получен + if desc_response is None: + error_msg = f"Не удалось получить ответ для категории {category_id}" + log_message(error_msg, "ERROR") + print(f"ОШИБКА: {error_msg}") + return False + + log_message(f"Получен JSON ответ: {json.dumps(desc_response, ensure_ascii=False, indent=2)}") + + # Шаг 4: Проверка версии API + log_message("ШАГ 4: Проверка версии API") + api_version = desc_response.get('api_version') + log_message(f"Получена версия API: {api_version}") + log_message(f"Ожидаемая версия API: {VOLK_API_VERSION}") + + if api_version != VOLK_API_VERSION: + error_msg = f"Несовпадение версии API при загрузке описания категории {category_id}: получена версия '{api_version}', ожидается '{VOLK_API_VERSION}'." + log_message(error_msg, "ERROR") + print(f"ОШИБКА: {error_msg}") + return False + + log_message("Версия API соответствует ожидаемой") + + # Шаг 5: Проверка наличия блока data + log_message("ШАГ 5: Проверка наличия блока data") + if 'data' not in desc_response: + error_msg = f"В JSON ответе для категории {category_id} отсутствует блок 'data'." + log_message(error_msg, "ERROR") + log_message(f"Структура JSON ответа: {list(desc_response.keys())}", "ERROR") + print(f"ОШИБКА: {error_msg}") + return False + + log_message("Блок 'data' найден в ответе") + + desc_data = desc_response['data'] + log_message(f"Тип desc_data: {type(desc_data)}") + + # Шаг 6: Проверка типа desc_data + log_message("ШАГ 6: Проверка типа desc_data") + if not isinstance(desc_data, dict): + error_msg = f"Блок 'data' для категории {category_id} не является словарём (тип: {type(desc_data)})." + log_message(error_msg, "ERROR") + print(f"ОШИБКА: {error_msg}") + return False + + log_message("desc_data является словарём") + log_message(f"Ключи в desc_data: {list(desc_data.keys())}") + + # Шаг 7: Получение description + log_message("ШАГ 7: Получение description из блока data") + description = desc_data.get('description') + log_message(f"description получен, тип: {type(description)}") + + if description: + log_message(f"Длина description: {len(description)} символов") + log_message(f"Первые 500 символов description:\n{description[:500]}") + log_message("Описание успешно получено") + else: + log_message("description пустое или отсутствует (будет установлено в None)") + description = None + + # Итоговый результат + log_message("=" * 80) + log_message(f"ТЕСТИРОВАНИЕ ЗАВЕРШЕНО УСПЕШНО ДЛЯ КАТЕГОРИИ: {category_id}") + log_message("=" * 80) + + print(f"\nУСПЕХ: Описание категории {category_id} успешно загружено") + print(f"Размер описания: {len(description) if description else 0} символов") + print(f"Лог сохранен в: {TEST_LOG_FILE}") + + return True + +def main(): + """Главная функция""" + parser = argparse.ArgumentParser( + description='Тестирование загрузки описания категории из API VOLK', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Пример использования: + python volk_category_testing.py volk26_nri + """ + ) + parser.add_argument( + 'category_id', + type=str, + help='ID категории для тестирования (например: volk26_nri)' + ) + + args = parser.parse_args() + category_id = args.category_id + + # Проверка наличия необходимых переменных окружения + if not VOLK_CATEGORY_DESC_PREFIX: + print("ОШИБКА: Переменная окружения VOLK_CATEGORY_DESC_PREFIX не установлена") + print("Проверьте файл .env") + sys.exit(1) + + if not VOLK_API_VERSION: + print("ОШИБКА: Переменная окружения VOLK_API_VERSION не установлена") + print("Проверьте файл .env") + sys.exit(1) + + log_message("=" * 80) + log_message("ЗАПУСК ТЕСТИРОВАНИЯ ЗАГРУЗКИ ОПИСАНИЯ КАТЕГОРИИ") + log_message(f"Время запуска: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + log_message(f"ID категории: {category_id}") + log_message(f"Префикс URL: {VOLK_CATEGORY_DESC_PREFIX}") + log_message(f"Версия API: {VOLK_API_VERSION}") + log_message(f"Лог-файл: {TEST_LOG_FILE}") + log_message("=" * 80) + + success = test_category_description_load(category_id) + + if success: + sys.exit(0) + else: + sys.exit(1) + +if __name__ == "__main__": + main()