#!/usr/bin/env python __author__ = "Philippe May " __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} {fromName} 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") # "from" is reserved word in Python, use an alias fromName: str = "" 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) # Make a dict for sms 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", type=str, default="0.0.0.0", help="Addess to listen to" ) parser.add_argument( "-p", "--port", type=int, default=8025, help="Port to listen to" ) parser.add_argument( "-v", "--version", action="store_true", help="Print version and exit" ) args = parser.parse_args() if args.version: import sys from importlib.metadata import version print(version("sms_handler")) sys.exit(0) run(app, host=args.host, port=args.port) if __name__ == "__main__": main()