2153 lines
72 KiB
Python
2153 lines
72 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import Depends, FastAPI, Header, HTTPException
|
|
from sqlalchemy import and_, case, inspect, or_, select, text, update
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from services.shared.core import Role, new_id, utc_now_iso
|
|
from services.shared.db import engine, get_session
|
|
from services.shared.models import (
|
|
AITelegramEnqueueIn,
|
|
EscalateRequest,
|
|
HealthResponse,
|
|
InteractionStatus,
|
|
TelegramThreadAISummaryOut,
|
|
TelegramThreadAIHandoffIn,
|
|
TelegramThreadAIReplyIn,
|
|
TelegramThreadEscalateIn,
|
|
TelegramThreadMessageOut,
|
|
TelegramThreadOut,
|
|
TelegramThreadReplyIn,
|
|
TelegramWebhookIn,
|
|
TelegramWebhookOut,
|
|
)
|
|
from services.shared.security import issue_app_token, require_roles
|
|
from services.shared.sql_init import init_sql_schema
|
|
from services.shared.sql_models import (
|
|
AISessionRow,
|
|
Customer,
|
|
CustomerExternalIdentity,
|
|
Interaction,
|
|
InteractionTimeline,
|
|
TelegramMessageRow,
|
|
TelegramThreadRow,
|
|
)
|
|
|
|
app = FastAPI(title="telegram-adapter-service", version="2.0.0")
|
|
|
|
init_sql_schema()
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TELEGRAM_BOT_DEDUP_INDEX = "ix_telegram_messages_chat_external_unique"
|
|
TELEGRAM_REPLY_NEXT_ATTEMPT_INDEX = "ix_telegram_messages_next_delivery_attempt_at"
|
|
TELEGRAM_REPLY_LOCKED_UNTIL_INDEX = "ix_telegram_messages_delivery_locked_until"
|
|
|
|
_reply_delivery_worker_lock = threading.Lock()
|
|
_reply_delivery_worker_thread: threading.Thread | None = None
|
|
_reply_delivery_worker_stop: threading.Event | None = None
|
|
_reply_delivery_worker_wakeup: threading.Event | None = None
|
|
_reply_delivery_worker_clients = 0
|
|
_HANDOFF_REASON_UNSET = object()
|
|
|
|
|
|
def _utc_now() -> datetime:
|
|
return datetime.now(timezone.utc).replace(microsecond=0)
|
|
|
|
|
|
def _utc_after_seconds(seconds: int) -> str:
|
|
return (_utc_now() + timedelta(seconds=max(seconds, 0))).isoformat()
|
|
|
|
|
|
def _text_column_type() -> str:
|
|
return "TEXT"
|
|
|
|
|
|
def _string_column_type(length: int = 64) -> str:
|
|
return f"VARCHAR({length})"
|
|
|
|
|
|
def _integer_env(name: str, default: int) -> int:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return int(raw.strip())
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def _ensure_telegram_message_indexes() -> None:
|
|
inspector = inspect(engine)
|
|
if "telegram_messages" not in inspector.get_table_names():
|
|
return
|
|
|
|
columns = {item["name"] for item in inspector.get_columns("telegram_messages")}
|
|
indexes = {item["name"] for item in inspector.get_indexes("telegram_messages")}
|
|
|
|
with engine.begin() as conn:
|
|
if "delivery_attempts" not in columns:
|
|
conn.execute(text("ALTER TABLE telegram_messages ADD COLUMN delivery_attempts INTEGER DEFAULT 0"))
|
|
if "next_delivery_attempt_at" not in columns:
|
|
conn.execute(
|
|
text(
|
|
f"ALTER TABLE telegram_messages ADD COLUMN next_delivery_attempt_at {_string_column_type()}"
|
|
)
|
|
)
|
|
if "delivery_locked_until" not in columns:
|
|
conn.execute(
|
|
text(
|
|
f"ALTER TABLE telegram_messages ADD COLUMN delivery_locked_until {_string_column_type()}"
|
|
)
|
|
)
|
|
if "last_delivery_error" not in columns:
|
|
conn.execute(
|
|
text(f"ALTER TABLE telegram_messages ADD COLUMN last_delivery_error {_text_column_type()}")
|
|
)
|
|
|
|
duplicates = conn.execute(
|
|
text(
|
|
"""
|
|
SELECT chat_id, telegram_message_id_external, MIN(id) AS keep_id
|
|
FROM telegram_messages
|
|
WHERE telegram_message_id_external IS NOT NULL
|
|
GROUP BY chat_id, telegram_message_id_external
|
|
HAVING COUNT(*) > 1
|
|
"""
|
|
)
|
|
).mappings()
|
|
for row in duplicates:
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
DELETE FROM telegram_messages
|
|
WHERE chat_id = :chat_id
|
|
AND telegram_message_id_external = :external_message_id
|
|
AND id <> :keep_id
|
|
"""
|
|
),
|
|
{
|
|
"chat_id": row["chat_id"],
|
|
"external_message_id": row["telegram_message_id_external"],
|
|
"keep_id": row["keep_id"],
|
|
},
|
|
)
|
|
if TELEGRAM_BOT_DEDUP_INDEX not in indexes:
|
|
conn.execute(
|
|
text(
|
|
f"""
|
|
CREATE UNIQUE INDEX IF NOT EXISTS {TELEGRAM_BOT_DEDUP_INDEX}
|
|
ON telegram_messages (chat_id, telegram_message_id_external)
|
|
"""
|
|
)
|
|
)
|
|
if TELEGRAM_REPLY_NEXT_ATTEMPT_INDEX not in indexes:
|
|
conn.execute(
|
|
text(
|
|
f"""
|
|
CREATE INDEX IF NOT EXISTS {TELEGRAM_REPLY_NEXT_ATTEMPT_INDEX}
|
|
ON telegram_messages (next_delivery_attempt_at)
|
|
"""
|
|
)
|
|
)
|
|
if TELEGRAM_REPLY_LOCKED_UNTIL_INDEX not in indexes:
|
|
conn.execute(
|
|
text(
|
|
f"""
|
|
CREATE INDEX IF NOT EXISTS {TELEGRAM_REPLY_LOCKED_UNTIL_INDEX}
|
|
ON telegram_messages (delivery_locked_until)
|
|
"""
|
|
)
|
|
)
|
|
|
|
|
|
_ensure_telegram_message_indexes()
|
|
|
|
|
|
def _bool_env(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _bot_enabled() -> bool:
|
|
return _bool_env("TELEGRAM_BOT_ENABLED", False)
|
|
|
|
|
|
def _bot_token() -> str:
|
|
return os.getenv("TELEGRAM_BOT_TOKEN", "").strip()
|
|
|
|
|
|
def _webhook_secret() -> str:
|
|
return os.getenv("TELEGRAM_WEBHOOK_SECRET", "").strip()
|
|
|
|
|
|
def _default_queue_id() -> str:
|
|
return os.getenv("TELEGRAM_DEFAULT_QUEUE_ID", "q_telegram").strip() or "q_telegram"
|
|
|
|
|
|
def _bot_api_base() -> str:
|
|
return os.getenv("TELEGRAM_BOT_API_BASE", "https://api.telegram.org").rstrip("/")
|
|
|
|
|
|
def _reply_delivery_worker_enabled() -> bool:
|
|
return _bool_env("TELEGRAM_REPLY_DELIVERY_WORKER_ENABLED", True)
|
|
|
|
|
|
def _reply_delivery_poll_seconds() -> int:
|
|
return max(1, _integer_env("TELEGRAM_REPLY_DELIVERY_POLL_SECONDS", 5))
|
|
|
|
|
|
def _reply_delivery_lease_seconds() -> int:
|
|
return max(5, _integer_env("TELEGRAM_REPLY_DELIVERY_LEASE_SECONDS", 30))
|
|
|
|
|
|
def _reply_delivery_retry_base_seconds() -> int:
|
|
return max(1, _integer_env("TELEGRAM_REPLY_DELIVERY_RETRY_BASE_SECONDS", 5))
|
|
|
|
|
|
def _reply_delivery_retry_cap_seconds() -> int:
|
|
return max(_reply_delivery_retry_base_seconds(), _integer_env("TELEGRAM_REPLY_DELIVERY_RETRY_CAP_SECONDS", 300))
|
|
|
|
|
|
def _reply_delivery_max_attempts() -> int:
|
|
return max(1, _integer_env("TELEGRAM_REPLY_DELIVERY_MAX_ATTEMPTS", 5))
|
|
|
|
|
|
def _interaction_service_url() -> str:
|
|
return os.getenv("INTERACTION_SERVICE_URL", "http://localhost:8004").rstrip("/")
|
|
|
|
|
|
def _ai_service_url() -> str:
|
|
return os.getenv("AI_ORCHESTRATOR_SERVICE_URL", "http://localhost:8017").rstrip("/")
|
|
|
|
|
|
def _sales_service_url() -> str:
|
|
return os.getenv("SALES_SERVICE_URL", "http://localhost:8020").rstrip("/")
|
|
|
|
|
|
def _sales_telegram_sync_tenant_id() -> str | None:
|
|
return os.getenv("SALES_TELEGRAM_SYNC_TENANT_ID", "").strip() or None
|
|
|
|
|
|
def _ai_telegram_enabled() -> bool:
|
|
return _bool_env("AI_TELEGRAM_ENABLED", False)
|
|
|
|
|
|
def _service_auth_headers() -> dict[str, str]:
|
|
token = issue_app_token(
|
|
subject="svc:telegram-adapter",
|
|
username="telegram-adapter",
|
|
role="admin",
|
|
auth_source="service",
|
|
provider="telegram-adapter",
|
|
ttl_seconds=300,
|
|
)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _internal_service_headers(
|
|
*,
|
|
subject: str,
|
|
username: str,
|
|
provider: str,
|
|
tenant_id: str | None = None,
|
|
) -> dict[str, str]:
|
|
token = issue_app_token(
|
|
subject=subject,
|
|
username=username,
|
|
role="admin",
|
|
auth_source="service",
|
|
provider=provider,
|
|
tenant_id=tenant_id,
|
|
ttl_seconds=300,
|
|
)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _interaction_request(method: str, path: str, *, payload: dict | None = None) -> dict:
|
|
with httpx.Client(timeout=5.0) as client:
|
|
response = client.request(
|
|
method,
|
|
f"{_interaction_service_url()}{path}",
|
|
json=payload,
|
|
headers=_service_auth_headers(),
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def _sales_sync_request(payload: dict[str, Any]) -> None:
|
|
try:
|
|
with httpx.Client(timeout=5.0) as client:
|
|
response = client.post(
|
|
f"{_sales_service_url()}/internal/sales-sync/telegram",
|
|
json=payload,
|
|
headers=_internal_service_headers(
|
|
subject="svc:telegram-adapter",
|
|
username="telegram-adapter",
|
|
provider="telegram-adapter",
|
|
tenant_id=_sales_telegram_sync_tenant_id(),
|
|
),
|
|
)
|
|
response.raise_for_status()
|
|
except Exception:
|
|
logger.exception(
|
|
"Telegram sales sync failed",
|
|
extra={
|
|
"thread_id": payload.get("thread_id"),
|
|
"message_id": payload.get("message_id"),
|
|
"direction": payload.get("direction"),
|
|
},
|
|
)
|
|
|
|
|
|
def _sync_sales_thread(thread: TelegramThreadRow, row: TelegramMessageRow | None = None) -> None:
|
|
payload = {
|
|
"thread_id": thread.thread_id,
|
|
"chat_id": thread.chat_id,
|
|
"interaction_id": thread.interaction_id,
|
|
"customer_id": getattr(row, "customer_id", None),
|
|
"phone_number": None,
|
|
"display_name": thread.display_name,
|
|
"queue_id": thread.queue_id,
|
|
"status": thread.status,
|
|
"ai_state": thread.ai_state,
|
|
"ai_handoff_reason": thread.ai_handoff_reason,
|
|
"message_id": getattr(row, "message_id", None),
|
|
"external_message_id": getattr(row, "telegram_message_id_external", None),
|
|
"text": getattr(row, "text", None),
|
|
"direction": getattr(row, "direction", "inbound"),
|
|
"author_type": getattr(row, "author_type", None),
|
|
"author_id": getattr(row, "author_id", None),
|
|
"happened_at": getattr(row, "created_at", None) or thread.updated_at,
|
|
"metadata": {
|
|
"telegram_user_id": thread.telegram_user_id,
|
|
"username": thread.username,
|
|
"claimed_by_user": thread.claimed_by_user,
|
|
"claimed_at": thread.claimed_at,
|
|
"ai_session_id": thread.ai_session_id,
|
|
"last_message_at": thread.last_message_at,
|
|
"last_message_preview": thread.last_message_preview,
|
|
},
|
|
}
|
|
_sales_sync_request(payload)
|
|
|
|
|
|
def _ai_enqueue_request(thread_id: str, trigger_message_id: str | None) -> None:
|
|
if not _ai_telegram_enabled():
|
|
return
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
response = client.post(
|
|
f"{_ai_service_url()}/ai/telegram/threads/{thread_id}/enqueue",
|
|
json=AITelegramEnqueueIn(trigger_message_id=trigger_message_id).model_dump(),
|
|
headers=_internal_service_headers(
|
|
subject="svc:telegram-adapter",
|
|
username="telegram-adapter",
|
|
provider="telegram-adapter",
|
|
),
|
|
)
|
|
response.raise_for_status()
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("Telegram AI enqueue failed", extra={"thread_id": thread_id})
|
|
|
|
|
|
def _preview_text(text: str, limit: int = 120) -> str:
|
|
value = str(text or "").strip()
|
|
if len(value) <= limit:
|
|
return value
|
|
return f"{value[: limit - 3]}..."
|
|
|
|
|
|
def _normalize_external_subject(value: str | None) -> str | None:
|
|
raw = str(value or "").strip()
|
|
if not raw:
|
|
return None
|
|
if raw.startswith("telegram:"):
|
|
return raw.split(":", 1)[1].strip() or None
|
|
return raw
|
|
|
|
|
|
def _first_external_subject(
|
|
telegram_user_id: str | None,
|
|
chat_id: str,
|
|
explicit: str | None = None,
|
|
) -> str:
|
|
return (
|
|
_normalize_external_subject(telegram_user_id)
|
|
or _normalize_external_subject(explicit)
|
|
or _normalize_external_subject(chat_id)
|
|
or chat_id
|
|
)
|
|
|
|
|
|
def _customer_external_id(telegram_user_id: str | None, chat_id: str, explicit: str | None = None) -> str:
|
|
return f"telegram:{_first_external_subject(telegram_user_id, chat_id, explicit)}"
|
|
|
|
|
|
def _customer_id_is_real(customer_id: str | None) -> bool:
|
|
return str(customer_id or "").startswith("cus_")
|
|
|
|
|
|
def _telegram_identity_subjects(
|
|
telegram_user_id: str | None,
|
|
chat_id: str,
|
|
explicit: str | None = None,
|
|
) -> list[str]:
|
|
ordered = [
|
|
_normalize_external_subject(telegram_user_id),
|
|
_normalize_external_subject(explicit),
|
|
_normalize_external_subject(chat_id),
|
|
]
|
|
seen: set[str] = set()
|
|
result: list[str] = []
|
|
for item in ordered:
|
|
if not item or item in seen:
|
|
continue
|
|
seen.add(item)
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def _ensure_customer_external_identity(
|
|
session,
|
|
*,
|
|
customer_id: str,
|
|
channel: str,
|
|
external_subject: str,
|
|
display_name_snapshot: str | None,
|
|
now: str,
|
|
) -> None:
|
|
row = session.execute(
|
|
select(CustomerExternalIdentity).where(
|
|
CustomerExternalIdentity.channel == channel,
|
|
CustomerExternalIdentity.external_subject == external_subject,
|
|
)
|
|
).scalar_one_or_none()
|
|
if row:
|
|
row.customer_id = customer_id
|
|
if display_name_snapshot:
|
|
row.display_name_snapshot = display_name_snapshot
|
|
row.updated_at = now
|
|
return
|
|
session.add(
|
|
CustomerExternalIdentity(
|
|
identity_id=new_id("cei"),
|
|
customer_id=customer_id,
|
|
channel=channel,
|
|
external_subject=external_subject,
|
|
display_name_snapshot=display_name_snapshot,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
|
|
|
|
def _resolve_or_create_customer_id(
|
|
session,
|
|
*,
|
|
telegram_user_id: str | None,
|
|
chat_id: str,
|
|
display_name: str,
|
|
explicit_customer_external_id: str | None = None,
|
|
) -> str:
|
|
now = utc_now_iso()
|
|
explicit_value = str(explicit_customer_external_id or "").strip()
|
|
subjects = _telegram_identity_subjects(telegram_user_id, chat_id, explicit_value)
|
|
|
|
for subject in subjects:
|
|
identity = session.execute(
|
|
select(CustomerExternalIdentity).where(
|
|
CustomerExternalIdentity.channel == "telegram",
|
|
CustomerExternalIdentity.external_subject == subject,
|
|
)
|
|
).scalar_one_or_none()
|
|
if identity:
|
|
if display_name and identity.display_name_snapshot != display_name:
|
|
identity.display_name_snapshot = display_name
|
|
identity.updated_at = now
|
|
return identity.customer_id
|
|
|
|
if _customer_id_is_real(explicit_value):
|
|
customer = session.execute(
|
|
select(Customer).where(Customer.customer_id == explicit_value)
|
|
).scalar_one_or_none()
|
|
if not customer:
|
|
customer = Customer(
|
|
customer_id=explicit_value,
|
|
display_name=display_name,
|
|
phones_json="[]",
|
|
preferred_phone=None,
|
|
tags_json=json.dumps(["telegram"], ensure_ascii=False),
|
|
created_at=now,
|
|
)
|
|
session.add(customer)
|
|
elif display_name and customer.display_name != display_name:
|
|
customer.display_name = display_name
|
|
customer_id = customer.customer_id
|
|
else:
|
|
customer = Customer(
|
|
customer_id=new_id("cus"),
|
|
display_name=display_name,
|
|
phones_json="[]",
|
|
preferred_phone=None,
|
|
tags_json=json.dumps(["telegram"], ensure_ascii=False),
|
|
created_at=now,
|
|
)
|
|
session.add(customer)
|
|
customer_id = customer.customer_id
|
|
|
|
for subject in subjects or [_normalize_external_subject(chat_id) or chat_id]:
|
|
_ensure_customer_external_identity(
|
|
session,
|
|
customer_id=customer_id,
|
|
channel="telegram",
|
|
external_subject=subject,
|
|
display_name_snapshot=display_name,
|
|
now=now,
|
|
)
|
|
return customer_id
|
|
|
|
|
|
def _display_name(
|
|
*,
|
|
username: str | None = None,
|
|
first_name: str | None = None,
|
|
last_name: str | None = None,
|
|
fallback: str | None = None,
|
|
) -> str:
|
|
parts = [str(first_name or "").strip(), str(last_name or "").strip()]
|
|
full_name = " ".join(part for part in parts if part).strip()
|
|
if full_name:
|
|
return full_name
|
|
if username:
|
|
return username
|
|
if fallback:
|
|
return fallback
|
|
return "Telegram user"
|
|
|
|
|
|
def _inbound_author_type(direction: str) -> str:
|
|
return "system" if direction == "system" else "customer"
|
|
|
|
|
|
def _message_can_trigger_ai(author_type: str | None) -> bool:
|
|
return _ai_telegram_enabled() and author_type == "customer"
|
|
|
|
|
|
def _normalize_phone_number(value: str | None) -> str | None:
|
|
raw = str(value or "").strip()
|
|
if not raw:
|
|
return None
|
|
digits = "".join(ch for ch in raw if ch.isdigit())
|
|
if not digits:
|
|
return None
|
|
if raw.startswith("+"):
|
|
return f"+{digits}"
|
|
return digits
|
|
|
|
|
|
def _contact_phone_number(contact: dict | None, *, sender_user_id: str | None) -> str | None:
|
|
if not isinstance(contact, dict):
|
|
return None
|
|
phone_number = _normalize_phone_number(contact.get("phone_number"))
|
|
if not phone_number:
|
|
return None
|
|
contact_user_id = str(contact.get("user_id") or "").strip() or None
|
|
if sender_user_id and contact_user_id and contact_user_id != sender_user_id:
|
|
return None
|
|
return phone_number
|
|
|
|
|
|
def _customer_phone_list(raw: str | None) -> list[str]:
|
|
try:
|
|
value = json.loads(raw or "[]")
|
|
except Exception: # noqa: BLE001
|
|
return []
|
|
if not isinstance(value, list):
|
|
return []
|
|
result: list[str] = []
|
|
for item in value:
|
|
normalized = _normalize_phone_number(item)
|
|
if normalized and normalized not in result:
|
|
result.append(normalized)
|
|
return result
|
|
|
|
|
|
def _ensure_customer_phone(session, *, customer_id: str, phone_number: str | None) -> None:
|
|
normalized_phone = _normalize_phone_number(phone_number)
|
|
if not normalized_phone or not _customer_id_is_real(customer_id):
|
|
return
|
|
session.flush()
|
|
customer = session.execute(
|
|
select(Customer).where(Customer.customer_id == customer_id)
|
|
).scalar_one_or_none()
|
|
if not customer:
|
|
return
|
|
phones = _customer_phone_list(customer.phones_json)
|
|
if normalized_phone not in phones:
|
|
phones.append(normalized_phone)
|
|
customer.phones_json = json.dumps(phones, ensure_ascii=False)
|
|
if not customer.preferred_phone:
|
|
customer.preferred_phone = normalized_phone
|
|
|
|
|
|
def _push_timeline(session, interaction_id: str, action: str, metadata: dict | None = None) -> None:
|
|
session.add(
|
|
InteractionTimeline(
|
|
interaction_id=interaction_id,
|
|
timestamp=utc_now_iso(),
|
|
action=action,
|
|
metadata_json=json.dumps(metadata or {}, ensure_ascii=False),
|
|
)
|
|
)
|
|
|
|
|
|
def _get_thread(session, thread_id: str) -> TelegramThreadRow:
|
|
row = session.execute(
|
|
select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)
|
|
).scalar_one_or_none()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Telegram thread not found")
|
|
return row
|
|
|
|
|
|
def _to_thread_out(row: TelegramThreadRow) -> TelegramThreadOut:
|
|
return TelegramThreadOut(
|
|
thread_id=row.thread_id,
|
|
chat_id=row.chat_id,
|
|
interaction_id=row.interaction_id,
|
|
telegram_user_id=row.telegram_user_id,
|
|
username=row.username,
|
|
display_name=row.display_name,
|
|
queue_id=row.queue_id,
|
|
status=row.status, # type: ignore[arg-type]
|
|
claimed_by_user=row.claimed_by_user,
|
|
claimed_at=row.claimed_at,
|
|
ai_session_id=row.ai_session_id,
|
|
ai_state=row.ai_state, # type: ignore[arg-type]
|
|
ai_handoff_reason=row.ai_handoff_reason,
|
|
ai_last_model_at=row.ai_last_model_at,
|
|
last_message_at=row.last_message_at,
|
|
last_message_preview=row.last_message_preview,
|
|
created_at=row.created_at,
|
|
updated_at=row.updated_at,
|
|
)
|
|
|
|
|
|
def _to_message_out(row: TelegramMessageRow) -> TelegramThreadMessageOut:
|
|
return TelegramThreadMessageOut(
|
|
message_id=row.message_id,
|
|
thread_id=row.thread_id or "",
|
|
interaction_id=row.interaction_id or "",
|
|
chat_id=row.chat_id,
|
|
direction=row.direction, # type: ignore[arg-type]
|
|
text=row.text,
|
|
telegram_message_id_external=row.telegram_message_id_external,
|
|
operator_user=row.operator_user,
|
|
author_type=row.author_type, # type: ignore[arg-type]
|
|
author_id=row.author_id,
|
|
delivery_status=row.delivery_status,
|
|
payload=json.loads(row.payload_json or "{}"),
|
|
created_at=row.created_at,
|
|
)
|
|
|
|
|
|
def _loads_payload(raw: str | None) -> dict[str, Any]:
|
|
try:
|
|
payload = json.loads(raw or "{}")
|
|
except Exception: # noqa: BLE001
|
|
return {}
|
|
return payload if isinstance(payload, dict) else {}
|
|
|
|
|
|
def _to_webhook_out(row: TelegramMessageRow) -> TelegramWebhookOut:
|
|
return TelegramWebhookOut(
|
|
message_id=row.message_id,
|
|
chat_id=row.chat_id,
|
|
text=row.text,
|
|
customer_external_id=row.customer_external_id,
|
|
payload=_loads_payload(row.payload_json),
|
|
thread_id=row.thread_id,
|
|
interaction_id=row.interaction_id,
|
|
direction=row.direction, # type: ignore[arg-type]
|
|
created_at=row.created_at,
|
|
)
|
|
|
|
|
|
def _to_bot_webhook_result(row: TelegramMessageRow, *, thread_created: bool) -> dict[str, Any]:
|
|
return {
|
|
"ok": True,
|
|
"thread_id": row.thread_id or "",
|
|
"interaction_id": row.interaction_id or "",
|
|
"message_id": row.message_id,
|
|
"thread_created": thread_created,
|
|
}
|
|
|
|
|
|
def _existing_bot_message(
|
|
session,
|
|
*,
|
|
chat_id: str,
|
|
external_message_id: str | None,
|
|
) -> TelegramMessageRow | None:
|
|
if not external_message_id:
|
|
return None
|
|
return session.execute(
|
|
select(TelegramMessageRow).where(
|
|
TelegramMessageRow.chat_id == chat_id,
|
|
TelegramMessageRow.telegram_message_id_external == external_message_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _latest_ai_session_for_thread(session, thread: TelegramThreadRow) -> AISessionRow | None:
|
|
if thread.ai_session_id:
|
|
ai_session = session.execute(
|
|
select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id)
|
|
).scalar_one_or_none()
|
|
if ai_session:
|
|
return ai_session
|
|
return session.execute(
|
|
select(AISessionRow)
|
|
.where(AISessionRow.thread_id == thread.thread_id)
|
|
.order_by(AISessionRow.updated_at.desc(), AISessionRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _summary_customer_request_text(
|
|
session,
|
|
*,
|
|
thread: TelegramThreadRow,
|
|
ai_session: AISessionRow,
|
|
generated_at: str,
|
|
) -> str:
|
|
if ai_session.last_user_message_id:
|
|
customer_message = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == ai_session.last_user_message_id)
|
|
).scalar_one_or_none()
|
|
if customer_message and customer_message.text.strip():
|
|
return customer_message.text.strip()
|
|
customer_message = session.execute(
|
|
select(TelegramMessageRow)
|
|
.where(
|
|
TelegramMessageRow.thread_id == thread.thread_id,
|
|
TelegramMessageRow.direction == "inbound",
|
|
TelegramMessageRow.author_type == "customer",
|
|
TelegramMessageRow.created_at <= generated_at,
|
|
)
|
|
.order_by(TelegramMessageRow.created_at.desc(), TelegramMessageRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if customer_message and customer_message.text.strip():
|
|
return customer_message.text.strip()
|
|
return "Последний запрос клиента недоступен."
|
|
|
|
|
|
def _summary_ai_outcome_text(
|
|
session,
|
|
*,
|
|
thread: TelegramThreadRow,
|
|
ai_session: AISessionRow,
|
|
generated_at: str,
|
|
) -> str:
|
|
if ai_session.last_ai_message_id:
|
|
ai_message = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == ai_session.last_ai_message_id)
|
|
).scalar_one_or_none()
|
|
if ai_message and ai_message.text.strip():
|
|
return ai_message.text.strip()
|
|
ai_message = session.execute(
|
|
select(TelegramMessageRow)
|
|
.where(
|
|
TelegramMessageRow.thread_id == thread.thread_id,
|
|
TelegramMessageRow.direction == "outbound",
|
|
TelegramMessageRow.author_type == "ai",
|
|
TelegramMessageRow.created_at <= generated_at,
|
|
)
|
|
.order_by(TelegramMessageRow.created_at.desc(), TelegramMessageRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if ai_message and ai_message.text.strip():
|
|
return ai_message.text.strip()
|
|
return "AI передал диалог оператору без ответа клиенту."
|
|
|
|
|
|
def _summary_status(
|
|
session,
|
|
*,
|
|
thread: TelegramThreadRow,
|
|
ai_session: AISessionRow,
|
|
generated_at: str,
|
|
) -> tuple[str, str]:
|
|
if ai_session.last_ai_message_id:
|
|
ai_message = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == ai_session.last_ai_message_id)
|
|
).scalar_one_or_none()
|
|
if ai_message and ai_message.text.strip():
|
|
return ("AI ответил клиенту", "answered")
|
|
ai_message = session.execute(
|
|
select(TelegramMessageRow.message_id)
|
|
.where(
|
|
TelegramMessageRow.thread_id == thread.thread_id,
|
|
TelegramMessageRow.direction == "outbound",
|
|
TelegramMessageRow.author_type == "ai",
|
|
TelegramMessageRow.created_at <= generated_at,
|
|
)
|
|
.order_by(TelegramMessageRow.created_at.desc(), TelegramMessageRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if ai_message:
|
|
return ("AI ответил клиенту", "answered")
|
|
return ("AI передал без ответа", "handoff")
|
|
|
|
|
|
def _summary_recommended_next_step(thread: TelegramThreadRow) -> str:
|
|
if thread.ai_state == "handoff_required":
|
|
return "Заберите чат и ответьте клиенту вручную."
|
|
if thread.ai_state == "human_owned":
|
|
return "Продолжайте диалог вручную; AI больше не отвечает в этот thread."
|
|
return ""
|
|
|
|
|
|
def _thread_supports_ai_summary(thread: TelegramThreadRow, ai_session: AISessionRow | None) -> bool:
|
|
if not ai_session:
|
|
return False
|
|
if thread.ai_state == "handoff_required":
|
|
return True
|
|
if thread.ai_state == "human_owned":
|
|
return bool(thread.ai_handoff_reason or ai_session.handoff_reason)
|
|
return False
|
|
|
|
|
|
def _build_ai_summary(session, thread: TelegramThreadRow) -> TelegramThreadAISummaryOut | None:
|
|
ai_session = _latest_ai_session_for_thread(session, thread)
|
|
if not _thread_supports_ai_summary(thread, ai_session):
|
|
return None
|
|
assert ai_session is not None
|
|
generated_at = thread.ai_last_model_at or ai_session.updated_at or thread.updated_at
|
|
handoff_reason = (thread.ai_handoff_reason or ai_session.handoff_reason or "").strip()
|
|
if not handoff_reason:
|
|
return None
|
|
status_label, status_tone = _summary_status(
|
|
session,
|
|
thread=thread,
|
|
ai_session=ai_session,
|
|
generated_at=generated_at,
|
|
)
|
|
return TelegramThreadAISummaryOut(
|
|
thread_id=thread.thread_id,
|
|
session_id=ai_session.session_id,
|
|
status_label=status_label,
|
|
status_tone=status_tone, # type: ignore[arg-type]
|
|
customer_request_text=_summary_customer_request_text(
|
|
session,
|
|
thread=thread,
|
|
ai_session=ai_session,
|
|
generated_at=generated_at,
|
|
),
|
|
ai_outcome_text=_summary_ai_outcome_text(
|
|
session,
|
|
thread=thread,
|
|
ai_session=ai_session,
|
|
generated_at=generated_at,
|
|
),
|
|
handoff_reason=handoff_reason,
|
|
recommended_next_step=_summary_recommended_next_step(thread),
|
|
generated_at=generated_at,
|
|
)
|
|
|
|
|
|
def _reactivate_thread_interaction(
|
|
session,
|
|
thread: TelegramThreadRow,
|
|
now: str,
|
|
*,
|
|
requeue_ai: bool,
|
|
) -> None:
|
|
interaction = session.execute(
|
|
select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
|
|
).scalar_one_or_none()
|
|
if not interaction:
|
|
return
|
|
interaction.status = "new"
|
|
interaction.assigned_to = None
|
|
interaction.updated_at = now
|
|
thread.status = "new"
|
|
thread.claimed_by_user = None
|
|
thread.claimed_at = None
|
|
thread.ai_handoff_reason = None
|
|
if thread.ai_state != "human_owned":
|
|
thread.ai_state = "queued" if requeue_ai else None
|
|
thread.updated_at = now
|
|
_push_timeline(
|
|
session,
|
|
interaction.interaction_id,
|
|
"interaction.status_changed",
|
|
{"status": "new", "source": "telegram-reactivation"},
|
|
)
|
|
_push_timeline(
|
|
session,
|
|
interaction.interaction_id,
|
|
"telegram.thread_reactivated",
|
|
{"thread_id": thread.thread_id, "chat_id": thread.chat_id},
|
|
)
|
|
|
|
|
|
def _create_interaction(session, *, subject: str, customer_id: str, queue_id: str, created_at: str) -> str:
|
|
interaction_id = new_id("int")
|
|
session.add(
|
|
Interaction(
|
|
interaction_id=interaction_id,
|
|
channel="telegram",
|
|
subject=_preview_text(subject, 120),
|
|
customer_id=customer_id,
|
|
queue_id=queue_id,
|
|
priority=3,
|
|
status="new",
|
|
assigned_to=None,
|
|
created_at=created_at,
|
|
updated_at=created_at,
|
|
)
|
|
)
|
|
_push_timeline(session, interaction_id, "interaction.created", {"channel": "telegram"})
|
|
return interaction_id
|
|
|
|
|
|
def _persist_message(
|
|
session,
|
|
*,
|
|
thread: TelegramThreadRow,
|
|
text: str,
|
|
payload: dict,
|
|
direction: str,
|
|
created_at: str,
|
|
customer_external_id: str | None = None,
|
|
telegram_message_id_external: str | None = None,
|
|
operator_user: str | None = None,
|
|
author_type: str | None = None,
|
|
author_id: str | None = None,
|
|
delivery_status: str | None = None,
|
|
) -> TelegramMessageRow:
|
|
resolved_author_type = author_type or ("human" if direction == "outbound" else ("system" if direction == "system" else "customer"))
|
|
row = TelegramMessageRow(
|
|
message_id=new_id("tgm"),
|
|
thread_id=thread.thread_id,
|
|
interaction_id=thread.interaction_id,
|
|
chat_id=thread.chat_id,
|
|
text=text,
|
|
customer_external_id=customer_external_id,
|
|
direction=direction,
|
|
telegram_message_id_external=telegram_message_id_external,
|
|
operator_user=operator_user,
|
|
author_type=resolved_author_type,
|
|
author_id=author_id,
|
|
delivery_status=delivery_status,
|
|
payload_json=json.dumps(payload, ensure_ascii=False),
|
|
created_at=created_at,
|
|
)
|
|
session.add(row)
|
|
thread.last_message_at = created_at
|
|
thread.last_message_preview = _preview_text(text, 140)
|
|
thread.updated_at = created_at
|
|
return row
|
|
|
|
|
|
def _mark_thread_ai_state(
|
|
thread: TelegramThreadRow,
|
|
*,
|
|
ai_state: str | None,
|
|
ai_handoff_reason: str | None = None,
|
|
ai_last_model_at: str | None = None,
|
|
) -> None:
|
|
thread.ai_state = ai_state
|
|
thread.ai_handoff_reason = ai_handoff_reason
|
|
if ai_last_model_at is not None:
|
|
thread.ai_last_model_at = ai_last_model_at
|
|
|
|
|
|
def _update_ai_session_state(
|
|
session,
|
|
*,
|
|
session_id: str | None,
|
|
status: str,
|
|
updated_at: str,
|
|
handoff_reason: object = _HANDOFF_REASON_UNSET,
|
|
closed: bool = False,
|
|
) -> None:
|
|
if not session_id:
|
|
return
|
|
ai_session = session.execute(
|
|
select(AISessionRow).where(AISessionRow.session_id == session_id)
|
|
).scalar_one_or_none()
|
|
if not ai_session:
|
|
return
|
|
ai_session.status = status
|
|
ai_session.updated_at = updated_at
|
|
if handoff_reason is not _HANDOFF_REASON_UNSET:
|
|
ai_session.handoff_reason = handoff_reason
|
|
if closed:
|
|
ai_session.closed_at = updated_at
|
|
|
|
|
|
def _maybe_enqueue_ai_for_thread(
|
|
thread: TelegramThreadRow,
|
|
trigger_message_id: str | None,
|
|
*,
|
|
author_type: str | None = None,
|
|
) -> None:
|
|
if not _ai_telegram_enabled():
|
|
return
|
|
if thread.ai_state == "human_owned":
|
|
return
|
|
if author_type and author_type != "customer":
|
|
return
|
|
threading.Thread(
|
|
target=_ai_enqueue_request,
|
|
args=(thread.thread_id, trigger_message_id),
|
|
daemon=True,
|
|
name=f"telegram-ai-enqueue-{thread.thread_id}",
|
|
).start()
|
|
|
|
|
|
def _upsert_thread_for_inbound(
|
|
session,
|
|
*,
|
|
chat_id: str,
|
|
telegram_user_id: str | None,
|
|
username: str | None,
|
|
display_name: str,
|
|
text: str,
|
|
payload: dict,
|
|
customer_external_id: str | None = None,
|
|
external_message_id: str | None = None,
|
|
phone_number: str | None = None,
|
|
queue_id: str | None = None,
|
|
direction: str = "inbound",
|
|
) -> tuple[TelegramThreadRow, TelegramMessageRow, bool]:
|
|
now = utc_now_iso()
|
|
author_type = _inbound_author_type(direction)
|
|
ai_eligible = _message_can_trigger_ai(author_type)
|
|
thread = session.execute(
|
|
select(TelegramThreadRow).where(TelegramThreadRow.chat_id == chat_id)
|
|
).scalar_one_or_none()
|
|
created = False
|
|
assigned_customer_id = _resolve_or_create_customer_id(
|
|
session,
|
|
telegram_user_id=telegram_user_id,
|
|
chat_id=chat_id,
|
|
display_name=display_name,
|
|
explicit_customer_external_id=customer_external_id,
|
|
)
|
|
_ensure_customer_phone(session, customer_id=assigned_customer_id, phone_number=phone_number)
|
|
resolved_queue_id = queue_id or (thread.queue_id if thread else None) or _default_queue_id()
|
|
|
|
if not thread:
|
|
interaction_id = _create_interaction(
|
|
session,
|
|
subject=text,
|
|
customer_id=assigned_customer_id,
|
|
queue_id=resolved_queue_id,
|
|
created_at=now,
|
|
)
|
|
thread = TelegramThreadRow(
|
|
thread_id=new_id("tgt"),
|
|
chat_id=chat_id,
|
|
interaction_id=interaction_id,
|
|
telegram_user_id=telegram_user_id,
|
|
username=username,
|
|
display_name=display_name,
|
|
queue_id=resolved_queue_id,
|
|
status="new",
|
|
claimed_by_user=None,
|
|
claimed_at=None,
|
|
ai_session_id=None,
|
|
ai_state="queued" if ai_eligible else None,
|
|
ai_handoff_reason=None,
|
|
ai_last_model_at=None,
|
|
last_message_at=now,
|
|
last_message_preview=_preview_text(text, 140),
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(thread)
|
|
created = True
|
|
else:
|
|
if thread.status == "closed":
|
|
_reactivate_thread_interaction(session, thread, now, requeue_ai=ai_eligible)
|
|
thread.telegram_user_id = telegram_user_id or thread.telegram_user_id
|
|
thread.username = username or thread.username
|
|
thread.display_name = display_name or thread.display_name
|
|
thread.queue_id = thread.queue_id or resolved_queue_id
|
|
interaction = session.execute(
|
|
select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
|
|
).scalar_one_or_none()
|
|
if interaction and interaction.customer_id != assigned_customer_id:
|
|
interaction.customer_id = assigned_customer_id
|
|
interaction.updated_at = now
|
|
if ai_eligible and thread.ai_state != "human_owned":
|
|
_mark_thread_ai_state(thread, ai_state="queued", ai_handoff_reason=None)
|
|
|
|
message_row = _persist_message(
|
|
session,
|
|
thread=thread,
|
|
text=text,
|
|
payload=payload,
|
|
direction=direction,
|
|
created_at=now,
|
|
customer_external_id=assigned_customer_id,
|
|
telegram_message_id_external=external_message_id,
|
|
author_type=author_type,
|
|
author_id=_first_external_subject(telegram_user_id, chat_id, customer_external_id),
|
|
delivery_status="received",
|
|
)
|
|
_push_timeline(
|
|
session,
|
|
thread.interaction_id,
|
|
"telegram.message_received",
|
|
{
|
|
"thread_id": thread.thread_id,
|
|
"chat_id": chat_id,
|
|
"direction": direction,
|
|
"username": username,
|
|
},
|
|
)
|
|
session.flush()
|
|
return thread, message_row, created
|
|
|
|
|
|
def _manual_webhook_metadata(payload: TelegramWebhookIn) -> dict[str, Any]:
|
|
raw_payload = payload.payload if isinstance(payload.payload, dict) else {}
|
|
sender = raw_payload.get("from") if isinstance(raw_payload.get("from"), dict) else {}
|
|
username = str(raw_payload.get("username") or sender.get("username") or "").strip() or None
|
|
first_name = str(raw_payload.get("first_name") or sender.get("first_name") or "").strip() or None
|
|
last_name = str(raw_payload.get("last_name") or sender.get("last_name") or "").strip() or None
|
|
telegram_user_id = str(raw_payload.get("telegram_user_id") or sender.get("id") or "").strip() or None
|
|
display_name = _display_name(
|
|
username=username,
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
fallback=f"Telegram {payload.chat_id}",
|
|
)
|
|
contact = raw_payload.get("contact") if isinstance(raw_payload.get("contact"), dict) else {}
|
|
phone_number = _contact_phone_number(contact, sender_user_id=telegram_user_id) or _normalize_phone_number(
|
|
raw_payload.get("phone_number")
|
|
)
|
|
queue_id = str(raw_payload.get("queue_id") or "").strip() or None
|
|
return {
|
|
"telegram_user_id": telegram_user_id,
|
|
"username": username,
|
|
"display_name": display_name,
|
|
"phone_number": phone_number,
|
|
"queue_id": queue_id,
|
|
}
|
|
|
|
|
|
def _parse_bot_update(update: dict) -> dict:
|
|
message = update.get("message") or update.get("edited_message")
|
|
if not isinstance(message, dict):
|
|
raise HTTPException(status_code=400, detail="Unsupported Telegram update")
|
|
|
|
chat = message.get("chat") if isinstance(message.get("chat"), dict) else {}
|
|
sender = message.get("from") if isinstance(message.get("from"), dict) else {}
|
|
chat_id = str(chat.get("id") or "").strip()
|
|
if not chat_id:
|
|
raise HTTPException(status_code=400, detail="Telegram chat_id is missing")
|
|
|
|
external_message_id = str(message.get("message_id") or "").strip() or None
|
|
username = str(sender.get("username") or chat.get("username") or "").strip() or None
|
|
first_name = str(sender.get("first_name") or "").strip() or None
|
|
last_name = str(sender.get("last_name") or "").strip() or None
|
|
telegram_user_id = str(sender.get("id") or chat.get("id") or "").strip() or None
|
|
contact = message.get("contact") if isinstance(message.get("contact"), dict) else {}
|
|
phone_number = _contact_phone_number(contact, sender_user_id=telegram_user_id)
|
|
text = str(message.get("text") or message.get("caption") or "").strip()
|
|
direction = "inbound"
|
|
if not text:
|
|
if phone_number:
|
|
text = "[Shared Telegram contact]"
|
|
direction = "system"
|
|
else:
|
|
content_kind = "message"
|
|
for candidate in ("photo", "document", "audio", "video", "voice", "sticker", "location", "contact"):
|
|
if candidate in message:
|
|
content_kind = candidate
|
|
break
|
|
text = f"[Unsupported Telegram content: {content_kind}]"
|
|
direction = "system"
|
|
|
|
display_name = _display_name(
|
|
username=username,
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
fallback=str(chat.get("title") or f"Telegram {chat_id}"),
|
|
)
|
|
return {
|
|
"chat_id": chat_id,
|
|
"telegram_user_id": telegram_user_id,
|
|
"username": username,
|
|
"display_name": display_name,
|
|
"text": text,
|
|
"payload": update,
|
|
"external_message_id": external_message_id,
|
|
"phone_number": phone_number,
|
|
"direction": direction,
|
|
}
|
|
|
|
|
|
def _send_telegram_message(chat_id: str, text: str) -> dict:
|
|
if not _bot_enabled() or not _bot_token():
|
|
raise HTTPException(status_code=503, detail="Telegram bot sending is not configured")
|
|
with httpx.Client(timeout=10.0) as client:
|
|
response = client.post(
|
|
f"{_bot_api_base()}/bot{_bot_token()}/sendMessage",
|
|
json={"chat_id": chat_id, "text": text},
|
|
)
|
|
if response.status_code >= 400:
|
|
raise HTTPException(status_code=502, detail=f"Telegram sendMessage failed: {response.text}")
|
|
payload = response.json()
|
|
if not payload.get("ok"):
|
|
raise HTTPException(status_code=502, detail=f"Telegram sendMessage failed: {payload}")
|
|
return payload
|
|
|
|
|
|
def _telegram_reply_delivery_filter(now: str):
|
|
return or_(
|
|
TelegramMessageRow.delivery_status == "pending",
|
|
TelegramMessageRow.delivery_status == "retrying",
|
|
and_(
|
|
TelegramMessageRow.delivery_status == "sending",
|
|
or_(
|
|
TelegramMessageRow.delivery_locked_until.is_(None),
|
|
TelegramMessageRow.delivery_locked_until < now,
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _telegram_reply_delivery_due_filter(now: str):
|
|
return or_(
|
|
TelegramMessageRow.next_delivery_attempt_at.is_(None),
|
|
TelegramMessageRow.next_delivery_attempt_at <= now,
|
|
)
|
|
|
|
|
|
def _delivery_error_text(error: Exception) -> str:
|
|
if isinstance(error, HTTPException):
|
|
detail = error.detail
|
|
if isinstance(detail, str):
|
|
return detail[:1000]
|
|
return str(detail)[:1000]
|
|
return str(error or error.__class__.__name__)[:1000]
|
|
|
|
|
|
def _telegram_reply_backoff_seconds(next_attempt: int) -> int:
|
|
base = _reply_delivery_retry_base_seconds()
|
|
cap = _reply_delivery_retry_cap_seconds()
|
|
return min(cap, base * (2 ** max(next_attempt - 1, 0)))
|
|
|
|
|
|
def _claim_telegram_reply_delivery(session, message_id: str) -> TelegramMessageRow | None:
|
|
now = utc_now_iso()
|
|
leased_until = _utc_after_seconds(_reply_delivery_lease_seconds())
|
|
claim = session.execute(
|
|
update(TelegramMessageRow)
|
|
.where(
|
|
TelegramMessageRow.message_id == message_id,
|
|
TelegramMessageRow.direction == "outbound",
|
|
TelegramMessageRow.thread_id.is_not(None),
|
|
TelegramMessageRow.interaction_id.is_not(None),
|
|
_telegram_reply_delivery_filter(now),
|
|
_telegram_reply_delivery_due_filter(now),
|
|
)
|
|
.values(
|
|
delivery_status="sending",
|
|
delivery_locked_until=leased_until,
|
|
)
|
|
)
|
|
session.commit()
|
|
if claim.rowcount != 1:
|
|
return None
|
|
return session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _mark_telegram_reply_delivery_failure(
|
|
session,
|
|
row: TelegramMessageRow,
|
|
error_text: str,
|
|
*,
|
|
allow_retry: bool = True,
|
|
) -> None:
|
|
attempts = int(row.delivery_attempts or 0) + 1
|
|
row.delivery_attempts = attempts
|
|
row.delivery_locked_until = None
|
|
row.last_delivery_error = error_text
|
|
should_retry = allow_retry and attempts < _reply_delivery_max_attempts()
|
|
if should_retry:
|
|
row.delivery_status = "retrying"
|
|
row.next_delivery_attempt_at = _utc_after_seconds(_telegram_reply_backoff_seconds(attempts))
|
|
else:
|
|
row.delivery_status = "failed"
|
|
row.next_delivery_attempt_at = None
|
|
if row.interaction_id and row.thread_id:
|
|
_push_timeline(
|
|
session,
|
|
row.interaction_id,
|
|
"telegram.message_failed",
|
|
{
|
|
"thread_id": row.thread_id,
|
|
"operator_user": row.operator_user,
|
|
"attempts": attempts,
|
|
"error": error_text,
|
|
},
|
|
)
|
|
session.commit()
|
|
|
|
|
|
def _mark_telegram_reply_delivery_success(session, row: TelegramMessageRow, telegram_payload: dict) -> None:
|
|
row.delivery_attempts = int(row.delivery_attempts or 0) + 1
|
|
row.delivery_status = "sent"
|
|
row.delivery_locked_until = None
|
|
row.next_delivery_attempt_at = None
|
|
row.last_delivery_error = None
|
|
result_message = telegram_payload.get("result") if isinstance(telegram_payload, dict) else {}
|
|
if isinstance(result_message, dict) and result_message.get("message_id") is not None:
|
|
row.telegram_message_id_external = str(result_message.get("message_id"))
|
|
_push_timeline(
|
|
session,
|
|
row.interaction_id,
|
|
"telegram.message_sent",
|
|
{"thread_id": row.thread_id, "operator_user": row.operator_user},
|
|
)
|
|
session.commit()
|
|
|
|
|
|
def _deliver_pending_telegram_reply(message_id: str) -> bool:
|
|
session = get_session()
|
|
try:
|
|
row = _claim_telegram_reply_delivery(session, message_id)
|
|
if not row or not row.thread_id or not row.interaction_id:
|
|
return False
|
|
|
|
thread = session.execute(
|
|
select(TelegramThreadRow).where(TelegramThreadRow.thread_id == row.thread_id)
|
|
).scalar_one_or_none()
|
|
if not thread:
|
|
_mark_telegram_reply_delivery_failure(
|
|
session,
|
|
row,
|
|
"Telegram thread not found during reply delivery",
|
|
allow_retry=False,
|
|
)
|
|
return False
|
|
|
|
try:
|
|
telegram_payload = _send_telegram_message(row.chat_id, row.text)
|
|
except Exception as exc: # noqa: BLE001
|
|
_mark_telegram_reply_delivery_failure(session, row, _delivery_error_text(exc))
|
|
return False
|
|
|
|
_mark_telegram_reply_delivery_success(session, row, telegram_payload)
|
|
return True
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _claim_next_due_telegram_reply_message_id(session) -> str | None:
|
|
now = utc_now_iso()
|
|
return session.execute(
|
|
select(TelegramMessageRow.message_id)
|
|
.where(
|
|
TelegramMessageRow.direction == "outbound",
|
|
TelegramMessageRow.thread_id.is_not(None),
|
|
TelegramMessageRow.interaction_id.is_not(None),
|
|
_telegram_reply_delivery_filter(now),
|
|
_telegram_reply_delivery_due_filter(now),
|
|
)
|
|
.order_by(
|
|
case(
|
|
(TelegramMessageRow.delivery_status == "sending", 0),
|
|
(TelegramMessageRow.delivery_status == "retrying", 1),
|
|
else_=2,
|
|
),
|
|
TelegramMessageRow.created_at.asc(),
|
|
TelegramMessageRow.id.asc(),
|
|
)
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _process_due_telegram_reply_batch(batch_size: int = 20) -> int:
|
|
processed = 0
|
|
attempts = 0
|
|
while processed < batch_size and attempts < batch_size * 2:
|
|
attempts += 1
|
|
session = get_session()
|
|
try:
|
|
message_id = _claim_next_due_telegram_reply_message_id(session)
|
|
finally:
|
|
session.close()
|
|
if not message_id:
|
|
break
|
|
if _deliver_pending_telegram_reply(message_id):
|
|
processed += 1
|
|
return processed
|
|
|
|
|
|
def _reply_delivery_worker_loop(stop_event: threading.Event, wakeup_event: threading.Event) -> None:
|
|
while not stop_event.is_set():
|
|
wakeup_event.wait(timeout=_reply_delivery_poll_seconds())
|
|
wakeup_event.clear()
|
|
if stop_event.is_set():
|
|
break
|
|
try:
|
|
_process_due_telegram_reply_batch()
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("Telegram reply delivery worker cycle failed")
|
|
|
|
|
|
def _wake_telegram_reply_delivery_worker() -> None:
|
|
with _reply_delivery_worker_lock:
|
|
wakeup_event = _reply_delivery_worker_wakeup
|
|
if wakeup_event:
|
|
wakeup_event.set()
|
|
|
|
|
|
def _start_reply_delivery_worker() -> None:
|
|
global _reply_delivery_worker_clients
|
|
global _reply_delivery_worker_stop
|
|
global _reply_delivery_worker_thread
|
|
global _reply_delivery_worker_wakeup
|
|
|
|
if not _reply_delivery_worker_enabled():
|
|
return
|
|
|
|
with _reply_delivery_worker_lock:
|
|
_reply_delivery_worker_clients += 1
|
|
if _reply_delivery_worker_thread and _reply_delivery_worker_thread.is_alive():
|
|
wakeup_event = _reply_delivery_worker_wakeup
|
|
else:
|
|
stop_event = threading.Event()
|
|
wakeup_event = threading.Event()
|
|
worker = threading.Thread(
|
|
target=_reply_delivery_worker_loop,
|
|
args=(stop_event, wakeup_event),
|
|
name="telegram-reply-delivery-worker",
|
|
daemon=True,
|
|
)
|
|
_reply_delivery_worker_stop = stop_event
|
|
_reply_delivery_worker_wakeup = wakeup_event
|
|
_reply_delivery_worker_thread = worker
|
|
worker.start()
|
|
if wakeup_event:
|
|
wakeup_event.set()
|
|
|
|
|
|
def _stop_reply_delivery_worker() -> None:
|
|
global _reply_delivery_worker_clients
|
|
global _reply_delivery_worker_stop
|
|
global _reply_delivery_worker_thread
|
|
global _reply_delivery_worker_wakeup
|
|
|
|
thread: threading.Thread | None = None
|
|
stop_event: threading.Event | None = None
|
|
wakeup_event: threading.Event | None = None
|
|
|
|
with _reply_delivery_worker_lock:
|
|
if _reply_delivery_worker_clients > 0:
|
|
_reply_delivery_worker_clients -= 1
|
|
if _reply_delivery_worker_clients > 0:
|
|
return
|
|
thread = _reply_delivery_worker_thread
|
|
stop_event = _reply_delivery_worker_stop
|
|
wakeup_event = _reply_delivery_worker_wakeup
|
|
_reply_delivery_worker_thread = None
|
|
_reply_delivery_worker_stop = None
|
|
_reply_delivery_worker_wakeup = None
|
|
|
|
if stop_event:
|
|
stop_event.set()
|
|
if wakeup_event:
|
|
wakeup_event.set()
|
|
if thread and thread.is_alive():
|
|
thread.join(timeout=2)
|
|
|
|
|
|
def _start_telegram_reply_delivery(message_id: str) -> None:
|
|
if _reply_delivery_worker_enabled():
|
|
_wake_telegram_reply_delivery_worker()
|
|
return
|
|
threading.Thread(
|
|
target=_deliver_pending_telegram_reply,
|
|
args=(message_id,),
|
|
daemon=True,
|
|
).start()
|
|
|
|
|
|
@app.on_event("startup")
|
|
def _start_delivery_worker_on_startup() -> None:
|
|
_start_reply_delivery_worker()
|
|
|
|
|
|
@app.on_event("shutdown")
|
|
def _stop_delivery_worker_on_shutdown() -> None:
|
|
_stop_reply_delivery_worker()
|
|
|
|
|
|
def _ensure_operator_can_manage(
|
|
actor: dict,
|
|
thread: TelegramThreadRow,
|
|
*,
|
|
for_reply: bool = False,
|
|
require_claim: bool = False,
|
|
) -> None:
|
|
if actor["role"] in {Role.ADMIN.value, Role.SUPERVISOR.value}:
|
|
return
|
|
if actor["role"] != Role.OPERATOR.value:
|
|
raise HTTPException(status_code=403, detail="Insufficient role")
|
|
if for_reply and thread.claimed_by_user != actor["user"]:
|
|
raise HTTPException(status_code=403, detail="Thread must be claimed by the current operator")
|
|
if require_claim and thread.claimed_by_user != actor["user"]:
|
|
raise HTTPException(status_code=403, detail="Thread must be claimed by the current operator")
|
|
if not for_reply and thread.claimed_by_user and thread.claimed_by_user != actor["user"]:
|
|
raise HTTPException(status_code=403, detail="Thread is already claimed by another operator")
|
|
|
|
|
|
def _raise_claim_conflict(thread: TelegramThreadRow, actor_user: str) -> None:
|
|
if thread.status == "closed":
|
|
raise HTTPException(status_code=409, detail="Telegram thread is closed")
|
|
if thread.claimed_by_user and thread.claimed_by_user != actor_user:
|
|
raise HTTPException(status_code=409, detail="Telegram thread is already claimed by another operator")
|
|
raise HTTPException(status_code=409, detail="Telegram thread claim conflict")
|
|
|
|
|
|
def _ensure_thread_available_for_ai(thread: TelegramThreadRow) -> None:
|
|
if thread.status == "closed":
|
|
raise HTTPException(status_code=409, detail="Telegram thread is closed")
|
|
if thread.ai_state == "human_owned" or thread.claimed_by_user:
|
|
raise HTTPException(status_code=409, detail="Telegram thread is owned by a human operator")
|
|
|
|
|
|
def _claim_thread_locally(
|
|
session,
|
|
*,
|
|
thread_id: str,
|
|
actor_user: str,
|
|
claimed_at: str,
|
|
) -> tuple[TelegramThreadRow, str, bool]:
|
|
thread = _get_thread(session, thread_id)
|
|
if thread.claimed_by_user == actor_user and thread.status == "in_progress":
|
|
return thread, thread.status, False
|
|
|
|
previous_status = thread.status
|
|
result = session.execute(
|
|
update(TelegramThreadRow)
|
|
.where(
|
|
TelegramThreadRow.thread_id == thread_id,
|
|
TelegramThreadRow.status != "closed",
|
|
or_(
|
|
TelegramThreadRow.claimed_by_user.is_(None),
|
|
TelegramThreadRow.claimed_by_user == actor_user,
|
|
),
|
|
)
|
|
.values(
|
|
claimed_by_user=actor_user,
|
|
claimed_at=claimed_at,
|
|
updated_at=claimed_at,
|
|
)
|
|
)
|
|
if not result.rowcount:
|
|
session.rollback()
|
|
_raise_claim_conflict(_get_thread(session, thread_id), actor_user)
|
|
|
|
session.commit()
|
|
return _get_thread(session, thread_id), previous_status, True
|
|
|
|
|
|
def _rollback_thread_claim(
|
|
session,
|
|
*,
|
|
thread_id: str,
|
|
actor_user: str,
|
|
claimed_at: str,
|
|
previous_status: str,
|
|
) -> None:
|
|
try:
|
|
session.execute(
|
|
update(TelegramThreadRow)
|
|
.where(
|
|
TelegramThreadRow.thread_id == thread_id,
|
|
TelegramThreadRow.claimed_by_user == actor_user,
|
|
TelegramThreadRow.claimed_at == claimed_at,
|
|
)
|
|
.values(
|
|
claimed_by_user=None,
|
|
claimed_at=None,
|
|
status=previous_status,
|
|
updated_at=utc_now_iso(),
|
|
)
|
|
)
|
|
session.commit()
|
|
except Exception:
|
|
session.rollback()
|
|
|
|
|
|
@app.get("/health", response_model=HealthResponse)
|
|
def health() -> HealthResponse:
|
|
return HealthResponse(status="ok", service="telegram-adapter-service", version="v2")
|
|
|
|
|
|
@app.post("/integrations/telegram/webhook", response_model=TelegramWebhookOut)
|
|
def webhook(payload: TelegramWebhookIn) -> TelegramWebhookOut:
|
|
session = get_session()
|
|
try:
|
|
metadata = _manual_webhook_metadata(payload)
|
|
thread, row, _ = _upsert_thread_for_inbound(
|
|
session,
|
|
chat_id=payload.chat_id,
|
|
telegram_user_id=metadata["telegram_user_id"],
|
|
username=metadata["username"],
|
|
display_name=metadata["display_name"],
|
|
text=payload.text,
|
|
payload=payload.payload,
|
|
customer_external_id=payload.customer_external_id,
|
|
phone_number=metadata["phone_number"],
|
|
queue_id=metadata["queue_id"],
|
|
)
|
|
session.commit()
|
|
session.refresh(row)
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread, row)
|
|
_maybe_enqueue_ai_for_thread(thread, row.message_id, author_type=row.author_type)
|
|
return _to_webhook_out(row)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post("/integrations/telegram/bot/webhook")
|
|
def bot_webhook(
|
|
payload: dict,
|
|
x_telegram_bot_api_secret_token: str | None = Header(
|
|
default=None,
|
|
alias="X-Telegram-Bot-Api-Secret-Token",
|
|
),
|
|
) -> dict:
|
|
expected_secret = _webhook_secret()
|
|
if expected_secret and x_telegram_bot_api_secret_token != expected_secret:
|
|
raise HTTPException(status_code=403, detail="Invalid Telegram webhook secret")
|
|
session = get_session()
|
|
try:
|
|
parsed = _parse_bot_update(payload)
|
|
existing = _existing_bot_message(
|
|
session,
|
|
chat_id=parsed["chat_id"],
|
|
external_message_id=parsed["external_message_id"],
|
|
)
|
|
if existing:
|
|
return _to_bot_webhook_result(existing, thread_created=False)
|
|
thread, row, created = _upsert_thread_for_inbound(
|
|
session,
|
|
chat_id=parsed["chat_id"],
|
|
telegram_user_id=parsed["telegram_user_id"],
|
|
username=parsed["username"],
|
|
display_name=parsed["display_name"],
|
|
text=parsed["text"],
|
|
payload=parsed["payload"],
|
|
external_message_id=parsed["external_message_id"],
|
|
phone_number=parsed["phone_number"],
|
|
direction=parsed["direction"],
|
|
)
|
|
try:
|
|
session.commit()
|
|
except IntegrityError:
|
|
session.rollback()
|
|
existing = _existing_bot_message(
|
|
session,
|
|
chat_id=parsed["chat_id"],
|
|
external_message_id=parsed["external_message_id"],
|
|
)
|
|
if existing:
|
|
return _to_bot_webhook_result(existing, thread_created=False)
|
|
raise
|
|
session.refresh(row)
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread, row)
|
|
_maybe_enqueue_ai_for_thread(thread, row.message_id, author_type=row.author_type)
|
|
return _to_bot_webhook_result(row, thread_created=created)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.get(
|
|
"/integrations/telegram/messages",
|
|
response_model=list[TelegramWebhookOut],
|
|
)
|
|
def list_messages(
|
|
limit: int = 100,
|
|
chat_id: str | None = None,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> list[TelegramWebhookOut]:
|
|
session = get_session()
|
|
try:
|
|
stmt = select(TelegramMessageRow).order_by(TelegramMessageRow.id.desc())
|
|
if chat_id:
|
|
stmt = stmt.where(TelegramMessageRow.chat_id == chat_id)
|
|
rows = session.execute(stmt.limit(max(limit, 1))).scalars().all()
|
|
return [_to_webhook_out(r) for r in reversed(rows)]
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.get(
|
|
"/integrations/telegram/threads",
|
|
response_model=list[TelegramThreadOut],
|
|
)
|
|
def list_threads(
|
|
limit: int = 100,
|
|
status: InteractionStatus | None = None,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> list[TelegramThreadOut]:
|
|
session = get_session()
|
|
try:
|
|
stmt = select(TelegramThreadRow).order_by(TelegramThreadRow.last_message_at.desc())
|
|
if status:
|
|
stmt = stmt.where(TelegramThreadRow.status == status)
|
|
rows = session.execute(stmt.limit(max(limit, 1))).scalars().all()
|
|
return [_to_thread_out(row) for row in rows]
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.get(
|
|
"/integrations/telegram/threads/{thread_id}",
|
|
response_model=TelegramThreadOut,
|
|
)
|
|
def get_thread(
|
|
thread_id: str,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> TelegramThreadOut:
|
|
session = get_session()
|
|
try:
|
|
return _to_thread_out(_get_thread(session, thread_id))
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.get(
|
|
"/integrations/telegram/threads/{thread_id}/ai-summary",
|
|
response_model=TelegramThreadAISummaryOut | None,
|
|
)
|
|
def get_thread_ai_summary(
|
|
thread_id: str,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> TelegramThreadAISummaryOut | None:
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
if actor["role"] == Role.OPERATOR.value and thread.claimed_by_user and thread.claimed_by_user != actor["user"]:
|
|
raise HTTPException(status_code=403, detail="Thread is already claimed by another operator")
|
|
return _build_ai_summary(session, thread)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.get(
|
|
"/integrations/telegram/threads/{thread_id}/messages",
|
|
response_model=list[TelegramThreadMessageOut],
|
|
)
|
|
def get_thread_messages(
|
|
thread_id: str,
|
|
limit: int = 200,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> list[TelegramThreadMessageOut]:
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
rows = session.execute(
|
|
select(TelegramMessageRow)
|
|
.where(TelegramMessageRow.thread_id == thread.thread_id)
|
|
.order_by(TelegramMessageRow.id.desc())
|
|
.limit(max(limit, 1))
|
|
).scalars().all()
|
|
return [_to_message_out(row) for row in reversed(rows)]
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post(
|
|
"/integrations/telegram/threads/{thread_id}/claim",
|
|
response_model=TelegramThreadOut,
|
|
)
|
|
def claim_thread(
|
|
thread_id: str,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> TelegramThreadOut:
|
|
session = get_session()
|
|
try:
|
|
claimed_at = utc_now_iso()
|
|
thread, previous_status, local_claimed = _claim_thread_locally(
|
|
session,
|
|
thread_id=thread_id,
|
|
actor_user=actor["user"],
|
|
claimed_at=claimed_at,
|
|
)
|
|
if not local_claimed:
|
|
return _to_thread_out(thread)
|
|
try:
|
|
result = _interaction_request(
|
|
"PATCH",
|
|
f"/interactions/{thread.interaction_id}/assign",
|
|
payload={"assignee": actor["user"]},
|
|
)
|
|
except Exception:
|
|
_rollback_thread_claim(
|
|
session,
|
|
thread_id=thread_id,
|
|
actor_user=actor["user"],
|
|
claimed_at=claimed_at,
|
|
previous_status=previous_status,
|
|
)
|
|
raise
|
|
thread = _get_thread(session, thread_id)
|
|
thread.status = result.get("status", "in_progress")
|
|
thread.updated_at = utc_now_iso()
|
|
_mark_thread_ai_state(thread, ai_state="human_owned", ai_handoff_reason=None)
|
|
_update_ai_session_state(
|
|
session,
|
|
session_id=thread.ai_session_id,
|
|
status="human_owned",
|
|
updated_at=thread.updated_at,
|
|
)
|
|
_push_timeline(
|
|
session,
|
|
thread.interaction_id,
|
|
"ai.human_takeover",
|
|
{"thread_id": thread.thread_id, "actor_user": actor["user"]},
|
|
)
|
|
session.commit()
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread)
|
|
return _to_thread_out(thread)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post(
|
|
"/integrations/telegram/threads/{thread_id}/return-to-ai",
|
|
response_model=TelegramThreadOut,
|
|
)
|
|
def return_thread_to_ai(
|
|
thread_id: str,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> TelegramThreadOut:
|
|
if not _ai_telegram_enabled():
|
|
raise HTTPException(status_code=409, detail="Telegram AI is disabled")
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
if thread.status == "closed":
|
|
raise HTTPException(status_code=409, detail="Telegram thread is closed")
|
|
_ensure_operator_can_manage(actor, thread, require_claim=True)
|
|
|
|
now = utc_now_iso()
|
|
interaction = session.execute(
|
|
select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
|
|
).scalar_one_or_none()
|
|
if interaction:
|
|
interaction.status = "new"
|
|
interaction.assigned_to = None
|
|
interaction.updated_at = now
|
|
_push_timeline(
|
|
session,
|
|
interaction.interaction_id,
|
|
"interaction.status_changed",
|
|
{"status": "new", "source": "telegram-return-to-ai"},
|
|
)
|
|
|
|
thread.status = "new"
|
|
thread.claimed_by_user = None
|
|
thread.claimed_at = None
|
|
thread.updated_at = now
|
|
_mark_thread_ai_state(thread, ai_state="queued", ai_handoff_reason=None)
|
|
_update_ai_session_state(
|
|
session,
|
|
session_id=thread.ai_session_id,
|
|
status="active",
|
|
updated_at=now,
|
|
handoff_reason=None,
|
|
)
|
|
_push_timeline(
|
|
session,
|
|
thread.interaction_id,
|
|
"ai.returned_to_ai",
|
|
{"thread_id": thread.thread_id, "actor_user": actor["user"]},
|
|
)
|
|
session.commit()
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread)
|
|
return _to_thread_out(thread)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post(
|
|
"/integrations/telegram/threads/{thread_id}/messages",
|
|
response_model=TelegramThreadMessageOut,
|
|
)
|
|
def reply_thread(
|
|
thread_id: str,
|
|
payload: TelegramThreadReplyIn,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> TelegramThreadMessageOut:
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
if thread.status == "closed":
|
|
raise HTTPException(status_code=409, detail="Telegram thread is closed")
|
|
_ensure_operator_can_manage(actor, thread, for_reply=True)
|
|
now = utc_now_iso()
|
|
row = _persist_message(
|
|
session,
|
|
thread=thread,
|
|
text=payload.text,
|
|
payload={"operator_user": actor["user"]},
|
|
direction="outbound",
|
|
created_at=now,
|
|
operator_user=actor["user"],
|
|
author_type="human",
|
|
author_id=actor["user"],
|
|
delivery_status="pending",
|
|
)
|
|
row.delivery_attempts = 0
|
|
row.next_delivery_attempt_at = now
|
|
row.delivery_locked_until = None
|
|
row.last_delivery_error = None
|
|
_mark_thread_ai_state(thread, ai_state="human_owned", ai_handoff_reason=None)
|
|
_update_ai_session_state(
|
|
session,
|
|
session_id=thread.ai_session_id,
|
|
status="human_owned",
|
|
updated_at=now,
|
|
)
|
|
session.commit()
|
|
session.refresh(row)
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread, row)
|
|
_start_telegram_reply_delivery(row.message_id)
|
|
return _to_message_out(row)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post(
|
|
"/integrations/telegram/threads/{thread_id}/ai/reply",
|
|
response_model=TelegramThreadMessageOut,
|
|
)
|
|
def ai_reply_thread(
|
|
thread_id: str,
|
|
payload: TelegramThreadAIReplyIn,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
|
|
) -> TelegramThreadMessageOut:
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
_ensure_thread_available_for_ai(thread)
|
|
now = utc_now_iso()
|
|
row = _persist_message(
|
|
session,
|
|
thread=thread,
|
|
text=payload.text,
|
|
payload={
|
|
"agent_profile": payload.agent_profile,
|
|
"model": payload.model,
|
|
"trigger_message_id": payload.trigger_message_id,
|
|
"language": payload.language,
|
|
"confidence": payload.confidence,
|
|
"kb_refs": payload.kb_refs,
|
|
**(payload.payload or {}),
|
|
},
|
|
direction="outbound",
|
|
created_at=now,
|
|
author_type="ai",
|
|
author_id=payload.agent_profile,
|
|
delivery_status="pending",
|
|
)
|
|
row.delivery_attempts = 0
|
|
row.next_delivery_attempt_at = now
|
|
row.delivery_locked_until = None
|
|
row.last_delivery_error = None
|
|
_mark_thread_ai_state(thread, ai_state="active", ai_handoff_reason=None, ai_last_model_at=now)
|
|
_update_ai_session_state(
|
|
session,
|
|
session_id=thread.ai_session_id,
|
|
status="active",
|
|
updated_at=now,
|
|
handoff_reason=None,
|
|
)
|
|
if thread.ai_session_id:
|
|
ai_session = session.execute(
|
|
select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id)
|
|
).scalar_one_or_none()
|
|
if ai_session:
|
|
ai_session.last_ai_message_id = row.message_id
|
|
if payload.language:
|
|
ai_session.language = payload.language
|
|
_push_timeline(
|
|
session,
|
|
thread.interaction_id,
|
|
"ai.reply_generated",
|
|
{
|
|
"thread_id": thread.thread_id,
|
|
"message_id": row.message_id,
|
|
"agent_profile": payload.agent_profile,
|
|
"model": payload.model,
|
|
"confidence": payload.confidence,
|
|
"kb_refs": payload.kb_refs,
|
|
"actor_user": actor["user"],
|
|
},
|
|
)
|
|
session.commit()
|
|
session.refresh(row)
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread, row)
|
|
_start_telegram_reply_delivery(row.message_id)
|
|
return _to_message_out(row)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post(
|
|
"/integrations/telegram/threads/{thread_id}/ai/handoff",
|
|
response_model=TelegramThreadOut,
|
|
)
|
|
def ai_handoff_thread(
|
|
thread_id: str,
|
|
payload: TelegramThreadAIHandoffIn,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
|
|
) -> TelegramThreadOut:
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
_ensure_thread_available_for_ai(thread)
|
|
now = utc_now_iso()
|
|
_mark_thread_ai_state(
|
|
thread,
|
|
ai_state="handoff_required",
|
|
ai_handoff_reason=payload.reason,
|
|
ai_last_model_at=now,
|
|
)
|
|
thread.updated_at = now
|
|
_update_ai_session_state(
|
|
session,
|
|
session_id=thread.ai_session_id,
|
|
status="handoff_required",
|
|
updated_at=now,
|
|
handoff_reason=payload.reason,
|
|
)
|
|
_push_timeline(
|
|
session,
|
|
thread.interaction_id,
|
|
"ai.handoff_requested",
|
|
{
|
|
"thread_id": thread.thread_id,
|
|
"reason": payload.reason,
|
|
"agent_profile": payload.agent_profile,
|
|
"trigger_message_id": payload.trigger_message_id,
|
|
"confidence": payload.confidence,
|
|
"actor_user": actor["user"],
|
|
},
|
|
)
|
|
session.commit()
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread)
|
|
return _to_thread_out(thread)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post(
|
|
"/integrations/telegram/threads/{thread_id}/close",
|
|
response_model=TelegramThreadOut,
|
|
)
|
|
def close_thread(
|
|
thread_id: str,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> TelegramThreadOut:
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
_ensure_operator_can_manage(actor, thread, require_claim=True)
|
|
_interaction_request(
|
|
"PATCH",
|
|
f"/interactions/{thread.interaction_id}/status",
|
|
payload={"status": "closed"},
|
|
)
|
|
thread.status = "closed"
|
|
thread.claimed_by_user = None
|
|
thread.claimed_at = None
|
|
thread.updated_at = utc_now_iso()
|
|
_mark_thread_ai_state(thread, ai_state="closed", ai_handoff_reason=None, ai_last_model_at=thread.ai_last_model_at)
|
|
_update_ai_session_state(
|
|
session,
|
|
session_id=thread.ai_session_id,
|
|
status="closed",
|
|
updated_at=thread.updated_at,
|
|
closed=True,
|
|
)
|
|
session.commit()
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread)
|
|
return _to_thread_out(thread)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post(
|
|
"/integrations/telegram/threads/{thread_id}/escalate",
|
|
response_model=TelegramThreadOut,
|
|
)
|
|
def escalate_thread(
|
|
thread_id: str,
|
|
payload: TelegramThreadEscalateIn,
|
|
actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
|
|
) -> TelegramThreadOut:
|
|
session = get_session()
|
|
try:
|
|
thread = _get_thread(session, thread_id)
|
|
_ensure_operator_can_manage(actor, thread, require_claim=True)
|
|
result = _interaction_request(
|
|
"POST",
|
|
f"/interactions/{thread.interaction_id}/escalate",
|
|
payload=EscalateRequest(target_queue_id=payload.target_queue_id).model_dump(),
|
|
)
|
|
thread.status = result.get("status", "escalated")
|
|
thread.queue_id = result.get("queue_id", payload.target_queue_id)
|
|
thread.updated_at = utc_now_iso()
|
|
_mark_thread_ai_state(thread, ai_state="human_owned", ai_handoff_reason=None)
|
|
_update_ai_session_state(
|
|
session,
|
|
session_id=thread.ai_session_id,
|
|
status="human_owned",
|
|
updated_at=thread.updated_at,
|
|
)
|
|
session.commit()
|
|
session.refresh(thread)
|
|
_sync_sales_thread(thread)
|
|
return _to_thread_out(thread)
|
|
finally:
|
|
session.close()
|