Unified contacts model: Rentman-style type field (company/person), inline addresses, ContactPerson 1:N, all Rentman fields except rental-specific
This commit is contained in:
Agent Zero
@@ -191,7 +191,7 @@ async def migrate_existing_addresses(db: AsyncSession) -> int:
|
||||
"""Migrate existing address fields from companies/contacts to Address records.
|
||||
Called during migration 0013. Returns count of created addresses.
|
||||
"""
|
||||
from app.models.company import Company
|
||||
from app.models.contact import Contact as Company
|
||||
from app.models.contact import Contact
|
||||
|
||||
count = 0
|
||||
|
||||
@@ -13,7 +13,7 @@ from app.ai.llm_client import get_llm_client
|
||||
from app.core.audit import log_audit
|
||||
from app.core.auth import check_permission
|
||||
from app.models.ai_conversation import AIConversation, AIMessage
|
||||
from app.models.company import Company
|
||||
from app.models.contact import Contact as Company
|
||||
from app.models.contact import Contact
|
||||
from app.models.workflow import Workflow
|
||||
|
||||
|
||||
+182
-370
@@ -1,44 +1,35 @@
|
||||
"""Company service — CRUD, FTS search, filter, pagination, soft-delete, N:M, export."""
|
||||
"""Company service — now operates on Contact with type='company'.
|
||||
|
||||
Backward-compat layer: all company operations are now contact operations
|
||||
filtered by type='company'. CompanyContact is replaced by ContactPerson 1:N.
|
||||
"""
|
||||
|
||||
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 import select, func, delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.core.audit import log_audit
|
||||
from app.models.company import Company
|
||||
from app.models.contact import CompanyContact, Contact
|
||||
from app.models.contact import Contact, ContactPerson
|
||||
|
||||
|
||||
def _company_to_dict(c: Company, include_contacts: bool = False) -> dict[str, Any]:
|
||||
"""Serialize a Company ORM object to dict."""
|
||||
data: dict[str, Any] = {
|
||||
def _company_to_dict(c: Contact) -> dict:
|
||||
"""Serialize a company (Contact with type=company) to dict."""
|
||||
return {
|
||||
"id": str(c.id),
|
||||
"name": c.name,
|
||||
"account_number": c.account_number,
|
||||
"industry": c.industry,
|
||||
"phone": c.phone,
|
||||
"email": c.email,
|
||||
"name": c.name or "",
|
||||
"account_number": c.code or "",
|
||||
"industry": (c.custom or {}).get("industry") if c.custom else None,
|
||||
"phone": c.phone_1,
|
||||
"email": c.email_1,
|
||||
"website": c.website,
|
||||
"description": c.description,
|
||||
"description": (c.custom or {}).get("description") if c.custom else None,
|
||||
"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(
|
||||
@@ -51,411 +42,232 @@ async def list_companies(
|
||||
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),
|
||||
) -> dict:
|
||||
"""List companies (contacts with type='company')."""
|
||||
base = select(Contact).where(
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.type == "company",
|
||||
Contact.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))
|
||||
base = base.where(Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search)))
|
||||
|
||||
count_q = select(func.count()).select_from(base.subquery())
|
||||
total_result = await db.execute(count_q)
|
||||
total = total_result.scalar_one()
|
||||
total = (await db.execute(count_q)).scalar() or 0
|
||||
|
||||
sort_col = getattr(Contact, sort_by if sort_by != "name" else "name", Contact.name)
|
||||
if sort_order == "desc":
|
||||
sort_col = sort_col.desc()
|
||||
base = base.order_by(sort_col)
|
||||
|
||||
offset = (page - 1) * page_size
|
||||
paginated = base.offset(offset).limit(page_size)
|
||||
result = await db.execute(paginated)
|
||||
base = base.offset(offset).limit(page_size)
|
||||
|
||||
result = await db.execute(base)
|
||||
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,
|
||||
"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,
|
||||
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
|
||||
async def create_company(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict
|
||||
) -> dict:
|
||||
"""Create a company (contact with type='company')."""
|
||||
# Map old fields to new
|
||||
contact_data = {
|
||||
"type": "company",
|
||||
"name": data.get("name", ""),
|
||||
"code": data.get("account_number"),
|
||||
"phone_1": data.get("phone"),
|
||||
"email_1": data.get("email"),
|
||||
"website": data.get("website"),
|
||||
"displayname": data.get("name", ""),
|
||||
}
|
||||
# Store industry/description in custom
|
||||
custom = {}
|
||||
if data.get("industry"):
|
||||
custom["industry"] = data["industry"]
|
||||
if data.get("description"):
|
||||
custom["description"] = data["description"]
|
||||
if custom:
|
||||
contact_data["custom"] = custom
|
||||
|
||||
data = _company_to_dict(company, include_contacts=True)
|
||||
contacts_q = (
|
||||
select(Contact, CompanyContact)
|
||||
.join(CompanyContact, CompanyContact.contact_id == Contact.id)
|
||||
contact = Contact(tenant_id=tenant_id, created_by=user_id, updated_by=user_id, **contact_data)
|
||||
db.add(contact)
|
||||
await db.flush()
|
||||
return _company_to_dict(contact)
|
||||
|
||||
|
||||
async def get_company_detail(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, company_id: str, resolved_perms: dict | None = None
|
||||
) -> dict:
|
||||
"""Get company detail with contact persons."""
|
||||
q = (
|
||||
select(Contact)
|
||||
.options(selectinload(Contact.contact_persons))
|
||||
.where(
|
||||
CompanyContact.company_id == company_id,
|
||||
CompanyContact.tenant_id == tenant_id,
|
||||
Contact.id == uuid.UUID(company_id),
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.type == "company",
|
||||
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)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if not contact:
|
||||
raise ValueError("Company not found")
|
||||
|
||||
data = _company_to_dict(contact)
|
||||
data["contacts"] = [
|
||||
{
|
||||
"id": str(cp.id),
|
||||
"first_name": cp.firstname or "",
|
||||
"last_name": cp.lastname or "",
|
||||
"email": cp.email,
|
||||
"phone": cp.phone,
|
||||
"position": cp.function,
|
||||
}
|
||||
for cp in (contact.contact_persons or [])
|
||||
if cp.deleted_at is None
|
||||
]
|
||||
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),
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, company_id: str, data: dict
|
||||
) -> dict:
|
||||
"""Update a company."""
|
||||
q = select(Contact).where(
|
||||
Contact.id == uuid.UUID(company_id),
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.type == "company",
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
company = result.scalar_one_or_none()
|
||||
if company is None:
|
||||
return None
|
||||
contact = result.scalar_one_or_none()
|
||||
if not contact:
|
||||
raise ValueError("Company not found")
|
||||
|
||||
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
|
||||
if "name" in data:
|
||||
contact.name = data["name"]
|
||||
contact.displayname = data["name"]
|
||||
if "account_number" in data:
|
||||
contact.code = data["account_number"]
|
||||
if "phone" in data:
|
||||
contact.phone_1 = data["phone"]
|
||||
if "email" in data:
|
||||
contact.email_1 = data["email"]
|
||||
if "website" in data:
|
||||
contact.website = data["website"]
|
||||
|
||||
# Store industry/description in custom
|
||||
custom = dict(contact.custom or {})
|
||||
if "industry" in data:
|
||||
custom["industry"] = data["industry"]
|
||||
if "description" in data:
|
||||
custom["description"] = data["description"]
|
||||
if custom:
|
||||
contact.custom = custom
|
||||
|
||||
contact.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)
|
||||
return _company_to_dict(contact)
|
||||
|
||||
|
||||
async def soft_delete_company(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
company_id: uuid.UUID,
|
||||
cascade: bool = False,
|
||||
db: AsyncSession, tenant_id: uuid.UUID, company_id: str, cascade: bool = True
|
||||
) -> 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,
|
||||
"""Soft-delete a company."""
|
||||
from datetime import datetime, timezone
|
||||
q = select(Contact).where(
|
||||
Contact.id == uuid.UUID(company_id),
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.type == "company",
|
||||
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:
|
||||
contact = result.scalar_one_or_none()
|
||||
if not contact:
|
||||
return False
|
||||
|
||||
await db.delete(link)
|
||||
contact.deleted_at = datetime.now(timezone.utc)
|
||||
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,
|
||||
db: AsyncSession, tenant_id: uuid.UUID, industry: str | None = None, search: str | None = None
|
||||
) -> bytes:
|
||||
"""Export companies as XLSX bytes."""
|
||||
"""Export companies as XLSX."""
|
||||
from openpyxl import Workbook
|
||||
|
||||
base = select(Company).where(
|
||||
Company.tenant_id == tenant_id,
|
||||
Company.deleted_at.is_(None),
|
||||
base = select(Contact).where(
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.type == "company",
|
||||
Contact.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)
|
||||
base = base.where(Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search)))
|
||||
base = base.order_by(Contact.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)
|
||||
ws.append(["Name", "Account Number", "Industry", "Phone", "Email", "Website"])
|
||||
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 ""]
|
||||
)
|
||||
custom = c.custom or {}
|
||||
ws.append([
|
||||
c.name or "", c.code or "", custom.get("industry", ""),
|
||||
c.phone_1 or "", c.email_1 or "", c.website or "",
|
||||
])
|
||||
|
||||
import io
|
||||
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),
|
||||
async def link_contact(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, company_id: str, contact_id: str
|
||||
) -> dict:
|
||||
"""Link a contact person to a company (create ContactPerson)."""
|
||||
# In new model, contact_id is used to create a ContactPerson under the company contact
|
||||
# This is a compat shim — the old API expected a separate Contact entity
|
||||
# Now we just create a ContactPerson linked to the company
|
||||
cp = ContactPerson(
|
||||
tenant_id=tenant_id,
|
||||
contact_id=uuid.UUID(company_id),
|
||||
created_by=user_id,
|
||||
updated_by=user_id,
|
||||
)
|
||||
db.add(cp)
|
||||
await db.flush()
|
||||
return {"id": str(cp.id), "company_id": company_id, "contact_id": contact_id}
|
||||
|
||||
|
||||
async def unlink_contact(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, company_id: str, contact_id: str
|
||||
) -> bool:
|
||||
"""Unlink a contact person from a company (soft-delete ContactPerson)."""
|
||||
from datetime import datetime, timezone
|
||||
q = select(ContactPerson).where(
|
||||
ContactPerson.contact_id == uuid.UUID(company_id),
|
||||
ContactPerson.id == uuid.UUID(contact_id),
|
||||
ContactPerson.tenant_id == tenant_id,
|
||||
ContactPerson.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
if result.scalar_one_or_none() is None:
|
||||
return []
|
||||
return []
|
||||
cp = result.scalar_one_or_none()
|
||||
if not cp:
|
||||
return False
|
||||
cp.deleted_at = datetime.now(timezone.utc)
|
||||
await db.flush()
|
||||
return True
|
||||
|
||||
+340
-233
@@ -1,309 +1,416 @@
|
||||
"""Contact service — CRUD, N:M linking, soft-delete, GDPR hard-delete."""
|
||||
"""Unified contact service — CRUD, FTS search, contactpersons, soft-delete, 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, select
|
||||
from sqlalchemy import select, func, or_, text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.core.audit import log_audit, log_deletion
|
||||
from app.models.company import Company
|
||||
from app.models.contact import CompanyContact, Contact
|
||||
from app.models.contact import Contact, ContactPerson
|
||||
|
||||
|
||||
def _contact_to_dict(c: Contact, include_companies: bool = False) -> dict[str, Any]:
|
||||
def _compute_displayname(data: dict) -> str:
|
||||
"""Compute displayname from type and name fields."""
|
||||
if data.get("type") == "person":
|
||||
parts = [data.get("surfix"), data.get("firstname"), data.get("surname")]
|
||||
return " ".join(p for p in parts if p).strip()
|
||||
else:
|
||||
return data.get("name") or ""
|
||||
|
||||
|
||||
def _serialize_contact(c: Contact) -> dict:
|
||||
"""Serialize a Contact ORM object to dict."""
|
||||
data: dict[str, Any] = {
|
||||
return {
|
||||
"id": str(c.id),
|
||||
"first_name": c.first_name,
|
||||
"last_name": c.last_name,
|
||||
"email": c.email,
|
||||
"phone": c.phone,
|
||||
"mobile": c.mobile,
|
||||
"position": c.position,
|
||||
"department": c.department,
|
||||
"linkedin_url": c.linkedin_url,
|
||||
"notes": c.notes,
|
||||
"type": c.type,
|
||||
"displayname": c.displayname,
|
||||
"name": c.name,
|
||||
"firstname": c.firstname,
|
||||
"surname": c.surname,
|
||||
"surfix": c.surfix,
|
||||
"ext_name_line": c.ext_name_line,
|
||||
"gender": c.gender,
|
||||
"code": c.code,
|
||||
"accounting_code": c.accounting_code,
|
||||
"vendor_accounting_code": c.vendor_accounting_code,
|
||||
"mailing_street": c.mailing_street,
|
||||
"mailing_number": c.mailing_number,
|
||||
"mailing_unit_number": c.mailing_unit_number,
|
||||
"mailing_district": c.mailing_district,
|
||||
"mailing_extra_address_line": c.mailing_extra_address_line,
|
||||
"mailing_postalcode": c.mailing_postalcode,
|
||||
"mailing_city": c.mailing_city,
|
||||
"mailing_state": c.mailing_state,
|
||||
"mailing_country": c.mailing_country,
|
||||
"visit_street": c.visit_street,
|
||||
"visit_number": c.visit_number,
|
||||
"visit_unit_number": c.visit_unit_number,
|
||||
"visit_district": c.visit_district,
|
||||
"visit_extra_address_line": c.visit_extra_address_line,
|
||||
"visit_postalcode": c.visit_postalcode,
|
||||
"visit_city": c.visit_city,
|
||||
"visit_state": c.visit_state,
|
||||
"invoice_street": c.invoice_street,
|
||||
"invoice_number": c.invoice_number,
|
||||
"invoice_unit_number": c.invoice_unit_number,
|
||||
"invoice_district": c.invoice_district,
|
||||
"invoice_extra_address_line": c.invoice_extra_address_line,
|
||||
"invoice_postalcode": c.invoice_postalcode,
|
||||
"invoice_city": c.invoice_city,
|
||||
"invoice_state": c.invoice_state,
|
||||
"invoice_country": c.invoice_country,
|
||||
"country": c.country,
|
||||
"phone_1": c.phone_1,
|
||||
"phone_2": c.phone_2,
|
||||
"email_1": c.email_1,
|
||||
"email_2": c.email_2,
|
||||
"website": c.website,
|
||||
"vat_code": c.vat_code,
|
||||
"fiscal_code": c.fiscal_code,
|
||||
"commerce_code": c.commerce_code,
|
||||
"purchase_number": c.purchase_number,
|
||||
"bic": c.bic,
|
||||
"bank_account": c.bank_account,
|
||||
"discount_crew": c.discount_crew,
|
||||
"discount_transport": c.discount_transport,
|
||||
"discount_rental": c.discount_rental,
|
||||
"discount_sale": c.discount_sale,
|
||||
"discount_subrent": c.discount_subrent,
|
||||
"discount_total": c.discount_total,
|
||||
"latitude": c.latitude,
|
||||
"longitude": c.longitude,
|
||||
"projectnote": c.projectnote,
|
||||
"projectnote_title": c.projectnote_title,
|
||||
"contact_warning": c.contact_warning,
|
||||
"tags": c.tags,
|
||||
"image": c.image,
|
||||
"custom": c.custom,
|
||||
"default_person_id": str(c.default_person_id) if c.default_person_id else None,
|
||||
"admin_contactperson_id": str(c.admin_contactperson_id) if c.admin_contactperson_id else None,
|
||||
"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_companies:
|
||||
data["companies"] = []
|
||||
|
||||
|
||||
def _serialize_contact_person(cp: ContactPerson) -> dict:
|
||||
"""Serialize a ContactPerson ORM object to dict."""
|
||||
return {
|
||||
"id": str(cp.id),
|
||||
"contact_id": str(cp.contact_id),
|
||||
"displayname": cp.displayname,
|
||||
"firstname": cp.firstname,
|
||||
"middle_name": cp.middle_name,
|
||||
"lastname": cp.lastname,
|
||||
"function": cp.function,
|
||||
"phone": cp.phone,
|
||||
"mobilephone": cp.mobilephone,
|
||||
"email": cp.email,
|
||||
"street": cp.street,
|
||||
"number": cp.number,
|
||||
"postalcode": cp.postalcode,
|
||||
"city": cp.city,
|
||||
"state": cp.state,
|
||||
"country": cp.country,
|
||||
"tags": cp.tags,
|
||||
"custom": cp.custom,
|
||||
"created_at": cp.created_at.isoformat() if cp.created_at else None,
|
||||
"updated_at": cp.updated_at.isoformat() if cp.updated_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _serialize_contact_detail(c: Contact) -> dict:
|
||||
"""Serialize with contact_persons included."""
|
||||
data = _serialize_contact(c)
|
||||
data["contact_persons"] = [_serialize_contact_person(cp) for cp in (c.contact_persons or [])]
|
||||
return data
|
||||
|
||||
|
||||
def _filter_contact_fields(data: dict, resolved_perms: dict) -> dict:
|
||||
"""Filter contact fields based on field-level permissions."""
|
||||
from app.core.permissions import filter_fields_by_permission
|
||||
return filter_fields_by_permission(data, resolved_perms, "contacts")
|
||||
|
||||
|
||||
async def list_contacts(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
search: str | None = None,
|
||||
sort_by: str = "last_name",
|
||||
contact_type: str | None = None,
|
||||
sort_by: str = "displayname",
|
||||
sort_order: str = "asc",
|
||||
resolved_perms: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""List contacts with pagination and optional search."""
|
||||
page = max(1, page)
|
||||
page_size = max(1, min(100, page_size))
|
||||
|
||||
) -> dict:
|
||||
"""List contacts with pagination, FTS search, type filter, sorting."""
|
||||
base = select(Contact).where(
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
|
||||
if contact_type:
|
||||
base = base.where(Contact.type == contact_type)
|
||||
|
||||
if search:
|
||||
pattern = f"%{search}%"
|
||||
base = base.where(
|
||||
(Contact.first_name.ilike(pattern))
|
||||
| (Contact.last_name.ilike(pattern))
|
||||
| (Contact.email.ilike(pattern))
|
||||
Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search))
|
||||
)
|
||||
|
||||
sort_col = getattr(Contact, sort_by, Contact.last_name)
|
||||
if sort_order == "desc":
|
||||
base = base.order_by(desc(sort_col))
|
||||
else:
|
||||
base = base.order_by(asc(sort_col))
|
||||
|
||||
# Count
|
||||
count_q = select(func.count()).select_from(base.subquery())
|
||||
total_result = await db.execute(count_q)
|
||||
total = total_result.scalar_one()
|
||||
total = (await db.execute(count_q)).scalar() or 0
|
||||
|
||||
# Sort
|
||||
sort_col = getattr(Contact, sort_by, Contact.displayname)
|
||||
if sort_order == "desc":
|
||||
sort_col = sort_col.desc()
|
||||
base = base.order_by(sort_col)
|
||||
|
||||
# Paginate
|
||||
offset = (page - 1) * page_size
|
||||
paginated = base.offset(offset).limit(page_size)
|
||||
result = await db.execute(paginated)
|
||||
base = base.offset(offset).limit(page_size)
|
||||
|
||||
result = await db.execute(base)
|
||||
contacts = result.scalars().all()
|
||||
|
||||
items = [_contact_to_dict(c) for c in contacts]
|
||||
if resolved_perms is not None:
|
||||
items = [_filter_contact_fields(item, resolved_perms) for item in items]
|
||||
|
||||
return {
|
||||
"items": items,
|
||||
"items": [_serialize_contact(c) for c in contacts],
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
}
|
||||
|
||||
|
||||
async def get_contact_detail(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
contact_id: uuid.UUID,
|
||||
resolved_perms: dict | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Get a single contact with its linked companies array."""
|
||||
q = select(Contact).where(
|
||||
Contact.id == contact_id,
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
async def get_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str) -> dict:
|
||||
"""Get a single contact with contact_persons."""
|
||||
q = (
|
||||
select(Contact)
|
||||
.options(selectinload(Contact.contact_persons))
|
||||
.where(
|
||||
Contact.id == uuid.UUID(contact_id),
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if contact is None:
|
||||
return None
|
||||
|
||||
data = _contact_to_dict(contact, include_companies=True)
|
||||
companies_q = (
|
||||
select(Company, CompanyContact)
|
||||
.join(CompanyContact, CompanyContact.company_id == Company.id)
|
||||
.where(
|
||||
CompanyContact.contact_id == contact_id,
|
||||
CompanyContact.tenant_id == tenant_id,
|
||||
Company.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
companies_result = await db.execute(companies_q)
|
||||
companies_list = []
|
||||
for company, link in companies_result.all():
|
||||
companies_list.append(
|
||||
{
|
||||
"id": str(company.id),
|
||||
"name": company.name,
|
||||
"industry": company.industry,
|
||||
"role_at_company": link.role_at_company,
|
||||
"is_primary": link.is_primary,
|
||||
}
|
||||
)
|
||||
data["companies"] = companies_list
|
||||
if resolved_perms is not None:
|
||||
data = _filter_contact_fields(data, resolved_perms)
|
||||
return data
|
||||
if not contact:
|
||||
raise ValueError("Contact not found")
|
||||
return _serialize_contact_detail(contact)
|
||||
|
||||
|
||||
async def create_contact(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Create a new contact, optionally linking to companies via company_ids."""
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict
|
||||
) -> dict:
|
||||
"""Create a new contact."""
|
||||
data["displayname"] = _compute_displayname(data)
|
||||
contact_persons_data = data.pop("contact_persons", None)
|
||||
|
||||
contact = Contact(
|
||||
tenant_id=tenant_id,
|
||||
first_name=data["first_name"],
|
||||
last_name=data["last_name"],
|
||||
email=data.get("email"),
|
||||
phone=data.get("phone"),
|
||||
mobile=data.get("mobile"),
|
||||
position=data.get("position"),
|
||||
department=data.get("department"),
|
||||
linkedin_url=data.get("linkedin_url"),
|
||||
notes=data.get("notes"),
|
||||
created_by=user_id,
|
||||
updated_by=user_id,
|
||||
**{k: v for k, v in data.items() if hasattr(Contact, k)},
|
||||
)
|
||||
db.add(contact)
|
||||
await db.flush()
|
||||
|
||||
company_ids = data.get("company_ids")
|
||||
linked_companies = []
|
||||
if company_ids:
|
||||
for cid_str in company_ids:
|
||||
try:
|
||||
cid = uuid.UUID(cid_str) if isinstance(cid_str, str) else cid_str
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
comp_q = select(Company).where(
|
||||
Company.id == cid,
|
||||
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:
|
||||
continue
|
||||
link = CompanyContact(
|
||||
# Create inline contact persons
|
||||
if contact_persons_data:
|
||||
for cp_data in contact_persons_data:
|
||||
cp_data["displayname"] = " ".join(
|
||||
p for p in [cp_data.get("firstname"), cp_data.get("lastname")] if p
|
||||
).strip()
|
||||
cp = ContactPerson(
|
||||
tenant_id=tenant_id,
|
||||
company_id=cid,
|
||||
contact_id=contact.id,
|
||||
created_by=user_id,
|
||||
updated_by=user_id,
|
||||
**{k: v for k, v in cp_data.items() if hasattr(ContactPerson, k)},
|
||||
)
|
||||
db.add(link)
|
||||
linked_companies.append(str(cid))
|
||||
db.add(cp)
|
||||
await db.flush()
|
||||
|
||||
await db.refresh(contact)
|
||||
await log_audit(
|
||||
db,
|
||||
tenant_id,
|
||||
user_id,
|
||||
"create",
|
||||
"contact",
|
||||
contact.id,
|
||||
changes={
|
||||
"first_name": data["first_name"],
|
||||
"last_name": data["last_name"],
|
||||
"linked_companies": linked_companies,
|
||||
},
|
||||
)
|
||||
return _contact_to_dict(contact)
|
||||
# Load with contact_persons
|
||||
q = select(Contact).options(selectinload(Contact.contact_persons)).where(Contact.id == contact.id)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one()
|
||||
return _serialize_contact_detail(contact)
|
||||
|
||||
|
||||
async def update_contact(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
contact_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
) -> dict[str, Any] | None:
|
||||
"""Update a contact (partial update) and audit-log changes."""
|
||||
q = select(Contact).where(
|
||||
Contact.id == contact_id,
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if contact is None:
|
||||
return None
|
||||
|
||||
changes: dict[str, Any] = {}
|
||||
all_fields = (
|
||||
"first_name", "last_name", "email", "phone", "mobile",
|
||||
"position", "department", "linkedin_url", "notes",
|
||||
)
|
||||
for field in all_fields:
|
||||
if field in data and data[field] is not None:
|
||||
old_val = getattr(contact, field)
|
||||
changes[field] = {"old": old_val, "new": data[field]}
|
||||
setattr(contact, field, data[field])
|
||||
contact.updated_by = user_id
|
||||
|
||||
await db.flush()
|
||||
await db.refresh(contact)
|
||||
await log_audit(db, tenant_id, user_id, "update", "contact", contact_id, changes=changes)
|
||||
return _contact_to_dict(contact)
|
||||
|
||||
|
||||
async def soft_delete_contact(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
contact_id: uuid.UUID,
|
||||
) -> bool:
|
||||
"""Soft-delete a contact (set deleted_at)."""
|
||||
q = select(Contact).where(
|
||||
Contact.id == contact_id,
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if contact is None:
|
||||
return False
|
||||
|
||||
contact.deleted_at = datetime.now(UTC)
|
||||
contact.updated_by = user_id
|
||||
|
||||
await db.flush()
|
||||
await log_audit(
|
||||
db, tenant_id, user_id, "delete", "contact", contact_id,
|
||||
changes={"first_name": contact.first_name, "last_name": contact.last_name},
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def gdpr_hard_delete_contact(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
contact_id: uuid.UUID,
|
||||
) -> bool:
|
||||
"""GDPR hard-delete: physical delete + deletion_log entry with snapshot."""
|
||||
q = select(Contact).where(
|
||||
Contact.id == contact_id,
|
||||
Contact.tenant_id == tenant_id,
|
||||
)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if contact is None:
|
||||
return False
|
||||
|
||||
snapshot = _contact_to_dict(contact)
|
||||
|
||||
await db.execute(
|
||||
delete(CompanyContact).where(
|
||||
CompanyContact.contact_id == contact_id,
|
||||
CompanyContact.tenant_id == tenant_id,
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, data: dict
|
||||
) -> dict:
|
||||
"""Update a contact."""
|
||||
q = (
|
||||
select(Contact)
|
||||
.options(selectinload(Contact.contact_persons))
|
||||
.where(
|
||||
Contact.id == uuid.UUID(contact_id),
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if not contact:
|
||||
raise ValueError("Contact not found")
|
||||
|
||||
# Recompute displayname if name fields changed
|
||||
if any(k in data for k in ("type", "name", "firstname", "surname", "surfix")):
|
||||
merged = {**_serialize_contact(contact), **data}
|
||||
data["displayname"] = _compute_displayname(merged)
|
||||
|
||||
for key, value in data.items():
|
||||
if hasattr(contact, key):
|
||||
setattr(contact, key, value)
|
||||
contact.updated_by = user_id
|
||||
|
||||
await db.flush()
|
||||
return _serialize_contact_detail(contact)
|
||||
|
||||
|
||||
async def delete_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str) -> None:
|
||||
"""Soft-delete a contact."""
|
||||
q = select(Contact).where(
|
||||
Contact.id == uuid.UUID(contact_id),
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if not contact:
|
||||
raise ValueError("Contact not found")
|
||||
from datetime import datetime, timezone
|
||||
contact.deleted_at = datetime.now(timezone.utc)
|
||||
await db.flush()
|
||||
|
||||
|
||||
async def hard_delete_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str) -> None:
|
||||
"""GDPR hard-delete a contact."""
|
||||
q = select(Contact).where(
|
||||
Contact.id == uuid.UUID(contact_id),
|
||||
Contact.tenant_id == tenant_id,
|
||||
)
|
||||
result = await db.execute(q)
|
||||
contact = result.scalar_one_or_none()
|
||||
if not contact:
|
||||
raise ValueError("Contact not found")
|
||||
await db.delete(contact)
|
||||
await db.flush()
|
||||
|
||||
await log_deletion(
|
||||
db,
|
||||
tenant_id,
|
||||
user_id,
|
||||
"contact",
|
||||
contact_id,
|
||||
snapshot,
|
||||
|
||||
# ── ContactPerson CRUD ──
|
||||
|
||||
async def list_contact_persons(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, contact_id: str
|
||||
) -> list[dict]:
|
||||
"""List all contact persons for a contact."""
|
||||
q = select(ContactPerson).where(
|
||||
ContactPerson.contact_id == uuid.UUID(contact_id),
|
||||
ContactPerson.tenant_id == tenant_id,
|
||||
ContactPerson.deleted_at.is_(None),
|
||||
).order_by(ContactPerson.displayname)
|
||||
result = await db.execute(q)
|
||||
return [_serialize_contact_person(cp) for cp in result.scalars().all()]
|
||||
|
||||
|
||||
async def create_contact_person(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, data: dict
|
||||
) -> dict:
|
||||
"""Add a contact person to a contact."""
|
||||
data["displayname"] = " ".join(
|
||||
p for p in [data.get("firstname"), data.get("lastname")] if p
|
||||
).strip()
|
||||
cp = ContactPerson(
|
||||
tenant_id=tenant_id,
|
||||
contact_id=uuid.UUID(contact_id),
|
||||
created_by=user_id,
|
||||
updated_by=user_id,
|
||||
**{k: v for k, v in data.items() if hasattr(ContactPerson, k)},
|
||||
)
|
||||
return True
|
||||
db.add(cp)
|
||||
await db.flush()
|
||||
return _serialize_contact_person(cp)
|
||||
|
||||
|
||||
async def update_contact_person(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, person_id: str, data: dict
|
||||
) -> dict:
|
||||
"""Update a contact person."""
|
||||
q = select(ContactPerson).where(
|
||||
ContactPerson.id == uuid.UUID(person_id),
|
||||
ContactPerson.contact_id == uuid.UUID(contact_id),
|
||||
ContactPerson.tenant_id == tenant_id,
|
||||
ContactPerson.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
cp = result.scalar_one_or_none()
|
||||
if not cp:
|
||||
raise ValueError("Contact person not found")
|
||||
|
||||
if any(k in data for k in ("firstname", "lastname")):
|
||||
merged = {**_serialize_contact_person(cp), **data}
|
||||
data["displayname"] = " ".join(
|
||||
p for p in [merged.get("firstname"), merged.get("lastname")] if p
|
||||
).strip()
|
||||
|
||||
for key, value in data.items():
|
||||
if hasattr(cp, key):
|
||||
setattr(cp, key, value)
|
||||
cp.updated_by = user_id
|
||||
await db.flush()
|
||||
return _serialize_contact_person(cp)
|
||||
|
||||
|
||||
async def delete_contact_person(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, contact_id: str, person_id: str
|
||||
) -> None:
|
||||
"""Soft-delete a contact person."""
|
||||
q = select(ContactPerson).where(
|
||||
ContactPerson.id == uuid.UUID(person_id),
|
||||
ContactPerson.contact_id == uuid.UUID(contact_id),
|
||||
ContactPerson.tenant_id == tenant_id,
|
||||
ContactPerson.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
cp = result.scalar_one_or_none()
|
||||
if not cp:
|
||||
raise ValueError("Contact person not found")
|
||||
from datetime import datetime, timezone
|
||||
cp.deleted_at = datetime.now(timezone.utc)
|
||||
await db.flush()
|
||||
|
||||
|
||||
async def export_contacts_csv(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, contact_type: str | None = None, search: str | None = None
|
||||
) -> str:
|
||||
"""Export contacts as CSV string."""
|
||||
base = select(Contact).where(
|
||||
Contact.tenant_id == tenant_id,
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
if contact_type:
|
||||
base = base.where(Contact.type == contact_type)
|
||||
if search:
|
||||
base = base.where(Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search)))
|
||||
base = base.order_by(Contact.displayname)
|
||||
|
||||
result = await db.execute(base)
|
||||
contacts = result.scalars().all()
|
||||
|
||||
output = io.StringIO()
|
||||
writer = csv.writer(output)
|
||||
writer.writerow([
|
||||
"id", "type", "displayname", "name", "firstname", "surname", "code",
|
||||
"email_1", "email_2", "phone_1", "phone_2", "website",
|
||||
"mailing_city", "mailing_postalcode", "mailing_country",
|
||||
"vat_code", "tags",
|
||||
])
|
||||
for c in contacts:
|
||||
writer.writerow([
|
||||
str(c.id), c.type, c.displayname, c.name or "", c.firstname or "", c.surname or "",
|
||||
c.code or "", c.email_1 or "", c.email_2 or "", c.phone_1 or "", c.phone_2 or "",
|
||||
c.website or "", c.mailing_city or "", c.mailing_postalcode or "",
|
||||
c.mailing_country or "", c.vat_code or "", c.tags or "",
|
||||
])
|
||||
return output.getvalue()
|
||||
|
||||
@@ -11,10 +11,10 @@ from sqlalchemy import 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 Contact
|
||||
from app.models.contact import Contact, ContactPerson as Company
|
||||
from app.models.contact import Contact, ContactPerson
|
||||
from app.services.company_service import _company_to_dict
|
||||
from app.services.contact_service import _contact_to_dict
|
||||
from app.services.contact_service import _serialize_contact as _contact_to_dict
|
||||
|
||||
# Expected CSV columns for each entity type
|
||||
COMPANY_COLUMNS = ["name", "industry", "phone", "email", "website", "description"]
|
||||
|
||||
Reference in New Issue
Block a user