Update for HF

.gitattributes

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,8 @@
+.env
+__pycache__/
+*.pyc
+.venv/
+venv/
+.python-version
+.DS_Store
+rest.http

Dockerfile

@@ -0,0 +1,41 @@
+# Use Python 3.11 slim image for smaller size
+FROM python:3.11-slim
+
+# Set working directory
+WORKDIR /app
+
+# Set environment variables for Hugging Face Spaces (can be overridden for local development)
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PORT=7860 \
+    HF_HOME=/app/cache \
+    TRANSFORMERS_CACHE=/app/cache
+
+# Install system dependencies including curl for health check
+RUN apt-get update && apt-get install -y \
+    gcc \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements first for better caching
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Copy all application code
+COPY . .
+
+# Create cache directory for Hugging Face models
+RUN mkdir -p /app/cache && chmod 777 /app/cache
+
+# Expose port (default 8000 for local, can be set to 7860 for Hugging Face Spaces)
+EXPOSE ${PORT}
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:${PORT}/ || exit 1
+
+# Run the application (uses PORT environment variable)
+CMD sh -c "uvicorn main:app --host 0.0.0.0 --port ${PORT}"

README.md

@@ -0,0 +1,10 @@
+---
+title: Customer Service AI
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
+sdk: docker
+sdk_version: "latest"
+app_file: main.py
+pinned: false
+---

README_HF.md

@@ -0,0 +1,10 @@
+---
+title: Customer Service AI
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
+sdk: docker
+sdk_version: "latest"
+app_file: main.py
+pinned: false
+---

ai_service.py

