init

.gitignore

@@ -0,0 +1,23 @@
+__pycache__/
+env/
+venv/
+.venv/
+.idea/
+*.log
+*.egg-info/
+pip-wheel-EntityData/
+.env
+.DS_Store
+static/
+test.py
+rsa_key.p8
+rsa_key.pub
+aws.pem
+.vscode/
+data/
+*.csv
+test.json
+voiceagentcbh.pem
+*.pem
+download
+investigation

cbh/__init__.py

@@ -0,0 +1,94 @@
+# pylint: disable=C0415
+"""
+ClipboardHealthAI application package.
+"""
+import asyncio
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
+from cbh.core.wrappers import CbhResponseWrapper, ErrorCbhResponse
+
+
+def create_app() -> FastAPI:
+    """
+    Create and configure the FastAPI application.
+    """
+    app = FastAPI(docs_url="/api/docs", openapi_url="/api/openapi.json")
+
+    from cbh.api.account import account_router
+
+    app.include_router(account_router, tags=["account"])
+
+    from cbh.api.calls import calls_router
+
+    app.include_router(calls_router, tags=["calls"])
+
+    from cbh.api.reports import reports_router
+
+    app.include_router(reports_router, tags=["reports"])
+
+    from cbh.api.reps import reps_router
+
+    app.include_router(reps_router, tags=["reps"])
+
+    from cbh.api.security import security_router
+
+    app.include_router(security_router, tags=["security"])
+
+    from cbh.api.userinsights import userinsights_router
+
+    app.include_router(userinsights_router, tags=["userinsights"])
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origin_regex=r"https?://([a-z0-9-]+\.)?(localhost|trainwitharena|cbhexp\.com)(:\d+)?",
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.exception_handler(StarletteHTTPException)
+    async def http_exception_handler(_, exc):
+        """
+        Handle HTTP exceptions and convert them to standardized error responses.
+        """
+        return CbhResponseWrapper(
+            data=None, successful=False, error=ErrorCbhResponse(message=str(exc.detail))
+        ).response(exc.status_code)
+
+    @app.on_event("startup")
+    async def startup_event():
+        """
+        Execute startup tasks when the application starts.
+        """
+        from cbh.api.calls.services import run_call_listener
+        #
+        asyncio.create_task(run_call_listener())
+
+    @app.get("/api/health")
+    async def health():
+        """
+        Health check endpoint for container orchestration and monitoring.
+        """
+        try:
+            return {"status": "healthy", "database": "connected"}
+        except Exception as e:
+            return {"status": "unhealthy", "database": "disconnected", "error": str(e)}
+
+    @app.get("/health")
+    async def root():
+        """
+        Root endpoint for the application.
+        """
+        return {"message": "hello!"}
+
+    @app.get("/api/test")
+    async def root():
+        """
+        Root endpoint for the application.
+        """
+        return {"message": "hi hello!"}
+
+    return app

cbh/api/account/__init__.py

@@ -0,0 +1,14 @@
+"""
+Account module initialization.
+
+This module defines the FastAPI router for account API endpoints
+and imports related views for account management.
+"""
+
+from fastapi import APIRouter
+
+account_router = APIRouter(
+    prefix="/api/account",
+)
+
+from . import views  # noqa # pylint: disable=C0413

cbh/api/account/dto.py

@@ -0,0 +1,30 @@
+"""
+Account DTOs.
+"""
+
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class AccountType(Enum):
+    """
+    Enum for account types.
+    """
+
+    USER = 1
+    ADMIN = 2
+    OWNER = 3
+    SUPER_ADMIN = 4
+
+
+class RegistrationType(Enum):
+    """
+    Enum for registration types.
+    """
+
+    ORGANIC = 1
+    GOOGLE = 2
+    GITHUB = 3
+    APPLE = 4
+

cbh/api/account/models.py

@@ -0,0 +1,71 @@
+"""
+Account models.
+"""
+
+from datetime import datetime
+
+from passlib.context import CryptContext
+from pydantic import Field, field_validator
+
+from cbh.api.account.dto import AccountType, RegistrationType
+from cbh.core.database import MongoBaseModel, MongoBaseShortenModel
+
+
+class AccountModel(MongoBaseModel):
+    """
+    Account model class.
+
+    This class represents a user account in the system.
+    It includes fields for email, password, and timestamps for creation and update.
+    """
+
+    name: str | None = None
+    email: str
+    password: str | None = Field(exclude=True, default=None)
+
+    accountType: AccountType = Field(default=AccountType.USER)
+    registrationType: RegistrationType | None = Field(
+        default=RegistrationType.ORGANIC, exclude=True
+    )
+
+    datetimeInserted: datetime = Field(default_factory=datetime.now)
+    datetimeUpdated: datetime = Field(default_factory=datetime.now)
+
+    @field_validator("password", mode="before", check_fields=False)
+    @classmethod
+    def set_password_hash(cls, v: str | None) -> str | None:
+        """
+        Set the password hash.
+
+        Args:
+            v (str): The password to hash.
+
+        Returns:
+            str: The hashed password.
+        """
+        if isinstance(v, str) and not v.startswith("$2b$"):
+            return CryptContext(schemes=["bcrypt"], deprecated="auto").hash(v)
+        return v
+
+
+    class Config:  # pylint: disable=R0903
+        """
+        Config for the AccountModel class.
+        """
+
+        arbitrary_types_allowed = True
+        populate_by_name = True
+        json_encoders = {datetime: lambda dt: dt.isoformat()}
+
+
+class AccountShorten(MongoBaseShortenModel):
+    """
+    Account shorten model.
+    """
+
+    id: str
+    name: str | None = None
+    email: str
+    pictureUrl: str | None = None
+
+    accountType: AccountType

cbh/api/account/views.py

@@ -0,0 +1,20 @@
+"""
+Account views module.
+"""
+
+from fastapi import Depends
+
+from cbh.api.account import account_router
+from cbh.api.account.models import AccountModel
+from cbh.core.security import PermissionDependency
+from cbh.core.wrappers import CbhResponseWrapper
+
+
+@account_router.get("")
+async def get_own_account(
+    account: AccountModel = Depends(PermissionDependency()),
+) -> CbhResponseWrapper[AccountModel]:
+    """
+    Get own account.
+    """
+    return CbhResponseWrapper(data=account)

cbh/api/ari/__init__.py

@@ -0,0 +1,5 @@
+from fastapi import APIRouter
+
+ari_router = APIRouter(prefix="/api/ari", tags=["ari"])
+
+from . import views

cbh/api/ari/db_requests.py

@@ -0,0 +1,41 @@
+from datetime import datetime
+from pymongo import ReturnDocument
+from cbh.api.ari.dto import Author
+from cbh.api.ari.models import ChatModel, MessageModel
+from cbh.api.account.models import AccountModel, AccountShorten
+from cbh.api.ari.schemas import ChatFilter, CreateMessageRequest
+from cbh.api.common.db_requests import get_all_objs
+from cbh.api.common.schemas import FilterRequest
+from cbh.core.config import settings
+from cbh.core.wrappers import background_task
+
+
+async def truncate_message_history(message_id: str, ids_to_delete: list[str]) -> None:
+    await settings.DB_CLIENT.messages.delete_many({"id": {"$in": [message_id, *ids_to_delete]}})
+
+
+@background_task()
+async def add_messages_obj(
+    chat_id: str,
+    account_id: str,
+    request: CreateMessageRequest,
+    message_args: tuple[str, str, datetime, datetime],
+) -> None:
+    user_message = MessageModel(
+        id=request.messageId,
+        chatId=chat_id,
+        accountId=account_id,
+        role=Author.HUMAN,
+        content=request.content,
+        datetimeInserted=message_args[2],
+    )
+    assistant_message = MessageModel(
+        id=message_args[1],
+        chatId=chat_id,
+        accountId=account_id,
+        role=Author.AI,
+        content=message_args[0],
+        datetimeInserted=message_args[3],
+    )
+    await settings.DB_CLIENT.messages.insert_one(user_message.to_mongo())
+    await settings.DB_CLIENT.messages.insert_one(assistant_message.to_mongo())

cbh/api/ari/dto.py

@@ -0,0 +1,7 @@
+from enum import Enum
+
+
+class Author(str, Enum):
+    HUMAN = 'human'
+    AI = 'ai'
+

cbh/api/ari/schemas.py

@@ -0,0 +1,15 @@
+from typing import Optional
+from pydantic import BaseModel
+
+
+class CreateMessageRequest(BaseModel):
+    content: str
+    messageId: Optional[str] = None
+
+
+class TranscribeResponse(BaseModel):
+    """
+    Response schema for the transcribe endpoint.
+    """
+
+    text: str

cbh/api/ari/services/__init__.py

@@ -0,0 +1,3 @@
+from .workflows import convert_audio_to_text
+
+__all__ = ["convert_audio_to_text"]

cbh/api/ari/services/agent/__init__.py

@@ -0,0 +1,8 @@
+"""
+This module contains the Ari Agent.
+"""
+
+from .agent import AriAgent
+
+
+__all__ = ["AriAgent"]

cbh/api/ari/services/agent/agent.py

