USSD Webhook

POST /api/v1/ussd/

{
  "sessionId": "ATUid_abc123xyz",
  "serviceCode": "*384*1234#",
  "phoneNumber": "+254712345678",
  "text": ""
}

{
  "sessionId": "ATUid_abc123xyz",
  "serviceCode": "*384*1234#",
  "phoneNumber": "+254712345678",
  "text": "1"
}

{
  "sessionId": "ATUid_abc123xyz",
  "serviceCode": "*384*1234#",
  "phoneNumber": "+254712345678",
  "text": "1*2"
}

CON Welcome to VoicePact
1. View My Contracts
2. Confirm Delivery
3. Check Payments
4. Help & Support
0. Exit

END Thank you for using VoicePact!

@router.post("/")
async def ussd_handler(
    request: Request,
    sessionId: str = Form(...),
    serviceCode: str = Form(...),
    phoneNumber: str = Form(...),
    text: str = Form(""),
    at_client: AfricasTalkingClient = Depends(get_africastalking_client),
    db: AsyncSession = Depends(get_db)
):
    """Main USSD handler for VoicePact"""
    try:
        # Get or create session
        session = await get_or_create_session(sessionId, phoneNumber, db)
        
        # Parse user input (last selection)
        user_input = text.split('*')[-1] if text else ""
        
        # Determine current menu
        if not text:  # First request
            response = main_menu()
            session.current_menu = "main"
        else:
            response = await handle_menu_navigation(
                session, user_input, phoneNumber, at_client, db
            )
        
        # Update session state
        session.last_input = user_input
        session.last_response = response
        session.updated_at = datetime.utcnow()
        session.expires_at = datetime.utcnow() + timedelta(minutes=5)
        
        await db.commit()
        
        return response
        
    except Exception as e:
        logger.error(f"USSD handler error: {e}")
        return at_client.build_ussd_response(
            "Service temporarily unavailable. Please try again later.",
            end_session=True
        )

def main_menu() -> str:
    """Generate main USSD menu"""
    return """Welcome to VoicePact
1. View My Contracts
2. Confirm Delivery
3. Check Payments
4. Help & Support
0. Exit"""

async def handle_main_menu(session, user_input, phone_number, at_client, db):
    if user_input == "1":  # View Active Contracts
        contracts = await get_user_contracts(phone_number, db)
        
        if not contracts:
            return at_client.build_ussd_response(
                "No active contracts found.\n0. Back to Main Menu",
                end_session=False
            )
        
        session.current_menu = "contracts"
        session.context_data = {"contracts": [c.id for c in contracts]}
        
        menu_text = "Your Contracts:\n"
        for i, contract in enumerate(contracts[:5], 1):
            status = contract.status.value
            menu_text += f"{i}. {contract.id[:12]}... ({status})\n"
        
        menu_text += "\n0. Back to Main Menu"
        return at_client.build_ussd_response(menu_text, end_session=False)

def contract_detail_menu(contract: Contract, at_client: AfricasTalkingClient) -> str:
    product = contract.terms.get('product', 'Product')
    amount = contract.total_amount or 0
    currency = contract.currency
    
    menu_text = f"""Contract Details
{contract.id[:12]}...
Product: {product}
Value: {currency} {amount:,.0f}
Status: {contract.status.value.title()}
"""
    
    if contract.status == ContractStatus.ACTIVE:
        menu_text += "1. Confirm Delivery\n"
    
    menu_text += """2. Report Issue
0. Back"""
    
    return at_client.build_ussd_response(menu_text, end_session=False)

async def handle_delivery_menu(session, user_input, phone_number, at_client, db):
    contract_id = session.context_data.get("selected_contract")
    
    if user_input == "1":  # Full delivery
        await update_contract_status(contract_id, ContractStatus.COMPLETED, db)
        
        return at_client.build_ussd_response(
            "Full delivery confirmed!\nBuyer will be notified.\nPayment will be processed.",
            end_session=True
        )
    
    elif user_input == "2":  # Partial delivery
        return at_client.build_ussd_response(
            "Partial delivery noted.\nSMS will be sent for details.\n0. Main Menu",
            end_session=False
        )
    
    elif user_input == "3":  # Report issue
        await update_contract_status(contract_id, ContractStatus.DISPUTED, db)
        
        return at_client.build_ussd_response(
            "Issue reported.\nContract marked for review.\nSupport will contact you.",
            end_session=True
        )

class USSDSession(Base):
    __tablename__ = "ussd_sessions"
    
    id = Column(UUID, primary_key=True, default=uuid4)
    session_id = Column(String, unique=True, index=True)
    phone_number = Column(String, index=True)
    current_menu = Column(String)  # Current menu state
    context_data = Column(JSON)    # Store temp data (contract IDs, etc.)
    last_input = Column(String)
    last_response = Column(Text)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow)
    expires_at = Column(DateTime)  # Auto-expire after 5 minutes

async def get_or_create_session(
    session_id: str,
    phone_number: str,
    db: AsyncSession
) -> USSDSession:
    result = await db.execute(
        select(USSDSession).where(USSDSession.session_id == session_id)
    )
    session = result.scalar_one_or_none()
    
    if not session:
        session = USSDSession(
            session_id=session_id,
            phone_number=phone_number,
            current_menu="main",
            context_data={},
            is_active=True,
            expires_at=datetime.utcnow() + timedelta(minutes=5)
        )
        db.add(session)
    
    return session

def build_ussd_response(text: str, end_session: bool = False) -> str:
    """Build USSD response with correct prefix"""
    if end_session:
        return f"END {text}"
    return f"CON {text}"

def parse_ussd_input(text: str) -> List[str]:
    """Parse user's navigation path"""
    return text.split('*') if text else []

# Usage
path = parse_ussd_input("1*2*3")  # ["1", "2", "3"]
current_selection = path[-1] if path else None  # "3"

ALLOWED_IPS = [
    "54.166.123.1",
    "54.166.123.2",
    # Add Africa's Talking IP ranges
]

@router.post("/")
async def ussd_handler(request: Request, ...):
    client_ip = request.client.host
    
    if client_ip not in ALLOWED_IPS:
        raise HTTPException(status_code=403, detail="Forbidden")
    
    # Process USSD request...

def validate_menu_selection(user_input: str, valid_options: List[str]) -> bool:
    return user_input in valid_options

# Usage
if not validate_menu_selection(user_input, ["0", "1", "2", "3", "4"]):
    return build_ussd_response(
        "Invalid selection. Please try again.",
        end_session=False
    )

async def cleanup_expired_sessions(db: AsyncSession):
    """Remove expired USSD sessions"""
    await db.execute(
        delete(USSDSession).where(
            USSDSession.expires_at < datetime.utcnow()
        )
    )
    await db.commit()

# Start server
uvicorn app.main:app --reload --port 8000

# Simulate USSD request
curl -X POST http://localhost:8000/api/v1/ussd/ \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "sessionId=ATUid_test123" \
  -d "serviceCode=*384*1234#" \
  -d "phoneNumber=%2B254712345678" \
  -d "text="

CON Welcome to VoicePact
1. View My Contracts
2. Confirm Delivery
3. Check Payments
4. Help & Support
0. Exit

# Log session analytics
logger.info(f"USSD Session: {session_id}, Menu: {current_menu}, User: {phone_number}")

# Track popular menu paths
logger.info(f"Navigation path: {text}")

# Monitor session duration
session_duration = (datetime.utcnow() - session.created_at).total_seconds()
logger.info(f"Session duration: {session_duration}s")