File size: 2,311 Bytes
e6f0cc8 b72e388 e6f0cc8 b72e388 e6f0cc8 b72e388 e6f0cc8 b72e388 e6f0cc8 b72e388 e6f0cc8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
# import vonage
# client = vonage.Client(key="712db024", secret="L4RYeQ2EWbHG6UTY")
# sms = vonage.Sms(client)
# responseData = sms.send_message(
# {
# "from": "Vonage APIs",
# "to": "917869844761",
# "text": "A text message sent using the Nexmo SMS API",
# }
# )
# if responseData["messages"][0]["status"] == "0":
# print("Message sent successfully.")
# else:
# print(f"Message failed with error: {responseData['messages'][0]['error-text']}")
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from vonage import Client, Sms
from datetime import datetime, timedelta
import schedule
import asyncio
import time
import vonage
app = FastAPI()
client = vonage.Client(key="712db024", secret="L4RYeQ2EWbHG6UTY")
sms = vonage.Sms(client)
class MessageRequest(BaseModel):
phone_num: str
text: str
scheduled_time: str # HH:MM format
async def send_scheduled_sms(message_request: MessageRequest):
try:
response = sms.send_message(
{
"from": "Vonage APIs",
"to": message_request.phone_num,
"text": f"This is your medicine reminder for {message_request.text}"
}
)
print(f"SMS sent to {message_request.phone_num} at {message_request.scheduled_time}")
except Exception as e:
print(f"Failed to send SMS to {message_request.phone_num}: {e}")
async def schedule_sms(message_request: MessageRequest):
scheduled_time = datetime.strptime(message_request.scheduled_time, "%H:%M")
now = datetime.now()
delta = scheduled_time - now
seconds_until_scheduled_time = delta.total_seconds()
if seconds_until_scheduled_time > 0:
print(f"Scheduling SMS for {message_request.phone_num} at {scheduled_time}")
await asyncio.sleep(seconds_until_scheduled_time)
await send_scheduled_sms(message_request)
@app.post("/send_scheduled_sms")
async def send_scheduled_sms_endpoint(message_request: MessageRequest):
asyncio.create_task(schedule_sms(message_request))
return {"message": f"SMS scheduled for {message_request.phone_num} at {message_request.scheduled_time}"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
# uvicorn test:app --reload
|