@@ -0,0 +1,247 @@
+import re
+import json
+from config import pc, index, EMBED_MODEL, hf_client, PROMPT, HF_MODEL, TRANSFER_PROMPT
+from database import db_manager
+from transfers import (
+    prepare_transfer,
+    confirm_transfer,
+    cancel_transfer,
+    get_pending_transfer,
+    get_account_balance,
+    get_sender_account,
+)
+
+
+MODEL_NAME = HF_MODEL
+BASE_PROMPT = PROMPT or "You are a helpful banking customer service assistant."
+TRANSFER_PROMPT = TRANSFER_PROMPT or (
+    "Current user telegram_id is {telegram_id}. "
+    "The model must never choose, guess, extract, or override any telegram_id for tool calls. "
+    "Always act only for the current authenticated user from server-side request context. "
+    "If the user asks for another person's balance or provides another person's telegram ID, refuse and explain that you can only access the current user's own account. "
+    "If the user asks for their balance, call check_account_balance. "
+    "For money transfers, first collect the receiver account serial ID and the amount if it is missing. "
+    "Then call prepare_money_transfer to fetch the receiver name and store the pending transfer. "
+    "Show the receiver name back to the user and ask for explicit confirmation. "
+    "Only call confirm_money_transfer after the user clearly agrees. "
+    "If the user rejects the receiver or wants to stop, call cancel_money_transfer. "
+    "Never claim a transfer is completed unless confirm_money_transfer returns success. "
+    "If a tool result returns 'need_user_name': true, ask the user for their name and then call create_illusion_account with their name. "
+    "If a tool result indicates an illusion account was created (contains 'is_illusion': true or mentions testing), inform the user that an illusion account with 2000 YER balance was created for testing purposes. "
+    "If confirm_money_transfer returns a result with 'is_illusion': true, make sure to display the testing disclaimer message to the user."
+)
+
+
+def clean_ai_response(text: str) -> str:
+    if not text:
+        return ""
+
+    text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL)
+    
+    text = re.sub(r'<br\s*/?>', '\n', text, flags=re.IGNORECASE)
+    text = re.sub(r'</?(p|div|span|section)[^>]*>', '\n', text, flags=re.IGNORECASE)
+    
+    text = re.sub(r'<[^>]+>', '', text)
+
+    text = re.sub(r'^\|.*\|\s*$', '', text, flags=re.MULTILINE)
+    text = re.sub(r'^[\s|:-]+$', '', text, flags=re.MULTILINE)
+
+    text = re.sub(r'^#{1,6}\s*', '', text, flags=re.MULTILINE)
+
+    text = re.sub(r'\n{3,}', '\n\n', text)
+    
+    return text.strip()
+
+async def search_bank_knowledge(query: str):
+    query_embedding = pc.inference.embed(
+        model=EMBED_MODEL,
+        inputs=[query],
+        parameters={"input_type": "query"}
+    )
+    search_results = index.query(
+        vector=query_embedding[0].values,
+        top_k=3,
+        include_metadata=True
+    )
+    return "\n".join([res.metadata['original_text'] for res in search_results.matches])
+
+TOOLS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "search_bank_knowledge",
+            "description": "Use this tool to search the official Hadhramout Bank profile for accurate information about services, organizational structure, capital, and policies.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "The search query (e.g., 'What is Hadhramout Bank capital?' or 'individual services')."
+                    }
+                },
+                "required": ["query"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "check_account_balance",
+            "description": "Use this tool when the user wants to check their own account balance. The server identifies the current user from request context.",
+            "parameters": {
+                "type": "object",
+                "properties": {},
+                "required": []
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "prepare_money_transfer",
+            "description": "Use this tool when the user wants to transfer money. The server identifies the sender from request context, looks up the receiver by account serial ID, stores a pending transfer, and returns the receiver name for confirmation before any money is sent.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "receiver_serial_id": {
+                        "type": "string",
+                        "description": "The serial ID of the receiver account."
+                    },
+                    "amount": {
+                        "type": "number",
+                        "description": "Amount to transfer. Ask the user for it if missing."
+                    }
+                },
+                "required": ["receiver_serial_id"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "confirm_money_transfer",
+            "description": "Use this tool only after the user confirms that the receiver account is correct and the transfer should proceed.",
+            "parameters": {
+                "type": "object",
+                "properties": {},
+                "required": []
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "cancel_money_transfer",
+            "description": "Use this tool when the user says the receiver is wrong or wants to stop a pending money transfer.",
+            "parameters": {
+                "type": "object",
+                "properties": {},
+                "required": []
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "get_pending_money_transfer",
+            "description": "Use this tool to inspect the current pending transfer for the current user before asking for confirmation or when the user asks about the transfer details.",
+            "parameters": {
+                "type": "object",
+                "properties": {},
+                "required": []
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "create_illusion_account",
+            "description": "Use this tool when the user needs an illusion account created for testing purposes. Requires the user's name.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "user_name": {
+                        "type": "string",
+                        "description": "The name of the user for the illusion account."
+                    }
+                },
+                "required": ["user_name"]
+            }
+        }
+    }
+]
+
+
+async def run_tool(tool_name: str, args: dict, telegram_id: int):
+    if tool_name == "search_bank_knowledge":
+        return await search_bank_knowledge(args["query"])
+    if tool_name == "check_account_balance":
+        return json.dumps(get_account_balance(telegram_id), ensure_ascii=False)
+    if tool_name == "prepare_money_transfer":
+        return json.dumps(
+            prepare_transfer(
+                telegram_id=telegram_id,
+                receiver_serial_id=args["receiver_serial_id"],
+                amount=args.get("amount"),
+            ),
+            ensure_ascii=False,
+        )
+    if tool_name == "confirm_money_transfer":
+        return json.dumps(confirm_transfer(telegram_id), ensure_ascii=False)
+    if tool_name == "cancel_money_transfer":
+        return json.dumps(cancel_transfer(telegram_id), ensure_ascii=False)
+    if tool_name == "get_pending_money_transfer":
+        return json.dumps(get_pending_transfer(telegram_id), ensure_ascii=False)
+    if tool_name == "create_illusion_account":
+        return json.dumps(get_sender_account(telegram_id, args["user_name"]), ensure_ascii=False)
+    return json.dumps({"success": False, "message": f"Unknown tool: {tool_name}"}, ensure_ascii=False)
+
+async def get_ai_response(user_query: str, telegram_id: int):
+    conversation_history = []
+    if db_manager:
+        raw_history = await db_manager.get_conversation_history(telegram_id, limit=6)
+        raw_history.reverse()
+        for msg in raw_history:
+            if msg.get('message_text'):
+                role = "user" if msg['message_type'] == 'user' else "assistant"
+                conversation_history.append({"role": role, "content": msg['message_text']})
+
+    transfer_instructions = TRANSFER_PROMPT.format(telegram_id=telegram_id)
+    messages = [{"role": "system", "content": f"{BASE_PROMPT}\n\n{transfer_instructions}"}] + conversation_history + [{"role": "user", "content": user_query}]
+    
+    
+    import asyncio
+    loop = asyncio.get_event_loop()
+    
+    
+    def call_hf(msgs):
+        return hf_client.chat.completions.create(
+            model=MODEL_NAME,
+            messages=msgs,
+            tools=TOOLS,
+            tool_choice="auto",
+            temperature=0.1,
+            max_tokens=800
+        )
+
+    completion = await loop.run_in_executor(None, lambda: call_hf(messages))
+    response_message = completion.choices[0].message
+
+    for _ in range(4):
+        if not response_message.tool_calls:
+            break
+
+        messages.append(response_message)
+        for tool_call in response_message.tool_calls:
+            args = json.loads(tool_call.function.arguments or "{}")
+            tool_result = await run_tool(tool_call.function.name, args, telegram_id)
+            messages.append({
+                "role": "tool",
+                "tool_call_id": tool_call.id,
+                "content": tool_result
+            })
+
+        completion = await loop.run_in_executor(None, lambda: call_hf(messages))
+        response_message = completion.choices[0].message
+
+    return clean_ai_response(response_message.content if response_message.content else "")

config.py

@@ -0,0 +1,49 @@
+import os
+from pinecone import Pinecone
+from openai import OpenAI
+from dotenv import load_dotenv
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Environment Variables
+PINECONE_API_KEY = os.environ.get("PINECONE_API_KEY")
+PINECONE_INDEX = os.environ.get("PINECONE_INDEX", "customerserviceindex")
+HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HF_API_KEY")
+TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN")
+TELEGRAM_WEBHOOK_SECRET = os.environ.get("TELEGRAM_WEBHOOK_SECRET")
+SUPABASE_URL = os.environ.get("SUPABASE_URL")
+SUPABASE_KEY = os.environ.get("SUPABASE_KEY")
+TELEGRAM_DOMAIN = os.environ.get("TELEGRAM_DOMAIN", "https://api.telegram.org").rstrip("/")
+
+# Only create TELEGRAM_URL if token exists
+TELEGRAM_URL = f"{TELEGRAM_DOMAIN}/bot{TELEGRAM_TOKEN}/sendMessage" if TELEGRAM_TOKEN and TELEGRAM_DOMAIN else None
+
+EMBED_MODEL = os.environ.get("EMBED_MODEL", "multilingual-e5-large")
+HF_MODEL = os.environ.get(
+    "HF_MODEL",
+    "dphn/Dolphin-Mistral-24B-Venice-Edition",
+)
+PROMPT = os.environ.get("PROMPT")
+TRANSFER_PROMPT = os.environ.get("TRANSFER_PROMPT")
+
+# Initialize clients only if API keys are available
+pc = None
+if PINECONE_API_KEY:
+    pc = Pinecone(api_key=PINECONE_API_KEY)
+
+hf_client = None
+if HF_TOKEN:
+    try:
+        hf_client = OpenAI(
+            base_url="https://router.huggingface.co/v1",
+            api_key=HF_TOKEN,
+        )
+    except Exception as e:
+        print(f"Warning: Failed to initialize Hugging Face OpenAI client: {e}")
+        hf_client = None
+
+# Initialize index only if Pinecone client is available
+index = None
+if pc:
+    index = pc.Index(PINECONE_INDEX)

