292 lines
14 KiB
Python
292 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Тестовый модуль для проверки загрузки описания категории.
|
|
Использование: python volk_category_testing.py <category_id>
|
|
"""
|
|
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()
|