@@ -0,0 +1,113 @@
+# pylint: disable=R0801
+"""
+This module contains the CBH Agent.
+"""
+import asyncio
+
+from langchain_classic.agents import AgentExecutor, create_openai_tools_agent
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+
+from cbh.api.account.models import AccountModel
+from cbh.api.ari.schemas import CreateMessageRequest
+from cbh.api.ari.services.agent.handler import StreamingAgentCallbackHandler
+from cbh.core.config import settings
+from .prompt import ARI_PROMPT
+from .tools import ScenarioAgentTools
+
+
+class AriAgent:
+    """
+    CBH Agent for handling schedule creation.
+    """
+
+    def __init__(self, account: AccountModel):
+        """
+        Initialize the CBHAgent with an account. The agent is stateless with
+        respect to chat history — history is provided per call to `stream`.
+        """
+        self.tools = ScenarioAgentTools.load_tools(account)
+        self.agent_executor = self._get_agent()
+
+    def _get_agent(self):
+        """
+        Get the agent_pd instance.
+        """
+        return AgentExecutor(
+            agent=create_openai_tools_agent(
+                llm=self._get_agent_model(),
+                tools=self.tools,
+                prompt=self._load_system_prompt(),
+            ),
+            tools=self.tools,
+            verbose=True,
+            return_intermediate_steps=True,
+            max_iterations=100,
+        )
+
+    def _get_agent_model(self):
+        """
+        Get the language model used by the agent_pd.
+        """
+        return settings.get_llm(
+            model="gpt-5.4", reasoning_effort="medium", reasoning_summary="auto"
+        )
+
+    def _load_system_prompt(self) -> ChatPromptTemplate:
+        """
+        Load the system prompt from file.
+        """
+        try:
+            return ChatPromptTemplate.from_messages(
+                [
+                    ("system", ARI_PROMPT),
+                    MessagesPlaceholder(variable_name="chat_history"),
+                    ("human", "{content}"),
+                    MessagesPlaceholder(variable_name="agent_scratchpad"),
+                ]
+            )
+        except Exception as e:
+            raise Exception(  # pylint: disable=W0719,W0707
+                f"Failed to load system prompt: {str(e)}"
+            )
+
+    async def stream(
+        self,
+        message_history: list[dict],
+        request: CreateMessageRequest,
+        stop_event: asyncio.Event = None,
+    ):
+        """
+        Stream the agent's response to the client.
+        """
+        queue = []
+
+        async def send(data):
+            queue.append(data)
+
+        handler = StreamingAgentCallbackHandler(send)
+
+        task = asyncio.create_task(
+            self.agent_executor.ainvoke(
+                {
+                    "content": request.content,
+                    "chat_history": message_history,
+                },
+                config={"callbacks": [handler]},
+            )
+        )
+
+        while not task.done() or queue:
+            if stop_event and stop_event.is_set():
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+                break
+            while queue:
+                data = queue.pop(0)
+                yield data
+            await asyncio.sleep(0.01)
+
+        if not task.cancelled():
+            await task

cbh/api/ari/services/agent/handler.py

@@ -0,0 +1,37 @@
+import pydash
+from langchain_core.callbacks import AsyncCallbackHandler
+
+
+class StreamingAgentCallbackHandler(AsyncCallbackHandler):  # pylint: disable=R0901
+
+    def __init__(self, send):
+        self.send = send
+
+    async def on_chat_model_start(self, *args, **kwargs):
+        await self.send({"type": "init", "content": ""})
+
+    async def on_tool_start(self, serialized, input_str, **kwargs):
+        pass
+
+    async def on_tool_end(self, output, **kwargs):
+        await self.send({"type": "tool_response", "content": output})
+
+    async def on_llm_new_token(self, token: str, **kwargs):
+        chunk = kwargs.get("chunk")
+        if chunk:
+            message = pydash.get(chunk.message.content, "[0]") or {}
+            if message.get("type") == "reasoning":
+                content = pydash.get(message, "summary[0].text")
+                if content:
+                    await self.send({"type": "thinking", "content": content})
+            elif message.get("type") == "text":
+                await self.send({"type": "ai_token", "content": message.get("text")})
+
+    async def on_chain_end(self, outputs, **kwargs):
+        if (
+            isinstance(outputs, dict)
+            and "intermediate_steps" in outputs.keys()
+            and "output" in outputs.keys()
+        ):
+            ai_message = ""
+            await self.send({"type": "ai", "content": ai_message})

cbh/api/ari/services/agent/prompt.py

@@ -0,0 +1,68 @@
+# pylint: disable-all
+# flake8: noqa
+"""
+This module contains the prompt for the CBH agent.
+"""
+ARI_PROMPT = """You are **Ari**, a friendly analytics assistant. You help managers and admins understand their organization's sales training performance by answering questions using live data.
+
+## Your Capabilities
+You can retrieve the following data by calling the appropriate tool:
+- **Accounts**: search users by name, role, or status
+- **Dashboard overview**: session counts, average scores, engagement, and trends
+- **Scenario performance**: per-scenario averages, top/bottom reps, most common mistakes
+- **Skills averages**: org-wide or per-user breakdown across 6 skill dimensions (communication, active listening, conversation, objection handling, empathy, overall)
+- **Leaderboard**: ranked list of reps by score, filterable by scenario and date
+- **Attention needs**: reps who are inactive, scenarios with low scores, or scenarios never attempted
+- **Scenario details**: statistics, skills breakdown, and leaderboard for a specific scenario
+- **Team details**: statistics, skills breakdown, and leaderboard for a specific team
+- **Insights**: top mistakes and achievements for a user, team, scenario, or the entire organization
+
+## Rules
+1. **Always call a tool before answering when user asks for data** — never guess or fabricate data. Every data point in your response must come from a tool result.
+2. **Resolve names to IDs first** — if the user references a person, team, or scenario by name and you need an ID, call the appropriate search tool first (`search_account`, `search_teams`, `search_scenarios`), then call the target tool with the resolved ID.
+3. **Use only the data returned** — summarize and present what the tool gives you. Do not add assumptions, projections, or context that was not returned.
+4. **Stay on topic** — only answer questions about the organization's training performance. Politely decline anything unrelated.
+5. **Never expose internals** — do not mention tool names, internal errors, raw JSON, or how you work under the hood.
+6. **Handle empty results gracefully** — if a tool returns no data, tell the user clearly and suggest a next step (e.g. broaden the date range or check the spelling).
+
+## Response Format
+- Be friendly, concise, and conversational
+- Keep replies short — 2–4 sentences or a brief list; avoid walls of text
+- Present numbers clearly (e.g. "Average score: **78/100**")
+- When showing multiple items (e.g. leaderboard, insights), use a short bullet list
+- Do NOT dump raw data — always interpret and summarize it for the user
+
+## Examples
+
+**User:** How is the team doing overall?
+**Ari:** *[calls retrieve_admin_intro_statistics]*
+The team completed **142 sessions** this month (+12% vs last month), with an average score of **74/100**. **38 out of 45 members** were active. Want a breakdown by scenario or skill?
+
+---
+
+**User:** Who are the top performers?
+**Ari:** *[calls retrieve_leaderboard with page_size=5]*
+Here are your top 5 reps this month:
+1. **Sarah M.** — 91
+2. **James T.** — 88
+3. **Priya K.** — 85
+...
+Want to see how they perform on a specific scenario?
+
+---
+
+**User:** What mistakes is John making?
+**Ari:** *[calls search_account with search_term="John"]*
+*[calls get_top_account_insights with account_id=..., type_="mistake"]*
+John's most common mistake is **rushing the discovery phase** — he tends to jump to the pitch before fully understanding the customer's situation. Want tips on how to coach this?
+
+---
+
+**User:** Which scenarios need attention?
+**Ari:** *[calls retrieve_attention_needs]*
+Three scenarios haven't been attempted by anyone yet, and **"Cold Call Objection Drill"** has an average score below the threshold. I'd recommend assigning it and scheduling a coaching session.
+
+---
+
+**User:** Write me a poem.
+**Ari:** I can only help with questions about your team's training performance. Ask me about scores, leaderboards, or coaching insights!"""

cbh/api/ari/services/agent/tools.py

@@ -0,0 +1,69 @@
+"""
+This module contains the tools for the CBH Agent.
+"""
+
+# pylint: disable=R0801,C0301
+# flake8: noqa
+from typing import Literal
+from langchain_core.tools import StructuredTool
+
+from cbh.api.account.models import AccountModel
+from cbh.api.common.dto import OrderType
+
+
+GET_TEAM_LEADERBOARD_DESCRIPTION = """Retrieve a ranked leaderboard of users within a specific team.
+
+Use this tool when the user wants to compare individual performance within a team —
+for example, to identify the best and worst performers or run intra-team competitions.
+Call search_teams first to obtain the team_id if needed.
+
+Parameters:
+- team_id: The unique ID of the team (required).
+- order: Sort direction. Accepted values:
+    "desc" — highest scores first (top performers at the top).
+    "asc" — lowest scores first (use to surface struggling members).
+- page_size: Number of leaderboard positions to return (default 1). Increase to see more users.
+
+Returns a ranked list of positions, each containing the user (id, name, picture)
+and their Scores across all 6 skill dimensions."""
+
+
+class ScenarioAgentTools:
+    """
+    Tools for the CBH Agent.
+    """
+
+    @staticmethod
+    def create_team_leaderboard(account: AccountModel):
+        async def get_team_leaderboard(
+            team_id: str,
+            order: Literal["asc", "desc"],
+            page_size: int = 1,
+        ):
+            sort_order = OrderType.DESCENDING.value
+            if order == "asc":
+                sort_order = OrderType.ASCENDING.value
+            results = await get_team_leaderboard_obj(
+                team_id=team_id,
+                account=account,
+                order=sort_order,
+                page_size=page_size,
+                page_index=0,
+            )
+            return {
+                "tool": "get_team_leaderboard",
+                "value": results.model_dump(mode="json"),
+            }
+
+        return get_team_leaderboard
+
+    @staticmethod
+    def load_tools(account: AccountModel) -> list[StructuredTool]:
+        return [
+
+            StructuredTool.from_function(
+                name="get_team_leaderboard",
+                description=GET_TEAM_LEADERBOARD_DESCRIPTION,
+                coroutine=ScenarioAgentTools.create_team_leaderboard(account),
+            ),
+        ]

