Add platform SMTP and firm wizard invite email

This commit is contained in:
A R R R Associates
2026-07-02 23:43:24 +05:30
parent b4b8dc97fd
commit b84090d8c9
12 changed files with 821 additions and 45 deletions
+110 -1
View File
@@ -12,7 +12,7 @@ from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
from app.core.security.session_auth import get_current_user
from app.core.templating import templates
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.email_integration.models import EmailIncomingAttachment, EmailIncomingMessage, EmailLog, EmailSetting, EmailTemplate
from app.modules.email_integration.models import EmailIncomingAttachment, EmailIncomingMessage, EmailLog, EmailSetting, EmailTemplate, PlatformEmailSetting
from app.modules.email_integration.imap_service import fetch_incoming_emails
from app.modules.email_integration.mapping_service import apply_email_mapping, auto_map_unmapped_emails
from app.modules.billing.models import BillingInvoice
@@ -23,6 +23,8 @@ from app.modules.email_integration.services import (
seed_default_email_templates,
send_template_email,
process_pending_email_queue,
get_or_create_platform_email_setting,
send_platform_template_email,
)
router = APIRouter(prefix="/email", tags=["email-integration-ui"])
@@ -32,6 +34,11 @@ def _redirect_login():
return RedirectResponse(url="/login", status_code=303)
def _user_is_system_admin(db, user) -> bool:
return "System Admin" in set(get_user_roles(db, int(user.id)))
def _user_can_manage_email(db, user) -> bool:
roles = set(get_user_roles(db, int(user.id)))
perms = set(get_user_permissions(db, int(user.id)))
@@ -58,6 +65,108 @@ def _ctx(request: Request, db, user, **extra):
return ctx
@router.get("/platform-smtp")
def platform_smtp_page(request: Request, flash: str | None = None):
db = CommonSessionLocal()
try:
user = get_current_user(request, db=db)
if not user:
return _redirect_login()
if not _user_is_system_admin(db, user):
return forbidden_response(request, "Access denied: platform SMTP is restricted to System Admin")
setting = get_or_create_platform_email_setting(db, actor_user_id=int(user.id))
db.commit()
recent_logs = db.execute(
select(EmailLog)
.where(EmailLog.tenant_id.is_(None), EmailLog.related_module.in_(["platform_smtp_test", "wizard_firm_invite"]))
.order_by(desc(EmailLog.created_at_utc))
.limit(10)
).scalars().all()
return templates.TemplateResponse(
"modules/email_integration/templates/email_integration/platform_smtp.html",
_ctx(request, db, user, title="Platform SMTP Settings", setting=setting, recent_logs=recent_logs, flash=flash),
)
finally:
db.close()
@router.post("/platform-smtp")
def save_platform_smtp(
request: Request,
csrf_token: str = Form(...),
smtp_host: str = Form(""),
smtp_port: int = Form(465),
smtp_username: str = Form(""),
smtp_password: str = Form(""),
smtp_security: str = Form("SSL"),
smtp_timeout_seconds: int = Form(20),
from_email: str = Form(""),
from_name: str = Form(""),
reply_to_email: str = Form(""),
send_auth_emails: str | None = Form(None),
is_active: str | None = Form(None),
):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
user = get_current_user(request, db=db)
if not user:
return _redirect_login()
if not _user_is_system_admin(db, user):
return forbidden_response(request, "Access denied: platform SMTP is restricted to System Admin")
setting = get_or_create_platform_email_setting(db, actor_user_id=int(user.id))
setting.smtp_host = smtp_host.strip() or None
setting.smtp_port = int(smtp_port or 465)
setting.smtp_username = smtp_username.strip() or None
if smtp_password.strip():
setting.smtp_password = smtp_password.strip()
setting.smtp_security = (smtp_security or "SSL").upper()
setting.smtp_timeout_seconds = max(5, min(int(smtp_timeout_seconds or 20), 120))
setting.from_email = from_email.strip() or None
setting.from_name = from_name.strip() or None
setting.reply_to_email = reply_to_email.strip() or None
setting.send_auth_emails = bool(send_auth_emails)
setting.is_active = bool(is_active)
setting.updated_by_user_id = int(user.id)
db.commit()
return RedirectResponse(url="/email/platform-smtp?flash=Platform SMTP settings saved.", status_code=303)
finally:
db.close()
@router.post("/platform-smtp/test")
def test_platform_smtp(request: Request, csrf_token: str = Form(...), test_email: str = Form("")):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
user = get_current_user(request, db=db)
if not user:
return _redirect_login()
if not _user_is_system_admin(db, user):
return forbidden_response(request, "Access denied: platform SMTP is restricted to System Admin")
recipient = (test_email or getattr(user, "email", "")).strip()
send_platform_template_email(
db,
recipient_email=recipient,
template_code="AUTH_USER_INVITE",
context={
"firm_name": "ERP Platform",
"user_name": getattr(user, "full_name", None) or recipient,
"invite_link": "This is a platform SMTP test email. No action is required.",
"expiry_hours": "0",
},
related_module="platform_smtp_test",
related_id=int(user.id),
queue_priority=1,
)
db.commit()
return RedirectResponse(url="/email/platform-smtp?flash=Test email attempted. Check recent logs below.", status_code=303)
finally:
db.close()
@router.get("/audit")