"""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 datetime, timezone from typing import Any from sqlalchemy import select, func, or_, desc, asc, delete from sqlalchemy.ext.asyncio import AsyncSession from app.core.audit import log_audit, log_deletion from app.models.company import Company from app.models.contact import Contact, CompanyContact 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"] = [] # populated by caller if needed return data 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", ) -> dict[str, Any]: """List companies with pagination, FTS search, industry filter, and sorting. Returns {items, total, page, page_size}. """ 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), ) # Industry filter if industry: base = base.where(Company.industry == industry) # FTS search using PostgreSQL tsvector @@ plainto_tsquery, with ILIKE fallback 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), ) ) # Sorting 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 total (without pagination) count_q = select(func.count()).select_from(base.subquery()) total_result = await db.execute(count_q) total = total_result.scalar_one() # Paginate offset = (page - 1) * page_size paginated = base.offset(offset).limit(page_size) result = await db.execute(paginated) companies = result.scalars().all() return { "items": [_company_to_dict(c) for c in companies], "total": total, "page": page, "page_size": page_size, } async def get_company_detail( db: AsyncSession, tenant_id: uuid.UUID, company_id: uuid.UUID, ) -> 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) # Fetch linked contacts 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 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] = {} for field in ("name", "account_number", "industry", "phone", "email", "website", "description"): 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(timezone.utc) company.updated_by = user_id if cascade: # Delete N:M links for this company 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.""" # Verify company exists and is in tenant 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 # Verify contact exists and is in tenant 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 # Check if link already exists 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: # Update existing link 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).""" # Verify company exists 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 [] # Caller should check existence separately return []