cbh/api/ari/services/workflows.py

@@ -0,0 +1,21 @@
+import io
+
+from fastapi import HTTPException
+from cbh.core.config import settings
+
+
+async def convert_audio_to_text(file: bytes, name: str) -> str:
+    """
+    Convert an audio file to text using OpenAI's Whisper model.
+    """
+    file_content = io.BytesIO(file)
+    file_content.name = name
+
+    transcription = await settings.OPENAI_CLIENT.audio.transcriptions.create(
+        file=file_content, model="whisper-1", language="en"
+    )
+    if isinstance(transcription, str):
+        return transcription
+    if transcription.text and isinstance(transcription.text, str):
+        return str(transcription.text)
+    raise HTTPException(status_code=500, detail="Failed to convert audio to text")

cbh/api/ari/utils.py

@@ -0,0 +1,207 @@
+import asyncio
+import io
+import statistics
+from datetime import datetime
+from typing import AsyncGenerator, Callable
+from fastapi import WebSocket, WebSocketDisconnect, HTTPException, UploadFile
+from cbh.api.ari.schemas import CreateMessageRequest
+from cbh.api.ari.db_requests import add_messages_obj, truncate_message_history
+from cbh.api.ari.models import MessageModel
+from pydub import AudioSegment
+
+async def send_exception_stream(websocket: WebSocket) -> None:
+    """Send an exception stream to the client."""
+    message = "An error occurred while processing your request. Please try again."
+    await websocket.send_json({"type": "init", "content": ""})
+    for c in message:
+        await websocket.send_json({"type": "ai_token", "content": c})
+        await asyncio.sleep(0.01)
+    await websocket.send_json({"type": "ai", "content": None})
+    await websocket.send_json({"type": "finish", "content": None, "messageId": None})
+
+
+async def handle_websocket_streaming(
+    websocket: WebSocket,
+    pipeline_executor: Callable[[asyncio.Event], AsyncGenerator[dict, None]],
+    message_id: str | None = None,
+    ai_message_id: str | None = None,
+) -> str:
+    """
+    Handle WebSocket streaming with stop signal support.
+
+    Args:
+        websocket: WebSocket connection
+        pipeline_executor: Async generator function that yields chunks and accepts stop_event
+    """
+    stop_event = asyncio.Event()
+    partial_text = ""
+
+    recv_task = asyncio.create_task(create_stop_receiver(websocket, stop_event))
+
+    try:
+        async for chunk in pipeline_executor(stop_event):
+            await websocket.send_json(chunk)
+
+            extracted_text = extract_partial_text(chunk)
+            if extracted_text:
+                partial_text += extracted_text
+    except asyncio.CancelledError:
+        stop_event.set()
+        raise
+    except RuntimeError:
+        pass
+    finally:
+        recv_task.cancel()
+        await asyncio.gather(recv_task, return_exceptions=True)
+
+        if stop_event.is_set() and partial_text.strip():
+            print(partial_text)
+    await websocket.send_json(
+        {"type": "finish", "content": None, "messageId": message_id, "aiMessageId": ai_message_id}
+    )
+    return partial_text
+
+
+async def create_stop_receiver(websocket: WebSocket, stop_event: asyncio.Event) -> None:
+    """
+    Create a receiver task that listens for stop signals from WebSocket.
+    """
+    while True:
+        try:
+            msg = await websocket.receive_json()
+            if msg.get("type") == "stop":
+                stop_event.set()
+                break
+        except WebSocketDisconnect:
+            break
+        except Exception:
+            break
+
+
+def extract_partial_text(chunk: dict) -> str:
+    """
+    Extract text content from a chunk for partial text accumulation.
+    """
+    try:
+        if chunk.get("type") == "ai_token" and chunk.get("content"):
+            return chunk.get("content", "")
+        elif chunk.get("type") == "ai" and chunk.get("content"):
+            return chunk.get("content", "")
+    except Exception:
+        pass
+    return ""
+
+
+def prepare_messages_from_history(message_history: list[MessageModel]) -> list[dict]:
+    """
+    Prepare messages from message history.
+    """
+    result = []
+    for message in message_history:
+        result.append(
+            {
+                "role": message.role.value,
+                "content": message.content,
+            }
+        )
+    return result
+
+
+async def truncate_from_message(
+    message_history: list[MessageModel], request: CreateMessageRequest
+) -> list[dict]:
+    result = []
+    ids_to_delete = []
+    found = False
+
+    for message in message_history:
+        msg_id = message.id
+        if found or msg_id == request.messageId:
+            found = True
+            ids_to_delete.append(msg_id)
+            continue
+        result.append({"role": message.role.value, "content": message.content})
+
+    asyncio.create_task(truncate_message_history(request.messageId, ids_to_delete))
+    return result
+
+
+async def add_messages(
+    chat_id: str,
+    account_id: str,
+    request: CreateMessageRequest,
+    messages: list[dict],
+    message_args: tuple[str, str, datetime, datetime],
+) -> list[dict]:
+    messages.append({"role": "human", "content": request.content})
+    messages.append({"role": "ai", "content": message_args[0]})
+    asyncio.create_task(
+        add_messages_obj(chat_id, account_id, request, message_args)
+    )
+    return messages
+
+
+
+
+async def compress_audio(audio_file: UploadFile) -> bytes | None:
+    """
+    Compress an uploaded audio file to MP3 format.
+    """
+    file_as_bytes = await audio_file.read()
+
+    try:
+        audio_segment = AudioSegment.from_file(io.BytesIO(file_as_bytes))
+        duration = audio_segment.duration_seconds
+
+        if duration > 300:
+            return None
+
+        mp3_data = io.BytesIO()
+        audio_segment.export(mp3_data, format="mp3")
+        mp3_data.seek(0)
+
+    except OSError:
+        return file_as_bytes
+
+    except Exception as e:  # pylint: disable=W0703
+        raise HTTPException(status_code=500, detail=str(e))  # pylint: disable=W0707
+
+    return mp3_data.read()
+
+
+def detect_silence(
+    audio_data: bytes, silence_threshold: float = -40.0, min_speech_duration: float = 1.0
+) -> bool:
+    try:
+        audio_segment = AudioSegment.from_file(io.BytesIO(audio_data))
+        chunk_length_ms = 2000
+        rms_values = []
+        for start_ms in range(0, len(audio_segment), chunk_length_ms):
+            end_ms = min(start_ms + chunk_length_ms, len(audio_segment))
+            chunk = audio_segment[start_ms:end_ms]
+            if len(chunk) > 0:
+                chunk_mono = chunk.set_channels(1)
+                rms_db = chunk_mono.dBFS
+                rms_values.append(rms_db)
+
+        if not rms_values:
+            return False
+
+        mean_rms = statistics.mean(rms_values)
+
+        non_silent_chunks = sum(1 for rms in rms_values if rms > silence_threshold)
+        total_chunks = len(rms_values)
+        non_silent_ratio = non_silent_chunks / total_chunks if total_chunks > 0 else 0
+
+        non_silent_duration = (non_silent_chunks * chunk_length_ms) / 1000.0  # Convert to seconds
+
+        has_speech = (
+            mean_rms > silence_threshold - 10
+            and non_silent_ratio > 0.1
+            and non_silent_duration >= min_speech_duration
+        )
+
+        return has_speech
+
+    except Exception:
+        return True

cbh/api/ari/views.py

