"""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