358 lines
17 KiB
Python
358 lines
17 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, category_number=None, total_count=None):
|
||
"""Тестирует загрузку описания категории
|
||
|
||
Args:
|
||
category_id: ID категории для тестирования
|
||
category_number: Номер категории в последовательности (опционально)
|
||
total_count: Общее количество категорий (опционально)
|
||
"""
|
||
if category_number and total_count:
|
||
header = f"НАЧАЛО ТЕСТИРОВАНИЯ [{category_number}/{total_count}] КАТЕГОРИИ: {category_id}"
|
||
else:
|
||
header = f"НАЧАЛО ТЕСТИРОВАНИЯ ЗАГРУЗКИ ОПИСАНИЯ КАТЕГОРИИ: {category_id}"
|
||
|
||
log_message("=" * 80)
|
||
log_message(header)
|
||
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
|
||
|
||
# Итоговый результат
|
||
if category_number and total_count:
|
||
footer = f"ТЕСТИРОВАНИЕ ЗАВЕРШЕНО УСПЕШНО [{category_number}/{total_count}] КАТЕГОРИИ: {category_id}"
|
||
else:
|
||
footer = f"ТЕСТИРОВАНИЕ ЗАВЕРШЕНО УСПЕШНО ДЛЯ КАТЕГОРИИ: {category_id}"
|
||
|
||
log_message("=" * 80)
|
||
log_message(footer)
|
||
log_message("=" * 80)
|
||
|
||
return True
|
||
|
||
def main():
|
||
"""Главная функция"""
|
||
parser = argparse.ArgumentParser(
|
||
description='Тестирование загрузки описания категории из API VOLK',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
Примеры использования:
|
||
python volk_category_testing.py volk26_nri
|
||
python volk_category_testing.py volk26_nri volk26_other volk26_third
|
||
"""
|
||
)
|
||
parser.add_argument(
|
||
'category_ids',
|
||
type=str,
|
||
nargs='+',
|
||
help='ID категорий для тестирования (от 1 до 10 штук, например: volk26_nri volk26_other)'
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
category_ids = args.category_ids
|
||
|
||
# Проверка количества ID
|
||
if len(category_ids) == 0:
|
||
print("ОШИБКА: Не указан ни один ID категории")
|
||
parser.print_help()
|
||
sys.exit(1)
|
||
|
||
if len(category_ids) > 10:
|
||
print(f"ОШИБКА: Указано слишком много ID категорий ({len(category_ids)}). Максимум 10.")
|
||
sys.exit(1)
|
||
|
||
# Проверка наличия необходимых переменных окружения
|
||
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"Количество категорий для обработки: {len(category_ids)}")
|
||
log_message(f"ID категорий: {', '.join(category_ids)}")
|
||
log_message(f"Префикс URL: {VOLK_CATEGORY_DESC_PREFIX}")
|
||
log_message(f"Версия API: {VOLK_API_VERSION}")
|
||
log_message(f"Лог-файл: {TEST_LOG_FILE}")
|
||
log_message("=" * 80)
|
||
|
||
# Обработка каждой категории последовательно
|
||
total_count = len(category_ids)
|
||
success_count = 0
|
||
fail_count = 0
|
||
|
||
for index, category_id in enumerate(category_ids, start=1):
|
||
log_message("")
|
||
log_message("")
|
||
log_message(f">>> ОБРАБОТКА КАТЕГОРИИ {index} из {total_count}: {category_id}")
|
||
log_message("")
|
||
|
||
print(f"\n[{index}/{total_count}] Обработка категории: {category_id}")
|
||
|
||
success = test_category_description_load(category_id, category_number=index, total_count=total_count)
|
||
|
||
if success:
|
||
success_count += 1
|
||
print(f"[{index}/{total_count}] ✓ УСПЕХ: Категория {category_id} обработана успешно")
|
||
else:
|
||
fail_count += 1
|
||
print(f"[{index}/{total_count}] ✗ ОШИБКА: Категория {category_id} не обработана")
|
||
|
||
# Итоговая статистика
|
||
log_message("")
|
||
log_message("")
|
||
log_message("=" * 80)
|
||
log_message("ИТОГОВАЯ СТАТИСТИКА")
|
||
log_message("=" * 80)
|
||
log_message(f"Всего категорий обработано: {total_count}")
|
||
log_message(f"Успешно: {success_count}")
|
||
log_message(f"С ошибками: {fail_count}")
|
||
log_message(f"Время завершения: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
log_message("=" * 80)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"ИТОГИ:")
|
||
print(f" Всего категорий: {total_count}")
|
||
print(f" Успешно: {success_count}")
|
||
print(f" С ошибками: {fail_count}")
|
||
print(f" Лог сохранен в: {TEST_LOG_FILE}")
|
||
print(f"{'='*60}")
|
||
|
||
# Возвращаем код ошибки, если были неудачи
|
||
if fail_count > 0:
|
||
sys.exit(1)
|
||
else:
|
||
sys.exit(0)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|