@@ -0,0 +1,99 @@
+import asyncio
+from datetime import datetime
+from fastapi import Depends, WebSocket, WebSocketDisconnect, UploadFile, File, HTTPException
+from bson import ObjectId
+
+from cbh.api.account.models import AccountModel
+from cbh.api.ari import ari_router
+from cbh.api.ari.services import convert_audio_to_text
+from cbh.api.ari.schemas import (
+    CreateMessageRequest, TranscribeResponse,
+)
+from cbh.api.ari.services.agent import AriAgent
+from cbh.api.ari.utils import (
+    prepare_messages_from_history,
+    send_exception_stream,
+    handle_websocket_streaming,
+    truncate_from_message,
+    add_messages, detect_silence, compress_audio,
+)
+from cbh.api.common.db_requests import get_all_objs, get_obj_by_id
+from cbh.core.security import PermissionDependency, check_account_token
+from cbh.core.wrappers import CbhResponseWrapper
+
+
+@ari_router.websocket("/{chatId}/send")
+async def send_ari_message(chatId: str, websocket: WebSocket):
+    await websocket.accept()
+    token = websocket.query_params.get("token")
+    token = check_account_token(token)
+    if not token:
+        await websocket.close(code=1008)
+        return
+
+    account, (message_history, _) = await asyncio.gather(
+        get_obj_by_id(AccountModel, token["account_id"]),
+        get_all_objs(
+            MessageModel,
+            100000,
+            0,
+            additional_filter={"accountId": token["account_id"], "chatId": chatId},
+            sort=("id", 1),
+        ),
+    )
+    agent = AriAgent(account)
+
+    messages = prepare_messages_from_history(message_history)
+    while True:
+        try:
+            request = await websocket.receive_json()
+            request = CreateMessageRequest(**request)
+            user_message_time = datetime.now()
+
+            if request.messageId:
+                messages = await truncate_from_message(message_history, request)
+            else:
+                request.messageId = str(ObjectId())
+
+            async def agent_executor(stop_event):
+                async for chunk in agent.stream(messages, request, stop_event):
+                    yield chunk
+                    await asyncio.sleep(0.01)
+
+            ai_message_id = str(ObjectId())
+            ai_response = await handle_websocket_streaming(
+                websocket=websocket,
+                pipeline_executor=agent_executor,
+                message_id=request.messageId,
+                ai_message_id=ai_message_id,
+            )
+            ai_message_time = datetime.now()
+
+            message_args = (ai_response, ai_message_id, ai_message_time, user_message_time)
+            messages = await add_messages(chatId, account.id, request, messages, message_args)
+
+        except WebSocketDisconnect:
+            return
+        except Exception as e:
+            await send_exception_stream(websocket)
+
+
+@ari_router.post("/voice/transcript")
+async def get_voice_transcript(
+    file: UploadFile = File(...),
+    _: AccountModel = Depends(PermissionDependency()),
+) -> CbhResponseWrapper[TranscribeResponse]:
+    """
+    Transcribe an uploaded audio file.
+    """
+    mp3_data = await compress_audio(file)
+    if mp3_data is None:
+        raise HTTPException(status_code=400, detail="Could not compress audio file")
+    if not detect_silence(mp3_data):
+        raise HTTPException(
+            status_code=422,
+            detail="Recording appears to contain only silence or background noise.",
+        )
+
+    transcribed_text = await convert_audio_to_text(mp3_data, file.filename)
+    return CbhResponseWrapper(data=TranscribeResponse(text=transcribed_text))

cbh/api/chats/__init__.py

@@ -0,0 +1,5 @@
+from fastapi import APIRouter
+
+chats_router = APIRouter(prefix="/chats", tags=["chats"])
+
+from . import views

cbh/api/chats/db_requests.py

@@ -0,0 +1,45 @@
+from pymongo import ReturnDocument
+
+from cbh.api.account.models import AccountModel, AccountShorten
+from cbh.api.chats.models import ChatModel
+from cbh.api.chats.schemas import ChatFilter
+from cbh.api.common.db_requests import get_all_objs
+from cbh.api.common.schemas import FilterRequest
+from cbh.core.config import settings
+
+
+async def create_chat_obj(account: AccountModel, name: str) -> ChatModel:
+    """
+    Create a chat object.
+    """
+    chat = ChatModel(account=AccountShorten(**account.model_dump()), name=name)
+    await settings.DB_CLIENT.chats.insert_one(chat.to_mongo())
+    return chat
+
+
+async def update_chat_obj(account: AccountModel, chat_id: str, name: str) -> ChatModel:
+    """
+    Update a chat object.
+    """
+    chat = await settings.DB_CLIENT.chats.find_one_and_update(
+        {"id": chat_id, "account.id": account.id},
+        {"$set": {"name": name}},
+        return_document=ReturnDocument.AFTER,
+    )
+    return ChatModel.from_mongo(chat)
+
+
+async def filter_chats_objs(
+    account: AccountModel, request: FilterRequest[ChatFilter]
+) -> tuple[list[ChatModel], int]:
+    """
+    Filter chats objects.
+    """
+    filter_ = {"account.id": account.id}
+    if request.filter.searchTerm:
+        filter_["name"] = {"$regex": f"^{request.filter.searchTerm}", "$options": "i"}
+
+    chats, total_count = await get_all_objs(
+        ChatModel, request.pageSize, request.pageIndex, additional_filter=filter_
+    )
+    return chats, total_count

cbh/api/chats/models.py

@@ -0,0 +1,12 @@
+from datetime import datetime
+
+from pydantic import Field
+
+from cbh.api.account.models import AccountShorten
+from cbh.core.database import MongoBaseModel
+
+
+class ChatModel(MongoBaseModel):
+    name: str
+    account: AccountShorten
+    datetimeInserted: datetime = Field(default_factory=datetime.now)

cbh/api/chats/schemas.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+
+
+class CreateChatRequest(BaseModel):
+    query: str
+
+
+class UpdateChatRequest(BaseModel):
+    name: str
+
+
+class ChatFilter(BaseModel):
+    searchTerm: str | None = None

cbh/api/chats/services/__init__.py

@@ -0,0 +1,3 @@
+from .workflows import generate_chat_name
+
+__all__ = ["generate_chat_name"]

cbh/api/chats/services/prompts.py

@@ -0,0 +1,57 @@
+from functools import lru_cache
+
+
+class ChatPrompts:
+    """
+    Ari prompts.
+    """
+
+    generate_chat_name = """You are a title generator for a sales team management assistant.
+
+Generate a concise, descriptive title (2–4 words) for a chat based on the user's first message.
+
+## User message:
+{query}
+
+## Rules
+- Use title case
+- No punctuation or filler words
+- Capture the specific intent, not just the topic
+- Prefer action + subject format when applicable
+- Output only the title, nothing else
+
+## Context
+The assistant helps admins who manage sales teams. Users ask for statistics, rep performance, scenario creation, and other team management tasks.
+
+## Examples
+<example>
+query: Who is the best sales rep in the "Connor's team" team?
+title: Top Rep Connor's Team
+</example>
+<example>
+query: What is the average score for Maksim Shymanouski?
+title: Maksim Average Score
+</example>
+<example>
+query: Create me a friendly scenario with 5 objections
+title: Friendly 5-Objection Scenario
+</example>
+<example>
+query: Show me the call conversion rate for last month
+title: Last Month Conversion Rate
+</example>
+<example>
+query: Which reps haven't hit their targets this quarter?
+title: Reps Missing Quarterly Targets
+</example>"""
+
+
+@lru_cache()
+def get_prompts() -> ChatPrompts:
+    """
+    Get prompts.
+    """
+    return ChatPrompts()
+
+
+chat_prompts = get_prompts()

cbh/api/chats/services/workflows.py

@@ -0,0 +1,23 @@
+from langchain_core.prompts import ChatPromptTemplate
+from pydantic import BaseModel, Field
+
+from cbh.api.chats.services.prompts import chat_prompts
+from cbh.core.config import settings
+
+
+class ChatNameSchema(BaseModel):
+    """
+    Chat name schema.
+    """
+
+    name: str = Field(description="A name for the chat.")
+
+
+async def generate_chat_name(query: str) -> str:
+    """
+    Generate a chat name.
+    """
+    prompt = ChatPromptTemplate.from_messages([("system", chat_prompts.generate_chat_name)])
+    chain = prompt | settings.get_llm(schema=ChatNameSchema, model="gpt-4.1-nano")
+    result = await chain.ainvoke({"query": query})
+    return result.name

cbh/api/chats/views.py

@@ -0,0 +1,49 @@
+from fastapi import Depends
+
+from cbh.api.account.dto import AccountType
+from cbh.api.account.models import AccountModel
+from cbh.api.chats import chats_router
+from cbh.api.chats.db_requests import create_chat_obj, update_chat_obj, filter_chats_objs
+from cbh.api.chats.models import ChatModel
+from cbh.api.chats.schemas import CreateChatRequest, UpdateChatRequest, ChatFilter
+from cbh.api.chats.services import generate_chat_name
+from cbh.api.common.schemas import AllObjectsResponse, Paging
+from cbh.api.common.schemas import FilterRequest
+from cbh.core.security import PermissionDependency
+from cbh.core.wrappers import CbhResponseWrapper
+
+@chats_router.post("/chats")
+async def create_new_chat(
+    request: CreateChatRequest,
+    account: AccountModel = Depends(PermissionDependency()),
+) -> CbhResponseWrapper[ChatModel]:
+    chat_name = await generate_chat_name(request.query)
+    chat = await create_chat_obj(account, chat_name)
+    return CbhResponseWrapper(data=chat)
+
+
+@chats_router.put("/chats/{chatId}")
+async def update_chat(
+    chatId: str,
+    request: UpdateChatRequest,
+    account: AccountModel = Depends(PermissionDependency([AccountType.ADMIN, AccountType.OWNER])),
+) -> CbhResponseWrapper[ChatModel]:
+    chat = await update_chat_obj(account, chatId, request.name)
+    return CbhResponseWrapper(data=chat)
+
+
+@chats_router.post("/chats/filter")
+async def filter_chats(
+    request: FilterRequest[ChatFilter],
+    account: AccountModel = Depends(PermissionDependency([AccountType.ADMIN, AccountType.OWNER])),
+) -> CbhResponseWrapper[AllObjectsResponse[ChatModel]]:
+    chats, total_count = await filter_chats_objs(account, request)
+    return CbhResponseWrapper(
+        data=AllObjectsResponse(
+            data=chats,
+            paging=Paging(
+                pageSize=request.pageSize, pageIndex=request.pageIndex, totalCount=total_count
+            ),
+        )
+    )
+

