sms-handler/src/sms_handler/server.py
phil 9558e45bed
Some checks failed
/ test (push) Failing after 3s
Slim down requirements
2024-12-01 05:01:45 +01:00

91 lines
2.3 KiB
Python
Executable file

#!/usr/bin/env python
__author__ = "Philippe May <phil.dev@philome.mooo.com>"
__license__ = "BSD"
__doc__ = "Listen to messages from the SMS Forwarder app on Android, and send mail."
import os
import textwrap
from datetime import datetime
from socket import gethostname
from email.message import EmailMessage
from fastapi import FastAPI
from aiosmtplib import send
from pydantic import BaseModel, Field, EmailStr
from pydantic_settings import BaseSettings, SettingsConfigDict
USER = os.environ.get("USER", "root")
HOST_NAME = gethostname()
## Definition of settings:
class MailSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="SMS_HANDLER_MAIL_")
enable: bool = True
sender: str = f"{USER}@{HOST_NAME}"
to: str = f"{USER}@{HOST_NAME}"
server: str = "localhost"
server_port: int = 25
server_start_tls: bool = False
template: str = textwrap.dedent(
"""\
{text}
---
From: {from}
Sent: {sentStamp:%c}
Received: {receivedStamp:%c}
Sim: {sim}
"""
)
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="SMS_HANDLER_")
mail: MailSettings = MailSettings()
class SmsAlert(BaseModel):
from_: str = Field(alias="from")
text: str
sentStamp: datetime
receivedStamp: datetime
sim: str
settings = Settings()
app = FastAPI()
def send_mail(smsAlert: SmsAlert):
msg = EmailMessage()
msg["Subject"] = f"SMS from {smsAlert.from_}"
msg["From"] = settings.mail.sender
msg["To"] = settings.mail.to
fmt = dict(smsAlert)
fmt["from"] = fmt.pop("from_")
msg.set_content(settings.mail.template.format_map(fmt))
return send(
msg,
hostname=settings.mail.server,
port=settings.mail.server_port,
start_tls=settings.mail.server_start_tls,
)
@app.post("/handle-sms")
async def handle_sms(smsAlert: SmsAlert):
if settings.mail.enable:
await send_mail(smsAlert)
return "OK"
def main():
from uvicorn import run
from argparse import ArgumentParser
parser = ArgumentParser(description=__doc__)
parser.add_argument("-l", "--host", default="0.0.0.0")
parser.add_argument("-p", "--port", type=int, default=8025)
args = parser.parse_args()
run(app, host=args.host, port=args.port)