from fastapi import FastAPI, APIRouter, HTTPException, Depends, status, Request, Query, UploadFile, File
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import RedirectResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from dotenv import load_dotenv
from starlette.middleware.cors import CORSMiddleware
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import text
import os
import logging
from pathlib import Path
from pydantic import BaseModel, Field, EmailStr, ConfigDict
from typing import List, Optional, Dict
import uuid
from datetime import datetime, timezone, timedelta
import jwt
import bcrypt
import asyncio
import httpx
import shutil
import json
# Twilio SMS
try:
from twilio.rest import Client as TwilioClient
TWILIO_AVAILABLE = True
except ImportError:
TWILIO_AVAILABLE = False
TwilioClient = None
# Resend Email
try:
import resend
RESEND_AVAILABLE = True
except ImportError:
RESEND_AVAILABLE = False
resend = None
ROOT_DIR = Path(__file__).parent
UPLOAD_DIR = ROOT_DIR / "uploads" / "photos"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
load_dotenv(ROOT_DIR / '.env')
# PostgreSQL connection
DATABASE_URL = os.environ.get('DATABASE_URL', 'postgresql+asyncpg://postgres:postgres@localhost:5432/opora_vie')
engine = create_async_engine(DATABASE_URL, echo=False)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# JWT Config
JWT_SECRET = os.environ.get('JWT_SECRET', 'opora-vie-secret-key-2024')
JWT_ALGORITHM = "HS256"
JWT_EXPIRATION_HOURS = 24
# Integration Config
RESEND_API_KEY = os.environ.get('RESEND_API_KEY', '')
SENDER_EMAIL = os.environ.get('SENDER_EMAIL', 'noreply@adora.net')
GOOGLE_CLIENT_ID = os.environ.get('GOOGLE_CLIENT_ID', '')
GOOGLE_CLIENT_SECRET = os.environ.get('GOOGLE_CLIENT_SECRET', '')
DAILY_API_KEY = os.environ.get('DAILY_API_KEY', '')
DAILY_DOMAIN = os.environ.get('DAILY_DOMAIN', '')
FRONTEND_URL = os.environ.get('FRONTEND_URL', 'https://apora.net')
STRIPE_API_KEY = os.environ.get('STRIPE_API_KEY', '')
STRIPE_PUBLIC_KEY = os.environ.get('STRIPE_PUBLIC_KEY', '')
STRIPE_PLATFORM_FEE = int(os.environ.get('STRIPE_PLATFORM_FEE', '300')) # 3€ in cents
STRIPE_SUBSCRIPTION_PRICE = 50.00 # 50€/month subscription
# Initialize Stripe
import stripe
stripe.api_key = STRIPE_API_KEY
# Twilio Config
TWILIO_ACCOUNT_SID = os.environ.get('TWILIO_ACCOUNT_SID', '')
TWILIO_AUTH_TOKEN = os.environ.get('TWILIO_AUTH_TOKEN', '')
TWILIO_PHONE_NUMBER = os.environ.get('TWILIO_PHONE_NUMBER', '')
# Create the main app
app = FastAPI(title="Opora Vie API")
api_router = APIRouter()
security = HTTPBearer()
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# ============== DATABASE INITIALIZATION ==============
async def init_db():
"""Initialize PostgreSQL tables"""
async with engine.begin() as conn:
# Users table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS users (
id VARCHAR(36) PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL,
first_name VARCHAR(255) NOT NULL,
last_name VARCHAR(255) NOT NULL,
phone VARCHAR(50),
password VARCHAR(255) NOT NULL,
role VARCHAR(50) NOT NULL DEFAULT 'client',
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
google_calendar_connected BOOLEAN DEFAULT FALSE,
google_tokens JSONB
)
"""))
# Practitioners table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS practitioners (
id VARCHAR(36) PRIMARY KEY,
user_id VARCHAR(36) REFERENCES users(id) ON DELETE CASCADE,
specialty VARCHAR(100) NOT NULL,
bio TEXT,
photo_url VARCHAR(500),
country VARCHAR(10) DEFAULT 'FR',
region VARCHAR(100),
department VARCHAR(100),
address VARCHAR(255),
city VARCHAR(100),
postal_code VARCHAR(20),
consultation_types JSONB DEFAULT '["cabinet"]',
price DECIMAL(10, 2),
consultation_duration INTEGER DEFAULT 60,
is_verified BOOLEAN DEFAULT FALSE,
years_experience INTEGER,
diplomas JSONB,
detailed_description TEXT,
languages JSONB,
stripe_customer_id VARCHAR(255),
stripe_subscription_id VARCHAR(255),
subscription_status VARCHAR(50) DEFAULT 'none',
subscription_enabled BOOLEAN DEFAULT FALSE
)
"""))
# Add subscription columns if they don't exist (for existing databases)
try:
await conn.execute(text("ALTER TABLE practitioners ADD COLUMN IF NOT EXISTS stripe_customer_id VARCHAR(255)"))
await conn.execute(text("ALTER TABLE practitioners ADD COLUMN IF NOT EXISTS stripe_subscription_id VARCHAR(255)"))
await conn.execute(text("ALTER TABLE practitioners ADD COLUMN IF NOT EXISTS subscription_status VARCHAR(50) DEFAULT 'none'"))
await conn.execute(text("ALTER TABLE practitioners ADD COLUMN IF NOT EXISTS subscription_enabled BOOLEAN DEFAULT FALSE"))
except Exception as e:
logger.info(f"Subscription columns might already exist: {e}")
# Time slots table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS time_slots (
id VARCHAR(36) PRIMARY KEY,
practitioner_id VARCHAR(36) REFERENCES practitioners(id) ON DELETE CASCADE,
day_of_week INTEGER NOT NULL,
start_time VARCHAR(10) NOT NULL,
end_time VARCHAR(10) NOT NULL
)
"""))
# Practitioner access codes table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS practitioner_access_codes (
id VARCHAR(36) PRIMARY KEY,
email VARCHAR(255) NOT NULL,
first_name VARCHAR(100),
last_name VARCHAR(100),
code VARCHAR(20) NOT NULL,
used BOOLEAN DEFAULT FALSE,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
expires_at TIMESTAMP WITH TIME ZONE
)
"""))
# Appointments table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS appointments (
id VARCHAR(36) PRIMARY KEY,
client_id VARCHAR(36) REFERENCES users(id) ON DELETE CASCADE,
practitioner_id VARCHAR(36) REFERENCES practitioners(id) ON DELETE CASCADE,
date VARCHAR(20) NOT NULL,
start_time VARCHAR(10) NOT NULL,
end_time VARCHAR(10) NOT NULL,
consultation_type VARCHAR(50) NOT NULL,
status VARCHAR(50) DEFAULT 'pending',
notes TEXT,
video_room_url VARCHAR(500),
google_event_id VARCHAR(255),
paid BOOLEAN DEFAULT FALSE,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
)
"""))
# Payment transactions table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS payment_transactions (
id VARCHAR(36) PRIMARY KEY,
session_id VARCHAR(255),
appointment_id VARCHAR(36) REFERENCES appointments(id) ON DELETE SET NULL,
client_id VARCHAR(36) REFERENCES users(id) ON DELETE SET NULL,
amount DECIMAL(10, 2),
currency VARCHAR(10) DEFAULT 'eur',
status VARCHAR(50) DEFAULT 'pending',
payment_status VARCHAR(50) DEFAULT 'initiated',
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
)
"""))
# Create indexes
await conn.execute(text("CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)"))
await conn.execute(text("CREATE INDEX IF NOT EXISTS idx_practitioners_user_id ON practitioners(user_id)"))
await conn.execute(text("CREATE INDEX IF NOT EXISTS idx_practitioners_specialty ON practitioners(specialty)"))
await conn.execute(text("CREATE INDEX IF NOT EXISTS idx_appointments_client_id ON appointments(client_id)"))
await conn.execute(text("CREATE INDEX IF NOT EXISTS idx_appointments_practitioner_id ON appointments(practitioner_id)"))
await conn.execute(text("CREATE INDEX IF NOT EXISTS idx_appointments_date ON appointments(date)"))
# Admin settings table (for SMS toggle, etc.)
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS admin_settings (
key VARCHAR(100) PRIMARY KEY,
value TEXT,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
)
"""))
# SMS reminders log table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS sms_reminders (
id VARCHAR(36) PRIMARY KEY,
appointment_id VARCHAR(36) REFERENCES appointments(id) ON DELETE CASCADE,
phone_number VARCHAR(50),
message TEXT,
status VARCHAR(50) DEFAULT 'pending',
sent_at TIMESTAMP WITH TIME ZONE,
error_message TEXT,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
)
"""))
# Blog articles table
await conn.execute(text("""
CREATE TABLE IF NOT EXISTS blog_articles (
id VARCHAR(36) PRIMARY KEY,
slug VARCHAR(255) UNIQUE NOT NULL,
category VARCHAR(100) NOT NULL,
author VARCHAR(255),
read_time VARCHAR(20),
image_url VARCHAR(500),
title_fr TEXT NOT NULL,
title_en TEXT,
title_uk TEXT,
title_ru TEXT,
title_it TEXT,
title_ro TEXT,
excerpt_fr TEXT,
excerpt_en TEXT,
excerpt_uk TEXT,
excerpt_ru TEXT,
excerpt_it TEXT,
excerpt_ro TEXT,
content_fr TEXT,
content_en TEXT,
content_uk TEXT,
content_ru TEXT,
content_it TEXT,
content_ro TEXT,
is_published BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
)
"""))
# Initialize default settings if not exist
await conn.execute(text("""
INSERT INTO admin_settings (key, value, updated_at)
VALUES ('sms_reminders_enabled', 'false', NOW())
ON CONFLICT (key) DO NOTHING
"""))
logger.info("Database tables initialized successfully")
# ============== MODELS ==============
class UserBase(BaseModel):
email: EmailStr
first_name: str
last_name: str
phone: Optional[str] = None
class UserCreate(UserBase):
password: str
role: str = "client"
class UserLogin(BaseModel):
email: EmailStr
password: str
class UserResponse(BaseModel):
model_config = ConfigDict(extra="ignore")
id: str
email: str
first_name: str
last_name: str
phone: Optional[str] = None
role: str
created_at: str
google_calendar_connected: Optional[bool] = False
is_admin: Optional[bool] = False
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
user: UserResponse
class PractitionerCreate(BaseModel):
specialty: str
bio: Optional[str] = None
photo_url: Optional[str] = None
country: str = "FR"
region: Optional[str] = None
department: Optional[str] = None
address: Optional[str] = None
city: Optional[str] = None
postal_code: Optional[str] = None
consultation_types: List[str] = ["cabinet"]
price: Optional[float] = None
consultation_duration: int = 60
years_experience: Optional[int] = None
diplomas: Optional[List[str]] = None
detailed_description: Optional[str] = None
languages: Optional[List[str]] = None
class PractitionerUpdate(BaseModel):
specialty: Optional[str] = None
bio: Optional[str] = None
photo_url: Optional[str] = None
country: Optional[str] = None
region: Optional[str] = None
department: Optional[str] = None
address: Optional[str] = None
city: Optional[str] = None
postal_code: Optional[str] = None
consultation_types: Optional[List[str]] = None
price: Optional[float] = None
consultation_duration: Optional[int] = None
years_experience: Optional[int] = None
diplomas: Optional[List[str]] = None
detailed_description: Optional[str] = None
languages: Optional[List[str]] = None
class PractitionerResponse(BaseModel):
model_config = ConfigDict(extra="ignore")
id: str
user_id: str
specialty: str
bio: Optional[str] = None
photo_url: Optional[str] = None
country: str = "FR"
region: Optional[str] = None
department: Optional[str] = None
address: Optional[str] = None
city: Optional[str] = None
postal_code: Optional[str] = None
consultation_types: List[str]
price: Optional[float] = None
consultation_duration: int
is_verified: bool = False
years_experience: Optional[int] = None
diplomas: Optional[List[str]] = None
detailed_description: Optional[str] = None
languages: Optional[List[str]] = None
user: Optional[UserResponse] = None
class TimeSlotCreate(BaseModel):
day_of_week: int
start_time: str
end_time: str
class TimeSlotResponse(BaseModel):
model_config = ConfigDict(extra="ignore")
id: str
practitioner_id: str
day_of_week: int
start_time: str
end_time: str
class AppointmentCreate(BaseModel):
practitioner_id: str
date: str
start_time: str
consultation_type: str
notes: Optional[str] = None
class AppointmentResponse(BaseModel):
model_config = ConfigDict(extra="ignore")
id: str
client_id: str
practitioner_id: str
date: str
start_time: str
end_time: str
consultation_type: str
status: str
notes: Optional[str] = None
created_at: str
video_room_url: Optional[str] = None
google_event_id: Optional[str] = None
client: Optional[UserResponse] = None
practitioner: Optional[PractitionerResponse] = None
# Blog Article Models
class BlogArticleCreate(BaseModel):
slug: str
category: str
author: Optional[str] = "Équipe Opora Vie"
read_time: Optional[str] = "5 min"
image_url: Optional[str] = None
title_fr: str
title_en: Optional[str] = None
title_uk: Optional[str] = None
title_ru: Optional[str] = None
title_it: Optional[str] = None
title_ro: Optional[str] = None
excerpt_fr: Optional[str] = None
excerpt_en: Optional[str] = None
excerpt_uk: Optional[str] = None
excerpt_ru: Optional[str] = None
excerpt_it: Optional[str] = None
excerpt_ro: Optional[str] = None
content_fr: Optional[str] = None
content_en: Optional[str] = None
content_uk: Optional[str] = None
content_ru: Optional[str] = None
content_it: Optional[str] = None
content_ro: Optional[str] = None
is_published: bool = True
class BlogArticleResponse(BaseModel):
model_config = ConfigDict(extra="ignore")
id: str
slug: str
category: str
author: Optional[str] = None
read_time: Optional[str] = None
image_url: Optional[str] = None
title_fr: str
title_en: Optional[str] = None
title_uk: Optional[str] = None
title_ru: Optional[str] = None
title_it: Optional[str] = None
title_ro: Optional[str] = None
excerpt_fr: Optional[str] = None
excerpt_en: Optional[str] = None
excerpt_uk: Optional[str] = None
excerpt_ru: Optional[str] = None
excerpt_it: Optional[str] = None
excerpt_ro: Optional[str] = None
content_fr: Optional[str] = None
content_en: Optional[str] = None
content_uk: Optional[str] = None
content_ru: Optional[str] = None
content_it: Optional[str] = None
content_ro: Optional[str] = None
is_published: bool = True
created_at: Optional[str] = None
updated_at: Optional[str] = None
# ============== DATABASE HELPERS ==============
def row_to_dict(row) -> dict:
"""Convert SQLAlchemy row to dictionary"""
if row is None:
return None
return dict(row._mapping)
def parse_json_field(value, default=None):
"""Parse JSON field from PostgreSQL"""
if value is None:
return default
if isinstance(value, (list, dict)):
return value
if isinstance(value, str):
try:
return json.loads(value)
except:
return default
return default
def format_practitioner_row(row_dict: dict) -> dict:
"""Format practitioner row with proper JSON fields"""
if row_dict is None:
return None
row_dict['consultation_types'] = parse_json_field(row_dict.get('consultation_types'), ['cabinet'])
row_dict['diplomas'] = parse_json_field(row_dict.get('diplomas'), [])
row_dict['languages'] = parse_json_field(row_dict.get('languages'), [])
row_dict['is_verified'] = row_dict.get('is_verified', False) or False
row_dict['consultation_duration'] = row_dict.get('consultation_duration', 60) or 60
row_dict['subscription_enabled'] = row_dict.get('subscription_enabled', False) or False
row_dict['subscription_status'] = row_dict.get('subscription_status', 'none') or 'none'
if row_dict.get('price'):
row_dict['price'] = float(row_dict['price'])
return row_dict
def format_user_row(row_dict: dict) -> dict:
"""Format user row"""
if row_dict is None:
return None
if row_dict.get('created_at'):
if hasattr(row_dict['created_at'], 'isoformat'):
row_dict['created_at'] = row_dict['created_at'].isoformat()
row_dict['google_calendar_connected'] = row_dict.get('google_calendar_connected', False) or False
return row_dict
# ============== EMAIL SERVICE (RESEND) ==============
async def send_email(to_email: str, subject: str, html_content: str) -> bool:
"""Send email using Resend API"""
if not RESEND_API_KEY:
logger.warning("RESEND_API_KEY not configured - email not sent")
return False
try:
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.resend.com/emails",
headers={
"Authorization": f"Bearer {RESEND_API_KEY}",
"Content-Type": "application/json"
},
json={
"from": SENDER_EMAIL,
"to": [to_email],
"subject": subject,
"html": html_content
},
timeout=10.0
)
if response.status_code == 200:
logger.info(f"Email sent to {to_email}")
return True
else:
logger.error(f"Email failed: {response.text}")
return False
except Exception as e:
logger.error(f"Email error: {str(e)}")
return False
def generate_appointment_email(appointment: dict, practitioner: dict, client: dict, is_reminder: bool = False) -> tuple:
"""Generate appointment confirmation or reminder email"""
date_formatted = appointment['date']
# Handle both dict and UserResponse object for practitioner['user']
pract_user = practitioner['user']
if hasattr(pract_user, 'first_name'):
pract_first = pract_user.first_name
pract_last = pract_user.last_name
else:
pract_first = pract_user['first_name']
pract_last = pract_user['last_name']
subject = f"{'Rappel: ' if is_reminder else ''}Votre RDV - {pract_first} {pract_last}"
video_section = ""
if appointment.get('consultation_type') == 'visio' and appointment.get('video_room_url'):
video_section = f"""
"""
html = f"""
Opora Vie
{'Rappel de votre rendez-vous' if is_reminder else 'Confirmation de rendez-vous'}
Praticien: {pract_first} {pract_last}
Spécialité: {practitioner['specialty'].capitalize()}
Date: {date_formatted}
Heure: {appointment['start_time']} - {appointment['end_time']}
Type: {'En cabinet' if appointment['consultation_type'] == 'cabinet' else 'En visio'}
{video_section}
{f"
Adresse: {practitioner.get('address', '')}, {practitioner.get('city', '')} {practitioner.get('postal_code', '')}
" if appointment['consultation_type'] == 'cabinet' and practitioner.get('address') else ''}
Pour annuler ou modifier votre rendez-vous, connectez-vous à votre espace client sur
Opora Vie.
"""
return subject, html
def generate_practitioner_notification_email(appointment: dict, practitioner: dict, client: dict) -> tuple:
"""Generate email notification for practitioner when a client books an appointment"""
date_formatted = appointment['date']
# Handle client info
if hasattr(client, 'first_name'):
client_first = client.first_name
client_last = client.last_name
client_email = client.email
client_phone = client.phone if hasattr(client, 'phone') else ''
else:
client_first = client.get('first_name', '')
client_last = client.get('last_name', '')
client_email = client.get('email', '')
client_phone = client.get('phone', '')
subject = f"Nouveau RDV - {client_first} {client_last} - {date_formatted}"
video_section = ""
if appointment.get('consultation_type') == 'visio' and appointment.get('video_room_url'):
video_section = f"""
"""
html = f"""
Opora Vie
Nouveau rendez-vous
Un nouveau client a pris rendez-vous avec vous.
Détails du rendez-vous
Date: {date_formatted}
Heure: {appointment['start_time']} - {appointment['end_time']}
Type: {'En cabinet' if appointment['consultation_type'] == 'cabinet' else 'En visio'}
{video_section}
{f"
Notes du client: {appointment.get('notes', '')}
" if appointment.get('notes') else ''}
Connectez-vous à votre espace praticien pour gérer vos rendez-vous.
"""
return subject, html
async def send_appointment_notification(appointment: dict, practitioner: dict, client: dict, is_reminder: bool = False):
"""Send appointment email notification to client AND practitioner"""
# Send to client
subject, html = generate_appointment_email(appointment, practitioner, client, is_reminder)
client_email = client.email if hasattr(client, 'email') else client['email']
await send_email(client_email, subject, html)
# Send to practitioner (only for new appointments, not reminders)
if not is_reminder:
pract_user = practitioner.get('user', {})
if pract_user:
pract_email = pract_user.email if hasattr(pract_user, 'email') else pract_user.get('email')
if pract_email:
pract_subject, pract_html = generate_practitioner_notification_email(appointment, practitioner, client)
await send_email(pract_email, pract_subject, pract_html)
logger.info(f"Practitioner notification sent to {pract_email}")
# ============== VIDEO CONFERENCE SERVICE (JITSI) ==============
async def create_video_room(appointment_id: str) -> Optional[str]:
"""Create a Jitsi video room for the appointment (free, no API key needed)"""
# Generate unique room name with prefix for security
jitsi_room = f"opora-vie-{appointment_id[:12]}"
jitsi_url = f"https://meet.jit.si/{jitsi_room}"
logger.info(f"Jitsi room created: {jitsi_url}")
return jitsi_url
# ============== GOOGLE CALENDAR SERVICE ==============
GOOGLE_SCOPES = ["https://www.googleapis.com/auth/calendar"]
GOOGLE_REDIRECT_URI = f"{os.environ.get('REACT_APP_BACKEND_URL', 'https://opora-vie.preview.emergentagent.com')}/api/oauth/calendar/callback"
@api_router.get("/oauth/calendar/login")
async def google_calendar_login():
"""Initiate Google Calendar OAuth flow"""
if not GOOGLE_CLIENT_ID or not GOOGLE_CLIENT_SECRET:
raise HTTPException(status_code=503, detail="Google Calendar non configuré")
auth_url = (
"https://accounts.google.com/o/oauth2/auth?"
f"client_id={GOOGLE_CLIENT_ID}&"
f"redirect_uri={GOOGLE_REDIRECT_URI}&"
"response_type=code&"
f"scope={'%20'.join(GOOGLE_SCOPES)}&"
"access_type=offline&"
"prompt=consent"
)
return {"authorization_url": auth_url}
@api_router.get("/oauth/calendar/callback")
async def google_calendar_callback(code: str, state: Optional[str] = None):
"""Handle Google Calendar OAuth callback"""
if not GOOGLE_CLIENT_ID or not GOOGLE_CLIENT_SECRET:
raise HTTPException(status_code=503, detail="Google Calendar non configuré")
try:
async with httpx.AsyncClient() as http_client:
token_response = await http_client.post(
"https://oauth2.googleapis.com/token",
data={
"code": code,
"client_id": GOOGLE_CLIENT_ID,
"client_secret": GOOGLE_CLIENT_SECRET,
"redirect_uri": GOOGLE_REDIRECT_URI,
"grant_type": "authorization_code"
},
timeout=10.0
)
if token_response.status_code != 200:
logger.error(f"Token exchange failed: {token_response.text}")
return RedirectResponse(f"{FRONTEND_URL}/practitioner/dashboard?calendar_error=true")
tokens = token_response.json()
user_response = await http_client.get(
"https://www.googleapis.com/oauth2/v2/userinfo",
headers={"Authorization": f"Bearer {tokens['access_token']}"},
timeout=10.0
)
if user_response.status_code == 200:
user_info = user_response.json()
google_email = user_info.get('email')
async with async_session() as session:
await session.execute(
text("""
UPDATE users
SET google_tokens = :tokens, google_calendar_connected = TRUE
WHERE email = :email
"""),
{"tokens": json.dumps(tokens), "email": google_email}
)
await session.commit()
logger.info(f"Google Calendar connected for {google_email}")
return RedirectResponse(f"{FRONTEND_URL}/practitioner/dashboard?calendar_connected=true")
except Exception as e:
logger.error(f"Google Calendar OAuth error: {str(e)}")
return RedirectResponse(f"{FRONTEND_URL}/practitioner/dashboard?calendar_error=true")
async def get_google_credentials(user_email: str) -> Optional[dict]:
"""Get and refresh Google credentials for a user"""
async with async_session() as session:
result = await session.execute(
text("SELECT google_tokens FROM users WHERE email = :email"),
{"email": user_email}
)
row = result.fetchone()
if not row or not row[0]:
return None
return parse_json_field(row[0])
async def add_to_google_calendar(user_email: str, appointment: dict, practitioner: dict, client: dict) -> Optional[str]:
"""Add appointment to practitioner's Google Calendar"""
tokens = await get_google_credentials(user_email)
if not tokens:
logger.info(f"No Google Calendar connected for {user_email}")
return None
try:
date_str = appointment['date']
start_time = appointment['start_time']
end_time = appointment['end_time']
start_datetime = f"{date_str}T{start_time}:00"
end_datetime = f"{date_str}T{end_time}:00"
event = {
"summary": f"RDV - {client['first_name']} {client['last_name']}",
"description": f"Consultation {appointment['consultation_type']}\nClient: {client['first_name']} {client['last_name']}\nEmail: {client['email']}\n{appointment.get('notes', '')}",
"start": {"dateTime": start_datetime, "timeZone": "Europe/Paris"},
"end": {"dateTime": end_datetime, "timeZone": "Europe/Paris"},
"reminders": {
"useDefault": False,
"overrides": [
{"method": "email", "minutes": 60},
{"method": "popup", "minutes": 30}
]
}
}
if appointment.get('video_room_url'):
event["description"] += f"\n\nLien visio: {appointment['video_room_url']}"
async with httpx.AsyncClient() as http_client:
response = await http_client.post(
"https://www.googleapis.com/calendar/v3/calendars/primary/events",
headers={
"Authorization": f"Bearer {tokens['access_token']}",
"Content-Type": "application/json"
},
json=event,
timeout=10.0
)
if response.status_code == 200:
event_data = response.json()
logger.info(f"Google Calendar event created: {event_data.get('id')}")
return event_data.get('id')
else:
logger.error(f"Google Calendar event creation failed: {response.text}")
return None
except Exception as e:
logger.error(f"Google Calendar error: {str(e)}")
return None
# ============== HELPERS ==============
def hash_password(password: str) -> str:
return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
def verify_password(password: str, hashed: str) -> bool:
return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8'))
def create_token(user_id: str, role: str) -> str:
payload = {
"sub": user_id,
"role": role,
"exp": datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRATION_HOURS)
}
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
try:
payload = jwt.decode(credentials.credentials, JWT_SECRET, algorithms=[JWT_ALGORITHM])
user_id = payload.get("sub")
if not user_id:
raise HTTPException(status_code=401, detail="Token invalide")
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM users WHERE id = :id"),
{"id": user_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=401, detail="Utilisateur non trouvé")
return format_user_row(row_to_dict(row))
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token expiré")
except jwt.InvalidTokenError:
raise HTTPException(status_code=401, detail="Token invalide")
ADMIN_EMAIL = os.environ.get('ADMIN_EMAIL', 'admin@opora-vie.fr')
def user_to_response(user: dict) -> UserResponse:
is_admin = user.get("email") == ADMIN_EMAIL or user.get("role") == "admin"
created_at = user.get("created_at", "")
if hasattr(created_at, 'isoformat'):
created_at = created_at.isoformat()
return UserResponse(
id=user["id"],
email=user["email"],
first_name=user["first_name"],
last_name=user["last_name"],
phone=user.get("phone"),
role=user["role"],
created_at=str(created_at),
google_calendar_connected=user.get("google_calendar_connected", False),
is_admin=is_admin
)
# ============== AUTH ROUTES ==============
PRACTITIONER_ACCESS_CODE = os.environ.get('PRACTITIONER_ACCESS_CODE', 'OporaVie2026!')
@api_router.post("/auth/verify-practitioner-access")
async def verify_practitioner_access(data: dict):
"""Verify the access code for practitioner registration"""
code = data.get("code", "").upper().strip()
email = data.get("email", "").lower().strip()
# First check database for generated codes
async with async_session() as session:
result = await session.execute(
text("""
SELECT id, code, used, expires_at
FROM practitioner_access_codes
WHERE email = :email AND code = :code AND used = FALSE
"""),
{"email": email, "code": code}
)
row = result.fetchone()
if row:
row_dict = dict(row._mapping)
# Check if expired
if row_dict.get('expires_at') and row_dict['expires_at'] < datetime.now(timezone.utc):
raise HTTPException(status_code=403, detail="Code d'accès expiré")
# Mark code as used
await session.execute(
text("UPDATE practitioner_access_codes SET used = TRUE WHERE id = :id"),
{"id": row_dict['id']}
)
await session.commit()
logger.info(f"Practitioner access code verified for {email}")
return {"valid": True}
# Fallback to static code (for backward compatibility)
if code == PRACTITIONER_ACCESS_CODE:
return {"valid": True}
raise HTTPException(status_code=403, detail="Code d'accès invalide")
@api_router.post("/auth/register", response_model=TokenResponse)
async def register(user_data: UserCreate):
async with async_session() as session:
# Check if user exists
result = await session.execute(
text("SELECT id FROM users WHERE email = :email"),
{"email": user_data.email}
)
if result.fetchone():
raise HTTPException(status_code=400, detail="Email déjà utilisé")
user_id = str(uuid.uuid4())
now = datetime.now(timezone.utc)
await session.execute(
text("""
INSERT INTO users (id, email, first_name, last_name, phone, password, role, created_at, google_calendar_connected)
VALUES (:id, :email, :first_name, :last_name, :phone, :password, :role, :created_at, FALSE)
"""),
{
"id": user_id,
"email": user_data.email,
"first_name": user_data.first_name,
"last_name": user_data.last_name,
"phone": user_data.phone,
"password": hash_password(user_data.password),
"role": user_data.role,
"created_at": now
}
)
await session.commit()
user_doc = {
"id": user_id,
"email": user_data.email,
"first_name": user_data.first_name,
"last_name": user_data.last_name,
"phone": user_data.phone,
"role": user_data.role,
"created_at": now.isoformat(),
"google_calendar_connected": False
}
token = create_token(user_id, user_data.role)
user_response = user_to_response(user_doc)
return TokenResponse(access_token=token, user=user_response)
@api_router.post("/auth/login", response_model=TokenResponse)
async def login(credentials: UserLogin):
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM users WHERE email = :email"),
{"email": credentials.email}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=401, detail="Email ou mot de passe incorrect")
user = format_user_row(row_to_dict(row))
if not verify_password(credentials.password, user["password"]):
raise HTTPException(status_code=401, detail="Email ou mot de passe incorrect")
token = create_token(user["id"], user["role"])
user_response = user_to_response(user)
return TokenResponse(access_token=token, user=user_response)
@api_router.get("/auth/me", response_model=UserResponse)
async def get_me(current_user: dict = Depends(get_current_user)):
return user_to_response(current_user)
@api_router.put("/auth/me", response_model=UserResponse)
async def update_me(update_data: dict, current_user: dict = Depends(get_current_user)):
allowed_fields = ["first_name", "last_name", "phone"]
update_fields = {k: v for k, v in update_data.items() if k in allowed_fields and v is not None}
if update_fields:
set_clause = ", ".join([f"{k} = :{k}" for k in update_fields])
update_fields["id"] = current_user["id"]
async with async_session() as session:
await session.execute(
text(f"UPDATE users SET {set_clause} WHERE id = :id"),
update_fields
)
await session.commit()
current_user.update(update_fields)
return user_to_response(current_user)
# ============== PRACTITIONER ROUTES ==============
@api_router.post("/practitioners", response_model=PractitionerResponse)
async def create_practitioner_profile(profile: PractitionerCreate, current_user: dict = Depends(get_current_user)):
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Seuls les praticiens peuvent créer un profil")
async with async_session() as session:
# Check if profile exists
result = await session.execute(
text("SELECT id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
if result.fetchone():
raise HTTPException(status_code=400, detail="Profil praticien déjà existant")
practitioner_id = str(uuid.uuid4())
await session.execute(
text("""
INSERT INTO practitioners (
id, user_id, specialty, bio, photo_url, country, region, department,
address, city, postal_code, consultation_types, price, consultation_duration,
is_verified, years_experience, diplomas, detailed_description, languages
) VALUES (
:id, :user_id, :specialty, :bio, :photo_url, :country, :region, :department,
:address, :city, :postal_code, :consultation_types, :price, :consultation_duration,
FALSE, :years_experience, :diplomas, :detailed_description, :languages
)
"""),
{
"id": practitioner_id,
"user_id": current_user["id"],
"specialty": profile.specialty,
"bio": profile.bio,
"photo_url": profile.photo_url,
"country": profile.country,
"region": profile.region,
"department": profile.department,
"address": profile.address,
"city": profile.city,
"postal_code": profile.postal_code,
"consultation_types": json.dumps(profile.consultation_types),
"price": profile.price,
"consultation_duration": profile.consultation_duration,
"years_experience": profile.years_experience,
"diplomas": json.dumps(profile.diplomas) if profile.diplomas else None,
"detailed_description": profile.detailed_description,
"languages": json.dumps(profile.languages) if profile.languages else None
}
)
await session.commit()
practitioner_doc = {
"id": practitioner_id,
"user_id": current_user["id"],
"is_verified": False,
**profile.model_dump()
}
return PractitionerResponse(**practitioner_doc, user=user_to_response(current_user))
@api_router.get("/practitioners", response_model=List[PractitionerResponse])
async def search_practitioners(
specialty: Optional[str] = None,
country: Optional[str] = None,
region: Optional[str] = None,
department: Optional[str] = None,
city: Optional[str] = None,
consultation_type: Optional[str] = None
):
conditions = []
params = {}
if specialty and specialty != "all":
conditions.append("p.specialty = :specialty")
params["specialty"] = specialty
if country and country != "all":
conditions.append("p.country = :country")
params["country"] = country
if region and region != "all":
conditions.append("p.region = :region")
params["region"] = region
if department and department != "all":
conditions.append("p.department = :department")
params["department"] = department
if city:
conditions.append("LOWER(p.city) LIKE LOWER(:city)")
params["city"] = f"%{city}%"
if consultation_type and consultation_type != "all":
conditions.append("p.consultation_types::text LIKE :consultation_type")
params["consultation_type"] = f"%{consultation_type}%"
where_clause = " AND ".join(conditions) if conditions else "1=1"
async with async_session() as session:
result = await session.execute(
text(f"""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE {where_clause}
ORDER BY p.is_verified DESC, u.created_at DESC
"""),
params
)
rows = result.fetchall()
practitioners = []
for row in rows:
row_dict = dict(row._mapping)
p_dict = format_practitioner_row({
"id": row_dict["id"],
"user_id": row_dict["user_id"],
"specialty": row_dict["specialty"],
"bio": row_dict["bio"],
"photo_url": row_dict["photo_url"],
"country": row_dict["country"],
"region": row_dict["region"],
"department": row_dict["department"],
"address": row_dict["address"],
"city": row_dict["city"],
"postal_code": row_dict["postal_code"],
"consultation_types": row_dict["consultation_types"],
"price": row_dict["price"],
"consultation_duration": row_dict["consultation_duration"],
"is_verified": row_dict["is_verified"],
"years_experience": row_dict["years_experience"],
"diplomas": row_dict["diplomas"],
"detailed_description": row_dict["detailed_description"],
"languages": row_dict["languages"]
})
user_dict = {
"id": row_dict["u_id"],
"email": row_dict["email"],
"first_name": row_dict["first_name"],
"last_name": row_dict["last_name"],
"phone": row_dict["phone"],
"role": row_dict["role"],
"created_at": row_dict["u_created_at"],
"google_calendar_connected": row_dict["google_calendar_connected"]
}
p_dict["user"] = user_to_response(format_user_row(user_dict))
practitioners.append(PractitionerResponse(**p_dict))
return practitioners
@api_router.get("/practitioners/{practitioner_id}", response_model=PractitionerResponse)
async def get_practitioner(practitioner_id: str):
async with async_session() as session:
result = await session.execute(
text("""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE p.id = :id
"""),
{"id": practitioner_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
row_dict = dict(row._mapping)
p_dict = format_practitioner_row({
"id": row_dict["id"],
"user_id": row_dict["user_id"],
"specialty": row_dict["specialty"],
"bio": row_dict["bio"],
"photo_url": row_dict["photo_url"],
"country": row_dict["country"],
"region": row_dict["region"],
"department": row_dict["department"],
"address": row_dict["address"],
"city": row_dict["city"],
"postal_code": row_dict["postal_code"],
"consultation_types": row_dict["consultation_types"],
"price": row_dict["price"],
"consultation_duration": row_dict["consultation_duration"],
"is_verified": row_dict["is_verified"],
"years_experience": row_dict["years_experience"],
"diplomas": row_dict["diplomas"],
"detailed_description": row_dict["detailed_description"],
"languages": row_dict["languages"]
})
user_dict = {
"id": row_dict["u_id"],
"email": row_dict["email"],
"first_name": row_dict["first_name"],
"last_name": row_dict["last_name"],
"phone": row_dict["phone"],
"role": row_dict["role"],
"created_at": row_dict["u_created_at"],
"google_calendar_connected": row_dict["google_calendar_connected"]
}
p_dict["user"] = user_to_response(format_user_row(user_dict))
return PractitionerResponse(**p_dict)
@api_router.put("/practitioners/me", response_model=PractitionerResponse)
async def update_practitioner_profile(update_data: PractitionerUpdate, current_user: dict = Depends(get_current_user)):
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Non autorisé")
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profil praticien non trouvé")
practitioner = format_practitioner_row(row_to_dict(row))
update_fields = {}
for k, v in update_data.model_dump().items():
if v is not None:
if k in ['consultation_types', 'diplomas', 'languages']:
update_fields[k] = json.dumps(v)
else:
update_fields[k] = v
if update_fields:
set_clause = ", ".join([f"{k} = :{k}" for k in update_fields])
update_fields["user_id"] = current_user["id"]
await session.execute(
text(f"UPDATE practitioners SET {set_clause} WHERE user_id = :user_id"),
update_fields
)
await session.commit()
# Refresh practitioner data
for k, v in update_data.model_dump().items():
if v is not None:
practitioner[k] = v
practitioner["user"] = user_to_response(current_user)
return PractitionerResponse(**practitioner)
@api_router.get("/practitioners/me/profile", response_model=PractitionerResponse)
async def get_my_practitioner_profile(current_user: dict = Depends(get_current_user)):
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Non autorisé")
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profil praticien non trouvé")
practitioner = format_practitioner_row(row_to_dict(row))
practitioner["user"] = user_to_response(current_user)
return PractitionerResponse(**practitioner)
# ============== PHOTO UPLOAD ==============
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"}
MAX_FILE_SIZE = 5 * 1024 * 1024 # 5 MB
@api_router.post("/upload/photo")
async def upload_photo(file: UploadFile = File(...), current_user: dict = Depends(get_current_user)):
"""Upload a profile photo for practitioners"""
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Seuls les praticiens peuvent uploader une photo")
file_ext = Path(file.filename).suffix.lower()
if file_ext not in ALLOWED_EXTENSIONS:
raise HTTPException(status_code=400, detail=f"Format non autorisé. Utilisez: {', '.join(ALLOWED_EXTENSIONS)}")
contents = await file.read()
if len(contents) > MAX_FILE_SIZE:
raise HTTPException(status_code=400, detail="Fichier trop volumineux (max 5 MB)")
unique_filename = f"{current_user['id']}_{uuid.uuid4().hex[:8]}{file_ext}"
file_path = UPLOAD_DIR / unique_filename
async with async_session() as session:
result = await session.execute(
text("SELECT photo_url FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if row and row[0] and "/api/photos/" in str(row[0]):
old_filename = row[0].split("/api/photos/")[-1]
old_path = UPLOAD_DIR / old_filename
if old_path.exists():
old_path.unlink()
with open(file_path, "wb") as f:
f.write(contents)
photo_url = f"/api/photos/{unique_filename}"
async with async_session() as session:
await session.execute(
text("UPDATE practitioners SET photo_url = :photo_url WHERE user_id = :user_id"),
{"photo_url": photo_url, "user_id": current_user["id"]}
)
await session.commit()
return {"photo_url": photo_url, "message": "Photo uploadée avec succès"}
@api_router.get("/photos/{filename}")
async def get_photo(filename: str):
"""Serve uploaded photos"""
file_path = UPLOAD_DIR / filename
if not file_path.exists():
raise HTTPException(status_code=404, detail="Photo non trouvée")
return FileResponse(file_path)
# ============== TIME SLOTS ROUTES ==============
@api_router.post("/slots", response_model=TimeSlotResponse)
async def create_time_slot(slot: TimeSlotCreate, current_user: dict = Depends(get_current_user)):
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Non autorisé")
async with async_session() as session:
result = await session.execute(
text("SELECT id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Créez d'abord votre profil praticien")
practitioner_id = row[0]
slot_id = str(uuid.uuid4())
await session.execute(
text("""
INSERT INTO time_slots (id, practitioner_id, day_of_week, start_time, end_time)
VALUES (:id, :practitioner_id, :day_of_week, :start_time, :end_time)
"""),
{
"id": slot_id,
"practitioner_id": practitioner_id,
"day_of_week": slot.day_of_week,
"start_time": slot.start_time,
"end_time": slot.end_time
}
)
await session.commit()
return TimeSlotResponse(
id=slot_id,
practitioner_id=practitioner_id,
day_of_week=slot.day_of_week,
start_time=slot.start_time,
end_time=slot.end_time
)
@api_router.get("/slots/practitioner/{practitioner_id}", response_model=List[TimeSlotResponse])
async def get_practitioner_slots(practitioner_id: str):
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM time_slots WHERE practitioner_id = :practitioner_id"),
{"practitioner_id": practitioner_id}
)
rows = result.fetchall()
return [TimeSlotResponse(**row_to_dict(row)) for row in rows]
@api_router.get("/slots/me", response_model=List[TimeSlotResponse])
async def get_my_slots(current_user: dict = Depends(get_current_user)):
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Non autorisé")
async with async_session() as session:
result = await session.execute(
text("SELECT id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
return []
result = await session.execute(
text("SELECT * FROM time_slots WHERE practitioner_id = :practitioner_id"),
{"practitioner_id": row[0]}
)
rows = result.fetchall()
return [TimeSlotResponse(**row_to_dict(r)) for r in rows]
@api_router.delete("/slots/{slot_id}")
async def delete_slot(slot_id: str, current_user: dict = Depends(get_current_user)):
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Non autorisé")
async with async_session() as session:
result = await session.execute(
text("SELECT id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Profil non trouvé")
result = await session.execute(
text("DELETE FROM time_slots WHERE id = :id AND practitioner_id = :practitioner_id RETURNING id"),
{"id": slot_id, "practitioner_id": row[0]}
)
await session.commit()
if not result.fetchone():
raise HTTPException(status_code=404, detail="Créneau non trouvé")
return {"message": "Créneau supprimé"}
# ============== APPOINTMENTS ROUTES ==============
@api_router.post("/appointments", response_model=AppointmentResponse)
async def create_appointment(appointment: AppointmentCreate, current_user: dict = Depends(get_current_user)):
async with async_session() as session:
# Get practitioner
result = await session.execute(
text("""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE p.id = :id
"""),
{"id": appointment.practitioner_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
row_dict = dict(row._mapping)
practitioner = format_practitioner_row({
"id": row_dict["id"],
"user_id": row_dict["user_id"],
"specialty": row_dict["specialty"],
"bio": row_dict["bio"],
"photo_url": row_dict["photo_url"],
"country": row_dict["country"],
"region": row_dict["region"],
"department": row_dict["department"],
"address": row_dict["address"],
"city": row_dict["city"],
"postal_code": row_dict["postal_code"],
"consultation_types": row_dict["consultation_types"],
"price": row_dict["price"],
"consultation_duration": row_dict["consultation_duration"],
"is_verified": row_dict["is_verified"],
"years_experience": row_dict["years_experience"],
"diplomas": row_dict["diplomas"],
"detailed_description": row_dict["detailed_description"],
"languages": row_dict["languages"]
})
practitioner_user = {
"id": row_dict["u_id"],
"email": row_dict["email"],
"first_name": row_dict["first_name"],
"last_name": row_dict["last_name"],
"phone": row_dict["phone"],
"role": row_dict["role"],
"created_at": row_dict["u_created_at"],
"google_calendar_connected": row_dict["google_calendar_connected"]
}
# Check if slot is available
result = await session.execute(
text("""
SELECT id FROM appointments
WHERE practitioner_id = :practitioner_id
AND date = :date
AND start_time = :start_time
AND status IN ('pending', 'confirmed')
"""),
{
"practitioner_id": appointment.practitioner_id,
"date": appointment.date,
"start_time": appointment.start_time
}
)
if result.fetchone():
raise HTTPException(status_code=400, detail="Ce créneau n'est plus disponible")
appointment_id = str(uuid.uuid4())
now = datetime.now(timezone.utc)
# Calculate end time
start_hour, start_min = map(int, appointment.start_time.split(":"))
end_datetime = datetime(2000, 1, 1, start_hour, start_min) + timedelta(minutes=practitioner["consultation_duration"])
end_time = end_datetime.strftime("%H:%M")
# Create video room for visio appointments
video_room_url = None
if appointment.consultation_type == "visio":
video_room_url = await create_video_room(appointment_id)
appointment_doc = {
"id": appointment_id,
"client_id": current_user["id"],
"practitioner_id": appointment.practitioner_id,
"date": appointment.date,
"start_time": appointment.start_time,
"end_time": end_time,
"consultation_type": appointment.consultation_type,
"status": "pending",
"notes": appointment.notes,
"video_room_url": video_room_url,
"created_at": now.isoformat()
}
# Add to Google Calendar if practitioner has connected
google_event_id = None
if practitioner_user.get("google_calendar_connected"):
practitioner["user"] = user_to_response(format_user_row(practitioner_user))
google_event_id = await add_to_google_calendar(
practitioner_user["email"],
appointment_doc,
practitioner,
current_user
)
await session.execute(
text("""
INSERT INTO appointments (id, client_id, practitioner_id, date, start_time, end_time,
consultation_type, status, notes, video_room_url, google_event_id, created_at)
VALUES (:id, :client_id, :practitioner_id, :date, :start_time, :end_time,
:consultation_type, :status, :notes, :video_room_url, :google_event_id, :created_at)
"""),
{
"id": appointment_id,
"client_id": current_user["id"],
"practitioner_id": appointment.practitioner_id,
"date": appointment.date,
"start_time": appointment.start_time,
"end_time": end_time,
"consultation_type": appointment.consultation_type,
"status": "pending",
"notes": appointment.notes,
"video_room_url": video_room_url,
"google_event_id": google_event_id,
"created_at": now
}
)
await session.commit()
# Send email notification to client
practitioner["user"] = user_to_response(format_user_row(practitioner_user))
asyncio.create_task(send_appointment_notification(appointment_doc, practitioner, current_user))
# Build response
appointment_doc["client"] = user_to_response(current_user)
appointment_doc["practitioner"] = PractitionerResponse(**practitioner)
appointment_doc["google_event_id"] = google_event_id
return AppointmentResponse(**appointment_doc)
@api_router.get("/appointments/me", response_model=List[AppointmentResponse])
async def get_my_appointments(current_user: dict = Depends(get_current_user)):
async with async_session() as session:
if current_user["role"] == "client":
result = await session.execute(
text("SELECT * FROM appointments WHERE client_id = :client_id ORDER BY date, start_time"),
{"client_id": current_user["id"]}
)
else:
# Get practitioner ID
p_result = await session.execute(
text("SELECT id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
p_row = p_result.fetchone()
if not p_row:
return []
result = await session.execute(
text("SELECT * FROM appointments WHERE practitioner_id = :practitioner_id ORDER BY date, start_time"),
{"practitioner_id": p_row[0]}
)
rows = result.fetchall()
appointments = []
for row in rows:
apt = row_to_dict(row)
if apt.get('created_at') and hasattr(apt['created_at'], 'isoformat'):
apt['created_at'] = apt['created_at'].isoformat()
# Get client
c_result = await session.execute(
text("SELECT * FROM users WHERE id = :id"),
{"id": apt["client_id"]}
)
c_row = c_result.fetchone()
if c_row:
apt["client"] = user_to_response(format_user_row(row_to_dict(c_row)))
# Get practitioner
p_result = await session.execute(
text("""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE p.id = :id
"""),
{"id": apt["practitioner_id"]}
)
p_row = p_result.fetchone()
if p_row:
p_dict = dict(p_row._mapping)
practitioner = format_practitioner_row({
"id": p_dict["id"],
"user_id": p_dict["user_id"],
"specialty": p_dict["specialty"],
"bio": p_dict["bio"],
"photo_url": p_dict["photo_url"],
"country": p_dict["country"],
"region": p_dict["region"],
"department": p_dict["department"],
"address": p_dict["address"],
"city": p_dict["city"],
"postal_code": p_dict["postal_code"],
"consultation_types": p_dict["consultation_types"],
"price": p_dict["price"],
"consultation_duration": p_dict["consultation_duration"],
"is_verified": p_dict["is_verified"],
"years_experience": p_dict["years_experience"],
"diplomas": p_dict["diplomas"],
"detailed_description": p_dict["detailed_description"],
"languages": p_dict["languages"]
})
practitioner["user"] = user_to_response(format_user_row({
"id": p_dict["u_id"],
"email": p_dict["email"],
"first_name": p_dict["first_name"],
"last_name": p_dict["last_name"],
"phone": p_dict["phone"],
"role": p_dict["role"],
"created_at": p_dict["u_created_at"],
"google_calendar_connected": p_dict["google_calendar_connected"]
}))
apt["practitioner"] = PractitionerResponse(**practitioner)
appointments.append(AppointmentResponse(**apt))
return appointments
@api_router.put("/appointments/{appointment_id}/status")
async def update_appointment_status(appointment_id: str, status: str, current_user: dict = Depends(get_current_user)):
valid_statuses = ["confirmed", "cancelled", "completed"]
if status not in valid_statuses:
raise HTTPException(status_code=400, detail=f"Statut invalide. Utilisez: {valid_statuses}")
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM appointments WHERE id = :id"),
{"id": appointment_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Rendez-vous non trouvé")
appointment = row_to_dict(row)
if current_user["role"] == "practitioner":
p_result = await session.execute(
text("SELECT id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
p_row = p_result.fetchone()
if not p_row or p_row[0] != appointment["practitioner_id"]:
raise HTTPException(status_code=403, detail="Non autorisé")
elif appointment["client_id"] != current_user["id"]:
raise HTTPException(status_code=403, detail="Non autorisé")
await session.execute(
text("UPDATE appointments SET status = :status WHERE id = :id"),
{"status": status, "id": appointment_id}
)
await session.commit()
# Send notification on confirmation
if status == "confirmed":
c_result = await session.execute(
text("SELECT * FROM users WHERE id = :id"),
{"id": appointment["client_id"]}
)
c_row = c_result.fetchone()
p_result = await session.execute(
text("""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE p.id = :id
"""),
{"id": appointment["practitioner_id"]}
)
p_row = p_result.fetchone()
if c_row and p_row:
client = format_user_row(row_to_dict(c_row))
p_dict = dict(p_row._mapping)
practitioner = format_practitioner_row({
"id": p_dict["id"],
"user_id": p_dict["user_id"],
"specialty": p_dict["specialty"],
"bio": p_dict["bio"],
"photo_url": p_dict["photo_url"],
"country": p_dict["country"],
"region": p_dict["region"],
"department": p_dict["department"],
"address": p_dict["address"],
"city": p_dict["city"],
"postal_code": p_dict["postal_code"],
"consultation_types": p_dict["consultation_types"],
"price": p_dict["price"],
"consultation_duration": p_dict["consultation_duration"],
"is_verified": p_dict["is_verified"],
"years_experience": p_dict["years_experience"],
"diplomas": p_dict["diplomas"],
"detailed_description": p_dict["detailed_description"],
"languages": p_dict["languages"]
})
practitioner["user"] = user_to_response(format_user_row({
"id": p_dict["u_id"],
"email": p_dict["email"],
"first_name": p_dict["first_name"],
"last_name": p_dict["last_name"],
"phone": p_dict["phone"],
"role": p_dict["role"],
"created_at": p_dict["u_created_at"],
"google_calendar_connected": p_dict["google_calendar_connected"]
}))
asyncio.create_task(send_appointment_notification(appointment, practitioner, client))
return {"message": f"Statut mis à jour: {status}"}
@api_router.get("/appointments/practitioner/{practitioner_id}/booked", response_model=List[dict])
async def get_booked_slots(practitioner_id: str, date: Optional[str] = None):
async with async_session() as session:
if date:
result = await session.execute(
text("""
SELECT date, start_time FROM appointments
WHERE practitioner_id = :practitioner_id
AND date = :date
AND status IN ('pending', 'confirmed')
"""),
{"practitioner_id": practitioner_id, "date": date}
)
else:
result = await session.execute(
text("""
SELECT date, start_time FROM appointments
WHERE practitioner_id = :practitioner_id
AND status IN ('pending', 'confirmed')
"""),
{"practitioner_id": practitioner_id}
)
rows = result.fetchall()
return [{"date": row[0], "start_time": row[1]} for row in rows]
# ============== STATISTICS ROUTES ==============
@api_router.get("/stats/practitioner")
async def get_practitioner_stats(current_user: dict = Depends(get_current_user)):
if current_user["role"] != "practitioner":
raise HTTPException(status_code=403, detail="Non autorisé")
async with async_session() as session:
result = await session.execute(
text("SELECT id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
return {"total": 0, "pending": 0, "confirmed": 0, "completed": 0, "cancelled": 0}
practitioner_id = row[0]
result = await session.execute(
text("""
SELECT status, COUNT(*) as count
FROM appointments
WHERE practitioner_id = :practitioner_id
GROUP BY status
"""),
{"practitioner_id": practitioner_id}
)
rows = result.fetchall()
stats = {"total": 0, "pending": 0, "confirmed": 0, "completed": 0, "cancelled": 0}
for row in rows:
status, count = row
if status in stats:
stats[status] = count
stats["total"] += count
return stats
# ============== SPECIALTIES ==============
@api_router.get("/specialties")
async def get_specialties():
return [
{"id": "therapeute", "name": "Thérapeute", "icon": "heart"},
{"id": "naturopathe", "name": "Naturopathe", "icon": "leaf"},
{"id": "sophrologue", "name": "Sophrologue", "icon": "wind"},
{"id": "psychologue", "name": "Psychologue", "icon": "brain"},
{"id": "osteopathe", "name": "Ostéopathe", "icon": "hand"},
{"id": "hypnotherapeute", "name": "Hypnothérapeute", "icon": "sparkles"},
{"id": "coach", "name": "Coach", "icon": "target"},
{"id": "orthophoniste", "name": "Orthophoniste", "icon": "mic"}
]
# ============== PRACTITIONER REGISTRATION REQUEST ==============
class PractitionerRequest(BaseModel):
first_name: str
last_name: str
email: str
class PractitionerRegistrationPayment(BaseModel):
email: str
first_name: str
last_name: str
origin_url: str
import random
import string
def generate_access_code(length=8):
"""Generate a random alphanumeric access code"""
return ''.join(random.choices(string.ascii_uppercase + string.digits, k=length))
@api_router.post("/practitioner-request")
async def submit_practitioner_request(request: PractitionerRequest):
"""Submit a request to become a practitioner - sends code to candidate and notification to admin"""
try:
# Generate unique access code
access_code = generate_access_code()
code_id = str(uuid.uuid4())
expires_at = datetime.now() + timedelta(days=7) # Code valid for 7 days (naive datetime)
# Store the code in database
async with async_session() as session:
await session.execute(
text("""
INSERT INTO practitioner_access_codes (id, email, first_name, last_name, code, expires_at)
VALUES (:id, :email, :first_name, :last_name, :code, :expires_at)
"""),
{
"id": code_id,
"email": request.email.lower(),
"first_name": request.first_name,
"last_name": request.last_name,
"code": access_code,
"expires_at": expires_at
}
)
await session.commit()
if RESEND_API_KEY:
import resend
resend.api_key = RESEND_API_KEY
# Determine sender format based on email domain
if 'resend.dev' in SENDER_EMAIL:
sender_from = SENDER_EMAIL # Use plain email for resend.dev
else:
sender_from = f"Opora Vie <{SENDER_EMAIL}>"
try:
# Send code to the candidate
resend.Emails.send({
"from": sender_from,
"to": [request.email],
"subject": "Votre code d'accès praticien - Opora Vie",
"html": f"""
Opora Vie
Bonjour {request.first_name},
Merci pour votre demande d'inscription en tant que praticien sur Opora Vie.
Voici votre code d'accès :
{access_code}
Ce code est valide pendant 7 jours. Retournez sur la page d'inscription praticien et entrez ce code pour continuer.
© 2025 Opora Vie - Portail des praticiens du bien-être
"""
})
logger.info(f"Email sent to candidate: {request.email}")
except Exception as e:
logger.error(f"Failed to send email to candidate: {str(e)}")
try:
# Send notification to admin
resend.Emails.send({
"from": sender_from,
"to": ["contact@apora.net"],
"subject": f"[INFO] Nouvelle demande praticien - {request.first_name} {request.last_name}",
"html": f"""
Nouvelle demande d'inscription praticien
Nom : {request.last_name}
Prénom : {request.first_name}
Email : {request.email}
Date de la demande : {datetime.now(timezone.utc).strftime('%d/%m/%Y à %H:%M')}
✅ Un code d'accès a été envoyé automatiquement au candidat.
Code généré : {access_code}
"""
})
logger.info("Notification email sent to admin")
except Exception as e:
logger.error(f"Failed to send admin notification: {str(e)}")
logger.info(f"Practitioner access code sent to {request.email}")
return {"message": "Code d'accès envoyé par email", "status": "code_sent"}
except Exception as e:
logger.error(f"Error sending practitioner request email: {e}")
raise HTTPException(status_code=500, detail=f"Erreur: {str(e)}")
@api_router.post("/payments/practitioner-registration")
async def create_practitioner_registration_payment(request: PractitionerRegistrationPayment, http_request: Request):
"""Create Stripe checkout session for practitioner registration fee (3€)"""
if not STRIPE_API_KEY:
raise HTTPException(status_code=503, detail="Paiement en ligne non configuré")
try:
from emergentintegrations.payments.stripe.checkout import StripeCheckout, CheckoutSessionRequest
webhook_url = f"{str(http_request.base_url).rstrip('/')}/api/webhook/stripe"
stripe_checkout = StripeCheckout(api_key=STRIPE_API_KEY, webhook_url=webhook_url)
success_url = f"{request.origin_url}/devenir-praticien?payment=success"
cancel_url = f"{request.origin_url}/devenir-praticien?payment=cancelled"
checkout_request = CheckoutSessionRequest(
amount=3.0, # 3€ registration fee
currency="eur",
success_url=success_url,
cancel_url=cancel_url,
metadata={
"type": "practitioner_registration",
"email": request.email,
"first_name": request.first_name,
"last_name": request.last_name
}
)
checkout_session = await stripe_checkout.create_checkout_session(checkout_request)
return {
"checkout_url": checkout_session.checkout_url,
"session_id": checkout_session.session_id
}
except Exception as e:
logger.error(f"Error creating practitioner registration checkout: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ============== STRIPE SUBSCRIPTIONS (50€/month) ==============
class SubscriptionRequest(BaseModel):
origin_url: str
@api_router.get("/subscription/status")
async def get_subscription_status(current_user: dict = Depends(get_current_user)):
"""Get practitioner's subscription status"""
if current_user.get("role") != "practitioner":
raise HTTPException(status_code=403, detail="Réservé aux praticiens")
async with async_session() as session:
# Get practitioner record
result = await session.execute(
text("SELECT id, stripe_customer_id, stripe_subscription_id, subscription_status, subscription_enabled FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
p = dict(row._mapping)
# If has active subscription, check with Stripe
subscription_data = None
if p.get("stripe_subscription_id") and STRIPE_API_KEY:
try:
subscription = stripe.Subscription.retrieve(p["stripe_subscription_id"])
subscription_data = {
"id": subscription.id,
"status": subscription.status,
"current_period_end": datetime.fromtimestamp(subscription.current_period_end).isoformat(),
"cancel_at_period_end": subscription.cancel_at_period_end
}
except Exception as e:
logger.error(f"Error fetching Stripe subscription: {e}")
return {
"subscription_enabled": p.get("subscription_enabled", False),
"subscription_status": p.get("subscription_status", "none"),
"stripe_subscription": subscription_data,
"price": STRIPE_SUBSCRIPTION_PRICE
}
@api_router.post("/subscription/create-checkout")
async def create_subscription_checkout(request: SubscriptionRequest, http_request: Request, current_user: dict = Depends(get_current_user)):
"""Create Stripe checkout session for monthly subscription"""
if current_user.get("role") != "practitioner":
raise HTTPException(status_code=403, detail="Réservé aux praticiens")
if not STRIPE_API_KEY:
raise HTTPException(status_code=503, detail="Paiement en ligne non configuré")
async with async_session() as session:
# Get practitioner record
result = await session.execute(
text("SELECT id, stripe_customer_id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
practitioner = dict(row._mapping)
practitioner_id = practitioner["id"]
try:
# Create or get Stripe customer
customer_id = practitioner.get("stripe_customer_id")
if not customer_id:
customer = stripe.Customer.create(
email=current_user.get("email"),
name=f"{current_user.get('first_name', '')} {current_user.get('last_name', '')}",
metadata={"practitioner_id": practitioner_id}
)
customer_id = customer.id
# Save customer ID
await session.execute(
text("UPDATE practitioners SET stripe_customer_id = :customer_id WHERE id = :id"),
{"customer_id": customer_id, "id": practitioner_id}
)
await session.commit()
# Create or get the subscription price
# First, try to find existing price
prices = stripe.Price.list(
lookup_keys=["opora_monthly_subscription"],
limit=1
)
if prices.data:
price_id = prices.data[0].id
else:
# Create product and price
product = stripe.Product.create(
name="Abonnement Opora Vie",
description="Cotisation mensuelle praticien - 50€/mois"
)
price = stripe.Price.create(
product=product.id,
unit_amount=5000, # 50€ in cents
currency="eur",
recurring={"interval": "month"},
lookup_key="opora_monthly_subscription"
)
price_id = price.id
# Create checkout session for subscription
success_url = f"{request.origin_url}/practitioner/dashboard?subscription=success&session_id={{CHECKOUT_SESSION_ID}}"
cancel_url = f"{request.origin_url}/practitioner/dashboard?subscription=cancelled"
checkout_session = stripe.checkout.Session.create(
customer=customer_id,
payment_method_types=["card"],
line_items=[{
"price": price_id,
"quantity": 1
}],
mode="subscription",
success_url=success_url,
cancel_url=cancel_url,
metadata={
"practitioner_id": practitioner_id,
"type": "monthly_subscription"
},
subscription_data={
"metadata": {
"practitioner_id": practitioner_id
}
}
)
logger.info(f"Subscription checkout created for practitioner {practitioner_id}")
return {
"checkout_url": checkout_session.url,
"session_id": checkout_session.id
}
except Exception as e:
logger.error(f"Error creating subscription checkout: {e}")
raise HTTPException(status_code=500, detail=str(e))
@api_router.post("/subscription/cancel")
async def cancel_subscription(current_user: dict = Depends(get_current_user)):
"""Cancel practitioner's subscription"""
if current_user.get("role") != "practitioner":
raise HTTPException(status_code=403, detail="Réservé aux praticiens")
if not STRIPE_API_KEY:
raise HTTPException(status_code=503, detail="Paiement en ligne non configuré")
async with async_session() as session:
result = await session.execute(
text("SELECT id, stripe_subscription_id FROM practitioners WHERE user_id = :user_id"),
{"user_id": current_user["id"]}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
practitioner = dict(row._mapping)
if not practitioner.get("stripe_subscription_id"):
raise HTTPException(status_code=400, detail="Aucun abonnement actif")
try:
# Cancel at period end (not immediately)
subscription = stripe.Subscription.modify(
practitioner["stripe_subscription_id"],
cancel_at_period_end=True
)
# Update database
await session.execute(
text("UPDATE practitioners SET subscription_status = 'cancelling' WHERE id = :id"),
{"id": practitioner["id"]}
)
await session.commit()
return {
"message": "Abonnement annulé. Il restera actif jusqu'à la fin de la période en cours.",
"cancel_at": datetime.fromtimestamp(subscription.current_period_end).isoformat()
}
except Exception as e:
logger.error(f"Error cancelling subscription: {e}")
raise HTTPException(status_code=500, detail=str(e))
@api_router.post("/webhook/stripe-subscription")
async def handle_stripe_subscription_webhook(request: Request):
"""Handle Stripe subscription webhooks"""
payload = await request.body()
sig_header = request.headers.get("Stripe-Signature")
# For simplicity, we'll process without signature verification
# In production, you should verify the webhook signature
try:
event = json.loads(payload)
event_type = event.get("type")
data = event.get("data", {}).get("object", {})
logger.info(f"Received Stripe webhook: {event_type}")
async with async_session() as session:
if event_type == "checkout.session.completed":
# Subscription checkout completed
if data.get("mode") == "subscription":
practitioner_id = data.get("metadata", {}).get("practitioner_id")
subscription_id = data.get("subscription")
if practitioner_id and subscription_id:
await session.execute(
text("""
UPDATE practitioners
SET stripe_subscription_id = :sub_id,
subscription_status = 'active',
subscription_enabled = true
WHERE id = :id
"""),
{"sub_id": subscription_id, "id": practitioner_id}
)
await session.commit()
logger.info(f"Subscription activated for practitioner {practitioner_id}")
elif event_type == "customer.subscription.updated":
subscription_id = data.get("id")
status = data.get("status")
practitioner_id = data.get("metadata", {}).get("practitioner_id")
if practitioner_id:
await session.execute(
text("UPDATE practitioners SET subscription_status = :status WHERE id = :id"),
{"status": status, "id": practitioner_id}
)
await session.commit()
elif event_type == "customer.subscription.deleted":
subscription_id = data.get("id")
practitioner_id = data.get("metadata", {}).get("practitioner_id")
if practitioner_id:
await session.execute(
text("""
UPDATE practitioners
SET subscription_status = 'cancelled',
subscription_enabled = false,
stripe_subscription_id = NULL
WHERE id = :id
"""),
{"id": practitioner_id}
)
await session.commit()
logger.info(f"Subscription cancelled for practitioner {practitioner_id}")
elif event_type == "invoice.payment_failed":
subscription_id = data.get("subscription")
# Find practitioner by subscription ID
result = await session.execute(
text("SELECT id FROM practitioners WHERE stripe_subscription_id = :sub_id"),
{"sub_id": subscription_id}
)
row = result.fetchone()
if row:
await session.execute(
text("UPDATE practitioners SET subscription_status = 'past_due' WHERE id = :id"),
{"id": row[0]}
)
await session.commit()
return {"received": True}
except Exception as e:
logger.error(f"Webhook error: {e}")
return {"received": True} # Always return 200 to Stripe
# ============== STRIPE PAYMENT ==============
class PaymentRequest(BaseModel):
appointment_id: str
origin_url: str
class PaymentStatusRequest(BaseModel):
session_id: str
@api_router.post("/payments/create-checkout")
async def create_payment_checkout(request: PaymentRequest, http_request: Request, current_user: dict = Depends(get_current_user)):
"""Create Stripe checkout session for appointment payment"""
if not STRIPE_API_KEY:
raise HTTPException(status_code=503, detail="Paiement en ligne non configuré")
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM appointments WHERE id = :id"),
{"id": request.appointment_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Rendez-vous non trouvé")
appointment = row_to_dict(row)
if appointment["client_id"] != current_user["id"]:
raise HTTPException(status_code=403, detail="Non autorisé")
p_result = await session.execute(
text("SELECT * FROM practitioners WHERE id = :id"),
{"id": appointment["practitioner_id"]}
)
p_row = p_result.fetchone()
if not p_row:
raise HTTPException(status_code=400, detail="Praticien non trouvé")
practitioner = format_practitioner_row(row_to_dict(p_row))
if not practitioner.get("price"):
raise HTTPException(status_code=400, detail="Prix non défini pour ce praticien")
try:
from emergentintegrations.payments.stripe.checkout import StripeCheckout, CheckoutSessionRequest
webhook_url = f"{str(http_request.base_url).rstrip('/')}/api/webhook/stripe"
stripe_checkout = StripeCheckout(api_key=STRIPE_API_KEY, webhook_url=webhook_url)
success_url = f"{request.origin_url}/client/dashboard?payment=success&session_id={{CHECKOUT_SESSION_ID}}"
cancel_url = f"{request.origin_url}/booking/{appointment['practitioner_id']}?payment=cancelled"
checkout_request = CheckoutSessionRequest(
amount=float(practitioner["price"]),
currency="eur",
success_url=success_url,
cancel_url=cancel_url,
metadata={
"appointment_id": appointment["id"],
"client_id": current_user["id"],
"practitioner_id": appointment["practitioner_id"]
}
)
checkout_session = await stripe_checkout.create_checkout_session(checkout_request)
payment_id = str(uuid.uuid4())
await session.execute(
text("""
INSERT INTO payment_transactions (id, session_id, appointment_id, client_id, amount, currency, status, payment_status, created_at)
VALUES (:id, :session_id, :appointment_id, :client_id, :amount, :currency, :status, :payment_status, :created_at)
"""),
{
"id": payment_id,
"session_id": checkout_session.session_id,
"appointment_id": appointment["id"],
"client_id": current_user["id"],
"amount": float(practitioner["price"]),
"currency": "eur",
"status": "pending",
"payment_status": "initiated",
"created_at": datetime.now(timezone.utc)
}
)
await session.commit()
logger.info(f"Checkout session created: {checkout_session.session_id}")
return {"checkout_url": checkout_session.url, "session_id": checkout_session.session_id}
except Exception as e:
logger.error(f"Stripe checkout error: {str(e)}")
raise HTTPException(status_code=500, detail="Erreur lors de la création du paiement")
@api_router.get("/payments/status/{session_id}")
async def get_payment_status(session_id: str, current_user: dict = Depends(get_current_user)):
"""Get payment status for a checkout session"""
if not STRIPE_API_KEY:
raise HTTPException(status_code=503, detail="Paiement en ligne non configuré")
try:
from emergentintegrations.payments.stripe.checkout import StripeCheckout
stripe_checkout = StripeCheckout(api_key=STRIPE_API_KEY, webhook_url="")
status = await stripe_checkout.get_checkout_status(session_id)
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM payment_transactions WHERE session_id = :session_id"),
{"session_id": session_id}
)
row = result.fetchone()
if row:
payment = row_to_dict(row)
new_status = "completed" if status.payment_status == "paid" else status.status
await session.execute(
text("UPDATE payment_transactions SET status = :status, payment_status = :payment_status WHERE session_id = :session_id"),
{"status": new_status, "payment_status": status.payment_status, "session_id": session_id}
)
if status.payment_status == "paid":
await session.execute(
text("UPDATE appointments SET status = 'confirmed', paid = TRUE WHERE id = :id"),
{"id": payment["appointment_id"]}
)
await session.commit()
return {
"status": status.status,
"payment_status": status.payment_status,
"amount": status.amount_total / 100,
"currency": status.currency
}
except Exception as e:
logger.error(f"Payment status error: {str(e)}")
raise HTTPException(status_code=500, detail="Erreur lors de la vérification du paiement")
@api_router.post("/webhook/stripe")
async def stripe_webhook(request: Request):
"""Handle Stripe webhooks"""
if not STRIPE_API_KEY:
return {"status": "ignored"}
try:
from emergentintegrations.payments.stripe.checkout import StripeCheckout
body = await request.body()
signature = request.headers.get("Stripe-Signature")
stripe_checkout = StripeCheckout(api_key=STRIPE_API_KEY, webhook_url="")
webhook_response = await stripe_checkout.handle_webhook(body, signature)
if webhook_response.payment_status == "paid":
async with async_session() as session:
result = await session.execute(
text("SELECT appointment_id FROM payment_transactions WHERE session_id = :session_id"),
{"session_id": webhook_response.session_id}
)
row = result.fetchone()
if row:
await session.execute(
text("UPDATE payment_transactions SET status = 'completed', payment_status = 'paid' WHERE session_id = :session_id"),
{"session_id": webhook_response.session_id}
)
await session.execute(
text("UPDATE appointments SET status = 'confirmed', paid = TRUE WHERE id = :id"),
{"id": row[0]}
)
await session.commit()
logger.info(f"Webhook processed: {webhook_response.event_type}")
return {"status": "ok"}
except Exception as e:
logger.error(f"Webhook error: {str(e)}")
return {"status": "error"}
# ============== INTEGRATION STATUS ==============
@api_router.get("/integrations/status")
async def get_integrations_status():
"""Check which integrations are configured"""
return {
"email_notifications": bool(RESEND_API_KEY),
"google_calendar": bool(GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET),
"video_daily": bool(DAILY_API_KEY and DAILY_DOMAIN),
"video_jitsi": True,
"stripe_payment": bool(STRIPE_API_KEY)
}
# ============== ADMIN BACKOFFICE ==============
async def verify_admin(current_user: dict = Depends(get_current_user)):
"""Verify user is admin"""
if current_user.get("email") != ADMIN_EMAIL and current_user.get("role") != "admin":
raise HTTPException(status_code=403, detail="Accès réservé à l'administrateur")
return current_user
@api_router.get("/admin/practitioners")
async def admin_get_all_practitioners(current_user: dict = Depends(verify_admin)):
"""Get all practitioners for admin"""
async with async_session() as session:
result = await session.execute(
text("""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
ORDER BY u.last_name, u.first_name
""")
)
rows = result.fetchall()
practitioners = []
for row in rows:
p_dict = dict(row._mapping)
practitioner = format_practitioner_row({
"id": p_dict["id"],
"user_id": p_dict["user_id"],
"specialty": p_dict["specialty"],
"bio": p_dict["bio"],
"photo_url": p_dict["photo_url"],
"country": p_dict["country"],
"region": p_dict["region"],
"department": p_dict["department"],
"address": p_dict["address"],
"city": p_dict["city"],
"postal_code": p_dict["postal_code"],
"consultation_types": p_dict["consultation_types"],
"price": p_dict["price"],
"consultation_duration": p_dict["consultation_duration"],
"is_verified": p_dict["is_verified"],
"years_experience": p_dict["years_experience"],
"diplomas": p_dict["diplomas"],
"detailed_description": p_dict["detailed_description"],
"languages": p_dict["languages"]
})
practitioner["user"] = user_to_response(format_user_row({
"id": p_dict["u_id"],
"email": p_dict["email"],
"first_name": p_dict["first_name"],
"last_name": p_dict["last_name"],
"phone": p_dict["phone"],
"role": p_dict["role"],
"created_at": p_dict["u_created_at"],
"google_calendar_connected": p_dict["google_calendar_connected"]
}))
practitioners.append(practitioner)
return practitioners
@api_router.post("/admin/practitioner/{practitioner_id}/reset-password")
async def admin_reset_practitioner_password(
practitioner_id: str,
data: dict,
current_user: dict = Depends(verify_admin)
):
"""Reset a practitioner's password (admin only)"""
new_password = data.get("new_password")
if not new_password or len(new_password) < 6:
raise HTTPException(status_code=400, detail="Le mot de passe doit contenir au moins 6 caractères")
async with async_session() as session:
# Get practitioner's user_id
result = await session.execute(
text("SELECT user_id FROM practitioners WHERE id = :id"),
{"id": practitioner_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
user_id = row[0]
# Hash new password and update
hashed_password = hash_password(new_password)
await session.execute(
text("UPDATE users SET password = :password WHERE id = :id"),
{"password": hashed_password, "id": user_id}
)
await session.commit()
return {"message": "Mot de passe réinitialisé avec succès"}
@api_router.get("/admin/practitioner/{practitioner_id}/report")
async def admin_get_practitioner_report(
practitioner_id: str,
month: int = Query(..., ge=1, le=12),
year: int = Query(..., ge=2020, le=2100),
current_user: dict = Depends(verify_admin)
):
"""Get monthly report data for a practitioner"""
async with async_session() as session:
result = await session.execute(
text("""
SELECT p.*, u.first_name, u.last_name, u.email
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE p.id = :id
"""),
{"id": practitioner_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
p_dict = dict(row._mapping)
practitioner = format_practitioner_row(p_dict)
start_date = f"{year}-{month:02d}-01"
if month == 12:
end_date = f"{year + 1}-01-01"
else:
end_date = f"{year}-{month + 1:02d}-01"
result = await session.execute(
text("""
SELECT * FROM appointments
WHERE practitioner_id = :practitioner_id
AND date >= :start_date AND date < :end_date
AND status IN ('confirmed', 'completed')
ORDER BY date, start_time
"""),
{"practitioner_id": practitioner_id, "start_date": start_date, "end_date": end_date}
)
appointments = [row_to_dict(r) for r in result.fetchall()]
total_consultations = len(appointments)
cabinet_count = sum(1 for a in appointments if a.get("consultation_type") == "cabinet")
visio_count = sum(1 for a in appointments if a.get("consultation_type") == "visio")
price_per_consultation = float(practitioner.get("price", 0) or 0)
total_revenue = total_consultations * price_per_consultation
cabinet_revenue = cabinet_count * price_per_consultation
visio_revenue = visio_count * price_per_consultation
paid_count = sum(1 for a in appointments if a.get("paid"))
online_revenue = paid_count * price_per_consultation
return {
"practitioner": {
"id": practitioner_id,
"name": f"{p_dict['first_name']} {p_dict['last_name']}",
"email": p_dict.get("email", "N/A"),
"specialty": practitioner.get("specialty", "N/A"),
"city": practitioner.get("city", "N/A"),
"price": price_per_consultation
},
"period": {
"month": month,
"year": year,
"month_name": ["", "Janvier", "Février", "Mars", "Avril", "Mai", "Juin",
"Juillet", "Août", "Septembre", "Octobre", "Novembre", "Décembre"][month]
},
"consultations": {
"total": total_consultations,
"cabinet": cabinet_count,
"visio": visio_count
},
"revenue": {
"total": total_revenue,
"cabinet": cabinet_revenue,
"visio": visio_revenue,
"online_paid": online_revenue,
"currency": "EUR"
},
"appointments": [
{
"date": a["date"],
"time": a["start_time"],
"type": a.get("consultation_type", "cabinet"),
"status": a.get("status", "confirmed"),
"paid": a.get("paid", False)
}
for a in appointments
]
}
@api_router.get("/admin/practitioner/{practitioner_id}")
async def admin_get_practitioner(practitioner_id: str, current_user: dict = Depends(verify_admin)):
"""Get a single practitioner's full details for admin editing"""
async with async_session() as session:
result = await session.execute(
text("""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE p.id = :id
"""),
{"id": practitioner_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
p_dict = dict(row._mapping)
practitioner = format_practitioner_row({
"id": p_dict["id"],
"user_id": p_dict["user_id"],
"specialty": p_dict["specialty"],
"bio": p_dict["bio"],
"photo_url": p_dict["photo_url"],
"country": p_dict["country"],
"region": p_dict["region"],
"department": p_dict["department"],
"address": p_dict["address"],
"city": p_dict["city"],
"postal_code": p_dict["postal_code"],
"consultation_types": p_dict["consultation_types"],
"price": p_dict["price"],
"consultation_duration": p_dict["consultation_duration"],
"is_verified": p_dict["is_verified"],
"years_experience": p_dict["years_experience"],
"diplomas": p_dict["diplomas"],
"detailed_description": p_dict["detailed_description"],
"languages": p_dict["languages"]
})
practitioner["user"] = user_to_response(format_user_row({
"id": p_dict["u_id"],
"email": p_dict["email"],
"first_name": p_dict["first_name"],
"last_name": p_dict["last_name"],
"phone": p_dict["phone"],
"role": p_dict["role"],
"created_at": p_dict["u_created_at"],
"google_calendar_connected": p_dict["google_calendar_connected"]
}))
return practitioner
class CreatePractitionerRequest(BaseModel):
email: str
password: str
first_name: str
last_name: str
phone: Optional[str] = None
specialty: Optional[str] = None
country: Optional[str] = None
city: Optional[str] = None
address: Optional[str] = None
postal_code: Optional[str] = None
price: Optional[str] = "0"
bio: Optional[str] = None
@api_router.post("/admin/practitioner/create")
async def admin_create_practitioner(request: CreatePractitionerRequest, current_user: dict = Depends(verify_admin)):
"""Create a new practitioner account from admin backoffice"""
user_id = str(uuid.uuid4())
practitioner_id = str(uuid.uuid4())
try:
# First, check if email exists and create user
async with async_session() as session:
# Check if email already exists
result = await session.execute(
text("SELECT id FROM users WHERE email = :email"),
{"email": request.email}
)
if result.fetchone():
raise HTTPException(status_code=400, detail="Un compte avec cet email existe déjà")
# Create user
password_hash = bcrypt.hashpw(request.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
await session.execute(
text("""
INSERT INTO users (id, email, password, first_name, last_name, phone, role, created_at)
VALUES (:id, :email, :password, :first_name, :last_name, :phone, 'practitioner', :created_at)
"""),
{
"id": user_id,
"email": request.email,
"password": password_hash,
"first_name": request.first_name,
"last_name": request.last_name,
"phone": request.phone or '',
"created_at": datetime.now(timezone.utc)
}
)
await session.commit()
# Then create practitioner in a separate transaction
async with async_session() as session:
price_value = float(request.price) if request.price else 0
await session.execute(
text("""
INSERT INTO practitioners (id, user_id, specialty, bio, country, city, address, postal_code, price, is_verified)
VALUES (:id, :user_id, :specialty, :bio, :country, :city, :address, :postal_code, :price, true)
"""),
{
"id": practitioner_id,
"user_id": user_id,
"specialty": request.specialty or '',
"bio": request.bio or '',
"country": request.country or '',
"city": request.city or '',
"address": request.address or '',
"postal_code": request.postal_code or '',
"price": price_value
}
)
await session.commit()
logger.info(f"Admin created new practitioner: {request.first_name} {request.last_name} ({request.email})")
return {
"message": "Praticien créé avec succès",
"practitioner_id": practitioner_id,
"user_id": user_id,
"email": request.email
}
except HTTPException:
raise
except Exception as e:
logger.error(f"Error creating practitioner: {str(e)}")
# If practitioner creation failed, try to delete the user we created
try:
async with async_session() as session:
await session.execute(text("DELETE FROM users WHERE id = :id"), {"id": user_id})
await session.commit()
except:
pass
raise HTTPException(status_code=500, detail=f"Erreur: {str(e)}")
@api_router.put("/admin/practitioner/{practitioner_id}")
async def admin_update_practitioner(practitioner_id: str, update_data: dict, current_user: dict = Depends(verify_admin)):
"""Update a practitioner's profile as admin"""
allowed_fields = [
"specialty", "bio", "photo_url", "country", "region", "department",
"address", "city", "postal_code", "consultation_types", "price",
"consultation_duration", "years_experience", "diplomas",
"detailed_description", "languages", "is_verified", "subscription_enabled"
]
update_dict = {}
for k, v in update_data.items():
if k in allowed_fields:
if k in ['consultation_types', 'diplomas', 'languages'] and v is not None:
update_dict[k] = json.dumps(v)
else:
update_dict[k] = v
async with async_session() as session:
result = await session.execute(
text("SELECT id FROM practitioners WHERE id = :id"),
{"id": practitioner_id}
)
if not result.fetchone():
raise HTTPException(status_code=404, detail="Praticien non trouvé")
if update_dict:
set_clause = ", ".join([f"{k} = :{k}" for k in update_dict])
update_dict["id"] = practitioner_id
await session.execute(
text(f"UPDATE practitioners SET {set_clause} WHERE id = :id"),
update_dict
)
await session.commit()
# Return updated practitioner
result = await session.execute(
text("""
SELECT p.*, u.id as u_id, u.email, u.first_name, u.last_name, u.phone,
u.role, u.created_at as u_created_at, u.google_calendar_connected
FROM practitioners p
JOIN users u ON p.user_id = u.id
WHERE p.id = :id
"""),
{"id": practitioner_id}
)
row = result.fetchone()
p_dict = dict(row._mapping)
practitioner = format_practitioner_row({
"id": p_dict["id"],
"user_id": p_dict["user_id"],
"specialty": p_dict["specialty"],
"bio": p_dict["bio"],
"photo_url": p_dict["photo_url"],
"country": p_dict["country"],
"region": p_dict["region"],
"department": p_dict["department"],
"address": p_dict["address"],
"city": p_dict["city"],
"postal_code": p_dict["postal_code"],
"consultation_types": p_dict["consultation_types"],
"price": p_dict["price"],
"consultation_duration": p_dict["consultation_duration"],
"is_verified": p_dict["is_verified"],
"years_experience": p_dict["years_experience"],
"diplomas": p_dict["diplomas"],
"detailed_description": p_dict["detailed_description"],
"languages": p_dict["languages"]
})
practitioner["user"] = user_to_response(format_user_row({
"id": p_dict["u_id"],
"email": p_dict["email"],
"first_name": p_dict["first_name"],
"last_name": p_dict["last_name"],
"phone": p_dict["phone"],
"role": p_dict["role"],
"created_at": p_dict["u_created_at"],
"google_calendar_connected": p_dict["google_calendar_connected"]
}))
return practitioner
@api_router.post("/admin/practitioner/{practitioner_id}/upload-photo")
async def admin_upload_practitioner_photo(
practitioner_id: str,
file: UploadFile = File(...),
current_user: dict = Depends(verify_admin)
):
"""Upload a photo for a practitioner as admin"""
async with async_session() as session:
result = await session.execute(
text("SELECT photo_url FROM practitioners WHERE id = :id"),
{"id": practitioner_id}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Praticien non trouvé")
file_ext = Path(file.filename).suffix.lower()
if file_ext not in ALLOWED_EXTENSIONS:
raise HTTPException(status_code=400, detail=f"Format non autorisé. Utilisez: {', '.join(ALLOWED_EXTENSIONS)}")
contents = await file.read()
if len(contents) > MAX_FILE_SIZE:
raise HTTPException(status_code=400, detail="Fichier trop volumineux (max 5 MB)")
unique_filename = f"{practitioner_id}_{uuid.uuid4().hex[:8]}{file_ext}"
file_path = UPLOAD_DIR / unique_filename
if row[0] and "/uploads/photos/" in str(row[0]):
old_filename = row[0].split("/uploads/photos/")[-1]
old_path = UPLOAD_DIR / old_filename
if old_path.exists():
old_path.unlink()
with open(file_path, "wb") as f:
f.write(contents)
photo_url = f"/api/photos/{unique_filename}"
await session.execute(
text("UPDATE practitioners SET photo_url = :photo_url WHERE id = :id"),
{"photo_url": photo_url, "id": practitioner_id}
)
await session.commit()
return {"photo_url": photo_url, "message": "Photo uploadée avec succès"}
@api_router.get("/admin/stats/overview")
async def admin_get_overview_stats(current_user: dict = Depends(verify_admin)):
"""Get overall platform statistics for admin"""
async with async_session() as session:
# Count practitioners
result = await session.execute(text("SELECT COUNT(*) FROM practitioners"))
total_practitioners = result.scalar()
# Count clients
result = await session.execute(text("SELECT COUNT(*) FROM users WHERE role = 'client'"))
total_clients = result.scalar()
# Count appointments
result = await session.execute(text("SELECT COUNT(*) FROM appointments"))
total_appointments = result.scalar()
result = await session.execute(text("SELECT COUNT(*) FROM appointments WHERE status = 'completed'"))
completed_appointments = result.scalar()
# Revenue from paid appointments
result = await session.execute(
text("""
SELECT SUM(p.price)
FROM appointments a
JOIN practitioners p ON a.practitioner_id = p.id
WHERE a.paid = TRUE
""")
)
total_revenue = result.scalar() or 0
return {
"practitioners": total_practitioners,
"clients": total_clients,
"appointments": {
"total": total_appointments,
"completed": completed_appointments
},
"revenue": {
"total": float(total_revenue),
"currency": "EUR"
}
}
# ============== SMS SERVICE (TWILIO) ==============
async def get_sms_enabled() -> bool:
"""Check if SMS reminders are enabled in admin settings"""
async with async_session() as session:
result = await session.execute(
text("SELECT value FROM admin_settings WHERE key = 'sms_reminders_enabled'")
)
row = result.fetchone()
return row and row[0] == 'true'
async def send_sms(phone_number: str, message: str) -> tuple:
"""Send SMS using Twilio"""
if not TWILIO_AVAILABLE:
logger.warning("Twilio library not installed")
return False, "Twilio library not installed"
if not all([TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_PHONE_NUMBER]):
logger.warning("Twilio credentials not configured")
return False, "Twilio credentials not configured"
try:
client = TwilioClient(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)
sms = client.messages.create(
body=message,
from_=TWILIO_PHONE_NUMBER,
to=phone_number
)
logger.info(f"SMS sent to {phone_number}: {sms.sid}")
return True, sms.sid
except Exception as e:
logger.error(f"SMS error: {str(e)}")
return False, str(e)
def format_phone_for_sms(phone: str, country: str = "FR") -> str:
"""Format phone number for SMS (E.164 format)"""
if not phone:
return None
# Remove spaces, dots, dashes
phone = phone.replace(" ", "").replace(".", "").replace("-", "")
# If already in international format
if phone.startswith("+"):
return phone
# Convert French numbers
if country == "FR":
if phone.startswith("0"):
return "+33" + phone[1:]
elif phone.startswith("33"):
return "+" + phone
return phone
async def send_appointment_reminder_sms(appointment_id: str):
"""Send SMS reminder for an appointment"""
sms_enabled = await get_sms_enabled()
if not sms_enabled:
logger.info("SMS reminders disabled - skipping")
return
async with async_session() as session:
# Get appointment with client and practitioner info
result = await session.execute(
text("""
SELECT a.*,
c.first_name as client_first_name, c.last_name as client_last_name, c.phone as client_phone,
p.specialty, p.country,
pu.first_name as pract_first_name, pu.last_name as pract_last_name
FROM appointments a
JOIN users c ON a.client_id = c.id
JOIN practitioners p ON a.practitioner_id = p.id
JOIN users pu ON p.user_id = pu.id
WHERE a.id = :id
"""),
{"id": appointment_id}
)
row = result.fetchone()
if not row:
return
apt = dict(row._mapping)
if not apt.get('client_phone'):
logger.info(f"No phone number for client - skipping SMS for appointment {appointment_id}")
return
phone = format_phone_for_sms(apt['client_phone'], apt.get('country', 'FR'))
if not phone:
return
# Format message
consultation_type = "en cabinet" if apt['consultation_type'] == 'cabinet' else "en visio"
message = (
f"Rappel Opora Vie: Votre RDV avec {apt['pract_first_name']} {apt['pract_last_name']} "
f"({apt['specialty']}) est prévu demain {apt['date']} à {apt['start_time']} ({consultation_type}). "
f"À bientôt !"
)
# Send SMS
success, result_msg = await send_sms(phone, message)
# Log the SMS
sms_id = str(uuid.uuid4())
await session.execute(
text("""
INSERT INTO sms_reminders (id, appointment_id, phone_number, message, status, sent_at, error_message, created_at)
VALUES (:id, :appointment_id, :phone_number, :message, :status, :sent_at, :error_message, NOW())
"""),
{
"id": sms_id,
"appointment_id": appointment_id,
"phone_number": phone,
"message": message,
"status": "sent" if success else "failed",
"sent_at": datetime.now(timezone.utc) if success else None,
"error_message": None if success else result_msg
}
)
await session.commit()
async def process_sms_reminders():
"""Process SMS reminders for appointments happening in 24 hours"""
sms_enabled = await get_sms_enabled()
if not sms_enabled:
return
tomorrow = (datetime.now(timezone.utc) + timedelta(days=1)).strftime("%Y-%m-%d")
async with async_session() as session:
# Get appointments for tomorrow that haven't been reminded yet
result = await session.execute(
text("""
SELECT a.id FROM appointments a
LEFT JOIN sms_reminders sr ON a.id = sr.appointment_id
WHERE a.date = :tomorrow
AND a.status IN ('pending', 'confirmed')
AND sr.id IS NULL
"""),
{"tomorrow": tomorrow}
)
rows = result.fetchall()
for row in rows:
await send_appointment_reminder_sms(row[0])
# ============== ADMIN SETTINGS ROUTES ==============
@api_router.get("/admin/settings")
async def admin_get_settings(current_user: dict = Depends(verify_admin)):
"""Get all admin settings"""
async with async_session() as session:
result = await session.execute(
text("SELECT key, value FROM admin_settings")
)
rows = result.fetchall()
settings = {}
for row in rows:
settings[row[0]] = row[1]
# Add Twilio configuration status
settings['twilio_configured'] = bool(TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN and TWILIO_PHONE_NUMBER)
return settings
@api_router.put("/admin/settings/{key}")
async def admin_update_setting(key: str, data: dict, current_user: dict = Depends(verify_admin)):
"""Update an admin setting"""
allowed_keys = ['sms_reminders_enabled']
if key not in allowed_keys:
raise HTTPException(status_code=400, detail=f"Invalid setting key. Allowed: {allowed_keys}")
value = data.get('value')
if value is None:
raise HTTPException(status_code=400, detail="Value is required")
async with async_session() as session:
await session.execute(
text("""
INSERT INTO admin_settings (key, value, updated_at)
VALUES (:key, :value, NOW())
ON CONFLICT (key) DO UPDATE SET value = :value, updated_at = NOW()
"""),
{"key": key, "value": str(value).lower()}
)
await session.commit()
return {"message": f"Setting {key} updated", "value": value}
@api_router.get("/admin/sms-logs")
async def admin_get_sms_logs(
limit: int = Query(50, ge=1, le=200),
current_user: dict = Depends(verify_admin)
):
"""Get SMS reminder logs"""
async with async_session() as session:
result = await session.execute(
text("""
SELECT sr.*, a.date, a.start_time,
u.first_name, u.last_name
FROM sms_reminders sr
JOIN appointments a ON sr.appointment_id = a.id
JOIN users u ON a.client_id = u.id
ORDER BY sr.created_at DESC
LIMIT :limit
"""),
{"limit": limit}
)
rows = result.fetchall()
logs = []
for row in rows:
r = dict(row._mapping)
if r.get('created_at'):
r['created_at'] = r['created_at'].isoformat() if hasattr(r['created_at'], 'isoformat') else str(r['created_at'])
if r.get('sent_at'):
r['sent_at'] = r['sent_at'].isoformat() if hasattr(r['sent_at'], 'isoformat') else str(r['sent_at'])
logs.append(r)
return logs
@api_router.post("/admin/sms-reminders/trigger")
async def admin_trigger_sms_reminders(current_user: dict = Depends(verify_admin)):
"""Manually trigger SMS reminders for tomorrow's appointments"""
sms_enabled = await get_sms_enabled()
if not sms_enabled:
raise HTTPException(status_code=400, detail="SMS reminders are disabled. Enable them in settings first.")
if not all([TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_PHONE_NUMBER]):
raise HTTPException(status_code=503, detail="Twilio credentials not configured in .env")
# Process reminders in background
asyncio.create_task(process_sms_reminders())
return {"message": "SMS reminders processing started for tomorrow's appointments"}
@api_router.post("/admin/sms-test")
async def admin_test_sms(data: dict, current_user: dict = Depends(verify_admin)):
"""Send a test SMS"""
phone = data.get('phone')
if not phone:
raise HTTPException(status_code=400, detail="Phone number is required")
if not all([TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_PHONE_NUMBER]):
raise HTTPException(status_code=503, detail="Twilio credentials not configured in .env")
formatted_phone = format_phone_for_sms(phone, "FR")
message = "Test Opora Vie: Si vous recevez ce message, les SMS de rappel fonctionnent correctement !"
success, result_msg = await send_sms(formatted_phone, message)
if success:
return {"success": True, "message": f"SMS envoyé à {formatted_phone}", "sid": result_msg}
else:
raise HTTPException(status_code=500, detail=f"Erreur d'envoi: {result_msg}")
# ============== BLOG ARTICLES API ==============
@api_router.get("/blog/articles")
async def get_blog_articles(category: Optional[str] = None, published_only: bool = True):
"""Get all blog articles"""
async with async_session() as session:
query = "SELECT * FROM blog_articles"
params = {}
conditions = []
if published_only:
conditions.append("is_published = TRUE")
if category and category != 'all':
conditions.append("category = :category")
params["category"] = category
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += " ORDER BY created_at DESC"
result = await session.execute(text(query), params)
rows = result.fetchall()
articles = []
for row in rows:
article = dict(row._mapping)
# Format dates
if article.get("created_at"):
article["created_at"] = str(article["created_at"])
if article.get("updated_at"):
article["updated_at"] = str(article["updated_at"])
articles.append(article)
return articles
@api_router.get("/blog/articles/{slug}")
async def get_blog_article(slug: str):
"""Get a single blog article by slug"""
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM blog_articles WHERE slug = :slug"),
{"slug": slug}
)
row = result.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Article non trouvé")
article = dict(row._mapping)
if article.get("created_at"):
article["created_at"] = str(article["created_at"])
if article.get("updated_at"):
article["updated_at"] = str(article["updated_at"])
return article
@api_router.post("/admin/blog/articles")
async def create_blog_article(article: BlogArticleCreate, current_user: dict = Depends(verify_admin)):
"""Create a new blog article (admin only)"""
article_id = str(uuid.uuid4())
async with async_session() as session:
# Check if slug already exists
result = await session.execute(
text("SELECT id FROM blog_articles WHERE slug = :slug"),
{"slug": article.slug}
)
if result.fetchone():
raise HTTPException(status_code=400, detail="Un article avec ce slug existe déjà")
await session.execute(
text("""
INSERT INTO blog_articles (
id, slug, category, author, read_time, image_url,
title_fr, title_en, title_uk, title_ru, title_it, title_ro,
excerpt_fr, excerpt_en, excerpt_uk, excerpt_ru, excerpt_it, excerpt_ro,
content_fr, content_en, content_uk, content_ru, content_it, content_ro,
is_published
) VALUES (
:id, :slug, :category, :author, :read_time, :image_url,
:title_fr, :title_en, :title_uk, :title_ru, :title_it, :title_ro,
:excerpt_fr, :excerpt_en, :excerpt_uk, :excerpt_ru, :excerpt_it, :excerpt_ro,
:content_fr, :content_en, :content_uk, :content_ru, :content_it, :content_ro,
:is_published
)
"""),
{
"id": article_id,
"slug": article.slug,
"category": article.category,
"author": article.author,
"read_time": article.read_time,
"image_url": article.image_url,
"title_fr": article.title_fr,
"title_en": article.title_en,
"title_uk": article.title_uk,
"title_ru": article.title_ru,
"title_it": article.title_it,
"title_ro": article.title_ro,
"excerpt_fr": article.excerpt_fr,
"excerpt_en": article.excerpt_en,
"excerpt_uk": article.excerpt_uk,
"excerpt_ru": article.excerpt_ru,
"excerpt_it": article.excerpt_it,
"excerpt_ro": article.excerpt_ro,
"content_fr": article.content_fr,
"content_en": article.content_en,
"content_uk": article.content_uk,
"content_ru": article.content_ru,
"content_it": article.content_it,
"content_ro": article.content_ro,
"is_published": article.is_published
}
)
await session.commit()
return {"id": article_id, "message": "Article créé avec succès"}
@api_router.put("/admin/blog/articles/{article_id}")
async def update_blog_article(article_id: str, update_data: dict, current_user: dict = Depends(verify_admin)):
"""Update a blog article (admin only)"""
allowed_fields = [
"slug", "category", "author", "read_time", "image_url",
"title_fr", "title_en", "title_uk", "title_ru", "title_it", "title_ro",
"excerpt_fr", "excerpt_en", "excerpt_uk", "excerpt_ru", "excerpt_it", "excerpt_ro",
"content_fr", "content_en", "content_uk", "content_ru", "content_it", "content_ro",
"is_published"
]
update_dict = {k: v for k, v in update_data.items() if k in allowed_fields}
if not update_dict:
raise HTTPException(status_code=400, detail="Aucun champ valide à mettre à jour")
async with async_session() as session:
result = await session.execute(
text("SELECT id FROM blog_articles WHERE id = :id"),
{"id": article_id}
)
if not result.fetchone():
raise HTTPException(status_code=404, detail="Article non trouvé")
# Check for slug conflict if updating slug
if "slug" in update_dict:
result = await session.execute(
text("SELECT id FROM blog_articles WHERE slug = :slug AND id != :id"),
{"slug": update_dict["slug"], "id": article_id}
)
if result.fetchone():
raise HTTPException(status_code=400, detail="Un article avec ce slug existe déjà")
set_clause = ", ".join([f"{k} = :{k}" for k in update_dict])
set_clause += ", updated_at = NOW()"
update_dict["id"] = article_id
await session.execute(
text(f"UPDATE blog_articles SET {set_clause} WHERE id = :id"),
update_dict
)
await session.commit()
return {"message": "Article mis à jour avec succès"}
@api_router.delete("/admin/blog/articles/{article_id}")
async def delete_blog_article(article_id: str, current_user: dict = Depends(verify_admin)):
"""Delete a blog article (admin only)"""
async with async_session() as session:
result = await session.execute(
text("SELECT id FROM blog_articles WHERE id = :id"),
{"id": article_id}
)
if not result.fetchone():
raise HTTPException(status_code=404, detail="Article non trouvé")
await session.execute(
text("DELETE FROM blog_articles WHERE id = :id"),
{"id": article_id}
)
await session.commit()
return {"message": "Article supprimé avec succès"}
@api_router.get("/admin/blog/articles")
async def admin_get_all_articles(current_user: dict = Depends(verify_admin)):
"""Get all blog articles for admin (including unpublished)"""
async with async_session() as session:
result = await session.execute(
text("SELECT * FROM blog_articles ORDER BY created_at DESC")
)
rows = result.fetchall()
articles = []
for row in rows:
article = dict(row._mapping)
if article.get("created_at"):
article["created_at"] = str(article["created_at"])
if article.get("updated_at"):
article["updated_at"] = str(article["updated_at"])
articles.append(article)
return articles
# Include router
app.include_router(api_router)
# Mount static files for uploaded photos
app.mount("/uploads", StaticFiles(directory=str(UPLOAD_DIR.parent)), name="uploads")
app.add_middleware(
CORSMiddleware,
allow_credentials=True,
allow_origins=os.environ.get('CORS_ORIGINS', '*').split(','),
allow_methods=["*"],
allow_headers=["*"],
)
@app.on_event("startup")
async def startup_event():
await init_db()
logger.info("Application started - PostgreSQL database initialized")
@app.on_event("shutdown")
async def shutdown_event():
await engine.dispose()
logger.info("Application shutdown - Database connections closed")