bmc_hub/app/services/email_sender_create.py

"""Reviewable sender extraction and transactional company/contact creation."""
import hashlib
import json
import re
from email.utils import parseaddr
from typing import Literal, Optional
from fastapi import HTTPException
from pydantic import BaseModel, Field
from app.core.database import atomic_transaction, execute_query, execute_query_single, execute_update


class SenderCompany(BaseModel):
    existing_id: Optional[int] = Field(None, gt=0)
    name: str = Field('', max_length=255)
    cvr_number: Optional[str] = Field(None, max_length=8)
    email: Optional[str] = Field(None, max_length=255)
    phone: Optional[str] = Field(None, max_length=50)
    address: Optional[str] = Field(None, max_length=500)
    postal_code: Optional[str] = Field(None, max_length=20)
    city: Optional[str] = Field(None, max_length=100)
    website: Optional[str] = Field(None, max_length=500)


class SenderContact(BaseModel):
    existing_id: Optional[int] = Field(None, gt=0)
    first_name: str = Field('', max_length=100)
    last_name: str = Field('', max_length=100)
    email: str = Field('', max_length=255)
    phone: Optional[str] = Field(None, max_length=50)
    mobile: Optional[str] = Field(None, max_length=50)
    title: Optional[str] = Field(None, max_length=100)
    department: Optional[str] = Field(None, max_length=100)


class SenderCreateRequest(BaseModel):
    kind: Literal['customer', 'vendor']
    company: SenderCompany
    contact: SenderContact


def sender_suggestions(mail):
    from app.contacts.backend.router_simple import _plain_email_body, _contact_suggestions_from_email
    # Isolate the newest message, preserving its signature, before extraction/AI.
    text = _plain_email_body(mail)[:24000]
    text = re.split(r'(?im)^\s*(?:Fra:|From:|[-_]{3,}\s*(?:Original|Oprindelig)|On .{1,200} wrote:|Den .{1,200} skrev.*:|>).*$', text, maxsplit=1)[0]
    parsed = dict(mail, body_text=text, body_html='')
    fields = _contact_suggestions_from_email({}, parsed)
    contact = {row['field']: row['suggested'] for row in fields}
    contact['email'] = parseaddr(str(mail.get('sender_email') or ''))[1].lower()
    lines = [line.strip(' *\t') for line in text.splitlines() if line.strip()]
    company = {}
    labelled = re.search(r'(?im)^(?:Firma|Company|Virksomhed)\s*:\s*(.{2,100})$', text)
    legal = next((line for line in lines if re.search(r'\b(?:ApS|A/S|A\.S\.|Ltd\.?|GmbH)\b', line)
                  and len(line) < 120 and '@' not in line), None)
    company['name'] = labelled.group(1).strip() if labelled else legal or ''
    cvr = re.search(r'(?i)\b(?:CVR(?:[- ]?nr\.?)?|VAT)\s*[:#]?\s*(?:DK\s*)?(\d{8})\b', text)
    if cvr: company['cvr_number'] = cvr.group(1)
    website = re.search(r'(?:https?://|www\.)[^\s<>"\)]+', text)
    if website: company['website'] = website.group(0).rstrip('.,;')
    for index, line in enumerate(lines):
        city = re.fullmatch(r'(\d{4})\s+([A-ZÆØÅa-zæøå][\wæøåÆØÅ .-]+)', line)
        if city:
            company.update(postal_code=city.group(1), city=city.group(2))
            if index and re.search(r'[A-Za-zÆØÅæøå].*\d', lines[index-1]) and not re.search(r'(?i)phone|tel|mobil|cvr|@', lines[index-1]):
                company['address'] = lines[index-1]
            break
    return {'company': company, 'contact': contact, 'evidence': text[:8000], 'parsed': parsed}


def matching_companies(company, sender_email):
    address = parseaddr(str(sender_email or ''))[1].lower()
    domain = address.rsplit('@', 1)[-1] if '@' in address else ''
    from app.emails.backend.router import IGNORED_SENDER_DOMAINS
    if domain in IGNORED_SENDER_DOMAINS: domain = ''
    name, cvr = str(company.get('name') or '').strip(), str(company.get('cvr_number') or '').strip()
    return execute_query("""SELECT id, name, 'customer' AS kind FROM customers
        WHERE is_active = true AND ((%s <> '' AND LOWER(TRIM(name)) = LOWER(%s))
          OR (%s <> '' AND REGEXP_REPLACE(COALESCE(cvr_number,''), '[^0-9]', '', 'g') = %s)
          OR (%s <> '' AND LOWER(TRIM(email_domain)) = %s)
          OR (%s <> '' AND LOWER(TRIM(email)) = %s))
        UNION ALL SELECT id, name, 'vendor' FROM vendors
        WHERE is_active = true AND ((%s <> '' AND LOWER(TRIM(name)) = LOWER(%s))
          OR (%s <> '' AND REGEXP_REPLACE(COALESCE(cvr_number,''), '[^0-9]', '', 'g') = %s)
          OR (%s <> '' AND LOWER(TRIM(domain)) = %s)
          OR (%s <> '' AND LOWER(TRIM(email)) = %s))""",
        (name, name, cvr, cvr, domain, domain, address, address) * 2) or []


