162 lines
8.4 KiB
Python
162 lines
8.4 KiB
Python
# sd_api.py
|
||
|
||
import requests
|
||
import logging
|
||
import json
|
||
import os
|
||
from typing import List, Dict, Any
|
||
from datetime import datetime
|
||
|
||
# Импортируем нашу конфигурацию
|
||
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}
|
||
|
||
def _make_request(self, method: str, url: str, params: Dict = None, json_data: Dict = None) -> Any:
|
||
"""
|
||
Внутренний метод для выполнения HTTP-запросов.
|
||
|
||
: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
|
||
|
||
# Некоторые ответы могут быть пустыми (например, при успешном редактировании)
|
||
if response.status_code == 204 or not response.text:
|
||
return None
|
||
|
||
return response.json()
|
||
|
||
except requests.exceptions.HTTPError as e:
|
||
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:
|
||
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)} записей о ФР.")
|
||
return frs
|
||
|
||
def get_all_workstations(self) -> List[Dict]:
|
||
"""Получает список всех рабочих станций."""
|
||
log.info("Запрос списка всех рабочих станций из ServiceDesk...")
|
||
params = {
|
||
'attrs': 'UUID,owner,AnyDesk,Teamviewerm,lastModifiedDate'
|
||
}
|
||
workstations = self._make_request('POST', config.FIND_WORKSTATIONS_URL, params=params)
|
||
log.info(f"Получено {len(workstations)} записей о рабочих станциях.")
|
||
return workstations
|
||
|
||
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
|
||
|
||
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}...")
|
||
# Примечание: старый код использовал POST с параметрами для редактирования.
|
||
# Если API требует form-encoded data, а не JSON, нужно использовать `data=data` вместо `json=data`.
|
||
# Судя по вашему коду, это POST-запрос с параметрами в URL, а не в теле.
|
||
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 |