FleetController/caller-api-manager
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[GUP]

Browse source
This commit is contained in:
Evgeny (Krymmy) Momotov 2025-11-12 15:59:45 +03:00
parent 493e2258fb
commit 7eeb4fb05b
7 changed files with 610 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

192
.gitignore vendored Normal file
View file

@ -0,0 +1,192 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
database/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sql
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
#Keys
*.pem
#bin
*.bin
#try_file
try_*
.vscode
.env_*
loggs/
mediafiles/
db.sqlite*

19
pyproject.toml Normal file
View file

@ -0,0 +1,19 @@
[project]
name = "callerapimanager"
version = "0.1.0"
description = ""
authors = [
{name = "Evgeny (Krymmy) Momotov",email = "evgeny.momotov@gmail.com"}
]
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
]
[tool.poetry]
packages = [{include = "CallerApiManager", from = "src"}]
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"

2
requirements.txt Normal file
View file

@ -0,0 +1,2 @@
aiohttp==3.12.15
pydantic==2.11.7

0
src/CallerApiManager/__init__.py Normal file
View file

107
src/CallerApiManager/api_manager.py Normal file
View file

@ -0,0 +1,107 @@
import uuid
from enum import Enum
from datetime import datetime
import aiohttp
from data_models import *
class CallerManagerClient:
def __init__(self, base_url: str, auth_key: Optional[str] = None, token: Optional[str] = None):
self.base_url = base_url.rstrip('/')
self.token: Optional[str] = None
if token:
self.token = token
if auth_key and not token:
self.auth_key = auth_key
else:
self.auth_key = None
if not self.token and not self.auth_key:
raise ValueError("Either token or auth_key must be provided")
async def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
headers = kwargs.get('headers', {})
if self.token:
headers['Authorization'] = f'Bearer {self.token}'
async with aiohttp.ClientSession() as session:
async with session.request(method, f"{self.base_url}{endpoint}", **kwargs) as response:
if response.status >= 400:
error_data = await response.json()
raise Exception(f"API Error {response.status}: {error_data}")
return await response.json()
async def register_manager(self, auth_key: Optional[str] = None) -> AccessResponse:
if auth_key is None and self.auth_key is None:
raise ValueError("Auth key is required for registration")
auth_key = auth_key or self.auth_key
data = Registration(auth_key=auth_key).dict()
response = await self._request('POST', '/caller_manager/register', json=data)
access_response = AccessResponse(**response)
self.token = access_response.token
return access_response
async def get_sip_accounts(self, **params) -> List[SIPAccount]:
response = await self._request('GET', '/sip/', params=params)
return [SIPAccount(**account) for account in response]
async def get_sip_account(self, account_id: int) -> SIPAccount:
response = await self._request('GET', f'/sip/{account_id}')
return SIPAccount(**response)
async def get_calls(self, **params) -> List[Call]:
response = await self._request('GET', '/calls/', params=params)
return [Call(**call) for call in response]
async def get_call(self, call_id: int) -> Call:
response = await self._request('GET', f'/calls/{call_id}')
return Call(**response)
async def update_call_status(self, call_id: int, status: CallStatus) -> Call:
response = await self._request('PATCH', f'/calls/{call_id}', params={'status': status.value})
return Call(**response)
async def update_call(self, call_id: int, call: CallIn) -> Call:
response = await self._request('PATCH', f'/calls/{call_id}', json=call.model_dump(exclude_none=True))
return Call(**response)
async def get_call_dialog(self, call_id: int) -> CallDialog:
response = await self._request('GET', f'/calls/{call_id}/dialog')
return CallDialog(**response)
async def create_call_dialog(self, dialog: CallDialogIn) -> CallDialog:
response = await self._request('POST', f'/calls/dialog', json=dialog.model_dump(exclude_none=True))
return CallDialog(**response)
async def get_dialog_parts(self, dialog_id: int) -> List[CallDialogPart]:
response = await self._request('GET', f'/calls/dialog/{dialog_id}/parts')
return [CallDialogPart(**part) for part in response]
async def create_dialog_part(self, part: CallDialogPartIn) -> CallDialogPart:
response = await self._request('POST', f'/calls/dialog/parts', json=part.model_dump(exclude_none=True))
return CallDialogPart(**response)
async def get_prompts(self, **params) -> List[Prompt]:
response = await self._request('GET', '/prompts/', params=params)
return [CallPrompt(**prompt) for prompt in response]
async def get_prompt(self, prompt_id: int) -> Prompt:
response = await self._request('GET', f'/prompts/{prompt_id}')
return CallPrompt(**response)
async def get_prompt_yandex_settings(self, prompt_id: int) -> YandexSpeechKit:
response = await self._request('GET', f'/prompts/{prompt_id}/yandex_settings')
return YandexSpeechKit(**response)
async def get_prompt_openai_settings(self, prompt_id: int) -> OpenAISettings:
response = await self._request('GET', f'/prompts/{prompt_id}/openai_settings')
return OpenAISettings(**response)
async def get_yandex_fleet_driver_profile(self, driver_id: int) -> YandexFleetDriverProfile:
response = await self._request('GET', f'/yandex_fleet/drivers/{driver_id}/profile')
return YandexFleetDriverProfile(**response)

