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
async def send_appointment_notification(appointment: dict, practitioner: dict, client: dict, is_reminder: bool = False):
"""Send appointment email notification to client"""
subject, html = generate_appointment_email(appointment, practitioner, client, is_reminder)
# Handle both dict and UserResponse object for client
client_email = client.email if hasattr(client, 'email') else client['email']
await send_email(client_email, subject, html)
# ============== VIDEO CONFERENCE SERVICE (DAILY.CO / JITSI) ==============
async def create_video_room(appointment_id: str) -> Optional[str]:
"""Create a video room for the appointment"""
if DAILY_API_KEY and DAILY_DOMAIN:
try:
room_name = f"soins-{appointment_id[:8]}"
async with httpx.AsyncClient() as http_client:
response = await http_client.post(
"https://api.daily.co/v1/rooms",
headers={
"Authorization": f"Bearer {DAILY_API_KEY}",
"Content-Type": "application/json"
},
json={
"name": room_name,
"privacy": "private",
"properties": {
"exp": int((datetime.now(timezone.utc) + timedelta(hours=24)).timestamp()),
"enable_screenshare": True,
"enable_chat": True,
"max_participants": 2
}
},
timeout=10.0
)
if response.status_code == 200:
room_url = f"https://{DAILY_DOMAIN}.daily.co/{room_name}"
logger.info(f"Daily.co room created: {room_url}")
return room_url
else:
logger.error(f"Daily.co room creation failed: {response.text}")
except Exception as e:
logger.error(f"Daily.co error: {str(e)}")
# Fallback to Jitsi (free, no API key needed)
jitsi_room = f"opora-vie-{appointment_id[:12]}"
jitsi_url = f"https://meet.jit.si/{jitsi_room}"
logger.info(f"Using Jitsi fallback: {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://apora.net')}/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", "")
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}
LIMIT 100
"""),
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 "/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"/uploads/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"}
# ============== 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": "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"}
]
# ============== 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.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
@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"
]
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"/uploads/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}")
# 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")