# sd_api.py import requests import logging import json import os from typing import List, Dict, Any from datetime import datetime from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # Импортируем нашу конфигурацию import config log = logging.getLogger(__name__) class ServiceDeskAPIError(Exception): """Кастомное исключение для ошибок API ServiceDesk.""" pass class ServiceDeskClient: """ Клиент для взаимодействия с REST API ServiceDesk. Оснащен механизмом повторных запросов для повышения отказоустойчивости. """ def __init__(self, access_key: str, base_url: str): if not access_key or not base_url: raise ValueError("Access key and base URL cannot be empty.") self.base_url = base_url self.session = requests.Session() # Устанавливаем accessKey как параметр по умолчанию для всех запросов self.session.params = {'accessKey': access_key} # --- НАСТРОЙКА ЛОГИКИ ПОВТОРНЫХ ЗАПРОСОВ --- retry_strategy = Retry( total=3, # Общее количество повторных попыток backoff_factor=1, # Множитель для задержки (sleep_time = {backoff_factor} * (2 ** ({number of total retries} - 1))) # Задержки будут примерно 0.5с, 1с, 2с status_forcelist=[500, 502, 503, 504], # Коды, которые нужно повторять allowed_methods=["POST", "GET"] # Методы, для которых будет работать retry. API использует POST даже для поиска. ) # Создаем адаптер с нашей стратегией adapter = HTTPAdapter(max_retries=retry_strategy) # "Монтируем" адаптер для всех http и https запросов self.session.mount("https://", adapter) self.session.mount("http://", adapter) log.info("Клиент ServiceDesk инициализирован с логикой повторных запросов (retries).") def _make_request(self, method: str, url: str, params: Dict = None, json_data: Dict = None) -> Any: """ Внутренний метод для выполнения HTTP-запросов. Теперь он автоматически использует логику retry, настроенную в __init__. :param method: HTTP метод ('GET', 'POST', etc.) :param url: Полный URL для запроса. :param params: Дополнительные параметры URL (кроме accessKey). :param json_data: Тело запроса в формате JSON для POST/PUT. :return: Ответ сервера в виде JSON. :raises ServiceDeskAPIError: в случае ошибки API или сети ПОСЛЕ всех попыток. """ try: # Теперь этот вызов будет автоматически повторяться в случае сбоев response = self.session.request(method, url, params=params, json=json_data, timeout=30) # Проверяем на ошибки ПОСЛЕ успешного выполнения запроса (или исчерпания попыток) response.raise_for_status() # Вызовет исключение для кодов 4xx/5xx # Некоторые ответы могут быть пустыми (например, при успешном редактировании - 204) # или успешном создании (201) if response.status_code in [204, 201] or not response.text: return None return response.json() except requests.exceptions.HTTPError as e: # Эта ошибка возникнет, если после всех попыток сервер все равно вернул 4xx или 5xx error_message = f"HTTP Error: {e.response.status_code} for URL {url}. Response: {e.response.text}" log.error(error_message) raise ServiceDeskAPIError(error_message) from e except requests.exceptions.RequestException as e: # Эта ошибка возникнет, если после всех попыток не удалось подключиться к серверу (например, DNS или таймаут) error_message = f"Request failed for URL {url}: {e}" log.error(error_message) raise ServiceDeskAPIError(error_message) from e def _ensure_test_output_dir(self): """Проверяет и создает директорию для тестовых файлов.""" if not os.path.exists(config.TEST_OUTPUT_PATH): os.makedirs(config.TEST_OUTPUT_PATH) log.info(f"Создана директория для тестовых данных: {config.TEST_OUTPUT_PATH}") def get_all_frs(self) -> List[Dict]: """Получает список всех фискальных регистраторов.""" log.info("Запрос списка всех ФР из ServiceDesk...") params = { 'attrs': 'UUID,FRSerialNumber,RNKKT,KKTRegDate,FNExpireDate,FNNumber,owner,FRDownloader,LegalName,OFDName,ModelKKT,FFD,lastModifiedDate' } frs = self._make_request('POST', config.FIND_FRS_URL, params=params) log.info(f"Получено {len(frs)} записей о ФР.") # frs может быть None если ответ пустой, вернем пустой список для консистентности return frs or [] def get_all_workstations(self) -> List[Dict]: """Получает список всех рабочих станций.""" log.info("Запрос списка всех рабочих станций из ServiceDesk...") params = { 'attrs': 'UUID,owner,AnyDesk,Teamviewer,lastModifiedDate' } workstations = self._make_request('POST', config.FIND_WORKSTATIONS_URL, params=params) log.info(f"Получено {len(workstations)} записей о рабочих станциях.") return workstations or [] def get_lookup_values(self, metaclass: str) -> List[Dict]: """ Получает значения из справочника по его metaClass. :param metaclass: metaClass справочника (e.g., 'ModeliFR', 'FFD'). :return: Список словарей с 'UUID' и 'title'. """ log.info(f"Запрос значений справочника для metaClass: {metaclass}...") url = f"{self.base_url}/find/{metaclass}" params = {'attrs': 'UUID,title'} lookups = self._make_request('POST', url, params=params) log.info(f"Получено {len(lookups)} значений для {metaclass}.") return lookups or [] def update_fr(self, uuid: str, data: Dict) -> None: """ Обновляет существующий фискальный регистратор. В режиме DRY_RUN сохраняет данные в файл. :param uuid: UUID объекта для редактирования. :param data: Словарь с полями для обновления. """ if config.DRY_RUN: self._ensure_test_output_dir() filename = f"update_{uuid}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" filepath = os.path.join(config.TEST_OUTPUT_PATH, filename) # Сохраняем и URL, и параметры для полного понимания output_data = { "action": "UPDATE", "target_uuid": uuid, "payload": data } with open(filepath, 'w', encoding='utf-8') as f: json.dump(output_data, f, indent=4, ensure_ascii=False) log.warning(f"[DRY RUN] Пропущено обновление {uuid}. Данные сохранены в {filepath}") return log.info(f"Обновление объекта ФР с UUID: {uuid}...") url = config.EDIT_FR_URL_TEMPLATE.format(uuid=uuid) self._make_request('POST', url, params=data) log.info(f"Объект {uuid} успешно обновлен.") def create_fr(self, data: Dict) -> Dict: """ Создает новый фискальный регистратор. В режиме DRY_RUN сохраняет тело запроса в файл. :param data: Словарь с данными для создания объекта (тело запроса). :return: JSON-ответ от сервера, содержащий UUID нового объекта. """ serial_number = data.get('FRSerialNumber', 'unknown_sn') if config.DRY_RUN: self._ensure_test_output_dir() filename = f"create_{serial_number}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" filepath = os.path.join(config.TEST_OUTPUT_PATH, filename) # Сохраняем только тело запроса with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, indent=4, ensure_ascii=False) log.warning(f"[DRY RUN] Пропущено создание ФР с S/N {serial_number}. Body запроса сохранено в {filepath}") # В режиме dry run возвращаем фейковый ответ, чтобы не сломать вызывающий код return {"action": "DRY_RUN_CREATE", "saved_to": filepath} log.info(f"Создание нового ФР с серийным номером: {data.get('FRSerialNumber')}...") # Параметр для получения UUID в ответе params = {'attrs': 'UUID'} response = self._make_request('POST', config.CREATE_FR_URL, params=params, json_data=data) log.info(f"ФР успешно создан. Ответ сервера: {response}") return response