diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..95d5f72 --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +# Настройки Telegram бота +TELEGRAM_TOKEN=ваш_токен_бота +ALLOWED_USER_IDS=123456,789101 # ID пользователей через запятую + +# Настройки Synology NAS +SYNOLOGY_HOST=192.168.1.100 # IP-адрес или домен NAS +SYNOLOGY_PORT=5000 # Порт для DSM API (обычно 5000 или 5001) +SYNOLOGY_USERNAME=admin # Имя пользователя +SYNOLOGY_PASSWORD=пароль # Пароль +SYNOLOGY_SECURE=true # Использовать HTTPS (true) или HTTP (false) +SYNOLOGY_VERIFY_SSL=false # Проверять SSL-сертификат (рекомендуется true для интернета) + +# Настройки логирования +LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR, CRITICAL diff --git a/.gitignore b/.gitignore index 8dd95cf..2622eb5 100644 --- a/.gitignore +++ b/.gitignore @@ -19,7 +19,8 @@ wheels/ *.egg-info/ .installed.cfg *.egg -.history +.history/ + # Environments .env .venv @@ -30,7 +31,8 @@ env.bak/ venv.bak/ # Logs -logs/ +logs/* +!logs/.gitkeep *.log # IDE specific files @@ -42,3 +44,14 @@ logs/ # OS specific files .DS_Store Thumbs.db + +# Тестовые файлы и скрипты отладки +test_*.py +*api_test.py +diagnose_*.py + +# Временные файлы +*.tmp +*.bak +*~ +Thumbs.db diff --git a/diagnose_api.py b/diagnose_api.py deleted file mode 100644 index 2ae4b6a..0000000 --- a/diagnose_api.py +++ /dev/null @@ -1,91 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -""" -Диагностический скрипт для определения совместимых API -""" - -import os -import sys -import logging -import argparse -from pathlib import Path - -# Добавляем родительскую директорию в sys.path -parent_dir = str(Path(__file__).resolve().parent.parent) -if parent_dir not in sys.path: - sys.path.insert(0, parent_dir) - -from src.api.api_discovery import discover_available_apis -from src.config.config import SYNOLOGY_HOST, SYNOLOGY_PORT, SYNOLOGY_SECURE - -# Настройка логгера -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) -logger = logging.getLogger(__name__) - -def main(): - """Точка входа для диагностического скрипта""" - parser = argparse.ArgumentParser(description='Synology API Diagnostic Tool') - parser.add_argument('--host', help='Synology host address', default=SYNOLOGY_HOST) - parser.add_argument('--port', type=int, help='Synology host port', default=SYNOLOGY_PORT) - parser.add_argument('--secure', action='store_true', help='Use HTTPS', default=SYNOLOGY_SECURE) - - args = parser.parse_args() - - protocol = "https" if args.secure else "http" - base_url = f"{protocol}://{args.host}:{args.port}/webapi" - - print(f"Scanning APIs at {base_url}...") - - apis = discover_available_apis(base_url) - - if not apis: - print("No APIs were discovered. Check connection parameters.") - return - - print(f"Discovered {len(apis)} APIs") - - # Анализ результатов - - # 1. Ищем API для управления питанием - print("\nPower Management APIs:") - power_apis = [name for name in apis.keys() if "power" in name.lower()] - for api in power_apis: - info = apis[api] - print(f" - {api} (v{info.get('minVersion', 1)}-{info.get('maxVersion', 1)}), path: {info.get('path', 'entry.cgi')}") - - # 2. Ищем API для информации о системе - print("\nSystem Information APIs:") - system_info_apis = [name for name in apis.keys() if ("system" in name.lower() or "dsm" in name.lower()) and "info" in name.lower()] - for api in system_info_apis: - info = apis[api] - print(f" - {api} (v{info.get('minVersion', 1)}-{info.get('maxVersion', 1)}), path: {info.get('path', 'entry.cgi')}") - - # 3. Ищем API для перезагрузки - print("\nReboot/Restart APIs:") - reboot_apis = [name for name in apis.keys() if any(word in name.lower() for word in ["restart", "reboot", "power"])] - for api in reboot_apis: - info = apis[api] - print(f" - {api} (v{info.get('minVersion', 1)}-{info.get('maxVersion', 1)}), path: {info.get('path', 'entry.cgi')}") - - print("\nRecommended API Settings:") - - if power_apis: - recommended_power_api = max(power_apis, key=lambda x: apis[x].get('maxVersion', 1)) - print(f"Power API: {recommended_power_api}, version: {apis[recommended_power_api].get('maxVersion', 1)}") - else: - print("Power API: Not found, falling back to SYNO.Core.System") - - if system_info_apis: - recommended_info_api = max(system_info_apis, key=lambda x: apis[x].get('maxVersion', 1)) - print(f"System Info API: {recommended_info_api}, version: {apis[recommended_info_api].get('maxVersion', 1)}") - else: - print("System Info API: Not found, falling back to SYNO.DSM.Info") - - print("\nThese settings should be added to your .env file.") - -if __name__ == "__main__": - main() diff --git a/direct_api_test.py b/direct_api_test.py deleted file mode 100644 index ca9c245..0000000 --- a/direct_api_test.py +++ /dev/null @@ -1,293 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -""" -Тестовый скрипт для прямого доступа к API Synology для получения информации о системе. -Используется для отладки и определения совместимых API. -""" - -import requests -import logging -import json -import sys -import os -import urllib3 -from requests.adapters import HTTPAdapter -from urllib3.util import Retry - -# Добавляем корневой каталог в путь для импорта -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from src.config.config import ( - SYNOLOGY_HOST, - SYNOLOGY_PORT, - SYNOLOGY_USERNAME, - SYNOLOGY_PASSWORD, - SYNOLOGY_SECURE -) - -# Отключение предупреждений о небезопасных SSL-соединениях -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - -# Настройка логирования -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) -logger = logging.getLogger(__name__) - -def direct_api_test(): - """Прямой тест API без использования классов для определения проблемы""" - # Создаем базовую сессию - session = requests.Session() - session.verify = False # Отключаем проверку SSL - - # Добавляем повторные попытки для HTTP-запросов - retry_strategy = Retry( - total=3, - status_forcelist=[429, 500, 502, 503, 504], - allowed_methods=["GET", "POST"], - backoff_factor=1.0 - ) - adapter = HTTPAdapter(max_retries=retry_strategy) - session.mount("http://", adapter) - session.mount("https://", adapter) - - # Формируем базовый URL - protocol = "https" if SYNOLOGY_SECURE else "http" - base_url = f"{protocol}://{SYNOLOGY_HOST}:{SYNOLOGY_PORT}/webapi" - logger.info(f"Тестирование прямого API доступа к {base_url}") - - # Шаг 1: Авторизация - logger.info("Шаг 1: Попытка авторизации...") - - # Сначала получаем информацию об API авторизации - api_info_url = f"{base_url}/entry.cgi" - api_info_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": "SYNO.API.Auth" - } - - try: - auth_info_response = session.get(api_info_url, params=api_info_params, timeout=10) - auth_info_data = auth_info_response.json() - - if auth_info_data.get("success"): - auth_info = auth_info_data.get("data", {}).get("SYNO.API.Auth", {}) - auth_path = auth_info.get("path", "auth.cgi") - auth_max_version = auth_info.get("maxVersion", 6) - - logger.info(f"API авторизации: путь={auth_path}, макс. версия={auth_max_version}") - - # Пробуем версию 6 или максимальную доступную - auth_version = min(6, auth_max_version) - - # Выполняем авторизацию - auth_url = f"{base_url}/{auth_path}" - auth_params = { - "api": "SYNO.API.Auth", - "version": str(auth_version), - "method": "login", - "account": SYNOLOGY_USERNAME, - "passwd": SYNOLOGY_PASSWORD, - "session": "DirectApiTest", - "format": "cookie" - } - - # Для версии 6+ используем немного другой формат - if auth_version >= 6: - auth_params["enable_syno_token"] = "yes" - - logger.info(f"Авторизация с использованием SYNO.API.Auth v{auth_version}") - auth_response = session.get(auth_url, params=auth_params, timeout=10) - auth_data = auth_response.json() - - if auth_data.get("success"): - sid = auth_data.get("data", {}).get("sid") - logger.info(f"Авторизация успешна! SID: {sid[:10]}...") - - # Шаг 2: Тестирование различных API для получения информации о системе - logger.info("Шаг 2: Тестирование различных API для получения информации о системе") - - # Создаем список API для тестирования - api_to_test = [ - {"name": "SYNO.DSM.Info", "method": "getinfo", "version": 1}, - {"name": "SYNO.DSM.Info", "method": "getinfo", "version": 2}, - {"name": "SYNO.Core.System", "method": "info", "version": 1}, - {"name": "SYNO.Core.System", "method": "info", "version": 2}, - {"name": "SYNO.Core.System.Status", "method": "get", "version": 1}, - {"name": "SYNO.Core.System.Status", "method": "get", "version": 2}, - {"name": "SYNO.Core.System.Utilization", "method": "get", "version": 1}, - {"name": "SYNO.Core.CurrentConnection", "method": "list", "version": 1} - ] - - # Перебираем все API и тестируем их - for api in api_to_test: - # Сначала получаем информацию о конкретном API - try: - api_info_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": api["name"] - } - - api_info_resp = session.get(api_info_url, params=api_info_params, timeout=10) - api_info_data = api_info_resp.json() - - if api_info_data.get("success") and api["name"] in api_info_data.get("data", {}): - api_details = api_info_data["data"][api["name"]] - api_path = api_details.get("path", "entry.cgi") - api_min_version = api_details.get("minVersion", 1) - api_max_version = api_details.get("maxVersion", 1) - - # Проверяем, поддерживается ли указанная версия - if api["version"] < api_min_version: - logger.warning(f"{api['name']} v{api['version']} ниже минимальной {api_min_version}, используем {api_min_version}") - test_version = api_min_version - elif api["version"] > api_max_version: - logger.warning(f"{api['name']} v{api['version']} выше максимальной {api_max_version}, используем {api_max_version}") - test_version = api_max_version - else: - test_version = api["version"] - - # Выполняем запрос API - test_url = f"{base_url}/{api_path}" - test_params = { - "api": api["name"], - "version": str(test_version), - "method": api["method"], - "_sid": sid # Используем sid для аутентификации - } - - logger.info(f"Тестирование {api['name']}.{api['method']} v{test_version}") - test_response = session.get(test_url, params=test_params, timeout=10) - test_data = test_response.json() - - if test_data.get("success"): - logger.info(f"API {api['name']}.{api['method']} v{test_version} РАБОТАЕТ!") - logger.info(f"Результат: {json.dumps(test_data.get('data', {}), indent=2)[:200]}...") - else: - error_code = test_data.get("error", {}).get("code", -1) - logger.error(f"API {api['name']}.{api['method']} v{test_version} ОШИБКА: {error_code}") - - # Если ошибка связана с сессией, попробуем еще раз авторизоваться - if error_code == 119: # Session timeout - logger.info("Повторная авторизация из-за ошибки 119...") - - # Создаем новую сессию - new_session = requests.Session() - new_session.verify = False - - auth_response = new_session.get(auth_url, params=auth_params, timeout=10) - auth_data = auth_response.json() - - if auth_data.get("success"): - new_sid = auth_data.get("data", {}).get("sid") - logger.info(f"Повторная авторизация успешна! Новый SID: {new_sid[:10]}...") - - # Пробуем запрос с новым SID - test_params["_sid"] = new_sid - logger.info(f"Повторное тестирование {api['name']}.{api['method']} v{test_version}") - test_response = new_session.get(test_url, params=test_params, timeout=10) - test_data = test_response.json() - - if test_data.get("success"): - logger.info(f"API {api['name']}.{api['method']} v{test_version} теперь РАБОТАЕТ!") - logger.info(f"Результат с новой сессией: {json.dumps(test_data.get('data', {}), indent=2)[:200]}...") - else: - error_code = test_data.get("error", {}).get("code", -1) - logger.error(f"API {api['name']}.{api['method']} v{test_version} ВСЕ ЕЩЕ С ОШИБКОЙ: {error_code}") - else: - logger.warning(f"API {api['name']} не найден в информации API") - - except Exception as e: - logger.error(f"Ошибка при тестировании {api['name']}.{api['method']} v{api['version']}: {str(e)}") - - # Шаг 3: Тестирование комбинации запросов для решения проблемы - logger.info("Шаг 3: Тестирование комбинации запросов для решения проблемы") - - # Создаем новую сессию для каждого запроса - for api in [{"name": "SYNO.DSM.Info", "method": "getinfo", "version": 1}]: - try: - fresh_session = requests.Session() - fresh_session.verify = False - - # Авторизуемся - auth_response = fresh_session.get(auth_url, params=auth_params, timeout=10) - auth_data = auth_response.json() - - if auth_data.get("success"): - fresh_sid = auth_data.get("data", {}).get("sid") - logger.info(f"Авторизация в новой сессии успешна! SID: {fresh_sid[:10]}...") - - # Сразу же делаем запрос для получения информации в той же сессии - test_params = { - "api": api["name"], - "version": str(api["version"]), - "method": api["method"], - "_sid": fresh_sid - } - - test_url = f"{base_url}/entry.cgi" # Используем entry.cgi по умолчанию - logger.info(f"Тест в свежей сессии: {api['name']}.{api['method']} v{api['version']}") - test_response = fresh_session.get(test_url, params=test_params, timeout=10) - test_data = test_response.json() - - if test_data.get("success"): - logger.info(f"API в свежей сессии РАБОТАЕТ!") - logger.info(f"Результат: {json.dumps(test_data.get('data', {}), indent=2)[:200]}...") - else: - error_code = test_data.get("error", {}).get("code", -1) - logger.error(f"API в свежей сессии ОШИБКА: {error_code}") - except Exception as e: - logger.error(f"Ошибка при тестировании свежей сессии: {str(e)}") - - # Шаг 4: Получаем информацию об остальных API - logger.info("Шаг 4: Получаем информацию о доступных API для уточнения проблемы") - - # Запрашиваем все API из SYNO.API.Info - try: - all_api_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": "all" - } - - all_api_response = session.get(api_info_url, params=all_api_params, timeout=15) # Больший таймаут для большого ответа - all_api_data = all_api_response.json() - - if all_api_data.get("success"): - api_list = all_api_data.get("data", {}) - logger.info(f"Получен список всех API. Найдено {len(api_list)} API.") - - # Ищем интересующие нас API для отладки - interested_in = ["SYNO.DSM.Info", "SYNO.Core.System", "SYNO.Core.Hardware", - "SYNO.Core.System.Status", "SYNO.API.Auth"] - - logger.info("Информация о важных API:") - for api_name in interested_in: - if api_name in api_list: - logger.info(f"{api_name}: {api_list[api_name]}") - else: - logger.warning(f"API {api_name} не найден") - else: - logger.error("Не удалось получить список всех API") - except Exception as e: - logger.error(f"Ошибка при получении списка API: {str(e)}") - - else: - error_code = auth_data.get("error", {}).get("code", -1) - logger.error(f"Авторизация не удалась! Код ошибки: {error_code}") - else: - logger.error("Не удалось получить информацию об API авторизации") - - except Exception as e: - logger.error(f"Произошла ошибка при выполнении теста: {str(e)}") - -if __name__ == "__main__": - logger.info("Запуск прямого теста API Synology") - direct_api_test() diff --git a/logs/.gitkeep b/logs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/test_api_headers.py b/test_api_headers.py deleted file mode 100644 index ad073d6..0000000 --- a/test_api_headers.py +++ /dev/null @@ -1,391 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -""" -Тестовый скрипт для диагностики проблемы с использованием специальных заголовков -""" - -import requests -import logging -import json -import sys -import os -import urllib3 -import time -import socket -from requests.adapters import HTTPAdapter -from urllib3.util import Retry - -# Отключение предупреждений о небезопасных SSL-соединениях -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - -# Настройка логирования -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) -logger = logging.getLogger(__name__) - -# Тестовые учетные данные (для примера) -SYNOLOGY_HOST = "192.168.0.102" -SYNOLOGY_PORT = 5000 -SYNOLOGY_USERNAME = "superadmin" -SYNOLOGY_PASSWORD = "Cl0ud_1985!" -SYNOLOGY_SECURE = False -SYNOLOGY_TIMEOUT = 10 - -def test_api_with_headers(): - """Тестирование API с использованием специальных заголовков для решения проблемы 119""" - - # Создаем сессию - session = requests.Session() - session.verify = False # Отключаем проверку SSL - - # Настройки повторных попыток - retry_strategy = Retry( - total=3, - status_forcelist=[429, 500, 502, 503, 504], - allowed_methods=["GET", "POST"], - backoff_factor=1.0 - ) - adapter = HTTPAdapter(max_retries=retry_strategy) - session.mount("http://", adapter) - session.mount("https://", adapter) - - # Формируем базовый URL - protocol = "https" if SYNOLOGY_SECURE else "http" - base_url = f"{protocol}://{SYNOLOGY_HOST}:{SYNOLOGY_PORT}/webapi" - - logger.info(f"Тестирование API с заголовками для {base_url}") - - # Тест 1: Получение SID с настройкой cookie и user-agent - logger.info("Тест 1: Авторизация с настройкой cookie и user-agent") - - # Добавление пользовательских заголовков - custom_headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', - 'Accept': 'application/json, text/javascript, */*; q=0.01', - 'Accept-Language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7', - 'X-Requested-With': 'XMLHttpRequest', - 'Connection': 'keep-alive', - 'Referer': f'{protocol}://{SYNOLOGY_HOST}:{SYNOLOGY_PORT}/' - } - - session.headers.update(custom_headers) - - try: - # Определяем путь для авторизации - auth_info_url = f"{base_url}/entry.cgi" - auth_info_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": "SYNO.API.Auth" - } - - auth_info_response = session.get(auth_info_url, params=auth_info_params) - auth_info_data = auth_info_response.json() - - if auth_info_data.get("success"): - auth_info = auth_info_data.get("data", {}).get("SYNO.API.Auth", {}) - auth_path = auth_info.get("path", "auth.cgi") - auth_max_version = auth_info.get("maxVersion", 6) - - logger.info(f"API авторизации: путь={auth_path}, макс. версия={auth_max_version}") - - # Используем версию 3 вместо 6 - тестирование на возможное решение проблемы - auth_version = min(3, auth_max_version) # Пробуем более старую версию API - - # Выполняем авторизацию - auth_url = f"{base_url}/{auth_path}" - auth_params = { - "api": "SYNO.API.Auth", - "version": str(auth_version), - "method": "login", - "account": SYNOLOGY_USERNAME, - "passwd": SYNOLOGY_PASSWORD, - "session": "DirectHeaderTest", - "format": "cookie" - } - - if auth_version >= 6: - auth_params["enable_syno_token"] = "yes" - - logger.info(f"Авторизация с использованием SYNO.API.Auth v{auth_version}") - auth_response = session.get(auth_url, params=auth_params) - auth_data = auth_response.json() - - if auth_data.get("success"): - sid = auth_data.get("data", {}).get("sid") - logger.info(f"Авторизация успешна! SID: {sid[:10]}...") - - # Теперь проверим, работает ли получение информации о системе - # с настройкой cookie и заголовков - - # Сначала настроим куки для сохранения SID - cookies = { - 'id': sid, - 'sid': sid - } - session.cookies.update(cookies) - - # Определяем путь для SYNO.DSM.Info - info_info_url = f"{base_url}/entry.cgi" - info_info_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": "SYNO.DSM.Info" - } - - info_info_response = session.get(info_info_url, params=info_info_params) - info_info_data = info_info_response.json() - - if info_info_data.get("success"): - info_info = info_info_data.get("data", {}).get("SYNO.DSM.Info", {}) - info_path = info_info.get("path", "entry.cgi") - info_max_version = info_info.get("maxVersion", 1) - info_min_version = info_info.get("minVersion", 1) - - logger.info(f"API SYNO.DSM.Info: путь={info_path}, версия={info_min_version}-{info_max_version}") - - # Используем правильную версию API - info_version = min(2, info_max_version) - - # Делаем запрос для получения информации о системе - # с использованием sid как параметр запроса - logger.info("Тест 2: Получение информации о системе с SID как параметром") - info_url = f"{base_url}/{info_path}" - info_params = { - "api": "SYNO.DSM.Info", - "version": str(info_version), - "method": "getinfo", - "_sid": sid - } - - logger.info(f"Запрос информации с использованием SYNO.DSM.Info v{info_version}") - info_response = session.get(info_url, params=info_params) - info_data = info_response.json() - - if info_data.get("success"): - logger.info("Успешно получена информация о системе!") - logger.info(f"Данные: {json.dumps(info_data.get('data', {}), indent=2)}") - else: - error_code = info_data.get("error", {}).get("code", -1) - logger.error(f"Не удалось получить информацию о системе. Ошибка: {error_code}") - - # Пробуем альтернативный способ - logger.info("Тест 3: Попытка получить базовую информацию через SYNO.Core.System") - - # Определяем путь для SYNO.Core.System - system_info_url = f"{base_url}/entry.cgi" - system_info_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": "SYNO.Core.System" - } - - system_info_response = session.get(system_info_url, params=system_info_params) - system_info_data = system_info_response.json() - - if system_info_data.get("success"): - system_info = system_info_data.get("data", {}).get("SYNO.Core.System", {}) - system_path = system_info.get("path", "entry.cgi") - system_max_version = system_info.get("maxVersion", 1) - system_min_version = system_info.get("minVersion", 1) - - logger.info(f"API SYNO.Core.System: путь={system_path}, версия={system_min_version}-{system_max_version}") - - # Используем правильную версию API - system_version = 1 - - # Пробуем альтернативную стратегию с X-SYNO-TOKEN - # Некоторые API Synology требуют специальный токен в заголовках - token = auth_data.get("data", {}).get("synotoken") - if token: - session.headers.update({'X-SYNO-TOKEN': token}) - logger.info(f"Добавлен X-SYNO-TOKEN: {token}") - - # Делаем запрос для получения информации о системе - system_url = f"{base_url}/{system_path}" - system_params = { - "api": "SYNO.Core.System", - "version": str(system_version), - "method": "info", - "_sid": sid - } - - logger.info(f"Запрос информации с использованием SYNO.Core.System v{system_version}") - system_response = session.get(system_url, params=system_params) - system_data = system_response.json() - - if system_data.get("success"): - logger.info("Успешно получена информация о системе через SYNO.Core.System!") - logger.info(f"Данные: {json.dumps(system_data.get('data', {}), indent=2)}") - else: - error_code = system_data.get("error", {}).get("code", -1) - logger.error(f"Не удалось получить информацию через SYNO.Core.System. Ошибка: {error_code}") - - # Пробуем другие методы для диагностики - logger.info("Тест 4: Попытка использовать различные методы и заголовки") - - # Пробуем создать полностью новую сессию - new_session = requests.Session() - new_session.verify = False - new_session.headers.update(custom_headers) - - # Пробуем версию 1 для авторизации - auth_version = 1 - auth_params["version"] = str(auth_version) - - auth_response = new_session.get(auth_url, params=auth_params) - auth_data = auth_response.json() - - if auth_data.get("success"): - sid = auth_data.get("data", {}).get("sid") - logger.info(f"Новая авторизация (v{auth_version}) успешна! SID: {sid[:10]}...") - - # Добавляем SID как куки - cookies = { - 'id': sid, - 'sid': sid - } - new_session.cookies.update(cookies) - - # Пробуем другой подход: разделение запросов во времени - logger.info("Тест 5: Разделение запросов во времени") - - # Даем некоторое время для инициализации сессии на сервере - time.sleep(2) - - # Пробуем получить список файлов - filestation_info_url = f"{base_url}/entry.cgi" - filestation_info_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": "SYNO.FileStation.List" - } - - filestation_info_response = new_session.get(filestation_info_url, params=filestation_info_params) - filestation_info_data = filestation_info_response.json() - - if filestation_info_data.get("success"): - filestation_info = filestation_info_data.get("data", {}).get("SYNO.FileStation.List", {}) - filestation_path = filestation_info.get("path", "entry.cgi") - filestation_max_version = filestation_info.get("maxVersion", 1) - - logger.info(f"API SYNO.FileStation.List: путь={filestation_path}, макс. версия={filestation_max_version}") - - # Используем правильную версию API - filestation_version = min(2, filestation_max_version) - - # Делаем запрос для получения списка общих папок - filestation_url = f"{base_url}/{filestation_path}" - filestation_params = { - "api": "SYNO.FileStation.List", - "version": str(filestation_version), - "method": "list_share", - "_sid": sid - } - - logger.info(f"Запрос списка общих папок с использованием SYNO.FileStation.List v{filestation_version}") - filestation_response = new_session.get(filestation_url, params=filestation_params) - filestation_data = filestation_response.json() - - if filestation_data.get("success"): - logger.info("Успешно получен список общих папок!") - shares = filestation_data.get("data", {}).get("shares", []) - logger.info(f"Общие папки: {json.dumps(shares, indent=2)[:200]}...") - else: - error_code = filestation_data.get("error", {}).get("code", -1) - logger.error(f"Не удалось получить список общих папок. Ошибка: {error_code}") - else: - error_code = auth_data.get("error", {}).get("code", -1) - logger.error(f"Новая авторизация не удалась! Код ошибки: {error_code}") - else: - logger.error("Не удалось получить информацию о SYNO.Core.System API") - else: - logger.error("Не удалось получить информацию о SYNO.DSM.Info API") - else: - error_code = auth_data.get("error", {}).get("code", -1) - logger.error(f"Авторизация не удалась! Код ошибки: {error_code}") - else: - logger.error("Не удалось получить информацию об API авторизации") - - except Exception as e: - logger.error(f"Произошла ошибка: {str(e)}") - - # Тест 6: Проверка сетевой доступности - logger.info("Тест 6: Проверка сетевой доступности") - - try: - # Проверка базового TCP-соединения - socket_obj = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - socket_obj.settimeout(SYNOLOGY_TIMEOUT) - result = socket_obj.connect_ex((SYNOLOGY_HOST, SYNOLOGY_PORT)) - socket_obj.close() - - if result == 0: - logger.info("TCP-соединение успешно установлено") - else: - logger.error(f"Не удалось установить TCP-соединение, код ошибки: {result}") - except Exception as e: - logger.error(f"Ошибка при проверке TCP-соединения: {str(e)}") - - # Тест 7: Запрос без аутентификации для проверки доступности API - logger.info("Тест 7: Запрос без аутентификации для проверки доступности API") - - try: - # Создаем новую сессию без аутентификации - simple_session = requests.Session() - simple_session.verify = False - - # Запрос к SYNO.API.Info не требует аутентификации - api_info_url = f"{base_url}/entry.cgi" - api_info_params = { - "api": "SYNO.API.Info", - "version": "1", - "method": "query", - "query": "all" - } - - logger.info("Запрос информации о всех API без аутентификации") - api_info_response = simple_session.get(api_info_url, params=api_info_params) - - if api_info_response.status_code == 200: - logger.info("API доступно без аутентификации") - api_info_data = api_info_response.json() - - if api_info_data.get("success"): - logger.info("Успешно получена информация о всех API") - api_count = len(api_info_data.get("data", {})) - logger.info(f"Количество доступных API: {api_count}") - - # Поиск API для управления питанием - power_apis = [] - for api_name, api_info in api_info_data.get("data", {}).items(): - if "power" in api_name.lower() or "reboot" in api_name.lower() or "shutdown" in api_name.lower(): - power_apis.append(f"{api_name}: {api_info}") - - logger.info(f"Найдены API для управления питанием: {power_apis}") - - # Поиск API для получения информации о системе - info_apis = [] - for api_name, api_info in api_info_data.get("data", {}).items(): - if "info" in api_name.lower() or "system" in api_name.lower() or "status" in api_name.lower(): - info_apis.append(f"{api_name}: {api_info}") - - logger.info(f"Найдены API для информации о системе: {info_apis[:5]} и еще {len(info_apis)-5}") - else: - error_code = api_info_data.get("error", {}).get("code", -1) - logger.error(f"Запрос к API без аутентификации не удался! Код ошибки: {error_code}") - else: - logger.error(f"API не доступно без аутентификации. HTTP статус: {api_info_response.status_code}") - except Exception as e: - logger.error(f"Ошибка при проверке доступности API: {str(e)}") - -if __name__ == "__main__": - logger.info("Запуск теста API с заголовками") - test_api_with_headers() diff --git a/test_reboot.py b/test_reboot.py deleted file mode 100644 index 90e896c..0000000 --- a/test_reboot.py +++ /dev/null @@ -1,73 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -""" -Скрипт для тестирования функции перезагрузки Synology NAS -""" - -import os -import sys -import logging -from pathlib import Path - -# Добавляем путь проекта в sys.path -project_dir = str(Path(__file__).resolve().parent) -if project_dir not in sys.path: - sys.path.insert(0, project_dir) - -from src.api.synology import SynologyAPI -from src.config.config import SYNOLOGY_POWER_API, SYNOLOGY_INFO_API, SYNOLOGY_API_VERSION - -# Настройка логирования -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) -logger = logging.getLogger(__name__) - -def main(): - """Основная функция для тестирования перезагрузки""" - logger.info("Тестирование функции перезагрузки Synology NAS") - - logger.info(f"Используемые API: POWER={SYNOLOGY_POWER_API}, INFO={SYNOLOGY_INFO_API}, VERSION={SYNOLOGY_API_VERSION}") - - # Инициализация API - synology = SynologyAPI() - - # Проверка онлайн статуса - logger.info("Проверка онлайн статуса...") - is_online = synology.is_online(force_check=True) - logger.info(f"Статус: {'Онлайн' if is_online else 'Оффлайн'}") - - if not is_online: - logger.error("NAS недоступен. Невозможно выполнить перезагрузку.") - return - - # Вывод информации о системе - logger.info("Получение информации о системе...") - system_info = synology.get_system_status() - - if system_info.get("status") == "error": - logger.error(f"Ошибка получения информации о системе: {system_info.get('error')}") - else: - logger.info(f"Информация о системе: {system_info}") - - # Запрос на подтверждение действия - confirm = input("Вы действительно хотите перезагрузить Synology NAS? (y/n): ") - if confirm.lower() != 'y': - logger.info("Операция отменена пользователем.") - return - - # Выполнение перезагрузки - logger.info("Выполнение перезагрузки...") - try: - result = synology.reboot_system() - if result: - logger.info("Перезагрузка выполнена успешно.") - else: - logger.error("Не удалось выполнить перезагрузку.") - except Exception as e: - logger.error(f"Ошибка при выполнении перезагрузки: {str(e)}") - -if __name__ == "__main__": - main() diff --git a/test_system_info.py b/test_system_info.py deleted file mode 100644 index 95a9bf3..0000000 --- a/test_system_info.py +++ /dev/null @@ -1,87 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -""" -Скрипт для тестирования получения информации о системе Synology NAS -""" - -import os -import sys -import logging -from pathlib import Path - -# Добавляем путь проекта в sys.path -project_dir = str(Path(__file__).resolve().parent) -if project_dir not in sys.path: - sys.path.insert(0, project_dir) - -from src.api.synology import SynologyAPI -from src.config.config import SYNOLOGY_POWER_API, SYNOLOGY_INFO_API, SYNOLOGY_API_VERSION - -# Настройка логирования -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) -logger = logging.getLogger(__name__) - -def main(): - """Основная функция для тестирования получения информации о системе""" - logger.info("Тестирование получения информации о системе Synology NAS") - - logger.info(f"Используемые API: POWER={SYNOLOGY_POWER_API}, INFO={SYNOLOGY_INFO_API}, VERSION={SYNOLOGY_API_VERSION}") - - # Инициализация API - synology = SynologyAPI() - - # Проверка онлайн статуса - logger.info("Проверка онлайн статуса...") - is_online = synology.is_online(force_check=True) - logger.info(f"Статус: {'Онлайн' if is_online else 'Оффлайн'}") - - if not is_online: - logger.error("NAS недоступен. Невозможно получить информацию о системе.") - return - - # Получение списка доступных API - logger.info("Получение списка доступных API...") - from src.api.api_discovery import discover_available_apis - from src.config.config import SYNOLOGY_HOST, SYNOLOGY_PORT, SYNOLOGY_SECURE - - protocol = "https" if SYNOLOGY_SECURE else "http" - base_url = f"{protocol}://{SYNOLOGY_HOST}:{SYNOLOGY_PORT}/webapi" - - apis = discover_available_apis(base_url) - if apis: - logger.info(f"Найдено {len(apis)} API") - - # Фильтрация API для управления питанием - power_apis = [name for name in apis.keys() if "power" in name.lower()] - system_apis = [name for name in apis.keys() if "system" in name.lower() or "dsm.info" in name.lower()] - - logger.info(f"API для управления питанием: {power_apis}") - logger.info(f"API для системной информации: {system_apis}") - - # Проверка конкретных API - for api_name in [SYNOLOGY_POWER_API, SYNOLOGY_INFO_API]: - if api_name in apis: - api_info = apis[api_name] - logger.info(f"API {api_name}: versions={api_info.get('minVersion')}-{api_info.get('maxVersion')}, path={api_info.get('path')}") - else: - logger.warning(f"API {api_name} не найден в списке доступных API") - else: - logger.error("Не удалось получить список доступных API") - - # Вывод информации о системе - logger.info("Получение информации о системе...") - system_info = synology.get_system_status() - - if system_info.get("status") == "error": - logger.error(f"Ошибка получения информации о системе: {system_info.get('error')}") - else: - logger.info(f"Информация о системе: {system_info}") - - logger.info("Тестирование завершено.") - -if __name__ == "__main__": - main()