database.py

@@ -0,0 +1,104 @@
+import os
+from datetime import datetime
+from typing import List, Dict, Optional
+from supabase import create_async_client, AsyncClient
+import logging
+from config import SUPABASE_URL, SUPABASE_KEY
+
+class DatabaseManager:
+    def __init__(self, supabase_url: str = SUPABASE_URL, supabase_key: str = SUPABASE_KEY):
+        if not supabase_url or not supabase_key:
+            raise ValueError("SUPABASE_URL and SUPABASE_KEY must be set")
+        self.supabase_url = supabase_url
+        self.supabase_key = supabase_key
+        self.supabase: Optional[AsyncClient] = None
+        self.logger = logging.getLogger(__name__)
+
+    async def _ensure_connection(self):
+        if self.supabase is None:
+            await self.connect()
+
+    async def connect(self):
+        """Initialize the async client"""
+        if not self.supabase:
+            self.supabase = await create_async_client(self.supabase_url, self.supabase_key)
+
+    async def create_or_update_user(self, telegram_id: int, username: str = None, 
+                                    first_name: str = None, last_name: str = None):
+        await self._ensure_connection()
+        try:
+            existing_user = await self.supabase.table("users").select("id").eq("telegram_id", telegram_id).execute()
+            
+            user_data = {
+                "telegram_id": telegram_id,
+                "username": username,
+                "first_name": first_name,
+                "last_name": last_name,
+                "updated_at": datetime.utcnow().isoformat()
+            }
+            
+            if existing_user.data:
+                result = await self.supabase.table("users").update(user_data).eq("telegram_id", telegram_id).execute()
+            else:
+                user_data["created_at"] = datetime.utcnow().isoformat()
+                result = await self.supabase.table("users").insert(user_data).execute()
+            
+            return result.data[0] if result.data else None
+        except Exception as e:
+            self.logger.error(f"Error creating/updating user: {e}")
+            return None
+
+    async def save_message(self, telegram_id: int, message_text: str, message_type: str):
+        await self._ensure_connection()
+        try:
+            await self.create_or_update_user(telegram_id)
+            
+            message_data = {
+                "telegram_id": telegram_id,
+                "message_text": message_text,
+                "message_type": message_type,
+                "created_at": datetime.utcnow().isoformat()
+            }
+            
+            result = await self.supabase.table("messages").insert(message_data).execute()
+            await self._ensure_active_session(telegram_id)
+            
+            return result.data[0] if result.data else None
+        except Exception as e:
+            self.logger.error(f"Error saving message: {e}")
+            return None
+
+    async def get_conversation_history(self, telegram_id: int, limit: int = 10) -> List[Dict]:
+        await self._ensure_connection()
+        try:
+            result = await (self.supabase.table("messages")
+                           .select("message_text, message_type, created_at")
+                           .eq("telegram_id", telegram_id)
+                           .order("created_at", desc=True)
+                           .limit(limit)
+                           .execute())
+            return result.data if result.data else []
+        except Exception as e:
+            self.logger.error(f"Error getting history: {e}")
+            return []
+
+    async def _ensure_active_session(self, telegram_id: int):
+        await self._ensure_connection()
+        try:
+            active = await (self.supabase.table("conversation_sessions")
+                            .select("id")
+                            .eq("telegram_id", telegram_id)
+                            .is_("session_end", "null")
+                            .execute())
+            
+            if not active.data:
+                session_data = {
+                    "telegram_id": telegram_id,
+                    "session_start": datetime.utcnow().isoformat(),
+                    "created_at": datetime.utcnow().isoformat()
+                }
+                await self.supabase.table("conversation_sessions").insert(session_data).execute()
+        except Exception as e:
+            self.logger.error(f"Error ensuring session: {e}")
+
+db_manager = DatabaseManager()

main.py

@@ -0,0 +1,39 @@
+from fastapi import FastAPI, Header, Request
+from schemas import AiTestRequest, WebhookData
+from security import validate_webhook_secret
+from telegram_handlers import telegram_webhook
+from utils import dns_test, test_ai_response
+
+app = FastAPI()
+
+@app.get("/")
+async def root():
+    return {"message": "Hadhramout Bank AI Backend is Live"}
+
+@app.post("/webhook")
+async def webhook(
+    request: Request,
+    data: WebhookData,
+    x_telegram_bot_api_secret_token: str | None = Header(default=None),
+):
+    print("-------message arive at this time \n")
+    validate_webhook_secret(request, x_telegram_bot_api_secret_token)
+
+    return await telegram_webhook(data)
+
+@app.get("/dns-test")
+async def dns(
+    request: Request,
+    x_telegram_bot_api_secret_token: str | None = Header(default=None),
+):
+    validate_webhook_secret(request, x_telegram_bot_api_secret_token)
+    return await dns_test()
+
+@app.post("/ai-test")
+async def ai(
+    request: Request,
+    data: AiTestRequest,
+    x_telegram_bot_api_secret_token: str | None = Header(default=None),
+):
+    validate_webhook_secret(request, x_telegram_bot_api_secret_token)
+    return await test_ai_response(data.message, data.telegram_id)

requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+uvicorn
+pinecone
+httpx
+python-dotenv
+supabase
+openai

rest.http

@@ -0,0 +1,22 @@
+POST http://localhost:8000/ai-test HTTP/1.1
+Content-Type: application/json
+X-Telegram-Bot-Api-Secret-Token: {{$dotenv TELEGRAM_WEBHOOK_SECRET}}
+
+{
+    "message": "confirm then tell me how much that he has in his account",
+    "telegram_id": 12
+}
+###
+GET https://codeboker-customer-service.hf.space/ai-test HTTP/1.1
+
+###
+POST https://codeboker-customer-service.hf.space/webhook HTTP/1.1
+Content-Type: application/json
+X-Telegram-Bot-Api-Secret-Token: {{$dotenv TELEGRAM_WEBHOOK_SECRET}}
+
+{
+    "message": {
+        "chat": {"id": 123},
+        "text": "السلام عليكم وؤحمة الله اريد ان استعلم عنلاالخدمات الالكترونية"
+    }
+}

schemas.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+
+
+class ChatInfo(BaseModel):
+    id: int
+    username: str = None
+    first_name: str = None
+    last_name: str = None
+
+
+class Message(BaseModel):
+    chat: ChatInfo
+    text: str = ""
+
+
+class WebhookData(BaseModel):
+    message: Message = None
+
+
+class AiTestRequest(BaseModel):
+    message: str
+    telegram_id: int = 12

security.py

@@ -0,0 +1,54 @@
+from collections import defaultdict, deque
+from time import monotonic
+
+from fastapi import HTTPException, Request
+
+from config import TELEGRAM_WEBHOOK_SECRET
+
+FAILED_ATTEMPT_WINDOW_SECONDS = 60
+FAILED_ATTEMPT_LIMIT = 5
+BLOCK_DURATION_SECONDS = 15 * 60
+
+_failed_secret_attempts: dict[str, deque[float]] = defaultdict(deque)
+_blocked_clients: dict[str, float] = {}
+
+
+def _get_client_key(request: Request) -> str:
+    forwarded_for = request.headers.get("x-forwarded-for")
+    if forwarded_for:
+        return forwarded_for.split(",")[0].strip()
+
+    if request.client and request.client.host:
+        return request.client.host
+
+    return "unknown"
+
+
+def _prune_failed_attempts(client_key: str, now: float) -> deque[float]:
+    attempts = _failed_secret_attempts[client_key]
+    cutoff = now - FAILED_ATTEMPT_WINDOW_SECONDS
+    while attempts and attempts[0] < cutoff:
+        attempts.popleft()
+    return attempts
+
+
+def validate_webhook_secret(request: Request, secret_header: str | None) -> None:
+    if not TELEGRAM_WEBHOOK_SECRET:
+        raise HTTPException(status_code=500, detail="Webhook secret is not configured")
+
+    client_key = _get_client_key(request)
+    now = monotonic()
+    blocked_until = _blocked_clients.get(client_key)
+    if blocked_until and now < blocked_until:
+        raise HTTPException(status_code=429, detail="Too many requests")
+
+    if secret_header != TELEGRAM_WEBHOOK_SECRET:
+        attempts = _prune_failed_attempts(client_key, now)
+        attempts.append(now)
+        if len(attempts) >= FAILED_ATTEMPT_LIMIT:
+            _blocked_clients[client_key] = now + BLOCK_DURATION_SECONDS
+            attempts.clear()
+        raise HTTPException(status_code=403, detail="Forbidden")
+
+    _blocked_clients.pop(client_key, None)
+    _failed_secret_attempts.pop(client_key, None)

supabase_schema.sql

