""" Сервис синхронизации данных из Intradesk API. Поддерживаемые типы синхронизации: - clients — организации/клиенты (/settings/odata/v2/Clients) - users — контактные лица (/settings/odata/ClientUsers) - tasks — заявки (/tasklist/odata/v3/tasks) - all — всё вместе Запуск: await run_sync("all") """ from __future__ import annotations import logging import urllib.parse import httpx import config from database.connection import get_db from database.queries.intradesk import ( finish_sync_log, start_sync_log, sync_whitelist_from_intradesk, upsert_client, upsert_task, upsert_user, ) logger = logging.getLogger(__name__) _PAGE_SIZE = 100 # ── Helpers ─────────────────────────────────────────────────────────────────── def _build_url(path: str, api_key: str, extra: dict | None = None) -> str: """ Собирает URL с параметрами. Ключи начинающиеся на '$' вставляются без URL-кодирования имени (OData требует именно $filter, $top и т.д.). """ base = f"{config.INTRADESK_BASE_URL.rstrip('/')}{path}?ApiKey={api_key}" if extra: for k, v in extra.items(): base += f"&{k}={urllib.parse.quote(str(v), safe='')}" return base async def _fetch_page( client: httpx.AsyncClient, path: str, api_key: str, extra: dict, skip: int ) -> tuple[list[dict], dict[int, str]]: """Запросить одну страницу OData и вернуть (items, dicts_map).""" params = dict(extra) params["$top"] = str(_PAGE_SIZE) params["$skip"] = str(skip) url = _build_url(path, api_key, params) resp = await client.get(url, timeout=60.0) resp.raise_for_status() data = resp.json() items: list[dict] = data.get("value", []) # Аккумулируем словари (статусы, приоритеты, пользователи, группы...) dicts_map: dict[int, str] = {} for d in data.get("dictionaries", []): raw_name = d.get("name") if isinstance(raw_name, dict): # name может быть {"ru": "...", "en": null} raw_name = raw_name.get("ru") or raw_name.get("en") or "" dicts_map[d["id"]] = raw_name or "" return items, dicts_map async def _fetch_all( path: str, api_key: str, extra: dict | None = None ) -> tuple[list[dict], dict[int, str]]: """Постранично скачать все записи OData-эндпоинта.""" all_items: list[dict] = [] all_dicts: dict[int, str] = {} extra = extra or {} async with httpx.AsyncClient() as client: skip = 0 while True: items, dicts_map = await _fetch_page(client, path, api_key, extra, skip) all_items.extend(items) all_dicts.update(dicts_map) if len(items) < _PAGE_SIZE: break skip += _PAGE_SIZE return all_items, all_dicts # ── Sync functions ──────────────────────────────────────────────────────────── async def sync_clients() -> int: """Синхронизировать клиентов/организации.""" items, _ = await _fetch_all( "/settings/odata/v2/Clients", config.INTRADESK_API_KEY, {"$filter": "(isarchived eq false)", "$orderby": "name asc"}, ) db = get_db() for item in items: await upsert_client(db, item) await db.commit() logger.info("sync_clients: %d records", len(items)) return len(items) async def sync_users() -> int: """Синхронизировать контактные лица клиентов.""" items, _ = await _fetch_all( "/settings/odata/ClientUsers", config.INTRADESK_API_KEY, {"$filter": "(isarchived eq false)", "$orderby": "id desc"}, ) db = get_db() for item in items: await upsert_user(db, item) await db.commit() logger.info("sync_users: %d records", len(items)) return len(items) async def sync_whitelist() -> int: """Обновить organizations + org_users из уже синхронизированных данных Intradesk.""" db = get_db() count = await sync_whitelist_from_intradesk(db) logger.info("sync_whitelist: %d org_users затронуто", count) return count async def sync_tasks() -> int: """Синхронизировать заявки по нашей группе исполнителей.""" if not config.INTRADESK_TASKS_API_KEY: logger.warning("sync_tasks: INTRADESK_TASKS_API_KEY не задан, пропуск") return 0 # Строим OData $filter conditions: list[str] = [] if config.INTRADESK_EXECUTOR_GROUP: conditions.append(f"executorgroup eq {config.INTRADESK_EXECUTOR_GROUP}") for eid in config.INTRADESK_EXECUTORS: conditions.append(f"executor eq {eid}") if not conditions: logger.warning("sync_tasks: не задан ни executor_group, ни executors — фильтр пуст") filter_str = "isarchived eq false" else: filter_str = "(" + " or ".join(conditions) + ")" items, dicts_map = await _fetch_all( "/tasklist/odata/v3/tasks", config.INTRADESK_TASKS_API_KEY, {"$filter": filter_str, "$orderby": "updatedat desc"}, ) db = get_db() for item in items: await upsert_task(db, item, dicts_map) await db.commit() logger.info("sync_tasks: %d records", len(items)) return len(items) # ── Orchestrator ────────────────────────────────────────────────────────────── _SYNC_FUNCS = { "clients": sync_clients, "users": sync_users, "tasks": sync_tasks, } # После каких типов синхронизации нужно обновить белый список _NEEDS_WHITELIST = {"clients", "users", "all"} async def run_sync(sync_type: str) -> dict: """ Запустить синхронизацию и вернуть результат. sync_type: 'clients' | 'users' | 'tasks' | 'all' Возвращает dict: {sync_type, status, records_synced, error_message} """ db = get_db() log_id = await start_sync_log(db, sync_type) if sync_type == "all": funcs = [sync_clients, sync_users, sync_tasks, sync_whitelist] elif sync_type in _NEEDS_WHITELIST: funcs = [_SYNC_FUNCS[sync_type], sync_whitelist] else: funcs = [_SYNC_FUNCS[sync_type]] total = 0 error_msg = "" try: for fn in funcs: total += await fn() status = "success" except Exception as exc: error_msg = str(exc) status = "error" logger.exception("Ошибка синхронизации '%s': %s", sync_type, exc) await finish_sync_log(db, log_id, status, total, error_msg) return { "sync_type": sync_type, "status": status, "records_synced": total, "error_message": error_msg, }