725 lines
26 KiB
Python
725 lines
26 KiB
Python
"""
|
|
Birthday Manager - self-hosted birthday reminder & e-mail application
|
|
(c) 2026 Martinhal IT - Joao Vaz - Version 1.2
|
|
"""
|
|
import base64
|
|
import calendar as cal
|
|
import imaplib
|
|
import io
|
|
import os
|
|
import re
|
|
import secrets
|
|
import smtplib
|
|
import sqlite3
|
|
import uuid
|
|
from datetime import date, datetime
|
|
from email.mime.image import MIMEImage
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
from functools import wraps
|
|
|
|
import pyotp
|
|
import qrcode
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from flask import (Flask, flash, g, redirect, render_template, request,
|
|
send_from_directory, session, url_for)
|
|
from werkzeug.security import check_password_hash, generate_password_hash
|
|
from werkzeug.utils import secure_filename
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DATA_DIR = os.environ.get("DATA_DIR", os.path.join(BASE_DIR, "data"))
|
|
UPLOAD_DIR = os.path.join(DATA_DIR, "uploads")
|
|
DB_PATH = os.path.join(DATA_DIR, "birthday.db")
|
|
ALLOWED_IMAGE_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
|
|
|
|
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.environ.get("SECRET_KEY", secrets.token_hex(32))
|
|
app.config["MAX_CONTENT_LENGTH"] = 15 * 1024 * 1024 # 15 MB total upload limit
|
|
|
|
APP_VERSION = "Version 1.2"
|
|
FOOTER_TEXT = "\u00a9 2026 Martinhal IT - Joao Vaz - " + APP_VERSION
|
|
|
|
# ---------------------------------------------------------------- database
|
|
|
|
def get_db():
|
|
if "db" not in g:
|
|
g.db = sqlite3.connect(DB_PATH)
|
|
g.db.row_factory = sqlite3.Row
|
|
g.db.execute("PRAGMA foreign_keys = ON")
|
|
return g.db
|
|
|
|
|
|
@app.teardown_appcontext
|
|
def close_db(_exc):
|
|
db = g.pop("db", None)
|
|
if db is not None:
|
|
db.close()
|
|
|
|
|
|
def db_connect():
|
|
"""Standalone connection for use outside a request (scheduler)."""
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
DEFAULT_TEMPLATE_MALE = (
|
|
"<p>Dear {first_name} {last_name},</p>"
|
|
"<p>Happy Birthday! Wishing you a fantastic day and a wonderful year ahead. "
|
|
"Congratulations on turning {age}!</p>"
|
|
"<p>Best regards,<br>Martinhal IT</p>"
|
|
)
|
|
DEFAULT_TEMPLATE_FEMALE = (
|
|
"<p>Dear {first_name} {last_name},</p>"
|
|
"<p>Happy Birthday! Wishing you a beautiful day filled with joy. "
|
|
"Congratulations on turning {age}!</p>"
|
|
"<p>Warm regards,<br>Martinhal IT</p>"
|
|
)
|
|
|
|
|
|
def init_db():
|
|
conn = db_connect()
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT UNIQUE NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
mfa_secret TEXT,
|
|
mfa_enabled INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS birthdays (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
first_name TEXT NOT NULL,
|
|
last_name TEXT NOT NULL,
|
|
dob TEXT NOT NULL, -- YYYY-MM-DD
|
|
gender TEXT NOT NULL, -- Male / Female
|
|
email TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS email_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
recipient_name TEXT NOT NULL,
|
|
recipient_email TEXT NOT NULL,
|
|
subject TEXT NOT NULL,
|
|
sent_at TEXT NOT NULL, -- YYYY-MM-DD HH:MM:SS
|
|
status TEXT NOT NULL, -- Sent / Failed
|
|
detail TEXT
|
|
);
|
|
CREATE TABLE IF NOT EXISTS templates (
|
|
gender TEXT PRIMARY KEY, -- Male / Female
|
|
subject TEXT NOT NULL,
|
|
body TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS settings (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
# Seed default admin user
|
|
if conn.execute("SELECT COUNT(*) c FROM users").fetchone()["c"] == 0:
|
|
conn.execute(
|
|
"INSERT INTO users (username, password_hash, created_at) VALUES (?,?,?)",
|
|
("admin", generate_password_hash("admin"),
|
|
datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
|
|
)
|
|
print(" * Created default user 'admin' with password 'admin' - change it!")
|
|
# Seed default templates
|
|
for gender, body in (("Male", DEFAULT_TEMPLATE_MALE),
|
|
("Female", DEFAULT_TEMPLATE_FEMALE)):
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO templates (gender, subject, body) VALUES (?,?,?)",
|
|
(gender, "Happy Birthday, {first_name}!", body),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def get_setting(conn, key, default=None):
|
|
row = conn.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
|
|
return row["value"] if row else default
|
|
|
|
|
|
def set_setting(conn, key, value):
|
|
conn.execute(
|
|
"INSERT INTO settings (key, value) VALUES (?,?)"
|
|
" ON CONFLICT(key) DO UPDATE SET value = excluded.value", (key, value))
|
|
conn.commit()
|
|
|
|
|
|
def get_send_time(conn) -> tuple[int, int]:
|
|
default = f'{int(os.environ.get("SEND_HOUR", "8")):02d}:00'
|
|
raw = get_setting(conn, "send_time", default)
|
|
try:
|
|
h, m = raw.split(":")
|
|
return max(0, min(23, int(h))), max(0, min(59, int(m)))
|
|
except ValueError:
|
|
return 8, 0
|
|
|
|
|
|
# ---------------------------------------------------------------- auth
|
|
|
|
def login_required(view):
|
|
@wraps(view)
|
|
def wrapped(*args, **kwargs):
|
|
if not session.get("user_id"):
|
|
return redirect(url_for("login", next=request.path))
|
|
return view(*args, **kwargs)
|
|
return wrapped
|
|
|
|
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
user = get_db().execute(
|
|
"SELECT * FROM users WHERE username = ?", (username,)
|
|
).fetchone()
|
|
if user and check_password_hash(user["password_hash"], password):
|
|
if user["mfa_enabled"]:
|
|
session.clear()
|
|
session["mfa_pending_user"] = user["id"]
|
|
return redirect(url_for("mfa_verify"))
|
|
session.clear()
|
|
session["user_id"] = user["id"]
|
|
session["username"] = user["username"]
|
|
return redirect(url_for("dashboard"))
|
|
flash("Invalid username or password.", "error")
|
|
return render_template("login.html", footer=FOOTER_TEXT)
|
|
|
|
|
|
@app.route("/mfa-verify", methods=["GET", "POST"])
|
|
def mfa_verify():
|
|
pending = session.get("mfa_pending_user")
|
|
if not pending:
|
|
return redirect(url_for("login"))
|
|
if request.method == "POST":
|
|
code = request.form.get("code", "").strip().replace(" ", "")
|
|
user = get_db().execute("SELECT * FROM users WHERE id = ?", (pending,)).fetchone()
|
|
if user and user["mfa_secret"] and pyotp.TOTP(user["mfa_secret"]).verify(code, valid_window=1):
|
|
session.clear()
|
|
session["user_id"] = user["id"]
|
|
session["username"] = user["username"]
|
|
return redirect(url_for("dashboard"))
|
|
flash("Invalid authentication code.", "error")
|
|
return render_template("mfa_verify.html", footer=FOOTER_TEXT)
|
|
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("login"))
|
|
|
|
|
|
# ---------------------------------------------------------------- helpers
|
|
|
|
def calc_age(dob: date, on: date) -> int:
|
|
years = on.year - dob.year
|
|
if (on.month, on.day) < (dob.month, dob.day):
|
|
years -= 1
|
|
return years
|
|
|
|
|
|
def birthday_matches_day(dob: date, day: date) -> bool:
|
|
"""Feb 29 birthdays are celebrated on Feb 28 in non-leap years."""
|
|
if dob.month == 2 and dob.day == 29 and not cal.isleap(day.year):
|
|
return day.month == 2 and day.day == 28
|
|
return dob.month == day.month and dob.day == day.day
|
|
|
|
|
|
def birthdays_on(conn, day: date):
|
|
rows = conn.execute("SELECT * FROM birthdays").fetchall()
|
|
result = []
|
|
for r in rows:
|
|
dob = datetime.strptime(r["dob"], "%Y-%m-%d").date()
|
|
if birthday_matches_day(dob, day):
|
|
result.append((r, calc_age(dob, day)))
|
|
return result
|
|
|
|
|
|
def todays_birthdays(conn):
|
|
return birthdays_on(conn, date.today())
|
|
|
|
|
|
def render_placeholders(text: str, row, age: int) -> str:
|
|
return (text.replace("{first_name}", row["first_name"])
|
|
.replace("{last_name}", row["last_name"])
|
|
.replace("{age}", str(age)))
|
|
|
|
|
|
def smtp_settings():
|
|
# Defaults are pre-configured for Purelymail (smtp.purelymail.com:465 SSL/TLS).
|
|
# If your client only supports STARTTLS, set SMTP_PORT=587 and SMTP_SECURITY=starttls.
|
|
return {
|
|
"host": os.environ.get("SMTP_HOST", "smtp.purelymail.com"),
|
|
"port": int(os.environ.get("SMTP_PORT", "465")),
|
|
"user": os.environ.get("SMTP_USER", ""),
|
|
"password": os.environ.get("SMTP_PASS", ""),
|
|
"sender": os.environ.get("SMTP_FROM", os.environ.get("SMTP_USER", "birthday@localhost")),
|
|
"security": os.environ.get("SMTP_SECURITY", "ssl").lower(), # ssl / starttls / none
|
|
}
|
|
|
|
|
|
def imap_settings():
|
|
# Defaults are pre-configured for Purelymail (imap.purelymail.com:993 SSL/TLS).
|
|
# IMAP credentials fall back to the SMTP ones if not set separately.
|
|
return {
|
|
"host": os.environ.get("IMAP_HOST", "imap.purelymail.com"),
|
|
"port": int(os.environ.get("IMAP_PORT", "993")),
|
|
"user": os.environ.get("IMAP_USER", os.environ.get("SMTP_USER", "")),
|
|
"password": os.environ.get("IMAP_PASS", os.environ.get("SMTP_PASS", "")),
|
|
"folder": os.environ.get("IMAP_SENT_FOLDER", "Sent"),
|
|
}
|
|
|
|
|
|
def save_to_sent_folder(msg) -> str:
|
|
"""Append the sent message to the mailbox's Sent folder via IMAP.
|
|
Returns '' on success or an error description on failure."""
|
|
cfg = imap_settings()
|
|
if not cfg["user"]:
|
|
return "IMAP not configured - message not copied to Sent folder"
|
|
try:
|
|
imap = imaplib.IMAP4_SSL(cfg["host"], cfg["port"], timeout=20)
|
|
imap.login(cfg["user"], cfg["password"])
|
|
folder = cfg["folder"]
|
|
# Create the folder if the server doesn't have it yet
|
|
status, _ = imap.select(folder)
|
|
if status != "OK":
|
|
imap.create(folder)
|
|
imap.append(folder, r"(\Seen)", imaplib.Time2Internaldate(datetime.now().timestamp()),
|
|
msg.as_bytes())
|
|
imap.logout()
|
|
return ""
|
|
except Exception as exc:
|
|
return f"Delivered, but copy to Sent folder failed: {exc}"
|
|
|
|
|
|
IMG_SRC_RE = re.compile(r'src="/uploads/([^"]+)"')
|
|
|
|
|
|
def build_email(row, age: int, conn) -> MIMEMultipart:
|
|
tpl = conn.execute(
|
|
"SELECT * FROM templates WHERE gender = ?", (row["gender"],)
|
|
).fetchone()
|
|
subject = render_placeholders(tpl["subject"], row, age)
|
|
body_html = render_placeholders(tpl["body"], row, age)
|
|
|
|
msg = MIMEMultipart("related")
|
|
msg["Subject"] = subject
|
|
msg["From"] = smtp_settings()["sender"]
|
|
msg["To"] = row["email"]
|
|
|
|
# Embed uploaded images referenced in the template as inline attachments
|
|
inline = []
|
|
def to_cid(match):
|
|
fname = match.group(1)
|
|
path = os.path.join(UPLOAD_DIR, secure_filename(fname))
|
|
if os.path.isfile(path):
|
|
cid = uuid.uuid4().hex
|
|
inline.append((path, cid))
|
|
return f'src="cid:{cid}"'
|
|
return match.group(0)
|
|
|
|
body_html = IMG_SRC_RE.sub(to_cid, body_html)
|
|
|
|
alt = MIMEMultipart("alternative")
|
|
alt.attach(MIMEText(re.sub(r"<[^>]+>", " ", body_html), "plain", "utf-8"))
|
|
alt.attach(MIMEText(body_html, "html", "utf-8"))
|
|
msg.attach(alt)
|
|
|
|
for path, cid in inline:
|
|
with open(path, "rb") as f:
|
|
img = MIMEImage(f.read())
|
|
img.add_header("Content-ID", f"<{cid}>")
|
|
img.add_header("Content-Disposition", "inline", filename=os.path.basename(path))
|
|
msg.attach(img)
|
|
return msg
|
|
|
|
|
|
def send_birthday_email(row, age: int, conn) -> None:
|
|
cfg = smtp_settings()
|
|
msg = build_email(row, age, conn)
|
|
status, detail = "Sent", ""
|
|
try:
|
|
if not cfg["host"]:
|
|
raise RuntimeError("SMTP is not configured (set SMTP_HOST etc.)")
|
|
if cfg["security"] == "ssl":
|
|
server = smtplib.SMTP_SSL(cfg["host"], cfg["port"], timeout=20)
|
|
else:
|
|
server = smtplib.SMTP(cfg["host"], cfg["port"], timeout=20)
|
|
if cfg["security"] == "starttls":
|
|
server.starttls()
|
|
if cfg["user"]:
|
|
server.login(cfg["user"], cfg["password"])
|
|
server.send_message(msg)
|
|
server.quit()
|
|
detail = save_to_sent_folder(msg) # keep a copy in the mailbox Sent folder
|
|
except Exception as exc: # log failures too
|
|
status, detail = "Failed", str(exc)
|
|
conn.execute(
|
|
"INSERT INTO email_log (recipient_name, recipient_email, subject, sent_at, status, detail)"
|
|
" VALUES (?,?,?,?,?,?)",
|
|
(f'{row["first_name"]} {row["last_name"]}', row["email"], msg["Subject"],
|
|
datetime.now().strftime("%Y-%m-%d %H:%M:%S"), status, detail),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def already_sent_today(conn, row) -> bool:
|
|
today = date.today().strftime("%Y-%m-%d")
|
|
hit = conn.execute(
|
|
"SELECT 1 FROM email_log WHERE recipient_email = ? AND status = 'Sent'"
|
|
" AND sent_at LIKE ?", (row["email"], today + "%"),
|
|
).fetchone()
|
|
return hit is not None
|
|
|
|
|
|
def daily_birthday_job():
|
|
conn = db_connect()
|
|
try:
|
|
for row, age in todays_birthdays(conn):
|
|
if not already_sent_today(conn, row):
|
|
send_birthday_email(row, age, conn)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------- page 1
|
|
|
|
MONTH_NAMES = ["January", "February", "March", "April", "May", "June",
|
|
"July", "August", "September", "October", "November", "December"]
|
|
|
|
|
|
def build_year_calendar(conn, year: int):
|
|
"""Return list of month dicts with week grids; birthday days carry names."""
|
|
rows = conn.execute("SELECT * FROM birthdays").fetchall()
|
|
marks = {}
|
|
for r in rows:
|
|
dob = datetime.strptime(r["dob"], "%Y-%m-%d").date()
|
|
m, d = dob.month, dob.day
|
|
if m == 2 and d == 29 and not cal.isleap(year):
|
|
d = 28
|
|
marks.setdefault((m, d), []).append(f'{r["first_name"]} {r["last_name"]}')
|
|
months = []
|
|
c = cal.Calendar(firstweekday=0) # Monday
|
|
today = date.today()
|
|
for m in range(1, 13):
|
|
weeks = []
|
|
for week in c.monthdayscalendar(year, m):
|
|
weeks.append([
|
|
{
|
|
"day": d or "",
|
|
"names": marks.get((m, d), []) if d else [],
|
|
"today": bool(d) and today.year == year and today.month == m and today.day == d,
|
|
} for d in week
|
|
])
|
|
months.append({"name": MONTH_NAMES[m - 1], "weeks": weeks})
|
|
return months
|
|
|
|
|
|
@app.route("/", methods=["GET"])
|
|
@login_required
|
|
def dashboard():
|
|
conn = get_db()
|
|
year = request.args.get("year", type=int) or date.today().year
|
|
entries = conn.execute(
|
|
"SELECT * FROM birthdays ORDER BY substr(dob,6) ASC"
|
|
).fetchall()
|
|
from datetime import timedelta
|
|
today = date.today()
|
|
today_list = [
|
|
{"row": r, "age": a, "sent": already_sent_today(conn, r)}
|
|
for r, a in todays_birthdays(conn)
|
|
]
|
|
yesterday_list = [{"row": r, "age": a} for r, a in birthdays_on(conn, today - timedelta(days=1))]
|
|
tomorrow_list = [{"row": r, "age": a} for r, a in birthdays_on(conn, today + timedelta(days=1))]
|
|
return render_template(
|
|
"dashboard.html",
|
|
footer=FOOTER_TEXT,
|
|
username=session.get("username"),
|
|
entries=entries,
|
|
today_list=today_list,
|
|
yesterday_list=yesterday_list,
|
|
tomorrow_list=tomorrow_list,
|
|
today=today,
|
|
months=build_year_calendar(conn, year),
|
|
year=year,
|
|
)
|
|
|
|
|
|
@app.route("/birthdays/add", methods=["POST"])
|
|
@login_required
|
|
def add_birthday():
|
|
f = request.form
|
|
first = f.get("first_name", "").strip()
|
|
last = f.get("last_name", "").strip()
|
|
dob = f.get("dob", "").strip()
|
|
gender = f.get("gender", "").strip()
|
|
email = f.get("email", "").strip()
|
|
try:
|
|
datetime.strptime(dob, "%Y-%m-%d")
|
|
assert first and last and email and gender in ("Male", "Female")
|
|
except (ValueError, AssertionError):
|
|
flash("Please fill in all fields with valid values.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
get_db().execute(
|
|
"INSERT INTO birthdays (first_name, last_name, dob, gender, email) VALUES (?,?,?,?,?)",
|
|
(first, last, dob, gender, email),
|
|
)
|
|
get_db().commit()
|
|
flash(f"Added {first} {last}.", "ok")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
|
|
@app.route("/birthdays/<int:bid>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete_birthday(bid):
|
|
get_db().execute("DELETE FROM birthdays WHERE id = ?", (bid,))
|
|
get_db().commit()
|
|
flash("Entry removed.", "ok")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
|
|
@app.route("/force-send", methods=["POST"])
|
|
@login_required
|
|
def force_send():
|
|
conn = get_db()
|
|
todays = todays_birthdays(conn)
|
|
if not todays:
|
|
flash("Nobody has a birthday today - nothing to send.", "error")
|
|
else:
|
|
for row, age in todays:
|
|
send_birthday_email(row, age, conn)
|
|
flash(f"Birthday e-mail resent to {len(todays)} recipient(s). Check the log on the Administration page.", "ok")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
|
|
# ---------------------------------------------------------------- page 2
|
|
|
|
@app.route("/admin")
|
|
@login_required
|
|
def admin():
|
|
conn = get_db()
|
|
users = conn.execute("SELECT id, username, mfa_enabled, created_at FROM users ORDER BY username").fetchall()
|
|
log = conn.execute("SELECT * FROM email_log ORDER BY sent_at DESC, id DESC LIMIT 500").fetchall()
|
|
tpl_male = conn.execute("SELECT * FROM templates WHERE gender='Male'").fetchone()
|
|
tpl_female = conn.execute("SELECT * FROM templates WHERE gender='Female'").fetchone()
|
|
images = sorted(os.listdir(UPLOAD_DIR))
|
|
h, m = get_send_time(conn)
|
|
return render_template(
|
|
"admin.html",
|
|
send_time=f"{h:02d}:{m:02d}",
|
|
footer=FOOTER_TEXT,
|
|
username=session.get("username"),
|
|
user_id=session.get("user_id"),
|
|
users=users,
|
|
log=log,
|
|
tpl_male=tpl_male,
|
|
tpl_female=tpl_female,
|
|
images=images,
|
|
)
|
|
|
|
|
|
@app.route("/users/add", methods=["POST"])
|
|
@login_required
|
|
def add_user():
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
if not username or len(password) < 4:
|
|
flash("Username and a password of at least 4 characters are required.", "error")
|
|
return redirect(url_for("admin"))
|
|
try:
|
|
get_db().execute(
|
|
"INSERT INTO users (username, password_hash, created_at) VALUES (?,?,?)",
|
|
(username, generate_password_hash(password),
|
|
datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
|
|
)
|
|
get_db().commit()
|
|
flash(f"User '{username}' created.", "ok")
|
|
except sqlite3.IntegrityError:
|
|
flash("That username already exists.", "error")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/users/<int:uid>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete_user(uid):
|
|
conn = get_db()
|
|
if uid == session.get("user_id"):
|
|
flash("You cannot delete the account you are logged in with.", "error")
|
|
elif conn.execute("SELECT COUNT(*) c FROM users").fetchone()["c"] <= 1:
|
|
flash("At least one user must remain.", "error")
|
|
else:
|
|
conn.execute("DELETE FROM users WHERE id = ?", (uid,))
|
|
conn.commit()
|
|
flash("User removed.", "ok")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/account/password", methods=["POST"])
|
|
@login_required
|
|
def change_password():
|
|
current = request.form.get("current", "")
|
|
new = request.form.get("new", "")
|
|
confirm = request.form.get("confirm", "")
|
|
conn = get_db()
|
|
user = conn.execute("SELECT * FROM users WHERE id = ?", (session["user_id"],)).fetchone()
|
|
if not check_password_hash(user["password_hash"], current):
|
|
flash("Current password is incorrect.", "error")
|
|
elif len(new) < 4:
|
|
flash("New password must be at least 4 characters.", "error")
|
|
elif new != confirm:
|
|
flash("New passwords do not match.", "error")
|
|
else:
|
|
conn.execute("UPDATE users SET password_hash = ? WHERE id = ?",
|
|
(generate_password_hash(new), user["id"]))
|
|
conn.commit()
|
|
flash("Password changed.", "ok")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/account/mfa/setup", methods=["GET", "POST"])
|
|
@login_required
|
|
def mfa_setup():
|
|
conn = get_db()
|
|
user = conn.execute("SELECT * FROM users WHERE id = ?", (session["user_id"],)).fetchone()
|
|
if request.method == "POST":
|
|
code = request.form.get("code", "").strip().replace(" ", "")
|
|
secret = session.get("mfa_setup_secret")
|
|
if secret and pyotp.TOTP(secret).verify(code, valid_window=1):
|
|
conn.execute("UPDATE users SET mfa_secret = ?, mfa_enabled = 1 WHERE id = ?",
|
|
(secret, user["id"]))
|
|
conn.commit()
|
|
session.pop("mfa_setup_secret", None)
|
|
flash("Two-factor authentication is now enabled.", "ok")
|
|
return redirect(url_for("admin"))
|
|
flash("That code did not match. Scan the QR code and try again.", "error")
|
|
secret = session.get("mfa_setup_secret") or pyotp.random_base32()
|
|
session["mfa_setup_secret"] = secret
|
|
uri = pyotp.TOTP(secret).provisioning_uri(
|
|
name=user["username"], issuer_name="Birthday Manager")
|
|
buf = io.BytesIO()
|
|
qrcode.make(uri).save(buf, format="PNG")
|
|
qr_b64 = base64.b64encode(buf.getvalue()).decode()
|
|
return render_template("mfa_setup.html", footer=FOOTER_TEXT,
|
|
username=session.get("username"),
|
|
qr=qr_b64, secret=secret)
|
|
|
|
|
|
@app.route("/account/mfa/disable", methods=["POST"])
|
|
@login_required
|
|
def mfa_disable():
|
|
conn = get_db()
|
|
conn.execute("UPDATE users SET mfa_enabled = 0, mfa_secret = NULL WHERE id = ?",
|
|
(session["user_id"],))
|
|
conn.commit()
|
|
flash("Two-factor authentication disabled for your account.", "ok")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/settings/send-time", methods=["POST"])
|
|
@login_required
|
|
def save_send_time():
|
|
raw = request.form.get("send_time", "").strip()
|
|
try:
|
|
h, m = raw.split(":")
|
|
h, m = int(h), int(m)
|
|
assert 0 <= h <= 23 and 0 <= m <= 59
|
|
except (ValueError, AssertionError):
|
|
flash("Please pick a valid time.", "error")
|
|
return redirect(url_for("admin"))
|
|
set_setting(get_db(), "send_time", f"{h:02d}:{m:02d}")
|
|
scheduler.reschedule_job("daily-birthday-mail", trigger="cron", hour=h, minute=m)
|
|
flash(f"Automatic birthday e-mails will now be sent daily at {h:02d}:{m:02d}.", "ok")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/templates/save", methods=["POST"])
|
|
@login_required
|
|
def save_template():
|
|
gender = request.form.get("gender")
|
|
if gender not in ("Male", "Female"):
|
|
flash("Unknown template.", "error")
|
|
return redirect(url_for("admin"))
|
|
subject = request.form.get("subject", "").strip() or "Happy Birthday, {first_name}!"
|
|
body = request.form.get("body", "").strip()
|
|
get_db().execute("UPDATE templates SET subject = ?, body = ? WHERE gender = ?",
|
|
(subject, body, gender))
|
|
get_db().commit()
|
|
flash(f"E-mail template ({gender}) saved.", "ok")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/images/upload", methods=["POST"])
|
|
@login_required
|
|
def upload_image():
|
|
files = [f for f in request.files.getlist("images") if f and f.filename]
|
|
if not files:
|
|
flash("Choose one or more image files to upload.", "error")
|
|
return redirect(url_for("admin"))
|
|
# Enforce the 15 MB total limit across all files in this upload
|
|
total = 0
|
|
for f in files:
|
|
f.stream.seek(0, os.SEEK_END)
|
|
total += f.stream.tell()
|
|
f.stream.seek(0)
|
|
if total > 15 * 1024 * 1024:
|
|
flash(f"Upload rejected: total size {total / (1024*1024):.1f} MB exceeds the 15 MB limit.", "error")
|
|
return redirect(url_for("admin"))
|
|
saved, skipped = [], []
|
|
for f in files:
|
|
name = secure_filename(f.filename)
|
|
ext = os.path.splitext(name)[1].lower()
|
|
if ext not in ALLOWED_IMAGE_EXT:
|
|
skipped.append(f.filename)
|
|
continue
|
|
unique = f"{os.path.splitext(name)[0]}-{uuid.uuid4().hex[:6]}{ext}"
|
|
f.save(os.path.join(UPLOAD_DIR, unique))
|
|
saved.append(unique)
|
|
if saved:
|
|
flash(f"{len(saved)} image(s) uploaded. Copy the <img> command shown under each image to use it in a template.", "ok")
|
|
if skipped:
|
|
flash(f"Skipped (not PNG/JPG/GIF/WEBP): {', '.join(skipped)}", "error")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.errorhandler(413)
|
|
def too_large(_e):
|
|
flash("Upload rejected: the total upload size exceeds the 15 MB limit.", "error")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/images/<path:fname>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete_image(fname):
|
|
path = os.path.join(UPLOAD_DIR, secure_filename(fname))
|
|
if os.path.isfile(path):
|
|
os.remove(path)
|
|
flash("Image deleted.", "ok")
|
|
return redirect(url_for("admin"))
|
|
|
|
|
|
@app.route("/uploads/<path:fname>")
|
|
@login_required
|
|
def serve_upload(fname):
|
|
return send_from_directory(UPLOAD_DIR, secure_filename(fname))
|
|
|
|
|
|
# ---------------------------------------------------------------- startup
|
|
|
|
init_db()
|
|
scheduler = BackgroundScheduler(daemon=True)
|
|
_conn = db_connect()
|
|
_h, _m = get_send_time(_conn)
|
|
_conn.close()
|
|
scheduler.add_job(daily_birthday_job, "cron", hour=_h, minute=_m,
|
|
id="daily-birthday-mail")
|
|
scheduler.start()
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8080")), debug=False)
|