mirror of
https://github.com/x1xhlol/system-prompts-and-models-of-ai-tools.git
synced 2026-06-18 15:29:36 +00:00
Full-stack AI-powered sales automation platform for Saudi SMEs: Backend (FastAPI + PostgreSQL): - Multi-tenant architecture with row-level isolation - JWT auth with RBAC (owner/manager/agent/admin) - Lead, Customer, Deal, Pipeline, Activity, Message, Proposal models - Dashboard analytics API (overview, pipeline, revenue) - WhatsApp Business API, Email (SMTP/SendGrid), SMS (Unifonic) integrations - Celery + Redis workers for automated follow-ups and scheduled messages - Property model for Real Estate module (Riyadh districts) - Hijri date utilities, Arabic/English localization Frontend (Next.js + Tailwind): - Professional Arabic RTL landing page with 10 sections - Brand identity: SalesMatic (سيلزماتك) with custom SVG logo - Color system: Trust Blue #0F4C81, Growth Teal #00BFA6, CTA Orange #FF6B35 - IBM Plex Sans Arabic + Inter typography - Responsive design, dark hero section, pricing table, FAQ Industry Templates: - Healthcare/Clinics: pipeline stages, WhatsApp message templates, auto-workflows - Real Estate Riyadh: 20 districts, property tours, payment plans, matching Infrastructure: - Docker Compose (PostgreSQL, Redis, Backend, Celery, Frontend, Nginx) - Nginx reverse proxy config - Makefile for common operations https://claude.ai/code/session_01LLR7jzpyNRwDA9kojtT3CW
51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from uuid import UUID
|
|
from app.database import get_db
|
|
from app.utils.security import decode_token
|
|
from app.models.user import User
|
|
from app.models.tenant import Tenant
|
|
|
|
security = HTTPBearer()
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> User:
|
|
payload = decode_token(credentials.credentials)
|
|
if not payload or payload.get("type") != "access":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
|
|
|
|
user_id = payload.get("sub")
|
|
if not user_id:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload")
|
|
|
|
result = await db.execute(select(User).where(User.id == UUID(user_id), User.is_active == True))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive")
|
|
|
|
return user
|
|
|
|
|
|
async def get_current_tenant(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> Tenant:
|
|
result = await db.execute(select(Tenant).where(Tenant.id == current_user.tenant_id, Tenant.is_active == True))
|
|
tenant = result.scalar_one_or_none()
|
|
if not tenant:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Tenant not found or inactive")
|
|
return tenant
|
|
|
|
|
|
def require_role(*roles: str):
|
|
async def role_checker(current_user: User = Depends(get_current_user)):
|
|
if current_user.role not in roles:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions")
|
|
return current_user
|
|
return role_checker
|