from typing import Literal from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.orm import Session, joinedload from .auth import ADMIN_COOKIE_NAME, create_admin_token, current_admin from .config import settings from .database import get_db from .dlt_service import refresh_dlt_application_status, serialize_dlt from .kyc_service import refresh_application_status, serialize_kyc from .routes_messaging import serialize_sender, serialize_template from .models import ( DltApplication, DltDocument, KycApplication, KycDocument, PhoneNumber, SenderConfig, TemplateConfig, UrmiOtpProject, UserAccount, WhatsAppMessage, ) from .phone_number_validators import normalize_phone_number, validate_per_min_price, validate_price from .routes_urmiotp import serialize_project from .whatsapp_config import ( resolved_whatsapp_app_secret, resolved_whatsapp_business_account_id, resolved_whatsapp_phone_number_id, whatsapp_ready_to_send, ) from .whatsapp_graph_client import WhatsAppGraphClient, WhatsAppGraphError from .whatsapp_messages import ( extract_wa_message_id, normalize_wa_number, record_outbound_message, serialize_message, ) from .platform_settings import set_setting router = APIRouter(prefix="/api/v1/admin", tags=["admin"]) class AdminLoginRequest(BaseModel): username: str password: str class DocumentReviewRequest(BaseModel): status: str = Field(pattern="^(approved|rejected)$") review_note: str | None = Field(default=None, max_length=500) class ProjectReviewRequest(BaseModel): status: str = Field(pattern="^(approved|rejected)$") review_note: str | None = Field(default=None, max_length=500) class CreatePhoneNumberRequest(BaseModel): phone_number: str = Field(min_length=10, max_length=10) monthly_number_price: float = Field(ge=0) inbound_price: float = Field(ge=0) outbound_price: float = Field(ge=0) user_email: str | None = Field(default=None, max_length=320) class UpdatePhoneNumberRequest(BaseModel): monthly_number_price: float | None = Field(default=None, ge=0) inbound_price: float | None = Field(default=None, ge=0) outbound_price: float | None = Field(default=None, ge=0) user_email: str | None = Field(default=None, max_length=320) release: bool = False def set_admin_session(response: Response) -> None: response.set_cookie( ADMIN_COOKIE_NAME, create_admin_token(settings.admin_username), httponly=True, secure=settings.auth_cookie_secure, samesite="lax", max_age=settings.access_token_minutes * 60, path="/", ) @router.post("/login") def admin_login(payload: AdminLoginRequest, response: Response) -> dict[str, str]: if payload.username != settings.admin_username or payload.password != settings.admin_password: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid admin credentials") set_admin_session(response) return {"username": settings.admin_username} @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) def admin_logout(response: Response) -> None: response.delete_cookie(ADMIN_COOKIE_NAME, path="/") @router.get("/me") def admin_me(username: str = Depends(current_admin)) -> dict[str, str]: return {"username": username} @router.get("/compliance/kyc") def list_kyc_applications( db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> list[dict]: applications = db.scalars( select(KycApplication) .options(joinedload(KycApplication.user), joinedload(KycApplication.documents)) .order_by(KycApplication.updated_at.desc()) ).unique().all() return [ { **serialize_kyc(app), "user": { "id": str(app.user.id), "name": app.user.display_name, "email": app.user.email, }, } for app in applications if app.status != "not_started" or app.pan_number or any(doc.status != "missing" for doc in app.documents) ] @router.patch("/compliance/kyc/{application_id}/documents/{document_id}") def review_document( application_id: UUID, document_id: UUID, payload: DocumentReviewRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: application = db.scalar( select(KycApplication) .options(joinedload(KycApplication.user), joinedload(KycApplication.documents)) .where(KycApplication.id == application_id) ) if not application: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="KYC application not found") document = db.get(KycDocument, document_id) if not document or document.application_id != application.id: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document not found") if document.status == "missing": raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Document not uploaded yet") document.status = payload.status document.review_note = payload.review_note document.reviewed_at = datetime.now(timezone.utc) refresh_application_status(application) db.commit() db.refresh(application) return { **serialize_kyc(application), "user": { "id": str(application.user.id), "name": application.user.display_name, "email": application.user.email, }, } @router.get("/compliance/dlt") def list_dlt_applications( db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> list[dict]: applications = db.scalars( select(DltApplication) .options(joinedload(DltApplication.user), joinedload(DltApplication.documents)) .order_by(DltApplication.updated_at.desc()) ).unique().all() return [ { **serialize_dlt(app), "user": { "id": str(app.user.id), "name": app.user.display_name, "email": app.user.email, }, } for app in applications if app.status != "not_started" or any(doc.status != "missing" for doc in app.documents) ] @router.patch("/compliance/dlt/{application_id}/documents/{document_id}") def review_dlt_document( application_id: UUID, document_id: UUID, payload: DocumentReviewRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: application = db.scalar( select(DltApplication) .options(joinedload(DltApplication.user), joinedload(DltApplication.documents)) .where(DltApplication.id == application_id) ) if not application: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="DLT application not found") document = db.get(DltDocument, document_id) if not document or document.application_id != application.id: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document not found") if document.status == "missing": raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Document not uploaded yet") document.status = payload.status document.review_note = payload.review_note document.reviewed_at = datetime.now(timezone.utc) refresh_dlt_application_status(application) db.commit() db.refresh(application) return { **serialize_dlt(application), "user": { "id": str(application.user.id), "name": application.user.display_name, "email": application.user.email, }, } @router.get("/compliance/senders") def list_sender_configs( db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> list[dict]: senders = db.scalars( select(SenderConfig) .options(joinedload(SenderConfig.user)) .order_by(SenderConfig.created_at.desc()) ).unique().all() return [ { **serialize_sender(sender), "user": { "id": str(sender.user.id), "name": sender.user.display_name, "email": sender.user.email, }, } for sender in senders ] @router.patch("/compliance/senders/{sender_id}") def review_sender( sender_id: UUID, payload: DocumentReviewRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: sender = db.scalar( select(SenderConfig) .options(joinedload(SenderConfig.user)) .where(SenderConfig.id == sender_id) ) if not sender: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Sender config not found") if sender.status != "pending": raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Sender already reviewed") sender.status = payload.status sender.review_note = payload.review_note sender.reviewed_at = datetime.now(timezone.utc) db.commit() db.refresh(sender) return { **serialize_sender(sender), "user": { "id": str(sender.user.id), "name": sender.user.display_name, "email": sender.user.email, }, } @router.get("/compliance/templates") def list_template_configs( db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> list[dict]: templates = db.scalars( select(TemplateConfig) .options(joinedload(TemplateConfig.user), joinedload(TemplateConfig.sender_config)) .order_by(TemplateConfig.created_at.desc()) ).unique().all() return [ { **serialize_template(template), "user": { "id": str(template.user.id), "name": template.user.display_name, "email": template.user.email, }, } for template in templates ] @router.patch("/compliance/templates/{template_id}") def review_template( template_id: UUID, payload: DocumentReviewRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: template = db.scalar( select(TemplateConfig) .options(joinedload(TemplateConfig.user), joinedload(TemplateConfig.sender_config)) .where(TemplateConfig.id == template_id) ) if not template: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template config not found") if template.status != "pending": raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Template already reviewed") template.status = payload.status template.review_note = payload.review_note template.reviewed_at = datetime.now(timezone.utc) db.commit() db.refresh(template) return { **serialize_template(template), "user": { "id": str(template.user.id), "name": template.user.display_name, "email": template.user.email, }, } @router.get("/compliance/urmiotp-projects") def list_urmiotp_projects( db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> list[dict]: projects = db.scalars( select(UrmiOtpProject) .options(joinedload(UrmiOtpProject.user)) .order_by(UrmiOtpProject.created_at.desc()) ).all() return [ { **serialize_project(project), "user": { "id": str(project.user.id), "name": project.user.display_name, "email": project.user.email, }, } for project in projects ] @router.patch("/compliance/urmiotp-projects/{project_id}") def review_urmiotp_project( project_id: UUID, payload: ProjectReviewRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: project = db.scalar( select(UrmiOtpProject) .options(joinedload(UrmiOtpProject.user)) .where(UrmiOtpProject.id == project_id) ) if not project: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="UrmiOTP project not found") if project.status != "pending": raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Project already reviewed") project.status = payload.status project.review_note = payload.review_note project.reviewed_at = datetime.now(timezone.utc) db.commit() db.refresh(project) return { **serialize_project(project), "user": { "id": str(project.user.id), "name": project.user.display_name, "email": project.user.email, }, } def serialize_admin_phone_number(number: PhoneNumber) -> dict: return { "id": str(number.id), "phone_number": number.phone_number, "monthly_number_price": float(number.monthly_number_price), "inbound_price": float(number.inbound_price), "outbound_price": float(number.outbound_price), "currency": number.currency, "status": number.status, "acquisition_type": number.acquisition_type, "assigned_at": number.assigned_at, "created_at": number.created_at, "user": ( { "id": str(number.user.id), "name": number.user.display_name, "email": number.user.email, } if number.user else None ), } @router.get("/phone-numbers") def list_phone_numbers( db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> list[dict]: numbers = db.scalars( select(PhoneNumber) .options(joinedload(PhoneNumber.user)) .order_by(PhoneNumber.created_at.desc()) ).all() return [serialize_admin_phone_number(number) for number in numbers] @router.post("/phone-numbers", status_code=status.HTTP_201_CREATED) def create_phone_number( payload: CreatePhoneNumberRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: try: phone_number = normalize_phone_number(payload.phone_number) monthly_number_price = validate_price(payload.monthly_number_price, "Number monthly price") inbound_price = validate_per_min_price(payload.inbound_price, "Inbound price per min") outbound_price = validate_per_min_price(payload.outbound_price, "Outbound price per min") except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc existing = db.scalar(select(PhoneNumber).where(PhoneNumber.phone_number == phone_number)) if existing: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="This phone number already exists") user = None if payload.user_email: user = db.scalar(select(UserAccount).where(UserAccount.email == payload.user_email.strip().lower())) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="No user account found for that email", ) now = datetime.now(timezone.utc) number = PhoneNumber( phone_number=phone_number, inbound_price=inbound_price, outbound_price=outbound_price, monthly_number_price=monthly_number_price, user_id=user.id if user else None, status="assigned" if user else "available", acquisition_type="admin_assigned" if user else None, assigned_at=now if user else None, ) db.add(number) db.commit() db.refresh(number) if user: number.user = user return serialize_admin_phone_number(number) @router.patch("/phone-numbers/{number_id}") def update_phone_number( number_id: UUID, payload: UpdatePhoneNumberRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: number = db.scalar( select(PhoneNumber) .options(joinedload(PhoneNumber.user)) .where(PhoneNumber.id == number_id) ) if not number: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Phone number not found") if payload.release: number.user_id = None number.status = "available" number.acquisition_type = None number.assigned_at = None number.user = None elif payload.user_email is not None: email = payload.user_email.strip().lower() if not email: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="User email cannot be empty", ) user = db.scalar(select(UserAccount).where(UserAccount.email == email)) if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No user account found for that email") number.user_id = user.id number.status = "assigned" number.acquisition_type = "admin_assigned" number.assigned_at = datetime.now(timezone.utc) number.user = user if payload.monthly_number_price is not None: try: number.monthly_number_price = validate_price(payload.monthly_number_price, "Number monthly price") except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc if payload.inbound_price is not None: try: number.inbound_price = validate_per_min_price(payload.inbound_price, "Inbound price per min") except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc if payload.outbound_price is not None: try: number.outbound_price = validate_per_min_price(payload.outbound_price, "Outbound price per min") except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc db.commit() db.refresh(number) return serialize_admin_phone_number(number) class AdminWhatsAppSendRequest(BaseModel): to: str = Field(min_length=8, max_length=20) message_type: Literal[ "text", "template", "image", "document", "audio", "video", "interactive_buttons", "interactive_list", "reaction", ] = "text" text: str | None = Field(default=None, max_length=4096) reply_to: str | None = None template_name: str | None = None template_language: str = "en" template_components: list[dict] | None = None media_link: str | None = None media_caption: str | None = None media_filename: str | None = None interactive_body: str | None = None interactive_header: str | None = None interactive_footer: str | None = None interactive_buttons: list[dict[str, str]] | None = None interactive_list_button: str | None = None interactive_list_sections: list[dict] | None = None reaction_message_id: str | None = None reaction_emoji: str | None = None class AdminWhatsAppConfigUpdate(BaseModel): phone_number_id: str | None = None business_account_id: str | None = None app_secret: str | None = None def _whatsapp_client(db: Session) -> WhatsAppGraphClient: return WhatsAppGraphClient( phone_number_id=resolved_whatsapp_phone_number_id(db), business_account_id=resolved_whatsapp_business_account_id(db), ) @router.get("/whatsapp/config") def admin_whatsapp_config(db: Session = Depends(get_db), _: str = Depends(current_admin)) -> dict: return { "business_phone": settings.whatsapp_business_phone, "phone_number_id": resolved_whatsapp_phone_number_id(db) or None, "business_account_id": resolved_whatsapp_business_account_id(db) or None, "meta_app_id": settings.meta_app_id or None, "access_token_configured": bool(settings.whatsapp_access_token), "app_secret_configured": bool(resolved_whatsapp_app_secret(db)), "webhook_url": settings.whatsapp_webhook_url, "verify_token_configured": bool(settings.whatsapp_verify_token), "ready_to_send": whatsapp_ready_to_send(db), "supported_operations": [ "text", "template", "image", "document", "audio", "video", "interactive_buttons", "interactive_list", "reaction", "mark_read", "media_download", "list_templates", ], } @router.patch("/whatsapp/config") def admin_update_whatsapp_config( payload: AdminWhatsAppConfigUpdate, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: if payload.phone_number_id is not None: set_setting(db, "whatsapp_phone_number_id", payload.phone_number_id.strip()) if payload.business_account_id is not None: set_setting(db, "whatsapp_business_account_id", payload.business_account_id.strip()) if payload.app_secret is not None: set_setting(db, "whatsapp_app_secret", payload.app_secret.strip()) return admin_whatsapp_config(db=db, _=_) @router.get("/whatsapp/messages") def admin_list_whatsapp_messages( limit: int = 50, peer: str | None = None, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> list[dict]: query = select(WhatsAppMessage).order_by(WhatsAppMessage.created_at.desc()).limit(min(limit, 500)) if peer: try: normalized_peer = normalize_wa_number(peer) except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc query = query.where( (WhatsAppMessage.from_number == normalized_peer) | (WhatsAppMessage.to_number == normalized_peer) ) messages = db.scalars(query).all() return [serialize_message(message) for message in messages] @router.get("/whatsapp/templates") async def admin_list_whatsapp_templates( db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: client = _whatsapp_client(db) try: return await client.list_templates() except WhatsAppGraphError as exc: raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc @router.get("/whatsapp/media/{media_id}") async def admin_download_whatsapp_media( media_id: str, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> Response: client = _whatsapp_client(db) try: content, mime_type = await client.download_media(media_id) except WhatsAppGraphError as exc: raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc return Response(content=content, media_type=mime_type) @router.post("/whatsapp/messages/{wa_message_id}/read") async def admin_mark_whatsapp_read( wa_message_id: str, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: client = _whatsapp_client(db) try: result = await client.mark_read(wa_message_id) except WhatsAppGraphError as exc: raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc return result @router.post("/whatsapp/messages/send") async def admin_send_whatsapp_message( payload: AdminWhatsAppSendRequest, db: Session = Depends(get_db), _: str = Depends(current_admin), ) -> dict: try: recipient = normalize_wa_number(payload.to) except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc client = _whatsapp_client(db) message_type = payload.message_type try: if message_type == "text": if not payload.text: raise HTTPException(status_code=422, detail="text is required") result = await client.send_text(recipient, payload.text.strip(), reply_to=payload.reply_to) body = payload.text.strip() elif message_type == "template": if not payload.template_name: raise HTTPException(status_code=422, detail="template_name is required") result = await client.send_template( recipient, payload.template_name, language_code=payload.template_language, components=payload.template_components, ) body = f"Template: {payload.template_name}" elif message_type in {"image", "document", "audio", "video"}: if not payload.media_link: raise HTTPException(status_code=422, detail="media_link is required") result = await client.send_media( recipient, message_type, link=payload.media_link, caption=payload.media_caption, filename=payload.media_filename, reply_to=payload.reply_to, ) body = payload.media_caption or f"[{message_type}]" elif message_type == "interactive_buttons": if not payload.interactive_body or not payload.interactive_buttons: raise HTTPException(status_code=422, detail="interactive_body and interactive_buttons are required") result = await client.send_interactive_buttons( recipient, payload.interactive_body, payload.interactive_buttons, header=payload.interactive_header, footer=payload.interactive_footer, reply_to=payload.reply_to, ) body = payload.interactive_body message_type = "interactive" elif message_type == "interactive_list": if not payload.interactive_body or not payload.interactive_list_button or not payload.interactive_list_sections: raise HTTPException( status_code=422, detail="interactive_body, interactive_list_button, and interactive_list_sections are required", ) result = await client.send_interactive_list( recipient, payload.interactive_body, payload.interactive_list_button, payload.interactive_list_sections, header=payload.interactive_header, footer=payload.interactive_footer, reply_to=payload.reply_to, ) body = payload.interactive_body message_type = "interactive" elif message_type == "reaction": if not payload.reaction_message_id or not payload.reaction_emoji: raise HTTPException(status_code=422, detail="reaction_message_id and reaction_emoji are required") result = await client.send_reaction(recipient, payload.reaction_message_id, payload.reaction_emoji) body = payload.reaction_emoji else: raise HTTPException(status_code=422, detail="Unsupported message type") except WhatsAppGraphError as exc: raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc message = record_outbound_message( db, to_number=recipient, body=body, message_type=message_type, wa_message_id=extract_wa_message_id(result), status="sent", payload=result if isinstance(result, dict) else None, ) return serialize_message(message)