A R R R Associates
226 lines
9.7 KiB
Python
226 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Iterable
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.billing.models import BillingInvoice
|
|
from app.modules.email_integration.models import EmailIncomingMessage
|
|
from app.modules.services.models import ClientServiceSubscription, ClientServiceTaskInstance, ServiceTaskComment
|
|
|
|
TRACKING_PATTERN = re.compile(r"\[(?:AF-)?(?P<kind>ENG|TASK|INV|INVOICE|CLIENT|NOTICE)-(?P<value>[A-Za-z0-9._/-]+)\]", re.IGNORECASE)
|
|
|
|
|
|
@dataclass
|
|
class EmailMappingResult:
|
|
matched: bool = False
|
|
status: str = "NO_MATCH"
|
|
tracking_code: str | None = None
|
|
related_module: str | None = None
|
|
related_id: int | None = None
|
|
engagement_id: int | None = None
|
|
task_id: int | None = None
|
|
invoice_id: int | None = None
|
|
client_id: int | None = None
|
|
notes: str | None = None
|
|
|
|
|
|
def _scope_filter(model, tenant_id: int | None, branch_id: int | None) -> list:
|
|
filters = []
|
|
if tenant_id is not None and hasattr(model, "tenant_id"):
|
|
filters.append(model.tenant_id == tenant_id)
|
|
if branch_id is not None and hasattr(model, "branch_id"):
|
|
filters.append(or_(model.branch_id == branch_id, model.branch_id.is_(None)))
|
|
return filters
|
|
|
|
|
|
def extract_tracking_codes(subject: str | None, body_text: str | None = None, body_html: str | None = None) -> list[tuple[str, str, str]]:
|
|
"""Return tracking codes as (raw_code, kind, value)."""
|
|
combined = "\n".join([subject or "", body_text or "", body_html or ""])
|
|
seen: set[str] = set()
|
|
rows: list[tuple[str, str, str]] = []
|
|
for match in TRACKING_PATTERN.finditer(combined):
|
|
raw = match.group(0).upper()
|
|
if raw in seen:
|
|
continue
|
|
seen.add(raw)
|
|
kind = match.group("kind").upper()
|
|
if kind == "INVOICE":
|
|
kind = "INV"
|
|
value = match.group("value").strip()
|
|
rows.append((raw, kind, value))
|
|
return rows
|
|
|
|
|
|
def _int_or_none(value: str | int | None) -> int | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return int(str(value).strip())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _find_engagement(db: Session, tenant_id: int | None, branch_id: int | None, value: str) -> ClientServiceSubscription | None:
|
|
ident = _int_or_none(value)
|
|
if ident is None:
|
|
return None
|
|
filters = [ClientServiceSubscription.id == ident, *_scope_filter(ClientServiceSubscription, tenant_id, branch_id)]
|
|
return db.execute(select(ClientServiceSubscription).where(*filters).limit(1)).scalar_one_or_none()
|
|
|
|
|
|
def _find_task(db: Session, tenant_id: int | None, branch_id: int | None, value: str) -> ClientServiceTaskInstance | None:
|
|
ident = _int_or_none(value)
|
|
if ident is None:
|
|
return None
|
|
filters = [ClientServiceTaskInstance.id == ident, *_scope_filter(ClientServiceTaskInstance, tenant_id, branch_id)]
|
|
return db.execute(select(ClientServiceTaskInstance).where(*filters).limit(1)).scalar_one_or_none()
|
|
|
|
|
|
def _find_invoice(db: Session, tenant_id: int | None, branch_id: int | None, value: str) -> BillingInvoice | None:
|
|
ident = _int_or_none(value)
|
|
filters = [*_scope_filter(BillingInvoice, tenant_id, branch_id)]
|
|
if ident is not None:
|
|
invoice = db.execute(select(BillingInvoice).where(BillingInvoice.id == ident, *filters).limit(1)).scalar_one_or_none()
|
|
if invoice:
|
|
return invoice
|
|
value_clean = value.strip()
|
|
if value_clean:
|
|
return db.execute(select(BillingInvoice).where(BillingInvoice.invoice_no == value_clean, *filters).limit(1)).scalar_one_or_none()
|
|
return None
|
|
|
|
|
|
def _create_task_timeline_comment(db: Session, message: EmailIncomingMessage, task: ClientServiceTaskInstance) -> None:
|
|
existing = db.execute(
|
|
select(ServiceTaskComment).where(
|
|
ServiceTaskComment.task_instance_id == task.id,
|
|
ServiceTaskComment.message.like(f"%Incoming Email ID: {message.id}%"),
|
|
).limit(1)
|
|
).scalar_one_or_none()
|
|
if existing:
|
|
return
|
|
body = (message.body_text or "").strip()
|
|
if len(body) > 1200:
|
|
body = body[:1200].rstrip() + "..."
|
|
sender = message.sender_email or "Unknown sender"
|
|
subject = message.subject or "No subject"
|
|
comment_text = (
|
|
f"Incoming email mapped from {sender}\n"
|
|
f"Subject: {subject}\n\n"
|
|
f"{body}\n\n"
|
|
f"Incoming Email ID: {message.id}"
|
|
).strip()
|
|
db.add(
|
|
ServiceTaskComment(
|
|
tenant_id=task.tenant_id,
|
|
branch_id=task.branch_id,
|
|
subscription_id=task.subscription_id,
|
|
task_instance_id=task.id,
|
|
comment_type="client_communication" if message.matched_client_id else "email_reply",
|
|
visibility="internal",
|
|
message=comment_text,
|
|
created_by_user_id=message.matched_user_id,
|
|
)
|
|
)
|
|
|
|
|
|
def apply_email_mapping(
|
|
db: Session,
|
|
message: EmailIncomingMessage,
|
|
*,
|
|
engagement_id: int | None = None,
|
|
task_id: int | None = None,
|
|
invoice_id: int | None = None,
|
|
manual: bool = False,
|
|
) -> EmailMappingResult:
|
|
"""Map one incoming email to engagement/task/invoice by manual selection or tracking code."""
|
|
tenant_id = message.tenant_id
|
|
branch_id = message.branch_id
|
|
result = EmailMappingResult(status="NO_MATCH")
|
|
|
|
# Manual mapping takes precedence.
|
|
if task_id:
|
|
task = _find_task(db, tenant_id, branch_id, str(task_id))
|
|
if task:
|
|
result = EmailMappingResult(True, "MANUAL_MAPPED" if manual else "AUTO_MAPPED", None, "task", int(task.id), int(task.subscription_id), int(task.id), None, int(task.client_id), "Mapped to task.")
|
|
_apply_result(db, message, result)
|
|
_create_task_timeline_comment(db, message, task)
|
|
return result
|
|
if engagement_id:
|
|
eng = _find_engagement(db, tenant_id, branch_id, str(engagement_id))
|
|
if eng:
|
|
result = EmailMappingResult(True, "MANUAL_MAPPED" if manual else "AUTO_MAPPED", None, "engagement", int(eng.id), int(eng.id), None, None, int(eng.client_id), "Mapped to engagement.")
|
|
_apply_result(db, message, result)
|
|
return result
|
|
if invoice_id:
|
|
inv = _find_invoice(db, tenant_id, branch_id, str(invoice_id))
|
|
if inv:
|
|
result = EmailMappingResult(True, "MANUAL_MAPPED" if manual else "AUTO_MAPPED", None, "invoice", int(inv.id), int(inv.engagement_id) if inv.engagement_id else None, None, int(inv.id), int(inv.client_id), "Mapped to invoice.")
|
|
_apply_result(db, message, result)
|
|
return result
|
|
|
|
# Auto mapping by tracking code in subject/body.
|
|
for raw, kind, value in extract_tracking_codes(message.subject, message.body_text, message.body_html):
|
|
if kind == "TASK":
|
|
task = _find_task(db, tenant_id, branch_id, value)
|
|
if task:
|
|
result = EmailMappingResult(True, "AUTO_MAPPED", raw, "task", int(task.id), int(task.subscription_id), int(task.id), None, int(task.client_id), "Auto-mapped using task tracking code.")
|
|
_apply_result(db, message, result)
|
|
_create_task_timeline_comment(db, message, task)
|
|
return result
|
|
if kind == "ENG":
|
|
eng = _find_engagement(db, tenant_id, branch_id, value)
|
|
if eng:
|
|
result = EmailMappingResult(True, "AUTO_MAPPED", raw, "engagement", int(eng.id), int(eng.id), None, None, int(eng.client_id), "Auto-mapped using engagement tracking code.")
|
|
_apply_result(db, message, result)
|
|
return result
|
|
if kind == "INV":
|
|
inv = _find_invoice(db, tenant_id, branch_id, value)
|
|
if inv:
|
|
result = EmailMappingResult(True, "AUTO_MAPPED", raw, "invoice", int(inv.id), int(inv.engagement_id) if inv.engagement_id else None, None, int(inv.id), int(inv.client_id), "Auto-mapped using invoice tracking code.")
|
|
_apply_result(db, message, result)
|
|
return result
|
|
|
|
message.mapping_status = "NO_MATCH"
|
|
message.mapping_notes = "No tracking code/manual match found."
|
|
message.mapped_at_utc = datetime.now(timezone.utc)
|
|
if message.status == "NEW":
|
|
message.status = "NEW"
|
|
return result
|
|
|
|
|
|
def _apply_result(db: Session, message: EmailIncomingMessage, result: EmailMappingResult) -> None:
|
|
message.tracking_code = result.tracking_code
|
|
message.related_module = result.related_module
|
|
message.related_id = result.related_id
|
|
message.matched_engagement_id = result.engagement_id
|
|
message.matched_task_id = result.task_id
|
|
message.matched_invoice_id = result.invoice_id
|
|
if result.client_id and not message.matched_client_id:
|
|
message.matched_client_id = result.client_id
|
|
message.mapping_status = result.status
|
|
message.mapping_notes = result.notes
|
|
message.mapped_at_utc = datetime.now(timezone.utc)
|
|
message.status = "MATCHED" if result.matched else message.status
|
|
|
|
|
|
def auto_map_unmapped_emails(db: Session, *, tenant_id: int | None, branch_id: int | None, limit: int = 100) -> dict[str, int]:
|
|
q = select(EmailIncomingMessage).where(EmailIncomingMessage.tenant_id == tenant_id)
|
|
if branch_id:
|
|
q = q.where(EmailIncomingMessage.branch_id == branch_id)
|
|
q = q.where(EmailIncomingMessage.mapping_status.in_(["UNMAPPED", "NO_MATCH"]))
|
|
rows = db.execute(q.order_by(EmailIncomingMessage.received_at_utc.desc(), EmailIncomingMessage.id.desc()).limit(max(1, min(limit, 500)))).scalars().all()
|
|
mapped = 0
|
|
no_match = 0
|
|
for row in rows:
|
|
res = apply_email_mapping(db, row)
|
|
if res.matched:
|
|
mapped += 1
|
|
else:
|
|
no_match += 1
|
|
return {"processed": len(rows), "mapped": mapped, "no_match": no_match}
|