"""Company service — CRUD, FTS search, filter, pagination, soft-delete, N:M, export.""" from __future__ import annotations import csv import io import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import asc, delete, desc, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.audit import log_audit from app.models.company import Company from app.models.contact import CompanyContact, Contact def _company_to_dict(c: Company, include_contacts: bool = False) -> dict[str, Any]: """Serialize a Company ORM object to dict.""" data: dict[str, Any] = { "id": str(c.id), "name": c.name, "account_number": c.account_number, "industry": c.industry, "phone": c.phone, "email": c.email, "website": c.website, "description": c.description, "created_at": c.created_at.isoformat() if c.created_at else None, "updated_at": c.updated_at.isoformat() if c.updated_at else None, } if include_contacts: data["contacts"] = [] return data def _filter_company_fields(data: dict, resolved_perms: dict) -> dict: """Filter company fields based on field-level permissions.""" from app.core.permissions import filter_fields_by_permission return filter_fields_by_permission(data, resolved_perms, "companies") async def list_companies( db: AsyncSession, tenant_id: uuid.UUID, page: int = 1, page_size: int = 20, search: str | None = None, industry: str | None = None, sort_by: str = "name", sort_order: str = "asc", resolved_perms: dict | None = None, ) -> dict[str, Any]: """List companies with pagination, FTS search, industry filter, and sorting.""" page = max(1, page) page_size = max(1, min(100, page_size)) base = select(Company).where( Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) if industry: base = base.where(Company.industry == industry) if search: pattern = f"%{search}%" base = base.where( or_( Company.search_tsv.op("@@")(func.plainto_tsquery("english", search)), Company.name.ilike(pattern), Company.industry.ilike(pattern), Company.description.ilike(pattern), ) ) sort_col = getattr(Company, sort_by, Company.name) if sort_order == "desc": base = base.order_by(desc(sort_col)) else: base = base.order_by(asc(sort_col)) count_q = select(func.count()).select_from(base.subquery()) total_result = await db.execute(count_q) total = total_result.scalar_one() offset = (page - 1) * page_size paginated = base.offset(offset).limit(page_size) result = await db.execute(paginated) companies = result.scalars().all() items = [_company_to_dict(c) for c in companies] if resolved_perms is not None: items = [_filter_company_fields(item, resolved_perms) for item in items] return { "items": items, "total": total, "page": page, "page_size": page_size, } async def get_company_detail( db: AsyncSession, tenant_id: uuid.UUID, company_id: uuid.UUID, resolved_perms: dict | None = None, ) -> dict[str, Any] | None: """Get a single company with its contacts array.""" q = select(Company).where( Company.id == company_id, Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) result = await db.execute(q) company = result.scalar_one_or_none() if company is None: return None data = _company_to_dict(company, include_contacts=True) contacts_q = ( select(Contact, CompanyContact) .join(CompanyContact, CompanyContact.contact_id == Contact.id) .where( CompanyContact.company_id == company_id, CompanyContact.tenant_id == tenant_id, Contact.deleted_at.is_(None), ) ) contacts_result = await db.execute(contacts_q) contacts_list = [] for contact, link in contacts_result.all(): contacts_list.append( { "id": str(contact.id), "first_name": contact.first_name, "last_name": contact.last_name, "email": contact.email, "phone": contact.phone, "position": contact.position, "role_at_company": link.role_at_company, "is_primary": link.is_primary, } ) data["contacts"] = contacts_list if resolved_perms is not None: data = _filter_company_fields(data, resolved_perms) return data async def create_company( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict[str, Any], ) -> dict[str, Any]: """Create a new company and audit-log it.""" company = Company( tenant_id=tenant_id, name=data["name"], account_number=data.get("account_number"), industry=data.get("industry"), phone=data.get("phone"), email=data.get("email"), website=data.get("website"), description=data.get("description"), created_by=user_id, updated_by=user_id, ) db.add(company) await db.flush() await db.refresh(company) await log_audit( db, tenant_id, user_id, "create", "company", company.id, changes={"name": data["name"]}, ) return _company_to_dict(company) async def update_company( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, company_id: uuid.UUID, data: dict[str, Any], ) -> dict[str, Any] | None: """Update a company (partial update) and audit-log changes.""" q = select(Company).where( Company.id == company_id, Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) result = await db.execute(q) company = result.scalar_one_or_none() if company is None: return None changes: dict[str, Any] = {} all_fields = ( "name", "account_number", "industry", "phone", "email", "website", "description", ) for field in all_fields: if field in data and data[field] is not None: old_val = getattr(company, field) changes[field] = {"old": old_val, "new": data[field]} setattr(company, field, data[field]) company.updated_by = user_id await db.flush() await db.refresh(company) await log_audit(db, tenant_id, user_id, "update", "company", company_id, changes=changes) return _company_to_dict(company) async def soft_delete_company( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, company_id: uuid.UUID, cascade: bool = False, ) -> bool: """Soft-delete a company. If cascade=True, also soft-delete linked CompanyContact rows.""" q = select(Company).where( Company.id == company_id, Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) result = await db.execute(q) company = result.scalar_one_or_none() if company is None: return False company.deleted_at = datetime.now(UTC) company.updated_by = user_id if cascade: await db.execute( delete(CompanyContact).where( CompanyContact.company_id == company_id, CompanyContact.tenant_id == tenant_id, ) ) await db.flush() await log_audit( db, tenant_id, user_id, "delete", "company", company_id, changes={"name": company.name, "cascade": cascade}, ) return True async def link_contact( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, company_id: uuid.UUID, contact_id: uuid.UUID, role_at_company: str | None = None, is_primary: bool = False, ) -> dict[str, Any] | None: """Link a contact to a company (N:M). Returns link data or None if either not found.""" comp_q = select(Company).where( Company.id == company_id, Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) comp_result = await db.execute(comp_q) if comp_result.scalar_one_or_none() is None: return None cont_q = select(Contact).where( Contact.id == contact_id, Contact.tenant_id == tenant_id, Contact.deleted_at.is_(None), ) cont_result = await db.execute(cont_q) if cont_result.scalar_one_or_none() is None: return None existing_q = select(CompanyContact).where( CompanyContact.company_id == company_id, CompanyContact.contact_id == contact_id, CompanyContact.tenant_id == tenant_id, ) existing_result = await db.execute(existing_q) existing = existing_result.scalar_one_or_none() if existing is not None: existing.role_at_company = role_at_company existing.is_primary = is_primary await db.flush() await log_audit( db, tenant_id, user_id, "link", "company_contact", existing.id, changes={"company_id": str(company_id), "contact_id": str(contact_id)}, ) return { "id": str(existing.id), "company_id": str(company_id), "contact_id": str(contact_id), "role_at_company": role_at_company, "is_primary": is_primary, } link = CompanyContact( tenant_id=tenant_id, company_id=company_id, contact_id=contact_id, role_at_company=role_at_company, is_primary=is_primary, ) db.add(link) await db.flush() await log_audit( db, tenant_id, user_id, "link", "company_contact", link.id, changes={"company_id": str(company_id), "contact_id": str(contact_id)}, ) return { "id": str(link.id), "company_id": str(company_id), "contact_id": str(contact_id), "role_at_company": role_at_company, "is_primary": is_primary, } async def unlink_contact( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, company_id: uuid.UUID, contact_id: uuid.UUID, ) -> bool: """Unlink a contact from a company (N:M).""" q = select(CompanyContact).where( CompanyContact.company_id == company_id, CompanyContact.contact_id == contact_id, CompanyContact.tenant_id == tenant_id, ) result = await db.execute(q) link = result.scalar_one_or_none() if link is None: return False await db.delete(link) await db.flush() await log_audit( db, tenant_id, user_id, "unlink", "company_contact", link.id, changes={"company_id": str(company_id), "contact_id": str(contact_id)}, ) return True async def export_companies_csv( db: AsyncSession, tenant_id: uuid.UUID, industry: str | None = None, search: str | None = None, ) -> str: """Export companies as CSV string.""" base = select(Company).where( Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) if industry: base = base.where(Company.industry == industry) if search: pattern = f"%{search}%" base = base.where( or_( Company.search_tsv.op("@@")(func.plainto_tsquery("english", search)), Company.name.ilike(pattern), Company.industry.ilike(pattern), Company.description.ilike(pattern), ) ) base = base.order_by(Company.name) result = await db.execute(base) companies = result.scalars().all() output = io.StringIO() writer = csv.writer(output) writer.writerow( ["id", "name", "account_number", "industry", "phone", "email", "website", "description"] ) for c in companies: writer.writerow( [str(c.id), c.name, c.account_number or "", c.industry or "", c.phone or "", c.email or "", c.website or "", c.description or ""] ) return output.getvalue() async def export_companies_xlsx( db: AsyncSession, tenant_id: uuid.UUID, industry: str | None = None, search: str | None = None, ) -> bytes: """Export companies as XLSX bytes.""" from openpyxl import Workbook base = select(Company).where( Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) if industry: base = base.where(Company.industry == industry) if search: pattern = f"%{search}%" base = base.where( or_( Company.search_tsv.op("@@")(func.plainto_tsquery("english", search)), Company.name.ilike(pattern), Company.industry.ilike(pattern), Company.description.ilike(pattern), ) ) base = base.order_by(Company.name) result = await db.execute(base) companies = result.scalars().all() wb = Workbook() ws = wb.active ws.title = "Companies" headers = [ "id", "name", "account_number", "industry", "phone", "email", "website", "description", ] ws.append(headers) for c in companies: ws.append( [str(c.id), c.name, c.account_number or "", c.industry or "", c.phone or "", c.email or "", c.website or "", c.description or ""] ) buf = io.BytesIO() wb.save(buf) return buf.getvalue() async def get_company_emails( db: AsyncSession, tenant_id: uuid.UUID, company_id: uuid.UUID, ) -> list[dict[str, Any]]: """Get emails for a company (mail plugin inactive — returns empty array).""" q = select(Company).where( Company.id == company_id, Company.tenant_id == tenant_id, Company.deleted_at.is_(None), ) result = await db.execute(q) if result.scalar_one_or_none() is None: return [] return []