File size: 2,307 Bytes
0c0e192 |
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 |
# 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 threading
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
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}")
def schedule_daily_sms(message_request: MessageRequest):
scheduled_time = datetime.strptime(message_request.scheduled_time, "%H:%M")
schedule.every().day.at(message_request.scheduled_time).do(send_scheduled_sms, message_request)
print(f"Scheduled daily SMS for {message_request.phone_num} at {scheduled_time}")
@app.post("/send_daily_sms")
async def send_daily_sms_endpoint(message_request: MessageRequest):
schedule_daily_sms(message_request)
return {"message": f"Daily SMS scheduled for {message_request.phone_num} at {message_request.scheduled_time}"}
def run_scheduler():
while True:
schedule.run_pending()
time.sleep(1)
if __name__ == "__main__":
scheduler_thread = threading.Thread(target=schedule_daily_sms)
scheduler_thread.start()
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
# uvicorn test:app --reload
|