cbh/api/common/db_requests.py

@@ -0,0 +1,196 @@
+"""
+Common database requests.
+"""
+
+import asyncio
+import re
+from datetime import timedelta, datetime
+from typing import TypeVar
+
+from fastapi import HTTPException
+from pydantic import BaseModel
+
+from cbh.api.common.schemas import (
+    SearchRequest,
+)
+from cbh.core.config import settings
+
+T = TypeVar("T", bound=BaseModel)
+
+collection_map = {
+    "AccountModel": "accounts",
+    "AccountShorten": "accounts",
+    "UserInsightModel": "userinsights",
+    "UserInsightShorten": "userinsights",
+    "CallModel": "calls",
+    "RepModel": "reps",
+}
+
+
+async def get_obj_by_id(
+    model: T,
+    obj_id: str | None,
+    additional_filter: dict | None = None,
+    projection: dict | None = None,
+    exception: bool = True,
+) -> T | None:
+    """
+    Get an object by ID.
+    """
+    filter_ = {"id": obj_id} if obj_id else {}
+    if additional_filter:
+        filter_.update(additional_filter)
+    obj = await settings.DB_CLIENT[collection_map[model.__name__]].find_one(filter_, projection)
+    if obj is None:
+        if exception:
+            raise HTTPException(status_code=404, detail="Object not found.")
+        else:
+            return None
+    return model.from_mongo(obj)
+
+
+async def get_all_objs(
+    model: T,
+    page_size: int,
+    page_index: int,
+    sort: tuple[str, int] = ("id", -1),
+    additional_filter: dict | None = None,
+    projection: dict | None = None,
+) -> tuple[list[T], int]:
+    """
+    Get all objects.
+    """
+    filter_ = additional_filter if additional_filter else {}
+    skip = page_index * page_size
+    objs, total_count = await asyncio.gather(
+        settings.DB_CLIENT[collection_map[model.__name__]]
+        .find(filter_, projection)
+        .sort(*sort)
+        .skip(skip)
+        .limit(page_size)
+        .to_list(page_size),
+        settings.DB_CLIENT[collection_map[model.__name__]].count_documents(filter_),
+    )
+    return [model.from_mongo(obj) for obj in objs], total_count
+
+
+async def delete_obj(
+    model: T, obj_id: str | None = None, additional_filter: dict | None = None
+) -> T:
+    """
+    Delete an object.
+    """
+    filter_ = {"id": obj_id} if obj_id else {}
+    if additional_filter:
+        filter_.update(additional_filter)
+    obj = await settings.DB_CLIENT[collection_map[model.__name__]].find_one(filter_)
+    if obj is None:
+        raise HTTPException(status_code=404, detail="Object not found.")
+    await settings.DB_CLIENT[collection_map[model.__name__]].delete_one(filter_)
+    return model.from_mongo(obj)
+
+
+async def search_objs(
+    model: T,
+    data: SearchRequest,
+    additional_filter: dict | None = None,
+    projection: dict | None = None,
+) -> tuple[list[T], int]:
+    """
+    Search for objects in a specified collection based on search filters.
+    """
+    filters = []
+    date_filters = {}
+
+    for search_filter in data.filter:
+        if isinstance(search_filter.value, str):
+            date_match = re.fullmatch(r"^(\d{4}-\d{2}-\d{2});([+-]\d{1,2})$", search_filter.value)
+
+            if date_match:
+                if search_filter.name not in date_filters:
+                    date_filters[search_filter.name] = []
+
+                date_filters[search_filter.name].append(
+                    {
+                        "date": datetime.strptime(date_match.group(1), "%Y-%m-%d"),
+                        "timezone_offset": int(date_match.group(2)),
+                    }
+                )
+            else:
+                filters.append(
+                    {
+                        search_filter.name: {
+                            "$regex": f"^{re.escape(search_filter.value)}",
+                            "$options": "i",
+                        }
+                    }
+                )
+        else:
+            filters.append({search_filter.name: search_filter.value})
+
+    for field_name, dates in date_filters.items():
+        if len(dates) == 1:
+            date_info = dates[0]
+            user_local_day_start = date_info["date"]
+            user_local_day_end = user_local_day_start + timedelta(days=1)
+            filters.append(
+                {
+                    field_name: {
+                        "$gte": (
+                            user_local_day_start - timedelta(hours=date_info["timezone_offset"])
+                        ).isoformat(),
+                        "$lt": (
+                            user_local_day_end - timedelta(hours=date_info["timezone_offset"])
+                        ).isoformat(),
+                    }
+                }
+            )
+        elif len(dates) == 2:
+            start_date = min(dates, key=lambda x: x["date"])
+            end_date = max(dates, key=lambda x: x["date"])
+
+            start_datetime = start_date["date"] - timedelta(hours=start_date["timezone_offset"])
+            end_datetime = (
+                end_date["date"] + timedelta(days=1) - timedelta(hours=end_date["timezone_offset"])
+            )
+
+            filters.append(
+                {
+                    field_name: {
+                        "$gte": start_datetime.isoformat(),
+                        "$lt": end_datetime.isoformat(),
+                    }
+                }
+            )
+        elif len(dates) > 2:
+            dates_sorted = sorted(dates, key=lambda x: x["date"])
+            start_date = dates_sorted[0]
+            end_date = dates_sorted[-1]
+
+            start_datetime = start_date["date"] - timedelta(hours=start_date["timezone_offset"])
+            end_datetime = (
+                end_date["date"] + timedelta(days=1) - timedelta(hours=end_date["timezone_offset"])
+            )
+
+            filters.append(
+                {
+                    field_name: {
+                        "$gte": start_datetime.isoformat(),
+                        "$lt": end_datetime.isoformat(),
+                    }
+                }
+            )
+
+    if additional_filter:
+        filters.append(additional_filter)
+    regex_filter = {"$and": filters} if filters else {}
+    objects, total_count = await asyncio.gather(
+        settings.DB_CLIENT[collection_map[model.__name__]]
+        .find(regex_filter, projection)
+        .sort("id", -1)
+        .skip(data.pageSize * data.pageIndex)
+        .limit(data.pageSize)
+        .to_list(length=data.pageSize),
+        settings.DB_CLIENT[collection_map[model.__name__]].count_documents(regex_filter),
+    )
+    return [model.from_mongo(obj) for obj in objects], total_count

cbh/api/common/dto.py

@@ -0,0 +1,149 @@
+"""
+Common DTOs.
+"""
+
+from datetime import datetime
+from enum import Enum
+
+from pydantic import BaseModel, field_validator
+
+from cbh.core.config import settings
+
+
+class Paging(BaseModel):
+    """
+    Pagination model for API responses.
+    """
+
+    pageSize: int
+    pageIndex: int
+    totalCount: int
+
+
+class SearchFilter(BaseModel):
+    """
+    Search filter model for constructing database queries.
+
+    Attributes:
+        name (str): Field name to filter on
+        value (str | int): Value to search for
+    """
+
+    name: str
+    value: str | int
+
+
+class Scores(BaseModel):
+    """
+    Scores for the recording.
+    """
+
+    communication: int
+    activeListening: int
+    conversation: int
+    objection: int
+    empathy: int
+    final: int
+
+    def __sub__(self, other: "Scores") -> "Scores":
+        return Scores(
+            communication=self.communication - other.communication,
+            activeListening=self.activeListening - other.activeListening,
+            conversation=self.conversation - other.conversation,
+            objection=self.objection - other.objection,
+            empathy=self.empathy - other.empathy,
+            final=self.final - other.final,
+        )
+
+    def __mod__(self, other: "Scores") -> "Scores":
+        def calc_percentage_diff(current: int, previous: int) -> int:
+            if previous == 0:
+                return 0
+            return int(((current - previous) / previous) * 100)
+
+        return Scores(
+            communication=calc_percentage_diff(self.communication, other.communication),
+            activeListening=calc_percentage_diff(
+                self.activeListening, other.activeListening
+            ),
+            conversation=calc_percentage_diff(self.conversation, other.conversation),
+            objection=calc_percentage_diff(self.objection, other.objection),
+            empathy=calc_percentage_diff(self.empathy, other.empathy),
+            final=calc_percentage_diff(self.final, other.final),
+        )
+
+
+class DateValue(BaseModel):
+    """
+    Date value.
+    """
+
+    date: datetime
+    value: int
+
+
+class ValueDelta(BaseModel):
+    """
+    Value delta.
+    """
+
+    value: int
+    delta: int
+
+
+class IDName(BaseModel):
+    id: str | int
+    name: str
+
+
+class IDNamePicture(IDName):
+    pictureUrl: str | None
+
+    @field_validator("pictureUrl", mode="before")
+    @classmethod
+    def serialize_picture_url(cls, v: str | None) -> str | None:
+        """
+        Serialize the picture URL.
+        """
+        if v:
+            if not v.startswith("https://"):
+                return settings.S3_CLIENT.generate_presigned_url(v, expiration=3600)
+            return v
+        return None
+
+
+class OrderType(Enum):
+    """
+    Order type.
+    """
+
+    ASCENDING = 1
+    DESCENDING = -1
+
+
+class SortBy(BaseModel):
+    """
+    Sort by.
+    """
+
+    name: str
+    order: OrderType
+
+
+class TokenUsage(BaseModel):
+    inputTokens: int
+    outputTokens: int
+
+
+class SkillStatistics(BaseModel):
+    """
+    Skill statistics model.
+    """
+
+    score: int
+    bestAccount: IDNamePicture | None = None
+
+
+class InsightType(Enum):
+    MISTAKE = 1
+    ACHIEVEMENT = 2

