241 lines
8.4 KiB
Python
241 lines
8.4 KiB
Python
"""
|
||
Сервис синхронизации данных из 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 ────────────────────────────────────────────────────────────
|
||
|
||
def _dry_run_print(label: str, items: list[dict], sample_keys: list[str]) -> None:
|
||
"""Вывод первых 3 записей в консоль при dry-run."""
|
||
print(f"\n[DRY-RUN] {label} — запись в БД НЕ выполнена, получено {len(items)} записей")
|
||
for i, item in enumerate(items[:3]):
|
||
preview = {k: item.get(k) for k in sample_keys if k in item}
|
||
print(f" [{i}] {preview}")
|
||
if len(items) > 3:
|
||
print(f" ... и ещё {len(items) - 3}")
|
||
print()
|
||
|
||
|
||
async def sync_clients() -> int:
|
||
"""Синхронизировать клиентов/организации."""
|
||
items, _ = await _fetch_all(
|
||
"/settings/odata/v2/Clients",
|
||
config.INTRADESK_API_KEY,
|
||
{"$filter": "(isarchived eq false)", "$orderby": "name asc"},
|
||
)
|
||
if config.DRY_RUN:
|
||
_dry_run_print("sync_clients", items, ["id", "name", "clientType", "email"])
|
||
return len(items)
|
||
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"},
|
||
)
|
||
if config.DRY_RUN:
|
||
_dry_run_print("sync_users", items, ["id", "name", "email", "phoneNumbers"])
|
||
return len(items)
|
||
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."""
|
||
if config.DRY_RUN:
|
||
print("[DRY-RUN] sync_whitelist — обновление org_users пропущено\n")
|
||
return 0
|
||
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"},
|
||
)
|
||
if config.DRY_RUN:
|
||
_dry_run_print(
|
||
"sync_tasks", items,
|
||
["id", "tasknumber", "name", "status", "clientid", "updatedat"],
|
||
)
|
||
return len(items)
|
||
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,
|
||
}
|