@@ -0,0 +1,90 @@
+-- Supabase Database Schema for Conversation History
+-- Run this in your Supabase SQL Editor
+
+-- Enable UUID extension
+CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
+
+-- Users table to store user information
+CREATE TABLE IF NOT EXISTS users (
+    id UUID DEFAULT uuid_generate_v4() PRIMARY KEY,
+    telegram_id BIGINT UNIQUE NOT NULL,
+    username VARCHAR(255),
+    first_name VARCHAR(255),
+    last_name VARCHAR(255),
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
+    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
+);
+
+-- Messages table to store conversation history
+CREATE TABLE IF NOT EXISTS messages (
+    id UUID DEFAULT uuid_generate_v4() PRIMARY KEY,
+    telegram_id BIGINT NOT NULL,
+    message_text TEXT NOT NULL,
+    message_type VARCHAR(20) NOT NULL CHECK (message_type IN ('user', 'assistant')),
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
+    FOREIGN KEY (telegram_id) REFERENCES users(telegram_id) ON DELETE CASCADE
+);
+
+-- Conversation sessions to group messages by session
+CREATE TABLE IF NOT EXISTS conversation_sessions (
+    id UUID DEFAULT uuid_generate_v4() PRIMARY KEY,
+    telegram_id BIGINT NOT NULL,
+    session_start TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
+    session_end TIMESTAMP WITH TIME ZONE,
+    message_count INTEGER DEFAULT 0,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
+    FOREIGN KEY (telegram_id) REFERENCES users(telegram_id) ON DELETE CASCADE
+);
+
+-- Indexes for better performance
+CREATE INDEX IF NOT EXISTS idx_messages_telegram_id_created_at ON messages(telegram_id, created_at);
+CREATE INDEX IF NOT EXISTS idx_users_telegram_id ON users(telegram_id);
+CREATE INDEX IF NOT EXISTS idx_sessions_telegram_id ON conversation_sessions(telegram_id);
+
+-- Function to update updated_at timestamp
+CREATE OR REPLACE FUNCTION update_updated_at_column()
+RETURNS TRIGGER AS $$
+BEGIN
+    NEW.updated_at = NOW();
+    RETURN NEW;
+END;
+$$ language 'plpgsql';
+
+-- Trigger to update user's updated_at timestamp
+CREATE TRIGGER update_user_updated_at 
+    BEFORE UPDATE ON users
+    FOR EACH ROW
+    EXECUTE FUNCTION update_updated_at_column();
+
+-- Trigger to update session message count
+CREATE OR REPLACE FUNCTION update_session_count()
+RETURNS TRIGGER AS $$
+BEGIN
+    UPDATE conversation_sessions 
+    SET message_count = message_count + 1 
+    WHERE telegram_id = NEW.telegram_id AND session_end IS NULL;
+    RETURN NEW;
+END;
+$$ language 'plpgsql';
+
+CREATE TRIGGER update_session_message_count
+    AFTER INSERT ON messages
+    FOR EACH ROW
+    EXECUTE FUNCTION update_session_count();
+
+-- Row Level Security (RLS) policies
+ALTER TABLE users ENABLE ROW LEVEL SECURITY;
+ALTER TABLE messages ENABLE ROW LEVEL SECURITY;
+ALTER TABLE conversation_sessions ENABLE ROW LEVEL SECURITY;
+
+-- Policy for users table (allow all operations for now)
+CREATE POLICY "Enable all operations for users" ON users
+    FOR ALL USING (true);
+
+-- Policy for messages table (allow all operations for now)
+CREATE POLICY "Enable all operations for messages" ON messages
+    FOR ALL USING (true);
+
+-- Policy for conversation_sessions table (allow all operations for now)
+CREATE POLICY "Enable all operations for conversation_sessions" ON conversation_sessions
+    FOR ALL USING (true);

telegram_handlers.py

@@ -0,0 +1,95 @@
+import httpx
+import json
+import html
+import re
+import asyncio
+from config import TELEGRAM_URL, TELEGRAM_TOKEN
+from ai_service import get_ai_response
+from database import db_manager
+from schemas import WebhookData
+
+TELEGRAM_IP = "149.154.167.220"
+MAX_TELEGRAM_MESSAGE_LENGTH = 4096
+
+def _sanitize_telegram_text(text: str) -> str:
+    if text is None:
+        return ""
+    normalized = str(text).replace("\r\n", "\n").replace("\r", "\n")
+    cleaned = "".join(
+        ch
+        for ch in normalized
+        if (ch in ("\n", "\t") or ord(ch) >= 32) and not (0xD800 <= ord(ch) <= 0xDFFF)
+    )
+    return cleaned.strip()
+
+def _format_telegram_message(text: str) -> str:
+    if not text:
+        return text
+    escaped = html.escape(text, quote=False)
+    formatted = re.sub(r'\*\*(.*?)\*\*', r'<b>\1</b>', escaped)
+    return formatted
+
+async def telegram_webhook(data: WebhookData):
+    try:
+        if not data.message or not data.message.text:
+            return {"status": "ok"}
+
+        telegram_id = data.message.chat.id
+        user_text = data.message.text
+        username = data.message.chat.username
+        first_name = data.message.chat.first_name
+
+        if db_manager:
+            await db_manager.create_or_update_user(telegram_id, username, first_name, data.message.chat.last_name)
+
+        if TELEGRAM_TOKEN:
+            async with httpx.AsyncClient(timeout=40.0, verify=False) as client:
+                action_payload = {
+                    "chat_id": telegram_id,
+                    "action": "typing",
+                }
+                action_url = f"https://{TELEGRAM_IP}/bot{TELEGRAM_TOKEN}/sendChatAction"
+                headers = {
+                    "Host": "api.telegram.org",
+                    "Content-Type": "application/json; charset=utf-8",
+                }
+                try:
+                    await client.post(action_url, json=action_payload, headers=headers)
+                except Exception:
+                    pass
+
+                ai_answer = await get_ai_response(user_text, telegram_id)
+                final_response = ai_answer or "Sorry, I couldn't generate a response."
+
+                prepared_text = _sanitize_telegram_text(final_response)
+                formatted_text = _format_telegram_message(prepared_text)
+                final_text = formatted_text[:MAX_TELEGRAM_MESSAGE_LENGTH]
+                
+                payload = {
+                    "chat_id": telegram_id,
+                    "text": final_text if final_text.strip() else ".",
+                    "parse_mode": "HTML",
+                }
+
+                forced_ip_url = f"https://{TELEGRAM_IP}/bot{TELEGRAM_TOKEN}/sendMessage"
+                headers = {
+                    "Host": "api.telegram.org",
+                    "Content-Type": "application/json; charset=utf-8",
+                }
+
+                response = await client.post(forced_ip_url, json=payload, headers=headers)
+
+                if response.status_code == 200:
+                    print("--- Success: Telegram message delivered ---")
+                    if db_manager:
+                        await asyncio.gather(
+                            db_manager.save_message(telegram_id, user_text, "user"),
+                            db_manager.save_message(telegram_id, final_response, "assistant")
+                        )
+                else:
+                    print(f"--- Telegram Rejected: {response.status_code} - {response.text} ---")
+        
+        return {"status": "ok"}
+    except Exception as e:
+        print(f"Error in webhook: {str(e)}")
+        return {"status": "error", "message": str(e)}