cbh/api/common/schemas.py

@@ -0,0 +1,121 @@
+"""
+Common schemas.
+"""
+
+from datetime import date
+from typing import TypeVar, Generic
+
+from pydantic import BaseModel
+
+from cbh.api.common.dto import (
+    OrderType,
+    Paging,
+    SearchFilter,
+    SkillStatistics,
+    SortBy,
+    DateValue, InsightType,
+)
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class AllObjectsResponse(BaseModel, Generic[T]):
+    """
+    Response model for all objects.
+    """
+
+    paging: Paging
+    data: list[T]
+
+
+class SearchRequest(BaseModel):
+    """
+    Request schema for searching calls or statistics.
+
+    Attributes:
+        filter (list[SearchFilter]): List of filters to apply
+        pageSize (int): Number of items to return per page
+        pageIndex (int): Page index to retrieve
+    """
+
+    filter: list[SearchFilter]
+    pageSize: int
+    pageIndex: int
+
+
+class FilterRequest(BaseModel, Generic[T]):
+    """
+    Filter request.
+    """
+
+    filter: T
+    sortBy: SortBy | None = None
+    pageSize: int = 10
+    pageIndex: int = 0
+
+
+class PlainTextResponse(BaseModel):
+    """
+    Response model for plain text.
+    """
+
+    text: str
+
+
+class BatchIdsRequest(BaseModel):
+    """
+    Batch ids request.
+    """
+
+    ids: list[str]
+
+
+class EmailRequest(BaseModel):
+    """
+    Email request.
+    """
+
+    email: str
+
+
+class OrderTypeRequest(BaseModel):
+    """
+    Order type request.
+    """
+
+    order: OrderType
+
+
+class SkillsStatisticsResponse(BaseModel):
+    """
+    Scenario skills statistics response.
+    """
+
+    communication: SkillStatistics
+    activeListening: SkillStatistics
+    conversation: SkillStatistics
+    objection: SkillStatistics
+    empathy: SkillStatistics
+    final: int
+
+
+class RepProgressResponse(BaseModel):
+    """
+    Rep progress response.
+    """
+
+    progress: list[DateValue]
+
+
+class AIInsightsResponse(BaseModel):
+    """
+    AI insights.
+    """
+
+    aiInsights: str | None = None
+
+
+class InsightFilter(BaseModel):
+    type: InsightType
+    startDate: date | None = None
+    endDate: date | None = None

cbh/api/common/utils.py

@@ -0,0 +1,244 @@
+"""
+Common utilities.
+"""
+
+import asyncio
+import base64
+import re
+from typing import Callable, TypeVar
+
+import httpx
+
+from cbh.api.account.dto import AccountType
+from cbh.api.account.models import AccountModel
+from cbh.api.common.dto import (
+    LeaderboardStatisticsPosition,
+    Scores,
+    OrderType,
+    SkillStatistics,
+    IDNamePicture,
+)
+from cbh.api.common.schemas import SkillsStatisticsResponse
+from cbh.core.config import settings
+
+T = TypeVar("T")
+
+
+def calculate_avg_scores(reports: list) -> Scores:
+    if not reports:
+        return Scores(
+            communication=0,
+            activeListening=0,
+            conversation=0,
+            objection=0,
+            empathy=0,
+            final=0,
+        )
+    return Scores(
+        communication=round(sum(report.scores.communication for report in reports) / len(reports)),
+        activeListening=round(
+            sum(report.scores.activeListening for report in reports) / len(reports)
+        ),
+        conversation=round(sum(report.scores.conversation for report in reports) / len(reports)),
+        objection=round(sum(report.scores.objection for report in reports) / len(reports)),
+        empathy=round(sum(report.scores.empathy for report in reports) / len(reports)),
+        final=round(sum(report.scores.final for report in reports) / len(reports)),
+    )
+
+
+def form_user_stats(session_reports: list) -> dict[str, dict]:
+    user_stats = {}
+    for report in session_reports:
+        user_id = report.account.id
+        if user_id not in user_stats:
+            user_stats[user_id] = {
+                "account": report.account,
+                "reports": [],
+                "attempts": 0,
+            }
+        user_stats[user_id]["reports"].append(report)
+        user_stats[user_id]["attempts"] += 1
+    return user_stats
+
+
+def leaderboard_sort_key(item: tuple, reverse: bool = False) -> tuple:
+    user_data, score = item
+    if reverse:
+        return -score, user_data["attempts"], user_data["account"].name.lower()
+
+    return score, -user_data["attempts"], user_data["account"].name.lower()
+
+
+def build_leaderboard(
+    session_reports: list,
+    position_builder: Callable[[dict, Scores], T],
+    order: OrderType | None = None,
+) -> list[T]:
+    user_stats = form_user_stats(session_reports)
+    user_scores = [
+        (user_data, calculate_avg_scores(user_data["reports"]).final)
+        for user_data in user_stats.values()
+    ]
+
+    is_descending = order is None or order == OrderType.DESCENDING
+    sorted_users = sorted(
+        user_scores, key=lambda item: leaderboard_sort_key(item, reverse=is_descending)
+    )
+
+    leaderboard = []
+    for user_data, _ in sorted_users:
+        avg_scores = calculate_avg_scores(user_data["reports"])
+        leaderboard.append(position_builder(user_data, avg_scores))
+
+    return leaderboard
+
+
+def build_leaderboard_simple(
+    session_reports: list,
+    pageSize: int | None = None,
+    pageIndex: int | None = None,
+    order: OrderType | None = None,
+) -> list[LeaderboardStatisticsPosition]:
+    def position_builder(user_data: dict, avg_scores: Scores) -> LeaderboardStatisticsPosition:
+        return LeaderboardStatisticsPosition(
+            account=IDNamePicture(
+                id=user_data["account"].id,
+                name=user_data["account"].name,
+                pictureUrl=user_data["account"].pictureUrl,
+            ),
+            scores=avg_scores,
+        )
+
+    leaderboard = build_leaderboard(session_reports, position_builder, order)
+    return paginate_list(leaderboard, pageSize, pageIndex)
+
+
+def paginate_list(
+    items: list[T], page_size: int | None = None, page_index: int | None = None
+) -> list[T]:
+    if page_size is None or page_index is None:
+        return items
+    return items[page_index * page_size : (page_index + 1) * page_size]
+
+
+def form_additional_scenario_filter(account: AccountModel, allow_demo: bool = False):
+    from cbh.api.scenario.dto import AssigneesType, ScenarioStatus
+
+    filter_ = {"owner.organization.id": account.organization.id}
+    if account.accountType == AccountType.USER:
+        filter_.update(
+            {
+                "$or": [
+                    {"assignees": {"$size": 0}},
+                    {
+                        "assignees": {
+                            "$elemMatch": {
+                                "type": AssigneesType.USER.value,
+                                "account.id": account.id,
+                            }
+                        }
+                    },
+                    {
+                        "assignees": {
+                            "$elemMatch": {
+                                "type": AssigneesType.TEAM.value,
+                                "team.members": {"$elemMatch": {"id": account.id}},
+                            }
+                        }
+                    },
+                ],
+                "isTemplate": False,
+                "status": ScenarioStatus.ACTIVE.value,
+            }
+        )
+    if not allow_demo or account.accountType != AccountType.USER:
+        filter_.update(
+            {
+                "isDemo": False,
+            }
+        )
+    return filter_
+
+
+async def convert_document_to_text(file: bytes, filename: str) -> str:
+    if filename.endswith(".txt"):
+        return file.decode("utf-8", errors="ignore")
+    filename = re.sub(r"[^\w\s.-]", "", filename)
+    base64_file = base64.b64encode(file).decode("utf-8")
+    headers = {"Content-Type": "application/json"}
+    data = {
+        "apikey": settings.CONVERTIO_API_KEY,
+        "input": "base64",
+        "file": base64_file,
+        "filename": filename,
+        "outputformat": "txt",
+    }
+    async with httpx.AsyncClient(timeout=httpx.Timeout(timeout=120)) as client:
+        response = await client.post("https://api.convertio.co/convert", json=data, headers=headers)
+        response = response.json()
+        if response["code"] == 200:
+            conversion_id = response["data"]["id"]
+            status = ""
+            attempt = 0
+            while status != "finish":
+                if attempt > 50:
+                    raise Exception("Please, try again")
+                get_status_response = await client.get(
+                    f"https://api.convertio.co/convert/{conversion_id}/status"
+                )
+                get_status_response = get_status_response.json()
+                if get_status_response["code"] != 200:
+                    raise Exception("Please, try again")
+                else:
+                    status = get_status_response["data"]["step"]
+                    await asyncio.sleep(1)
+                attempt += 1
+            file_url = get_status_response["data"]["output"]["url"]
+            response = await client.get(file_url)
+            response.raise_for_status()
+            return response.content.decode("utf-8", errors="ignore")
+        else:
+            return ""
+
+
+def calculate_skills_statistics(session_reports: list) -> SkillsStatisticsResponse:
+    """
+    Calculate team skills statistics.
+    """
+    if not session_reports:
+        empty_skill = SkillStatistics(score=0, bestAccount=None)
+        return SkillsStatisticsResponse(
+            communication=empty_skill,
+            activeListening=empty_skill,
+            conversation=empty_skill,
+            objection=empty_skill,
+            empathy=empty_skill,
+            final=0,
+        )
+    avg_scores = calculate_avg_scores(session_reports)
+    skills = ["communication", "activeListening", "conversation", "objection", "empathy"]
+
+    skill_stats = {}
+    for skill in skills:
+        best_report = sorted(
+            session_reports,
+            key=lambda r, s=skill: (getattr(r.scores, s), r.datetimeInserted.timestamp()),
+            reverse=True,
+        )[0]
+        skill_stats[skill] = SkillStatistics(
+            score=getattr(avg_scores, skill),
+            bestAccount=IDNamePicture(
+                id=best_report.account.id,
+                name=best_report.account.name,
+                pictureUrl=best_report.account.pictureUrl,
+            ),
+        )
+
+    return SkillsStatisticsResponse(
+        communication=skill_stats["communication"],
+        activeListening=skill_stats["activeListening"],
+        conversation=skill_stats["conversation"],
+        objection=skill_stats["objection"],
+        empathy=skill_stats["empathy"],
+        final=avg_scores.final,
+    )

