call-to-text/storage/contacts.py
ed0ss 7f03949593 Первая версия: транскрипция звонков из Novofon DATA API в Directus
- Клиент Novofon DATA API (JSON-RPC, get.calls_report, чанки по 90 дней)
- faster-whisper (модель turbo) на CUDA с разделением стереоканалов
- Маркировка спикеров: левый канал = caller, правый = callee
- Directus: коллекции calls, calls_74951284933, contacts
- CLI: process, process-file, export, daemon (polling 5 мин)
- Docker Compose для Directus + PostgreSQL
- Архив email-парсинга в old/
2026-06-26 12:44:50 +03:00

41 lines
1.4 KiB
Python

from typing import Optional
from storage.directus import DirectusClient
def resolve_contact_name(phone_number: Optional[str]) -> Optional[str]:
if not phone_number:
return None
try:
dc = DirectusClient()
if dc.is_available():
contact = dc.find_contact(phone_number)
dc.close()
if contact:
name = contact.get("name") or ""
company = contact.get("company") or ""
parts = [p for p in [name, f"({company})" if company else ""] if p]
return " ".join(parts)
except Exception:
pass
return None
def resolve_contacts(caller_number: Optional[str], callee_number: Optional[str]) -> dict:
result = {}
try:
dc = DirectusClient()
if dc.is_available():
if caller_number:
contact = dc.find_contact(caller_number)
if contact:
result["caller_name"] = contact.get("name", "")
result["caller_company"] = contact.get("company", "")
if callee_number:
contact = dc.find_contact(callee_number)
if contact:
result["callee_name"] = contact.get("name", "")
result["callee_company"] = contact.get("company", "")
dc.close()
except Exception:
pass
return result