86 lines
2 KiB
Python
86 lines
2 KiB
Python
|
#!/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
|
||
|
|
||
|
run(app, host="0.0.0.0", port=8025)
|