cbh/api/messages/__init__.py

@@ -0,0 +1,5 @@
+from fastapi import APIRouter
+
+messages_router = APIRouter(prefix="/messages", tags=["messages"])
+
+from . import views

cbh/api/messages/db_requests.py

@@ -0,0 +1,23 @@
+from cbh.api.common.db_requests import get_all_objs
+from cbh.api.messages.models import MessageModel
+
+
+async def get_chat_message_history(chat_id: str, account_id: str) -> list[dict]:
+    """
+    Get the message history for a chat.
+    """
+    messages = await get_all_objs(
+        MessageModel,
+        100000,
+        0,
+        additional_filter={"chat.account.id": account_id, "chat.id": chat_id},
+    )
+    response = []
+    for message in messages:
+        response.append(
+            {
+                "role": message.role,
+                "content": message.content,
+            }
+        )
+    return messages

cbh/api/messages/models.py

@@ -0,0 +1,13 @@
+from datetime import datetime
+from pydantic import Field
+
+from cbh.api.ari.dto import Author
+from cbh.core.database import MongoBaseModel
+
+
+class MessageModel(MongoBaseModel):
+    role: Author
+    content: str
+    chatId: str
+    accountId: str
+    datetimeInserted: datetime = Field(default_factory=datetime.now)

cbh/api/messages/views.py

@@ -0,0 +1,28 @@
+from fastapi import Depends
+from cbh.api.account.models import AccountModel
+from cbh.api.account.dto import AccountType
+from cbh.api.messages import messages_router
+from cbh.api.messages.models import MessageModel
+from cbh.api.common.schemas import AllObjectsResponse, Paging
+from cbh.api.common.db_requests import get_all_objs
+from cbh.core.security import PermissionDependency
+from cbh.core.wrappers import CbhResponseWrapper
+
+@messages_router.get("/messages/{chatId}")
+async def get_chat_messages(
+    chatId: str,
+    account: AccountModel = Depends(PermissionDependency([AccountType.ADMIN, AccountType.OWNER])),
+) -> CbhResponseWrapper[AllObjectsResponse[MessageModel]]:
+    messages, total_count = await get_all_objs(
+        MessageModel,
+        100000,
+        0,
+        additional_filter={"accountId": account.id, "chatId": chatId},
+        sort=("id", 1),
+    )
+    return CbhResponseWrapper(
+        data=AllObjectsResponse(
+            data=messages,
+            paging=Paging(pageSize=len(messages), pageIndex=0, totalCount=total_count),
+        )
+    )

cbh/api/platforms/__init__.py

@@ -0,0 +1,5 @@
+from fastapi import APIRouter
+
+platforms_router = APIRouter(prefix="/platforms", tags=["platforms"])
+
+from . import views

cbh/api/platforms/dto.py

@@ -0,0 +1,44 @@
+from enum import Enum
+
+
+class Category(int, Enum):
+    WEB_APPS_SAAS_MVP = 1
+    WEBSITES_LANDING_PAGES = 2
+    MOBILE_APPS = 3
+    UI_UX_DESIGN = 4
+    AI_CODING_TOOLS = 5
+    AUTOMATION_AI_AGENTS = 6
+    VIDEO_CREATIVE = 7
+    SEO_GEO = 8
+    GROWTH_SOCIAL_REDDIT = 9
+    RESEARCH_ANALYTICS = 10
+
+
+class Level(int, Enum):
+    LOW = 1
+    LOW_TO_MEDIUM = 2
+    MEDIUM = 3
+    MEDIUM_TO_HIGH = 4
+    HIGH = 5
+
+
+class ToolType(int, Enum):
+    NO_CODE = 1
+    HYBRID = 2
+    DEV = 3
+
+
+class Focus(int, Enum):
+    WEB = 1
+    MOBILE = 2
+    DESKTOP = 3
+    MULTI_PLATFORM = 4
+    MOBILE_DESIGN = 5
+    DEVELOPER_WORKFLOW = 6
+    DESKTOP_MULTI_PLATFORM_DEV = 7
+
+
+class MonetizationPriority(int, Enum):
+    LOW = 1
+    MEDIUM = 2
+    HIGH = 3

cbh/api/platforms/models.py

@@ -0,0 +1,23 @@
+from pydantic import Field
+
+from cbh.api.platforms.dto import Focus, ToolType, Level, Category, MonetizationPriority
+from cbh.core.database import MongoBaseModel
+
+class PlatformModel(MongoBaseModel):
+    name: str = Field(description="Platform name, e.g. 'v0', 'Lovable'")
+    category: Category = Field(description="Primary category. One of: 1=Web apps/SaaS/MVP, 2=Websites/Landing pages, 3=Mobile apps, 4=UI/UX Design, 5=AI Coding tools, 6=Automation/AI agents, 7=Video/Creative, 8=SEO/GEO, 9=Growth/Social/Reddit, 10=Research/Analytics")
+    subcategory: str = Field(description="Specific subcategory within the category, e.g. 'UI + app generation', 'Prompt-based app builder'")
+    oneLinePos: str = Field(description="One-line positioning statement describing what the tool does")
+    description: str = Field(description="Detailed description of the tool: when to use it, strengths, and best-fit scenarios")
+    userQueries: list[str] = Field(description="List of typical user queries/intents this tool covers, e.g. ['I want to build a SaaS', 'I need a dashboard']")
+    idealCases: str = Field(description="Description of the ideal client scenario, e.g. 'the client wants a fast visual MVP and is comfortable refining later'")
+    personas: list[str] = Field(description="List of recommended user personas, e.g. ['Founder', 'PM', 'Developer']")
+    level: Level = Field(description="Required skill level. One of: 1=Low, 2=Low-to-Medium, 3=Medium, 4=Medium-to-High, 5=High")
+    toolType: ToolType = Field(description="Tool type. One of: 1=No-code, 2=Hybrid, 3=Dev")
+    focus: list[Focus] = Field(description="Platform focus areas. List of: 1=Web, 2=Mobile, 3=Desktop, 4=Multi-platform, 5=Mobile design, 6=Developer workflow, 7=Desktop/Multi-platform dev")
+    productStage: list[str] = Field(description="Best product stages, e.g. ['Ideation', 'MVP', 'MVP UI']")
+    keyStrengths: list[str] = Field(description="List of key strengths of the tool")
+    caveats: list[str] = Field(description="List of main caveats or limitations")
+    monetizationPriority: MonetizationPriority = Field(description="Monetization priority. One of: 1=Low, 2=Medium, 3=High")
+    website: str = Field(description="Official website URL")
+    internalNotes: str = Field(description="Internal notes about when/how to recommend this tool")

cbh/api/security/__init__.py

@@ -0,0 +1,11 @@
+"""
+Security module initialization.
+"""
+
+from fastapi import APIRouter
+
+security_router = APIRouter(
+    prefix="/api/security",
+)
+
+from . import views  # pylint: disable=C0413  # noqa: E402,F401

cbh/api/security/db_requests.py

