A R R R Associates
720 lines
32 KiB
Python
720 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from app.core.http_responses import forbidden_response
|
|
from sqlalchemy import desc, func, or_, select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
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, 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
|
|
from app.modules.services.models import ClientServiceSubscription, ClientServiceTaskInstance
|
|
from app.modules.email_integration.services import (
|
|
DEFAULT_TEMPLATES,
|
|
get_or_create_email_setting,
|
|
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"])
|
|
|
|
|
|
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)))
|
|
return "System Admin" in roles or "Firm Admin" in roles or "system.settings.manage" in perms or "system.settings.edit" in perms
|
|
|
|
|
|
def _scope_from_request(request: Request, user):
|
|
tenant_id = request.session.get("active_tenant_id") or getattr(user, "tenant_id", None)
|
|
branch_id = request.session.get("active_branch_id") or getattr(user, "branch_id", None)
|
|
if branch_id in ("", "0", 0):
|
|
branch_id = None
|
|
return (int(tenant_id) if tenant_id else None, int(branch_id) if branch_id else None)
|
|
|
|
|
|
def _ctx(request: Request, db, user, **extra):
|
|
ctx = {
|
|
"request": request,
|
|
"current_user": user,
|
|
"current_user_roles": get_user_roles(db, int(user.id)),
|
|
"current_user_permissions": get_user_permissions(db, int(user.id)),
|
|
"csrf_token": get_or_create_csrf_token(request),
|
|
}
|
|
ctx.update(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")
|
|
def email_audit_dashboard_page(request: Request):
|
|
"""Operational dashboard for email health, queue, IMAP and mapping status.
|
|
|
|
Phase 7S.4 intentionally uses the existing email_settings, email_logs,
|
|
email_templates and incoming email tables. No schema change is required.
|
|
"""
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return _redirect_login()
|
|
if not _user_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
now = datetime.now(timezone.utc)
|
|
since_24h = now - timedelta(hours=24)
|
|
since_7d = now - timedelta(days=7)
|
|
|
|
def _log_filters(extra=None):
|
|
filters = [EmailLog.tenant_id == tenant_id]
|
|
if branch_id:
|
|
filters.append(EmailLog.branch_id == branch_id)
|
|
if extra:
|
|
filters.extend(extra)
|
|
return filters
|
|
|
|
def _incoming_filters(extra=None):
|
|
filters = [EmailIncomingMessage.tenant_id == tenant_id]
|
|
if branch_id:
|
|
filters.append(EmailIncomingMessage.branch_id == branch_id)
|
|
if extra:
|
|
filters.extend(extra)
|
|
return filters
|
|
|
|
def _count(model_id, filters):
|
|
return int(db.execute(select(func.count(model_id)).where(*filters)).scalar() or 0)
|
|
|
|
setting = db.execute(
|
|
select(EmailSetting).where(EmailSetting.tenant_id == tenant_id, EmailSetting.branch_id == branch_id)
|
|
).scalar_one_or_none()
|
|
if not setting and branch_id:
|
|
setting = db.execute(
|
|
select(EmailSetting).where(EmailSetting.tenant_id == tenant_id, EmailSetting.branch_id.is_(None))
|
|
).scalar_one_or_none()
|
|
|
|
summary = {
|
|
"sent_24h": _count(EmailLog.id, _log_filters([EmailLog.status == "SENT", EmailLog.created_at_utc >= since_24h])),
|
|
"failed_24h": _count(EmailLog.id, _log_filters([EmailLog.status == "FAILED", EmailLog.created_at_utc >= since_24h])),
|
|
"pending_now": _count(EmailLog.id, _log_filters([EmailLog.status == "PENDING"])),
|
|
"sent_7d": _count(EmailLog.id, _log_filters([EmailLog.status == "SENT", EmailLog.created_at_utc >= since_7d])),
|
|
"failed_7d": _count(EmailLog.id, _log_filters([EmailLog.status == "FAILED", EmailLog.created_at_utc >= since_7d])),
|
|
"skipped_7d": _count(EmailLog.id, _log_filters([EmailLog.status == "SKIPPED", EmailLog.created_at_utc >= since_7d])),
|
|
"queue_retryable": _count(EmailLog.id, _log_filters([EmailLog.status.in_(["PENDING", "FAILED"]), EmailLog.is_retryable.is_(True)])),
|
|
"queue_exhausted": _count(EmailLog.id, _log_filters([EmailLog.status == "FAILED", EmailLog.attempt_count >= EmailLog.max_attempts])),
|
|
"incoming_7d": _count(EmailIncomingMessage.id, _incoming_filters([EmailIncomingMessage.fetched_at_utc >= since_7d])),
|
|
"incoming_unmapped": _count(EmailIncomingMessage.id, _incoming_filters([EmailIncomingMessage.mapping_status.in_(["UNMAPPED", "NO_MATCH", "ERROR"])])),
|
|
"incoming_mapped": _count(EmailIncomingMessage.id, _incoming_filters([EmailIncomingMessage.mapping_status.in_(["AUTO_MAPPED", "MANUAL_MAPPED"])])),
|
|
"incoming_with_attachments": _count(EmailIncomingMessage.id, _incoming_filters([EmailIncomingMessage.has_attachments.is_(True)])),
|
|
"active_templates": _count(EmailTemplate.id, [EmailTemplate.tenant_id == tenant_id, EmailTemplate.branch_id == branch_id, EmailTemplate.is_active.is_(True)]),
|
|
"inactive_templates": _count(EmailTemplate.id, [EmailTemplate.tenant_id == tenant_id, EmailTemplate.branch_id == branch_id, EmailTemplate.is_active.is_(False)]),
|
|
}
|
|
|
|
status_rows = db.execute(
|
|
select(EmailLog.status, func.count(EmailLog.id))
|
|
.where(*_log_filters([EmailLog.created_at_utc >= since_7d]))
|
|
.group_by(EmailLog.status)
|
|
.order_by(EmailLog.status)
|
|
).all()
|
|
status_summary = [{"status": row[0] or "UNKNOWN", "count": int(row[1] or 0)} for row in status_rows]
|
|
|
|
template_rows = db.execute(
|
|
select(EmailLog.template_code, EmailLog.status, func.count(EmailLog.id))
|
|
.where(*_log_filters([EmailLog.created_at_utc >= since_7d]))
|
|
.group_by(EmailLog.template_code, EmailLog.status)
|
|
.order_by(desc(func.count(EmailLog.id)))
|
|
.limit(30)
|
|
).all()
|
|
template_summary = {}
|
|
for code, status, count in template_rows:
|
|
key = code or "MANUAL / NO TEMPLATE"
|
|
template_summary.setdefault(key, {"template_code": key, "SENT": 0, "FAILED": 0, "PENDING": 0, "SKIPPED": 0, "total": 0})
|
|
normalized_status = status or "UNKNOWN"
|
|
template_summary[key][normalized_status] = int(count or 0)
|
|
template_summary[key]["total"] += int(count or 0)
|
|
template_summary_rows = sorted(template_summary.values(), key=lambda row: row["total"], reverse=True)[:12]
|
|
|
|
recent_failed = db.execute(
|
|
select(EmailLog)
|
|
.where(*_log_filters([EmailLog.status == "FAILED"]))
|
|
.order_by(desc(EmailLog.created_at_utc))
|
|
.limit(10)
|
|
).scalars().all()
|
|
queue_due = db.execute(
|
|
select(EmailLog)
|
|
.where(*_log_filters([EmailLog.status.in_(["PENDING", "FAILED"]), EmailLog.is_retryable.is_(True)]))
|
|
.order_by(EmailLog.queue_priority.asc(), EmailLog.next_retry_at.asc().nullsfirst(), EmailLog.created_at_utc.asc())
|
|
.limit(10)
|
|
).scalars().all()
|
|
incoming_attention = db.execute(
|
|
select(EmailIncomingMessage)
|
|
.where(*_incoming_filters([EmailIncomingMessage.mapping_status.in_(["UNMAPPED", "NO_MATCH", "ERROR"])]))
|
|
.order_by(desc(EmailIncomingMessage.fetched_at_utc), desc(EmailIncomingMessage.id))
|
|
.limit(10)
|
|
).scalars().all()
|
|
|
|
smtp_configured = bool(setting and setting.is_active and setting.smtp_host and setting.smtp_port and setting.from_email)
|
|
imap_configured = bool(setting and setting.imap_host and setting.imap_port and setting.imap_username)
|
|
|
|
return templates.TemplateResponse(
|
|
"modules/email_integration/templates/email_integration/audit_dashboard.html",
|
|
_ctx(
|
|
request, db, user,
|
|
title="Email Audit Dashboard",
|
|
setting=setting,
|
|
summary=summary,
|
|
status_summary=status_summary,
|
|
template_summary_rows=template_summary_rows,
|
|
recent_failed=recent_failed,
|
|
queue_due=queue_due,
|
|
incoming_attention=incoming_attention,
|
|
smtp_configured=smtp_configured,
|
|
imap_configured=imap_configured,
|
|
generated_at=now,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/settings")
|
|
def email_settings_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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
if not tenant_id:
|
|
return RedirectResponse(url="/system-settings", status_code=303)
|
|
setting = get_or_create_email_setting(db, tenant_id, branch_id, actor_user_id=int(user.id))
|
|
# get_or_create_email_setting already seeds default templates for new settings.
|
|
# seed_default_email_templates is idempotent, but avoiding a second same-request
|
|
# call keeps the first load of /email/settings clean on SQLite.
|
|
db.commit()
|
|
recent_logs = db.execute(
|
|
select(EmailLog)
|
|
.where(EmailLog.tenant_id == tenant_id)
|
|
.order_by(desc(EmailLog.created_at_utc))
|
|
.limit(10)
|
|
).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/email_integration/templates/email_integration/settings.html",
|
|
_ctx(request, db, user, title="Email Settings", setting=setting, recent_logs=recent_logs, flash=flash),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/settings")
|
|
def save_email_settings(
|
|
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"),
|
|
from_email: str = Form(""),
|
|
from_name: str = Form(""),
|
|
reply_to_email: str = Form(""),
|
|
imap_host: str = Form(""),
|
|
imap_port: int = Form(993),
|
|
imap_username: str = Form(""),
|
|
imap_password: str = Form(""),
|
|
imap_security: str = Form("SSL"),
|
|
send_auth_emails: str | None = Form(None),
|
|
send_alert_emails: str | None = Form(None),
|
|
send_billing_emails: str | None = Form(None),
|
|
send_task_emails: str | None = Form(None),
|
|
send_invoice_emails: str | None = Form(None),
|
|
send_payment_emails: str | None = Form(None),
|
|
send_client_emails: str | None = Form(None),
|
|
send_document_emails: str | None = Form(None),
|
|
send_consultant_emails: str | None = Form(None),
|
|
send_partner_review_emails: str | None = Form(None),
|
|
send_leave_attendance_emails: str | None = Form(None),
|
|
send_online_payment_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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
setting = get_or_create_email_setting(db, int(tenant_id), branch_id, 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.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.imap_host = imap_host.strip() or None
|
|
setting.imap_port = int(imap_port or 993)
|
|
setting.imap_username = imap_username.strip() or None
|
|
if imap_password.strip():
|
|
setting.imap_password = imap_password.strip()
|
|
setting.imap_security = (imap_security or "SSL").upper()
|
|
setting.send_auth_emails = bool(send_auth_emails)
|
|
setting.send_alert_emails = bool(send_alert_emails)
|
|
setting.send_billing_emails = bool(send_billing_emails)
|
|
setting.send_task_emails = bool(send_task_emails)
|
|
setting.send_invoice_emails = bool(send_invoice_emails)
|
|
setting.send_payment_emails = bool(send_payment_emails)
|
|
setting.send_client_emails = bool(send_client_emails)
|
|
setting.send_document_emails = bool(send_document_emails)
|
|
setting.send_consultant_emails = bool(send_consultant_emails)
|
|
setting.send_partner_review_emails = bool(send_partner_review_emails)
|
|
setting.send_leave_attendance_emails = bool(send_leave_attendance_emails)
|
|
setting.send_online_payment_emails = bool(send_online_payment_emails)
|
|
setting.is_active = bool(is_active)
|
|
setting.updated_by_user_id = int(user.id)
|
|
db.commit()
|
|
return RedirectResponse(url="/email/settings?flash=Email settings saved.", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/settings/test")
|
|
def test_email_settings(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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
recipient = (test_email or getattr(user, "email", "")).strip()
|
|
send_template_email(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
recipient_email=recipient,
|
|
template_code="AUTH_LOGIN_OTP",
|
|
context={"user_name": getattr(user, "full_name", None) or recipient, "otp_code": "123456"},
|
|
related_module="email_settings_test",
|
|
related_id=int(user.id),
|
|
force_send=True,
|
|
)
|
|
db.commit()
|
|
return RedirectResponse(url="/email/settings?flash=Test email attempted. Please check Email Logs for SENT/FAILED status.", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/logs")
|
|
def email_logs_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return _redirect_login()
|
|
if not _user_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
q = select(EmailLog).where(EmailLog.tenant_id == tenant_id)
|
|
if branch_id:
|
|
q = q.where(EmailLog.branch_id == branch_id)
|
|
logs = db.execute(q.order_by(desc(EmailLog.created_at_utc)).limit(100)).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/email_integration/templates/email_integration/logs.html",
|
|
_ctx(request, db, user, title="Email Logs", logs=logs),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
|
|
@router.get("/queue")
|
|
def email_queue_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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
q = select(EmailLog).where(EmailLog.tenant_id == tenant_id, EmailLog.status.in_(["PENDING", "FAILED"]))
|
|
if branch_id:
|
|
q = q.where(EmailLog.branch_id == branch_id)
|
|
queue_rows = db.execute(q.order_by(EmailLog.queue_priority.asc(), EmailLog.created_at_utc.asc()).limit(100)).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/email_integration/templates/email_integration/queue.html",
|
|
_ctx(request, db, user, title="Email Queue", queue_rows=queue_rows, flash=flash),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/queue/process")
|
|
def process_email_queue_page(request: Request, csrf_token: str = Form(...), limit: int = Form(25)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return _redirect_login()
|
|
if not _user_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
result = process_pending_email_queue(db, tenant_id=tenant_id, branch_id=branch_id, limit=limit)
|
|
db.commit()
|
|
flash = f"Email queue processed: {result.get('processed', 0)} processed, {result.get('sent', 0)} sent, {result.get('failed', 0)} failed, {result.get('skipped', 0)} skipped."
|
|
return RedirectResponse(url=f"/email/queue?flash={flash}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/inbox")
|
|
def email_inbox_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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
q = select(EmailIncomingMessage).where(EmailIncomingMessage.tenant_id == tenant_id)
|
|
if branch_id:
|
|
q = q.where(EmailIncomingMessage.branch_id == branch_id)
|
|
incoming_rows = db.execute(q.order_by(desc(EmailIncomingMessage.received_at_utc), desc(EmailIncomingMessage.id)).limit(100)).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/email_integration/templates/email_integration/inbox.html",
|
|
_ctx(request, db, user, title="Incoming Emails", incoming_rows=incoming_rows, flash=flash),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/inbox/fetch")
|
|
def fetch_email_inbox_page(
|
|
request: Request,
|
|
csrf_token: str = Form(...),
|
|
folder: str = Form("INBOX"),
|
|
limit: int = Form(25),
|
|
unread_only: str | None = Form("1"),
|
|
mark_seen: 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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
setting = get_or_create_email_setting(db, int(tenant_id), branch_id, actor_user_id=int(user.id))
|
|
result = fetch_incoming_emails(
|
|
db,
|
|
setting=setting,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
folder=(folder or "INBOX").strip() or "INBOX",
|
|
unread_only=bool(unread_only),
|
|
limit=max(1, min(int(limit or 25), 100)),
|
|
mark_seen=bool(mark_seen),
|
|
)
|
|
db.commit()
|
|
if result.error:
|
|
flash = f"IMAP fetch failed: {result.error}"
|
|
else:
|
|
flash = f"IMAP fetch completed: {result.imported} imported, {result.skipped_existing} skipped, {result.failed} failed."
|
|
return RedirectResponse(url=f"/email/inbox?flash={flash}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/inbox/{message_id}")
|
|
def email_inbox_detail_page(request: Request, message_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return _redirect_login()
|
|
if not _user_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
filters = [EmailIncomingMessage.id == message_id, EmailIncomingMessage.tenant_id == tenant_id]
|
|
if branch_id:
|
|
filters.append(EmailIncomingMessage.branch_id == branch_id)
|
|
row = db.execute(select(EmailIncomingMessage).where(*filters)).scalar_one_or_none()
|
|
if not row:
|
|
return RedirectResponse(url="/email/inbox", status_code=303)
|
|
attachments = db.execute(
|
|
select(EmailIncomingAttachment).where(EmailIncomingAttachment.incoming_message_id == row.id).order_by(EmailIncomingAttachment.id)
|
|
).scalars().all()
|
|
scope_filters = [ClientServiceSubscription.tenant_id == tenant_id]
|
|
task_filters = [ClientServiceTaskInstance.tenant_id == tenant_id]
|
|
invoice_filters = [BillingInvoice.tenant_id == tenant_id]
|
|
if branch_id:
|
|
scope_filters.append(or_(ClientServiceSubscription.branch_id == branch_id, ClientServiceSubscription.branch_id.is_(None)))
|
|
task_filters.append(or_(ClientServiceTaskInstance.branch_id == branch_id, ClientServiceTaskInstance.branch_id.is_(None)))
|
|
invoice_filters.append(or_(BillingInvoice.branch_id == branch_id, BillingInvoice.branch_id.is_(None)))
|
|
if row.matched_client_id:
|
|
scope_filters.append(ClientServiceSubscription.client_id == row.matched_client_id)
|
|
task_filters.append(ClientServiceTaskInstance.client_id == row.matched_client_id)
|
|
invoice_filters.append(BillingInvoice.client_id == row.matched_client_id)
|
|
engagements = db.execute(select(ClientServiceSubscription).where(*scope_filters).order_by(desc(ClientServiceSubscription.updated_at_utc)).limit(50)).scalars().all()
|
|
tasks = db.execute(select(ClientServiceTaskInstance).where(*task_filters).order_by(desc(ClientServiceTaskInstance.updated_at_utc)).limit(50)).scalars().all()
|
|
invoices = db.execute(select(BillingInvoice).where(*invoice_filters).order_by(desc(BillingInvoice.invoice_date), desc(BillingInvoice.id)).limit(50)).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/email_integration/templates/email_integration/inbox_detail.html",
|
|
_ctx(
|
|
request, db, user, title="Incoming Email", row=row, attachments=attachments,
|
|
engagements=engagements, tasks=tasks, invoices=invoices,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/inbox/map-all")
|
|
def email_inbox_map_all(request: Request, csrf_token: str = Form(...), limit: int = Form(100)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return _redirect_login()
|
|
if not _user_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
result = auto_map_unmapped_emails(db, tenant_id=tenant_id, branch_id=branch_id, limit=limit)
|
|
db.commit()
|
|
flash = f"Email mapping completed: {result.get('processed', 0)} processed, {result.get('mapped', 0)} mapped, {result.get('no_match', 0)} no match."
|
|
return RedirectResponse(url=f"/email/inbox?flash={flash}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/inbox/{message_id}/map")
|
|
def email_inbox_manual_map_page(
|
|
request: Request,
|
|
message_id: int,
|
|
csrf_token: str = Form(...),
|
|
engagement_id: int | None = Form(None),
|
|
task_id: int | None = Form(None),
|
|
invoice_id: int | 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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
filters = [EmailIncomingMessage.id == message_id, EmailIncomingMessage.tenant_id == tenant_id]
|
|
if branch_id:
|
|
filters.append(EmailIncomingMessage.branch_id == branch_id)
|
|
row = db.execute(select(EmailIncomingMessage).where(*filters)).scalar_one_or_none()
|
|
if row:
|
|
apply_email_mapping(
|
|
db, row,
|
|
engagement_id=engagement_id or None,
|
|
task_id=task_id or None,
|
|
invoice_id=invoice_id or None,
|
|
manual=True,
|
|
)
|
|
db.commit()
|
|
return RedirectResponse(url=f"/email/inbox/{message_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/templates")
|
|
def email_templates_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return _redirect_login()
|
|
if not _user_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
seed_default_email_templates(db, tenant_id=tenant_id, branch_id=branch_id)
|
|
db.commit()
|
|
templates_rows = db.execute(
|
|
select(EmailTemplate)
|
|
.where(EmailTemplate.tenant_id == tenant_id, EmailTemplate.branch_id == branch_id)
|
|
.order_by(EmailTemplate.template_name)
|
|
).scalars().all()
|
|
return templates.TemplateResponse(
|
|
"modules/email_integration/templates/email_integration/templates.html",
|
|
_ctx(request, db, user, title="Email Templates", templates_rows=templates_rows, default_templates=DEFAULT_TEMPLATES),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/templates/{template_id}")
|
|
def update_email_template(
|
|
request: Request,
|
|
template_id: int,
|
|
csrf_token: str = Form(...),
|
|
subject_template: str = Form(...),
|
|
body_template: str = Form(...),
|
|
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_can_manage_email(db, user):
|
|
return forbidden_response(request, "Access denied: email settings require administrator permission")
|
|
tenant_id, branch_id = _scope_from_request(request, user)
|
|
row = db.execute(
|
|
select(EmailTemplate).where(
|
|
EmailTemplate.id == template_id,
|
|
EmailTemplate.tenant_id == tenant_id,
|
|
EmailTemplate.branch_id == branch_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if row:
|
|
row.subject_template = subject_template
|
|
row.body_template = body_template
|
|
row.is_active = bool(is_active)
|
|
db.commit()
|
|
return RedirectResponse(url="/email/templates", status_code=303)
|
|
finally:
|
|
db.close()
|