tests/test_telegram_formatting.py

@@ -0,0 +1,53 @@
+import unittest
+import html
+import re
+from telegram_handlers import _format_telegram_message
+
+
+class TestTelegramFormatting(unittest.TestCase):
+
+    def test_bold_conversion(self):
+        """Test that **text** is converted to <b>text</b>."""
+        input_text = "This is **bold** text."
+        expected = "This is <b>bold</b> text."
+        result = _format_telegram_message(input_text)
+        self.assertEqual(result, expected)
+
+    def test_multiple_bold(self):
+        """Test multiple bold sections."""
+        input_text = "**First** and **second** bold."
+        expected = "<b>First</b> and <b>second</b> bold."
+        result = _format_telegram_message(input_text)
+        self.assertEqual(result, expected)
+
+    def test_html_escaping(self):
+        """Test that HTML characters are escaped."""
+        input_text = "Text with <script> and **bold**."
+        expected = "Text with &lt;script&gt; and <b>bold</b>."
+        result = _format_telegram_message(input_text)
+        self.assertEqual(result, expected)
+
+    def test_no_bold(self):
+        """Test text without bold markers."""
+        input_text = "Plain text without formatting."
+        expected = "Plain text without formatting."
+        result = _format_telegram_message(input_text)
+        self.assertEqual(result, expected)
+
+    def test_empty_string(self):
+        """Test empty string."""
+        input_text = ""
+        expected = ""
+        result = _format_telegram_message(input_text)
+        self.assertEqual(result, expected)
+
+    def test_nested_asterisks(self):
+        """Test that single * are not converted."""
+        input_text = "Italic *text* but **bold**."
+        expected = "Italic *text* but <b>bold</b>."
+        result = _format_telegram_message(input_text)
+        self.assertEqual(result, expected)
+
+
+if __name__ == '__main__':
+    unittest.main()

transfers/__init__.py

@@ -0,0 +1,9 @@
+from .service import (
+    cancel_transfer,
+    confirm_transfer,
+    get_account_balance,
+    get_pending_transfer,
+    get_sender_account,
+    prepare_transfer,
+)
+

transfers/mock_bank_accounts.json

@@ -0,0 +1,91 @@
+{
+  "accounts": [
+    {
+      "telegram_id": 12,
+      "account_id": "ACC-00012",
+      "serial_id": "1001",
+      "name": "Test Sender",
+      "balance": 1900.0,
+      "currency": "YER",
+      "is_real": true
+    },
+    {
+      "telegram_id": 991001,
+      "account_id": "ACC-991001",
+      "serial_id": "2001",
+      "name": "Ahmed Salem",
+      "balance": 1990.0,
+      "currency": "YER",
+      "is_real": true
+    },
+    {
+      "telegram_id": 991002,
+      "account_id": "ACC-991002",
+      "serial_id": "2002",
+      "name": "Mona Ali",
+      "balance": 800.0,
+      "currency": "YER",
+      "is_real": true
+    },
+    {
+      "telegram_id": 991003,
+      "account_id": "ACC-991003",
+      "serial_id": "2003",
+      "name": "Khaled Omar",
+      "balance": 300.0,
+      "currency": "YER",
+      "is_real": true
+    },
+    {
+      "telegram_id": 34,
+      "account_id": "ACC-34",
+      "serial_id": "9034",
+      "name": "Test User Ali",
+      "balance": 1810.0,
+      "currency": "YER",
+      "is_real": false
+    }
+  ],
+  "transactions": [
+    {
+      "transaction_id": "TX-20260410193334671954",
+      "telegram_id": 12,
+      "sender_serial_id": "1001",
+      "receiver_serial_id": "2001",
+      "receiver_name": "Ahmed Salem",
+      "amount": 200.0,
+      "currency": "YER",
+      "created_at": "2026-04-10T19:33:34.671995"
+    },
+    {
+      "transaction_id": "TX-20260410193344247493",
+      "telegram_id": 12,
+      "sender_serial_id": "1001",
+      "receiver_serial_id": "2001",
+      "receiver_name": "Ahmed Salem",
+      "amount": 200.0,
+      "currency": "YER",
+      "created_at": "2026-04-10T19:33:44.247529"
+    },
+    {
+      "transaction_id": "TX-20260410193354495302",
+      "telegram_id": 12,
+      "sender_serial_id": "1001",
+      "receiver_serial_id": "2001",
+      "receiver_name": "Ahmed Salem",
+      "amount": 200.0,
+      "currency": "YER",
+      "created_at": "2026-04-10T19:33:54.495350"
+    },
+    {
+      "transaction_id": "TX-20260420140247475948",
+      "telegram_id": 34,
+      "sender_serial_id": "9034",
+      "receiver_serial_id": "2001",
+      "receiver_name": "Ahmed Salem",
+      "amount": 190.0,
+      "currency": "YER",
+      "created_at": "2026-04-20T14:02:47.475998"
+    }
+  ]
+}

transfers/pending_transfers.json

