Project: Deploy AI API on AWS

import json
import time
from datetime import datetime
from dataclasses import dataclass, field
from typing import List, Dict, Any

# ─── Models ────────────────────────────────────────────────
@dataclass
class ChatMessage:
    role:    str
    content: str
    ts:      str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))

@dataclass
class ChatSession:
    session_id: str
    messages:   List[ChatMessage] = field(default_factory=list)
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())

    def add(self, role: str, content: str):
        self.messages.append(ChatMessage(role, content))

    def to_history(self) -> List[dict]:
        return [{"role": m.role, "content": m.content, "ts": m.ts}
                for m in self.messages]

# ─── Services (محاكاة) ─────────────────────────────────────
class S3SessionStore:
    """تخزين الجلسات في S3"""
    def __init__(self):
        self._store: Dict[str, ChatSession] = {}

    def save(self, s: ChatSession):
        self._store[s.session_id] = s
        print(f"  💾 S3: saved '{s.session_id}'")

    def load(self, sid: str):
        return self._store.get(sid)

    def list_sessions(self) -> List[str]:
        return list(self._store.keys())

    def total_messages(self) -> int:
        return sum(len(s.messages) for s in self._store.values())

class BedrockClaude:
    """Claude 3 Haiku عبر AWS Bedrock"""
    def __init__(self):
        self._calls   = 0
        self._in_tok  = 0
        self._out_tok = 0

    def invoke(self, messages: List[dict], system: str = "") -> str:
        last = messages[-1]["content"] if messages else ""
        self._calls   += 1
        self._in_tok  += int(len(last.split()) * 1.3) + len(messages) * 5
        self._out_tok += 60
        return f"[Claude/Bedrock] ردّ على: {last[:55]}..."

    def cost(self) -> float:
        return round(
            (self._in_tok  / 1000) * 0.00025 +
            (self._out_tok / 1000) * 0.00125, 6
        )

# ─── Shared State (يتشارك فيها عبر Warm Containers)
_s3     = S3SessionStore()
_claude = BedrockClaude()

# ─── Helpers ───────────────────────────────────────────────
def _ok(data: dict)        -> dict: return _wrap(200, data)
def _err(code: int, msg)   -> dict: return _wrap(code, {"error": msg})
def _wrap(code: int, data) -> dict:
    return {
        "statusCode": code,
        "headers": {
            "Content-Type":                "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        "body": json.dumps(data, ensure_ascii=False),
    }

# ─── Lambda Handler ────────────────────────────────────────
def lambda_handler(event: dict, context: Any) -> dict:
    try:
        raw  = event.get("body", "{}")
        body = json.loads(raw) if isinstance(raw, str) else raw
    except json.JSONDecodeError:
        return _err(400, "JSON غير صالح")

    action     = body.get("action", "chat")
    session_id = body.get("session_id", "default")

    if action == "chat":
        msg = body.get("message", "").strip()
        if not msg:
            return _err(400, "حقل 'message' مطلوب")

        session = _s3.load(session_id) or ChatSession(session_id)
        session.add("user", msg)

        ai_reply = _claude.invoke(
            session.to_history(),
            system="أنت مساعد AWS خبير. أجب بالعربية بإيجاز.",
        )
        session.add("assistant", ai_reply)
        _s3.save(session)

        return _ok({
            "reply":      ai_reply,
            "session_id": session_id,
            "turn":       len(session.messages) // 2,
        })

    elif action == "history":
        s = _s3.load(session_id)
        if not s:
            return _err(404, f"الجلسة '{session_id}' غير موجودة")
        return _ok({"session_id": session_id, "messages": s.to_history()})

    elif action == "stats":
        cost_str = "$" + f"{_claude.cost():.6f}"
        return _ok({
            "sessions":         len(_s3.list_sessions()),
            "total_messages":   _s3.total_messages(),
            "bedrock_calls":    _claude._calls,
            "estimated_cost":   cost_str,
        })

    return _err(400, f"action '{action}' غير معروف")

# ─── اختبار المشروع ────────────────────────────────────────
class Ctx:
    aws_request_id = "proj-test-001"

ctx = Ctx()
print("🚀 اختبار AI Chatbot — Lambda + Bedrock + S3")
print("=" * 58)

turns = [
    ("alice", "ما هو الفرق بين Lambda وEC2؟"),
    ("alice", "متى أستخدم Lambda؟"),
    ("bob",   "كيف أتصل بـ Bedrock من Lambda؟"),
    ("alice", "شكراً على التوضيح!"),
]

for sid, msg in turns:
    ev  = {"body": json.dumps({"action": "chat", "session_id": sid, "message": msg})}
    res = json.loads(lambda_handler(ev, ctx)["body"])
    print(f"\n[{sid}] 💬 {msg}")
    print(f"[{sid}] 🤖 {res['reply'][:70]}...")
    print(f"[{sid}] 🔄 Turn #{res['turn']}")

# تاريخ جلسة alice
print("\n\n📜 تاريخ محادثة alice:")
hist_ev = {"body": json.dumps({"action": "history", "session_id": "alice"})}
hist    = json.loads(lambda_handler(hist_ev, ctx)["body"])
for m in hist["messages"]:
    icon = "👤" if m["role"] == "user" else "🤖"
    print(f"  {icon} [{m['ts']}] {m['content'][:60]}...")

# إحصائيات
print("\n\n📊 إحصائيات المشروع:")
stats_ev = {"body": json.dumps({"action": "stats"})}
stats    = json.loads(lambda_handler(stats_ev, ctx)["body"])
print(f"  الجلسات          : {stats['sessions']}")
print(f"  إجمالي الرسائل   : {stats['total_messages']}")
print(f"  مكالمات Bedrock  : {stats['bedrock_calls']}")
print(f"  التكلفة التقديرية: {stats['estimated_cost']}")
print("\n✅ المشروع جاهز للنشر على AWS Lambda!")