290
src/CallerApiManager/data_models.py Normal file
View file

@ -0,0 +1,290 @@
import re
import enum
from pydantic import BaseModel, Field, validator
from typing import Optional, Dict, Any
class CallSide(str, Enum):
DRIVER = "driver"
BOT = "bot"
class CallStatus(str, Enum):
PENDING = "PENDING"
RUNNING = "RUNNING"
COMPLETED = "COMPLETED"
ALL_MISSED = "ALL_MISSED"
FAILED = "FAILED"
class AccessResponse(BaseModel):
success: bool
token: str
class Registration(BaseModel):
auth_key: str
class User(BaseModel):
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
username: str = Field(max_length=150, description="Пользовательское имя", example="john_doe")
#password: str = Field(max_length=255, description="Пароль", example="secret123")
email: str = Field(max_length=254, default="", description="Электронная почта", example="john@example.com")
first_name: str = Field(max_length=150, default="", description="Имя", example="John")
last_name: str = Field(max_length=150, default="", description="Фамилия", example="Doe")
#is_staff: bool = Field(default=False, description="Статус администратора")
#is_active: bool = Field(default=False, description="Активен")
#is_superuser: bool = Field(default=False, description="Суперпользователь")
#date_joined: str = Field(description="Дата создания", example="2024-01-01T12:00:00")
#last_login: Optional[str] = Field(null=True, default=None, description="Последний вход")
@validator("email")
def validate_email(cls, v):
if v and "@" not in v or "." not in v:
raise ValueError("Некорректный email")
return v
@validator("username")
def validate_username(cls, v):
if len(v) > 150:
raise ValueError("Имя пользователя слишком длинное (максимум 150 символов)")
return v
@validator("first_name")
def validate_first_name(cls, v):
if v and len(v) > 150:
raise ValueError("Имя слишком длинное (максимум 150 символов)")
return v
@validator("last_name")
def validate_last_name(cls, v):
if v and len(v) > 150:
raise ValueError("Фамилия слишком длинная (максимум 150 символов)")
return v
class Config:
extra = "forbid"
from_attributes = True
class SIPAccount(BaseModel):
"""
SIP-аккаунт.
"""
unique_id: str = Field(description="Уникальный идентификатор", example="00000000-0000-0000-0000-000000000000")
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
created_at: Optional[str] = Field(description="Дата создания", example="2024-01-01T12:00:00")
updated_at: Optional[str] = Field(description="Дата обновления", example="2024-01-01T12:00:00")
owner: Optional[User] = Field(description="Владелец")
name: str = Field(max_length=255, description="Название", example="SIP-аккаунт-1")
is_registered: bool = Field(default=False, description="Зарегистрирован")
username: str = Field(max_length=40, description="Имя SIP клиента", example="user123")
password: Optional[str] = Field(max_length=80, null=True, blank=True, description="Пароль (открытый, временно)")
server_url: str = Field(
max_length=215,
description="URL сервера",
example="https://192.168.1.100:5060"
)
port: Optional[int] = Field(null=True, blank=True, description="Порт", ge=0, le=65535)
internal_number: Optional[str] = Field(max_length=255, null=True, blank=True, description="Внутренний номер")
endpoint_name: Optional[str] = Field(max_length=255, null=True, blank=True, description="Имя конечной точки")
@validator("server_url")
def validate_server_url(cls, v):
if not v:
return v
# Проверяем, содержит ли строка URL (в формате http:// или https://)
if v.startswith("http://") or v.startswith("https://"):
return v
# Проверяем, является ли это IPv4 (простая проверкаDetails)
parts = v.strip().split(':')
if len(parts) == 1:
ip_part = parts[0]
if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", ip_part):
# Проверяем, что каждая часть от 0 до 255
ip_parts = [int(x) for x in ip_part.split('.')]
if all(0 <= part <= 255 for part in ip_parts):
return v
raise ValueError("Invalid server URL format (must be http(s):// or IPv4 address)")
@validator("username")
def validate_username(cls, v):
if len(v) > 40:
raise ValueError("Username too long (max 40 characters)")
return v
@validator("name")
def validate_name(cls, v):
if len(v) > 255:
raise ValueError("Name too long (max 255 characters)")
return v
@validator("internal_number")
def validate_internal_number(cls, v):
if v and len(v) > 255:
raise ValueError("Internal number too long (max 255 characters)")
return v
@validator("endpoint_name")
def validate_endpoint_name(cls, v):
if v and len(v) > 255:
raise ValueError("Endpoint name too long (max 255 characters)")
return v
@validator("port")
def validate_port(cls, v):
if v is not None and not (0 <= v <= 65535):
raise ValueError("Port must be between 0 and 65535")
return v
class Config:
extra = "forbid"
from_attributes = True
class CallPrompt(BaseModel):
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
owner: Optional[User] = Field(description="Владелец")
name: str = Field(max_length=255, description="Название промпта", example="Приветствие клиенту")
content: str = Field(description="Содержание промпта", example="Добро пожаловать в службу поддержки")
speak_first: bool = Field(default=True, description="Говорить первым")
can_interrupt_greeting: bool = Field(default=False, description="Может прерывать приветствие")
created_at: Optional[str] = Field(description="Дата создания", example="2024-01-01T12:00:00")
updated_at: Optional[str] = Field(description="Дата обновления", example="2024-01-01T12:00:00")
class Config:
extra = "forbid"
from_attributes = True
class YandexSpeechKit(BaseModel):
prompt: Optional[CallPrompt] = Field(description="Промпт",)
api_key: str = Field(..., title="Api Key")
language: str = Field(..., title="Language")
voice: str = Field(..., title="Voice")
emotion: Optional[str] = Field(None, title="Emotion")
speed: float = Field(..., title="Speed")
class OpenAISettings(BaseModel):
prompt: Optional[CallPrompt] = Field(description="Промпт",)
api_key: Optional[str] = Field(None, title="Api Key")
api_url: str = Field(..., title="Api Url")
model: str = Field(..., title="Model")
max_tokens: int = Field(..., title="Max Tokens")
temperature: float = Field(..., title="Temperature")
top_p: float = Field(..., title="Top P")
class YandexFleet(BaseModel):
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
owner: Optional[User] = Field(description="Владелец")
name: str = Field(max_length=255, description="Название флота", example="Флот такси")
park_id: str = Field(max_length=255, description="ID флота в Yandex", example="park-12345")
timezone: str = Field(
max_length=255,
default="Europe/Moscow",
description="Часовой пояс",
example="Europe/Moscow"
)
created_at: Optional[str] = Field(description="Дата создания", example="2024-01-01T12:00:00")
updated_at: Optional[str] = Field(description="Дата обновления", example="2024-01-01T12:00:00")
class Config:
extra = "forbid"
from_attributes = True
class Driver(BaseModel):
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
yandex_fleet: Optional[YandexFleet] = Field(description="ID или название флота Yandex", example="fleet-123")
driver_id: str = Field(max_length=64, description="ID драйвера в Yandex", example="drv-abc123")
created_at: Optional[str] = Field(description="Дата создания", example="2024-01-01T12:00:00")
updated_at: Optional[str] = Field(description="Дата обновления", example="2024-01-01T12:00:00")
class Config:
extra = "forbid"
from_attributes = True
class Call(BaseModel):
unique_id: str = Field(description="Уникальный идентификатор", example="00000000-0000-0000-0000-000000000000")
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
owner: Optional[User] = Field(description="Владелец")
call_datetime: Optional[str] = Field(description="Время последнего звонка (по МСК)", example="2024-01-01T12:00:00")
prompt: Optional[CallPrompt] = Field(description="Промпт",)
driver: Optional[Driver] = Field(description="Водитель")
status: str = Field(description="Статус звонка", example="PENDING", default="PENDING")
data: Optional[Dict[str, Any]] = Field(description="Дополнительные данные", nullable=True)
channel_id: Optional[str] = Field(description="ID канала", example=" tweak-456", null=True)
audio_file: Optional[str] = Field(description="Путь к аудио", example="/audio/call-123.mp3", null=True)
created_at: Optional[str] = Field(description="Дата создания", example="2024-01-01T12:00:00")
updated_at: Optional[str] = Field(description="Дата обновления", example="2024-01-01T12:00:00")
class Config:
extra = "forbid"
from_attributes = True
class CallIn(BaseModel):
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
owner_id: optional[int] = Field(description="Владелец")
call_datetime: Optional[str] = Field(description="Время последнего звонка (по МСК)", example="2024-01-01T12:00:00")
prompt_id: Optional[int] = Field(description="Промпт",)
driver_id: Optional[int] = Field(description="Водитель")
status: Optional[CallStatus] = Field(default=CallStatus.PENDING, description="Статус звонка")
data: Optional[Dict[str, Any]] = Field(description="Дополнительные данные", nullable=True)
channel_id: Optional[str] = Field(description="ID канала", example=" tweak-456", null=True)
audio_file: Optional[str] = Field(description="Путь к аудио", example="/audio/call-123.mp3", null=True)
class Config:
extra = "forbid"
from_attributes = True
class CallDialog(BaseModel):
unique_id: str = Field(description="Уникальный идентификатор", example="00000000-0000-0000-0000-000000000000")
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
call: Optional[Call] = Field(description="ID звонка", example="call-123")
dialog: Optional[str] = Field(description="Содержание диалога", null=True)
data: Optional[Dict[str, Any]] = Field(description="Дополнительные данные", nullable=True)
call_time: int = Field(default=0, description="Время звонка (в секундах)", ge=0)
created_at: Optional[str] = Field(description="Дата создания", example="2024-01-01T12:00:00")
updated_at: Optional[str] = Field(description="Дата обновления", example="2024-01-01T12:00:00")
class Config:
extra = "forbid"
from_attributes = True
class CallDialogIn(BaseModel):
id: Optional[int] = Field(description="Идентификатор (автоматический)", default=None)
call_id: Optional[int] = Field(description="ID звонка", example="call-123")
dialog: Optional[str] = Field(description="Содержание диалога", null=True)
data: Optional[Dict[str, Any]] = Field(description="Дополнительные данные", nullable=True)
call_time: Optional[int] = Field(default=0, description="Время звонка (в секундах)", ge=0)
class Config:
extra = "forbid"
from_attributes = True
class CallDialogPart(BaseModel):
call_dialog: Optional[CallDialog]
side: CallSide = Field(default=CallSide.DRIVER, description="Сторона")
text: str = Field(description="Текст диалога", example="Добрый день")
call_time: int = Field(default=0, description="Время звонка (в секундах)", ge=0)
class Config:
extra = "forbid"
from_attributes = True
class CallDialogPartIn(BaseModel):
call_dialog_id: Optional[int] = Field(description="ID диалога", example="call-123")
side: CallSide = Field(default=CallSide.DRIVER, description="Сторона")
text: str = Field(description="Текст диалога", example="Добрый день")
call_time: Optional[int] = Field(default=0, description="Время звонка (в секундах)", ge=0)
class Config:
extra = "forbid"
from_attributes = True

0
tests/__init__.py Normal file
View file