@@ -0,0 +1,13 @@
+{
+  "12": {
+    "telegram_id": 12,
+    "sender_name": "Test Sender",
+    "sender_serial_id": "1001",
+    "receiver_name": "Ahmed Salem",
+    "receiver_serial_id": "2001",
+    "receiver_account_id": "ACC-991001",
+    "amount": 200,
+    "currency": "YER",
+    "created_at": "2026-04-11T05:28:56.362317"
+  }
+}

transfers/service.py

@@ -0,0 +1,303 @@
+import json
+from datetime import datetime
+from pathlib import Path
+from threading import Lock
+from typing import Any, Dict, Optional
+
+
+DATA_FILE = Path(__file__).with_name("mock_bank_accounts.json")
+PENDING_FILE = Path(__file__).with_name("pending_transfers.json")
+_LOCK = Lock()
+
+
+def _read_json(path: Path, default: Any) -> Any:
+    if not path.exists():
+        return default
+
+    with path.open("r", encoding="utf-8") as file:
+        return json.load(file)
+
+
+def _write_json(path: Path, data: Any) -> None:
+    with path.open("w", encoding="utf-8") as file:
+        json.dump(data, file, ensure_ascii=True, indent=2)
+
+
+def _load_bank_data() -> Dict[str, Any]:
+    return _read_json(DATA_FILE, {"accounts": [], "transactions": []})
+
+
+def _save_bank_data(data: Dict[str, Any]) -> None:
+    _write_json(DATA_FILE, data)
+
+
+def _load_pending_transfers() -> Dict[str, Any]:
+    return _read_json(PENDING_FILE, {})
+
+
+def _save_pending_transfers(data: Dict[str, Any]) -> None:
+    _write_json(PENDING_FILE, data)
+
+
+def _find_account_by_telegram_id(accounts: list[Dict[str, Any]], telegram_id: int) -> Optional[Dict[str, Any]]:
+    return next((account for account in accounts if account.get("telegram_id") == telegram_id), None)
+
+
+def _find_account_by_serial_id(accounts: list[Dict[str, Any]], serial_id: str) -> Optional[Dict[str, Any]]:
+    normalized_id = str(serial_id).strip()
+    return next((account for account in accounts if str(account.get("serial_id")) == normalized_id), None)
+
+
+def get_sender_account(telegram_id: int, user_name: str = None) -> Dict[str, Any]:
+    with _LOCK:
+        data = _load_bank_data()
+        sender = _find_account_by_telegram_id(data["accounts"], telegram_id)
+        if sender:
+            is_real = sender.get("is_real", True)
+            is_illusion = not is_real or sender.get("name", "").startswith("Test User")
+            return {
+                "success": True,
+                "account": sender,
+                "is_real": is_real,
+                "is_illusion": is_illusion,
+            }
+
+        # Ask for user name before creating illusion account
+        if not user_name:
+            return {
+                "success": False,
+                "need_user_name": True,
+                "message": "I need to create an illusion account for testing purposes. What is your name?",
+            }
+
+        # Create illusion account for testing purposes
+        new_account = {
+            "telegram_id": telegram_id,
+            "account_id": f"ACC-{telegram_id}",
+            "serial_id": str(9000 + telegram_id % 1000),
+            "name": f"Test User {user_name}",
+            "balance": 2000.0,
+            "currency": "YER",
+            "is_real": False
+        }
+        
+        data["accounts"].append(new_account)
+        _save_bank_data(data)
+        
+        return {
+            "success": True,
+            "account": new_account,
+            "is_real": False,
+            "is_illusion": True,
+            "message": f"I've created an illusion account for {user_name} with 2000 YER balance to try sending money. This is for testing purposes only."
+        }
+
+
+def get_account_balance(telegram_id: int) -> Dict[str, Any]:
+    sender_result = get_sender_account(telegram_id)
+    if not sender_result["success"]:
+        return sender_result
+
+    account = sender_result["account"]
+    result = {
+        "success": True,
+        "telegram_id": telegram_id,
+        "account_id": account["account_id"],
+        "serial_id": account["serial_id"],
+        "name": account["name"],
+        "balance": account.get("balance", 0.0),
+        "currency": account.get("currency", "YER"),
+        "is_real": sender_result.get("is_real", True),
+        "is_illusion": sender_result.get("is_illusion", False),
+    }
+    
+    if result["is_illusion"]:
+        result["message"] = "This is an illusion account created for testing purposes with 2000 YER balance."
+    
+    return result
+
+
+def get_receiver_account_name(receiver_serial_id: str) -> Dict[str, Any]:
+    with _LOCK:
+        data = _load_bank_data()
+        receiver = _find_account_by_serial_id(data["accounts"], receiver_serial_id)
+
+        if not receiver:
+            return {
+                "success": False,
+                "message": f"No account was found for ID {receiver_serial_id}.",
+            }
+
+        return {
+            "success": True,
+            "serial_id": receiver["serial_id"],
+            "name": receiver["name"],
+            "account_id": receiver["account_id"],
+            "currency": receiver.get("currency", "YER"),
+        }
+
+
+def prepare_transfer(telegram_id: int, receiver_serial_id: str, amount: Optional[float] = None) -> Dict[str, Any]:
+    with _LOCK:
+        data = _load_bank_data()
+        sender = _find_account_by_telegram_id(data["accounts"], telegram_id)
+        if not sender:
+            return {
+                "success": False,
+                "message": "You do not have an account in the bank system. Please visit the bank to create an account first.",
+            }
+
+        receiver = _find_account_by_serial_id(data["accounts"], receiver_serial_id)
+        if not receiver:
+            return {
+                "success": False,
+                "message": f"No account was found for ID {receiver_serial_id}.",
+            }
+
+        if receiver["serial_id"] == sender["serial_id"]:
+            return {
+                "success": False,
+                "message": "You cannot transfer money to the same account.",
+            }
+
+        pending_transfers = _load_pending_transfers()
+        pending_payload = {
+            "telegram_id": telegram_id,
+            "sender_name": sender["name"],
+            "sender_serial_id": sender["serial_id"],
+            "receiver_name": receiver["name"],
+            "receiver_serial_id": receiver["serial_id"],
+            "receiver_account_id": receiver["account_id"],
+            "amount": amount,
+            "currency": sender.get("currency", "YER"),
+            "created_at": datetime.utcnow().isoformat(),
+        }
+        pending_transfers[str(telegram_id)] = pending_payload
+        _save_pending_transfers(pending_transfers)
+
+        return {
+            "success": True,
+            "pending_transfer": pending_payload,
+            "message": f"Receiver found: {receiver['name']}. Waiting for user confirmation.",
+        }
+
+
+def get_pending_transfer(telegram_id: int) -> Dict[str, Any]:
+    with _LOCK:
+        pending_transfers = _load_pending_transfers()
+        pending_transfer = pending_transfers.get(str(telegram_id))
+
+        if not pending_transfer:
+            return {
+                "success": False,
+                "message": "No pending transfer was found for this Telegram user.",
+            }
+
+        return {
+            "success": True,
+            "pending_transfer": pending_transfer,
+        }
+
+
+def confirm_transfer(telegram_id: int) -> Dict[str, Any]:
+    with _LOCK:
+        data = _load_bank_data()
+        pending_transfers = _load_pending_transfers()
+        pending_transfer = pending_transfers.get(str(telegram_id))
+
+        if not pending_transfer:
+            return {
+                "success": False,
+                "message": "There is no pending transfer to confirm.",
+            }
+
+        sender = _find_account_by_telegram_id(data["accounts"], telegram_id)
+        receiver = _find_account_by_serial_id(data["accounts"], pending_transfer["receiver_serial_id"])
+
+        if not sender or not receiver:
+            return {
+                "success": False,
+                "message": "The sender or receiver account could not be found.",
+            }
+
+        amount = pending_transfer.get("amount")
+        if amount is None:
+            return {
+                "success": False,
+                "message": "The transfer amount is missing. Ask the user for the amount before confirming.",
+            }
+
+        try:
+            amount_value = float(amount)
+        except (TypeError, ValueError):
+            return {
+                "success": False,
+                "message": "The transfer amount is invalid.",
+            }
+
+        if amount_value <= 0:
+            return {
+                "success": False,
+                "message": "The transfer amount must be greater than zero.",
+            }
+
+        if float(sender.get("balance", 0.0)) < amount_value:
+            return {
+                "success": False,
+                "message": f"Insufficient balance. Available balance is {sender.get('balance', 0.0):.2f} {sender.get('currency', 'YER')}.",
+            }
+
+        sender["balance"] = round(float(sender["balance"]) - amount_value, 2)
+        receiver["balance"] = round(float(receiver.get("balance", 0.0)) + amount_value, 2)
+
+        transaction = {
+            "transaction_id": f"TX-{datetime.utcnow().strftime('%Y%m%d%H%M%S%f')}",
+            "telegram_id": telegram_id,
+            "sender_serial_id": sender["serial_id"],
+            "receiver_serial_id": receiver["serial_id"],
+            "receiver_name": receiver["name"],
+            "amount": amount_value,
+            "currency": sender.get("currency", "YER"),
+            "created_at": datetime.utcnow().isoformat(),
+        }
+        data.setdefault("transactions", []).append(transaction)
+        _save_bank_data(data)
+
+        pending_transfers.pop(str(telegram_id), None)
+        _save_pending_transfers(pending_transfers)
+
+        # Check if sender is an illusion account using is_real variable
+        is_real = sender.get("is_real", True)
+        is_illusion = not is_real
+        
+        message = f"Transfer completed successfully to {receiver['name']}."
+        if is_illusion:
+            message += " ⚠️ Please note: This process was not real - it was for testing purposes only. No actual money was transferred."
+        
+        return {
+            "success": True,
+            "transaction": transaction,
+            "sender_balance": sender["balance"],
+            "is_real": is_real,
+            "is_illusion": is_illusion,
+            "message": message,
+        }
+
+
+def cancel_transfer(telegram_id: int) -> Dict[str, Any]:
+    with _LOCK:
+        pending_transfers = _load_pending_transfers()
+        removed = pending_transfers.pop(str(telegram_id), None)
+        _save_pending_transfers(pending_transfers)
+
+        if not removed:
+            return {
+                "success": False,
+                "message": "There is no pending transfer to cancel.",
+            }
+
+        return {
+            "success": True,
+            "message": "The pending transfer has been canceled.",
+        }
+

utils.py

@@ -0,0 +1,20 @@
+import socket
+
+async def dns_test():
+    try:
+        ip = socket.gethostbyname("api.telegram.org")
+        return {"resolved_ip": ip}
+    except Exception as e:
+        return {"error": str(e)}
+
+async def test_ai_response(message: str, telegram_id: int):
+    try:
+        from ai_service import get_ai_response
+        response = await get_ai_response(message, telegram_id)
+        return {
+            "message": message,
+            "telegram_id": telegram_id,
+            "response": response,
+        }
+    except Exception as e:
+        return {"error": str(e)}