@@ -0,0 +1,186 @@
+"""
+Database requests module for security functionality.
+"""
+
+import asyncio
+from datetime import datetime, timedelta
+
+from fastapi import HTTPException
+from passlib.context import CryptContext
+from pydantic import EmailStr
+from pymongo import ReturnDocument
+
+from cbh.api.account.dto import AccountType, RegistrationType
+from cbh.api.account.models import AccountModel, AccountShorten
+from cbh.api.security.dto import VerificationCodeStatus, VerificationCodeType
+from cbh.api.security.models import VerificationCodeModel
+from cbh.api.security.schemas import (
+    LoginAccountRequest,
+    RegisterAccountRequest,
+)
+from cbh.core.config import settings
+from cbh.core.security import verify_password
+from cbh.core.wrappers import background_task
+
+
+async def check_unique_email(email: EmailStr | str) -> AccountModel | None:
+    """
+    Check if a field value already exists in the database to ensure uniqueness.
+    """
+    account = await settings.DB_CLIENT.accounts.find_one(
+        {"email": {"$regex": f"^{str(email)}$", "$options": "i"}}
+    )
+    account = AccountModel.from_mongo(account) if account else None
+
+    return account
+
+
+async def authenticate_account(data: LoginAccountRequest) -> AccountModel:
+    """
+    Authenticate a user account using mail and password.
+    """
+    account = await settings.DB_CLIENT.accounts.find_one(
+        {"email": {"$regex": f"^{data.email}$", "$options": "i"}}
+    )
+    if account is None:
+        raise HTTPException(status_code=404, detail="Invalid email or password.")
+
+    account = AccountModel.from_mongo(account)
+    if account.registrationType != RegistrationType.ORGANIC:
+        raise HTTPException(status_code=422, detail="Please sign in with social providers.")
+
+    if not verify_password(data.password, account.password):
+        raise HTTPException(status_code=400, detail="Invalid email or password.")
+
+    return account
+
+
+async def get_account_by_email(email: str) -> AccountModel | None:
+    """
+    Verify if an account exists.
+    """
+    account = await settings.DB_CLIENT.accounts.find_one(
+        {"email": {"$regex": f"^{email}$", "$options": "i"}}
+    )
+    return AccountModel.from_mongo(account) if account else None
+
+
+async def create_code_obj(
+    account: AccountShorten, type_: VerificationCodeType, time_delta: timedelta
+) -> VerificationCodeModel:
+    """
+    Create a code object.
+    """
+    prev_code = (
+        await settings.DB_CLIENT.verificationcodes.find(
+            {
+                "account.id": account.id,
+                "type": type_.value,
+            }
+        )
+        .sort("_id", -1)
+        .to_list(length=1)
+    )
+    prev_code = VerificationCodeModel.from_mongo(prev_code[0]) if prev_code else None
+
+    if prev_code and prev_code.datetimeInserted > datetime.now() - timedelta(minutes=1):
+        raise HTTPException(status_code=429, detail="Too many requests")
+
+    code = VerificationCodeModel(
+        account=account,
+        type=type_,
+        expiresAt=datetime.now() + time_delta,
+    )
+    await settings.DB_CLIENT.verificationcodes.insert_one(code.to_mongo())
+    return code
+
+
+@background_task()
+async def set_used_code(code: VerificationCodeModel | None):
+    """
+    Set a code object as used.
+    """
+    if code:
+        await settings.DB_CLIENT.verificationcodes.update_one(
+            {"id": code.id}, {"$set": {"status": VerificationCodeStatus.USED.value}}
+        )
+
+
+async def verify_code_obj(
+    code_: str, types: list[VerificationCodeType], exception: bool = True, set_used: bool = True
+) -> VerificationCodeModel:
+    """
+    Verify a code object.
+    """
+    code = (
+        await settings.DB_CLIENT.verificationcodes.find(
+            {"id": code_, "type": {"$in": [t.value for t in types]}},
+        )
+        .sort("_id", -1)
+        .to_list(length=1)
+    )
+    code = VerificationCodeModel.from_mongo(code[0]) if code else None
+
+    if not code and exception:
+        error_msg = "Invalid invitation link. Please ask your manager to resend the invite."
+        if VerificationCodeType.PASSWORD_RESET in types:
+            error_msg = "Invalid password reset link. Please request a new one."
+        raise HTTPException(
+            status_code=404,
+            detail=error_msg,
+        )
+
+    if code and code.status == VerificationCodeStatus.USED and exception:
+        error_map = {
+            VerificationCodeType.ORG_INVITATION: "You already created an account. Please sign in.",
+            VerificationCodeType.TEAM_INVITATION: "You already accepted this invitation. Please sign in.",
+            VerificationCodeType.PASSWORD_RESET: "You already used this reset link. Please request a new one.",
+            VerificationCodeType.ORG_CREATION: "You already created an organization. Please sign in.",
+        }
+        raise HTTPException(status_code=400, detail=error_map[code.type])
+
+    if code and code.expiresAt < datetime.now() and exception:
+        error_msg = "Expired invitation link. Please ask your manager to resend the invite."
+        if VerificationCodeType.PASSWORD_RESET in types:
+            error_msg = "Expired password reset link. Please request a new one."
+        raise HTTPException(status_code=410, detail=error_msg)
+
+    if code and set_used:
+        asyncio.create_task(set_used_code(code))
+    return code
+
+
+async def reset_password_obj(account: AccountShorten, password: str) -> AccountShorten:
+    """
+    Reset a password object.
+    """
+    password = CryptContext(schemes=["bcrypt"], deprecated="auto").hash(password)
+    await settings.DB_CLIENT.accounts.update_one(
+        {"id": account.id},
+        {"$set": {"password": password}},
+    )
+    return account
+
+
+
+async def create_google_account(user_info: dict) -> AccountModel:
+    account = AccountModel(
+        email=user_info["email"],
+        name=user_info.get("name"),
+        accountType=AccountType.USER,
+        registrationType=RegistrationType.GOOGLE,
+    )
+    await settings.DB_CLIENT.accounts.insert_one(account.to_mongo())
+    return account
+
+
+async def create_account(data: RegisterAccountRequest) -> AccountModel:
+    account = AccountModel(
+        email=data.email,
+        password=data.password,
+        name=data.name,
+        accountType=AccountType.USER,
+        registrationType=RegistrationType.ORGANIC,
+    )
+    await settings.DB_CLIENT.accounts.insert_one(account.to_mongo())
+    return account

cbh/api/security/dto.py

@@ -0,0 +1,48 @@
+"""
+Data Transfer Objects (DTOs) for security functionality.
+"""
+
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class AccessToken(BaseModel):
+    """
+    Access token model for authentication.
+    """
+
+    type: str = "Bearer"
+    value: str
+
+
+class GoogleCallbackError(Enum):
+    """
+    Error model for Google callback.
+    """
+
+    ORGANIZATION_NOT_FOUND = "OrganizationNotFound"
+    BLOCKED_ACCOUNT = "BlockedAccount"
+    INVALID_CODE = "InvalidCode"
+    EMAIL_MISMATCH = "EmailMismatch"
+    EXPIRED_CODE = "ExpiredCode"
+
+
+class VerificationCodeType(Enum):
+    """
+    Enum for verification code types.
+    """
+
+    PASSWORD_RESET = 1
+    TEAM_INVITATION = 2
+    ORG_INVITATION = 3
+    ORG_CREATION = 4
+
+
+class VerificationCodeStatus(Enum):
+    """
+    Enum for verification code status.
+    """
+
+    PENDING = 1
+    USED = 2

cbh/api/security/models.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+
+from pydantic import Field
+
+from cbh.api.account.models import AccountShorten
+from cbh.api.security.dto import VerificationCodeStatus, VerificationCodeType
+from cbh.core.database import MongoBaseModel
+
+
+class VerificationCodeModel(MongoBaseModel):
+    account: AccountShorten
+    expiresAt: datetime
+    type: VerificationCodeType
+    status: VerificationCodeStatus = VerificationCodeStatus.PENDING
+    datetimeInserted: datetime = Field(default_factory=datetime.now)

cbh/api/security/schemas.py

@@ -0,0 +1,48 @@
+"""
+Schema definitions for security API endpoints.
+"""
+
+from pydantic import BaseModel, EmailStr
+
+from cbh.api.account.models import AccountModel
+from cbh.api.account.dto import AccountType
+from cbh.api.security.dto import AccessToken
+
+
+
+class LoginAccountRequest(BaseModel):
+    """
+    Request model for account login.
+    """
+
+    email: EmailStr
+    password: str
+
+
+class LoginAccountResponse(BaseModel):
+    """
+    Response model for successful login.
+    """
+
+    accessToken: AccessToken | None = None
+    account: AccountModel
+    code: str | None = None
+
+
+class ResetPasswordConfirmRequest(BaseModel):
+    """
+    Request model for confirming a password reset.
+    """
+
+    code: str
+    password: str
+
+
+class RegisterAccountRequest(BaseModel):
+    """
+    Request model for registering a new account.
+    """
+
+    name: str
+    email: EmailStr
+    password: str

cbh/api/security/services/__init__.py

@@ -0,0 +1,11 @@
+from .oauth import handle_account_oauth
+from .utils import (
+    form_google_user_info,
+    form_google_login_url,
+)
+
+__all__ = [
+    "handle_account_oauth",
+    "form_google_user_info",
+    "form_google_login_url",
+]