Leopoldadmin/leocrm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 1C: Frontend unified contact UI

Browse Source
This commit is contained in:
Agent Zero
2026-07-23 17:17:32 +02:00
parent a8331fbc2b
commit 879106c4eb
62 changed files with 552 additions and 1276 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/address_service.py
+3 -3
View File
@@ -13,7 +13,7 @@ from app.core.audit import log_audit
from app.models.address import Address
VALID_ENTITY_TYPES = {"company", "contact"}
VALID_ENTITY_TYPES = {"contact"}
VALID_ADDRESS_TYPES = {"billing", "shipping", "headquarters", "branch", "private", "other"}
@@ -206,7 +206,7 @@ async def migrate_existing_addresses(db: AsyncSession) -> int:
existing = await db.execute(
select(Address).where(
Address.tenant_id == company.tenant_id,
Address.entity_type == "company",
Address.entity_type == "contact",
Address.entity_id == company.id,
Address.address_type == "headquarters",
Address.deleted_at.is_(None),
@@ -217,7 +217,7 @@ async def migrate_existing_addresses(db: AsyncSession) -> int:
addr = Address(
tenant_id=company.tenant_id,
entity_type="company",
entity_type="contact",
entity_id=company.id,
label="Hauptsitz",
address_type="headquarters",
app/services/ai_copilot_service.py
+10 -103
View File
@@ -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.contact import Contact as Company
from app.models.contact import Contact
from app.models.contact import Contact
from app.models.workflow import Workflow
@@ -306,16 +306,14 @@ async def _execute_api_action(
) -> dict[str, Any]:
"""Execute an API action directly against the database.
Supports companies and contacts CRUD, plus workflow listing.
Supports contacts CRUD (including company-type contacts), plus workflow listing.
"""
# Parse path to determine entity and operation
parts = path.replace("/api/v1/", "").strip("/").split("/")
entity = parts[0] if parts else ""
entity_id = parts[1] if len(parts) > 1 else None
if entity == "companies":
return await _exec_companies(db, tenant_id, user_id, method, entity_id, body)
elif entity == "contacts":
if entity in ("companies", "contacts"):
return await _exec_contacts(db, tenant_id, user_id, method, entity_id, body)
elif entity == "workflows":
return await _exec_workflows(db, tenant_id, user_id, method, entity_id, body)
@@ -323,7 +321,7 @@ async def _execute_api_action(
return {"error": f"Unsupported entity: {entity}", "status_code": 400, "success": False}
async def _exec_companies(
async def _exec_contacts\(
db: AsyncSession,
tenant_id: uuid.UUID,
user_id: uuid.UUID,
@@ -331,99 +329,7 @@ async def _exec_companies(
entity_id: str | None,
body: dict[str, Any],
) -> dict[str, Any]:
"""Execute company operations."""
if method == "GET":
result = await db.execute(
select(Company).where(
Company.tenant_id == tenant_id,
Company.deleted_at.is_(None),
)
)
companies = result.scalars().all()
return {
"success": True,
"status_code": 200,
"data": [{"id": str(c.id), "name": c.name, "industry": c.industry} for c in companies],
}
elif method == "POST":
company = Company(
tenant_id=tenant_id,
name=body.get("name", "Untitled"),
industry=body.get("industry"),
phone=body.get("phone"),
email=body.get("email"),
website=body.get("website"),
description=body.get("description"),
created_by=user_id,
updated_by=user_id,
)
db.add(company)
await db.flush()
return {
"success": True,
"status_code": 201,
"data": {"id": str(company.id), "name": company.name},
}
elif method == "DELETE":
if not entity_id or entity_id == "{id}":
return {"error": "Company ID required", "status_code": 400, "success": False}
comp_uuid = uuid.UUID(entity_id)
result = await db.execute(
select(Company).where(
Company.id == comp_uuid,
Company.tenant_id == tenant_id,
)
)
company = result.scalar_one_or_none()
if company is None:
return {"error": "Company not found", "status_code": 404, "success": False}
company.deleted_at = datetime.now(UTC)
await db.flush()
return {
"success": True,
"status_code": 200,
"data": {"id": str(company.id), "deleted": True},
}
elif method == "PATCH":
if not entity_id or entity_id == "{id}":
return {"error": "Company ID required", "status_code": 400, "success": False}
comp_uuid = uuid.UUID(entity_id)
result = await db.execute(
select(Company).where(
Company.id == comp_uuid,
Company.tenant_id == tenant_id,
Company.deleted_at.is_(None),
)
)
company = result.scalar_one_or_none()
if company is None:
return {"error": "Company not found", "status_code": 404, "success": False}
for key in ("name", "industry", "phone", "email", "website", "description"):
if key in body:
setattr(company, key, body[key])
company.updated_by = user_id
await db.flush()
return {
"success": True,
"status_code": 200,
"data": {"id": str(company.id), "name": company.name, "industry": company.industry},
}
return {"error": f"Unsupported method: {method}", "status_code": 400, "success": False}
async def _exec_contacts(
db: AsyncSession,
tenant_id: uuid.UUID,
user_id: uuid.UUID,
method: str,
entity_id: str | None,
body: dict[str, Any],
) -> dict[str, Any]:
"""Execute contact operations."""
"""Execute contact operations (unified: company + person)."""
if method == "GET":
result = await db.execute(
select(Contact).where(
@@ -435,15 +341,16 @@ async def _exec_contacts(
return {
"success": True,
"status_code": 200,
"data": [{"id": str(c.id), "name": c.name, "email": c.email} for c in contacts],
"data": [{"id": str(c.id), "name": c.name or c.displayname, "email": c.email_1, "type": c.type} for c in contacts],
}
elif method == "POST":
contact = Contact(
tenant_id=tenant_id,
type=body.get("type", "company"),
name=body.get("name", "Untitled"),
email=body.get("email"),
phone=body.get("phone"),
email_1=body.get("email"),
phone_1=body.get("phone"),
created_by=user_id,
updated_by=user_id,
)
@@ -452,7 +359,7 @@ async def _exec_contacts(
return {
"success": True,
"status_code": 201,
"data": {"id": str(contact.id), "name": contact.name},
"data": {"id": str(contact.id), "name": contact.name, "type": contact.type},
}
return {"error": f"Unsupported method: {method}", "status_code": 400, "success": False}
app/services/company_service.py
-273
View File
@@ -1,273 +0,0 @@
"""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 uuid
from typing import Any
from sqlalchemy import select, func, delete
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.models.contact import Contact, ContactPerson
def _company_to_dict(c: Contact) -> dict:
"""Serialize a company (Contact with type=company) to dict."""
return {
"id": str(c.id),
"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.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,
}
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:
"""List companies (contacts with type='company')."""
base = select(Contact).where(
Contact.tenant_id == tenant_id,
Contact.type == "company",
Contact.deleted_at.is_(None),
)
if search:
base = base.where(Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search)))
count_q = select(func.count()).select_from(base.subquery())
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
base = base.offset(offset).limit(page_size)
result = await db.execute(base)
companies = result.scalars().all()
return {
"items": [_company_to_dict(c) for c in companies],
"total": total,
"page": page,
"page_size": page_size,
}
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
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(
Contact.id == uuid.UUID(company_id),
Contact.tenant_id == tenant_id,
Contact.type == "company",
Contact.deleted_at.is_(None),
)
)
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 update_company(
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)
contact = result.scalar_one_or_none()
if not contact:
raise ValueError("Company not found")
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()
return _company_to_dict(contact)
async def soft_delete_company(
db: AsyncSession, tenant_id: uuid.UUID, company_id: str, cascade: bool = True
) -> bool:
"""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),
)
result = await db.execute(q)
contact = result.scalar_one_or_none()
if not contact:
return False
contact.deleted_at = datetime.now(timezone.utc)
await db.flush()
return True
async def export_companies_xlsx(
db: AsyncSession, tenant_id: uuid.UUID, industry: str | None = None, search: str | None = None
) -> bytes:
"""Export companies as XLSX."""
from openpyxl import Workbook
base = select(Contact).where(
Contact.tenant_id == tenant_id,
Contact.type == "company",
Contact.deleted_at.is_(None),
)
if search:
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"
ws.append(["Name", "Account Number", "Industry", "Phone", "Email", "Website"])
for c in companies:
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 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)
cp = result.scalar_one_or_none()
if not cp:
return False
cp.deleted_at = datetime.now(timezone.utc)
await db.flush()
return True