async def create_sender(email_id, payload, user_id=None):
    from app.customers.backend.router import create_customer, CustomerCreate
    from app.vendors.backend.router import create_vendor
    from app.models.schemas import VendorCreate
    data = payload.model_dump()
    company, contact = payload.company, payload.contact
    address = contact.email.strip().lower()
    if not contact.existing_id and (not contact.first_name.strip() or not re.fullmatch(r'[^\s@]+@[^\s@]+\.[^\s@]+', address)):
        raise HTTPException(422, 'Kontaktens fornavn og gyldige mailadresse skal udfyldes')
    if not company.existing_id and not company.name.strip():
        raise HTTPException(422, 'Virksomhedsnavn skal udfyldes')
    if company.cvr_number and not re.fullmatch(r'\d{8}', company.cvr_number):
        raise HTTPException(422, 'CVR skal bestå af 8 cifre')
    fingerprint = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
    with atomic_transaction():
        # Serialize this command by mail, then exact contact email and company identity.
        mail = execute_query_single('SELECT id, customer_id, supplier_id FROM email_messages WHERE id=%s AND deleted_at IS NULL FOR UPDATE', (email_id,))
        if not mail: raise HTTPException(404, 'Mailen blev ikke fundet')
        previous = execute_query_single("SELECT metadata FROM email_activity_log WHERE email_id=%s AND event_type='sender_created' ORDER BY id DESC LIMIT 1", (email_id,))
        if previous and previous['metadata'].get('fingerprint') == fingerprint:
            return dict(previous['metadata']['result'], idempotent=True)
        for key in sorted({'sender-contact:' + address, 'sender-company:' + payload.kind + ':' + (company.cvr_number or company.name.strip().lower())}):
            execute_query_single('SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))', (key,))
        table = 'customers' if payload.kind == 'customer' else 'vendors'
        company_id = company.existing_id
        if company_id:
            row = execute_query_single(f'SELECT id, name FROM {table} WHERE id=%s AND is_active=true FOR UPDATE', (company_id,))
            if not row: raise HTTPException(404, 'Den valgte virksomhed findes ikke eller er inaktiv')
        else:
            duplicates = execute_query(f"SELECT id, name FROM {table} WHERE LOWER(TRIM(name))=LOWER(%s) OR (%s::text IS NOT NULL AND REGEXP_REPLACE(COALESCE(cvr_number,''),'[^0-9]','','g')=%s)", (company.name.strip(), company.cvr_number, company.cvr_number)) or []
            if duplicates: raise HTTPException(409, 'Virksomheden findes allerede. Vælg den eksisterende virksomhed i feltet øverst.')
        existing_contacts = execute_query('SELECT id, is_active FROM contacts WHERE LOWER(TRIM(email))=%s ORDER BY id FOR UPDATE', (address,)) or []
        contact_id = contact.existing_id
        if contact_id:
            row = execute_query_single('SELECT id FROM contacts WHERE id=%s AND is_active=true FOR UPDATE', (contact_id,))
            if not row: raise HTTPException(404, 'Den valgte kontakt findes ikke eller er inaktiv')
        elif len(existing_contacts) > 1 or (existing_contacts and not existing_contacts[0]['is_active']):
            raise HTTPException(409, 'Mailadressen findes på flere eller inaktive kontakter. Vælg en eksisterende aktiv kontakt.')
        elif existing_contacts:
            contact_id = existing_contacts[0]['id']
        linked_field = 'customer_id' if payload.kind == 'customer' else 'supplier_id'
        if mail.get(linked_field) and mail[linked_field] != company_id:
            raise HTTPException(409, 'Mailen er allerede tilknyttet en virksomhed. Skift tilknytning først.')
        if not company_id:
            values = {k: (v.strip() or None if isinstance(v, str) else v) for k, v in company.model_dump(exclude={'existing_id'}).items()}
            # Sender's personal address stays on the contact, unless explicitly entered on the company.
            created = await (create_customer(CustomerCreate(**values)) if payload.kind == 'customer'
                             else create_vendor(VendorCreate(**values)))
            company_id = int(created['id'])
        reused_contact = bool(contact_id)
        if not contact_id:
            contact_id = execute_query_single('''INSERT INTO contacts
                (first_name,last_name,email,phone,mobile,title,department,is_active)
                VALUES (%s,%s,%s,%s,%s,%s,%s,true) RETURNING id''',
                (contact.first_name.strip(), contact.last_name.strip(), address, contact.phone, contact.mobile, contact.title, contact.department))['id']
        relation = ('INSERT INTO contact_companies (customer_id,contact_id) VALUES (%s,%s) ON CONFLICT DO NOTHING'
                    if payload.kind == 'customer' else 'INSERT INTO vendor_contacts (vendor_id,contact_id) VALUES (%s,%s) ON CONFLICT DO NOTHING')
        execute_update(relation, (company_id, contact_id))
        execute_update(f'UPDATE email_messages SET {linked_field}=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s', (company_id, email_id))
        result = {'success': True, 'company_id': company_id, 'contact_id': contact_id, 'reused_contact': reused_contact}
        execute_update("""INSERT INTO email_activity_log (email_id,event_type,event_category,description,metadata,user_id,created_by)
            VALUES (%s,'sender_created','user',%s,%s::jsonb,%s,'mailindbakke')""",
            (email_id, 'Virksomhed og kontakt tilknyttet fra mail', json.dumps({'fingerprint': fingerprint, 'result': result}), user_id))
        return result