from __future__ import annotations from datetime import date, datetime, timezone, timedelta import secrets from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import Session, selectinload from app.modules.clients.models import Client from app.modules.clients.models import Client, ClientAuditLog from app.modules.client_groups.models import ClientGroup from app.modules.consultants.models import ClientConsultantLink, ConsultantManagedClient, ConsultantProfile, ConsultantWorkspace, ConsultantServiceRequest from app.core.security.passwords import hash_password from app.modules.core.iam.invite_service import issue_invite_token from app.modules.core.iam.models import User from app.modules.core.tenancy.models import Branch from app.modules.core.rbac.models import Role, UserRole from app.modules.services.models import ( ClientServiceSubscription, ClientServiceTaskInstance, ServiceCatalogue, ServiceTaskComment, ) CONSULTANT_TYPES = [ ("external_consultant", "External Consultant"), ("gst_consultant", "GST Consultant"), ("tax_consultant", "Tax Consultant"), ("roc_consultant", "ROC Consultant"), ("payroll_consultant", "Payroll Consultant"), ("franchise_partner", "Franchise Partner"), ("saas_customer", "SaaS Customer"), ] CONSULTANT_RELATIONSHIP_TYPES = [ ("accounts_consultant", "Accounts Consultant"), ("gst_consultant", "GST Consultant"), ("tax_consultant", "Tax Consultant"), ("roc_consultant", "ROC Consultant"), ("payroll_consultant", "Payroll Consultant"), ("audit_coordination", "Audit Coordination"), ("other", "Other"), ] CONSULTANT_MANAGED_CLIENT_STATUSES = [ ("active", "Active"), ("prospect", "Prospect"), ("on_hold", "On Hold"), ("closed", "Closed"), ] CONSULTANT_MANAGED_CLIENT_STAGES = [ ("managed", "Managed Client"), ("lead", "Lead / Prospect"), ("referred_to_firm", "Referred to Audit Firm"), ("linked_to_firm", "Linked to Firm Client"), ] CONSULTANT_CLIENT_CONVERSION_STATUSES = [ ("not_requested", "Not Requested"), ("requested", "Requested"), ("under_review", "Under Review"), ("approved", "Approved / Converted"), ("rejected", "Rejected"), ] CONSULTANT_WORKSPACE_TYPES = [ ("consultant_saas", "Consultant SaaS Workspace"), ("franchise_partner", "Franchise / Referral Partner"), ("platform_partner", "Platform Ecosystem Partner"), ] CONSULTANT_WORKSPACE_PLANS = [ ("starter", "Starter"), ("professional", "Professional"), ("business", "Business"), ("franchise", "Franchise"), ] CONSULTANT_SUBSCRIPTION_STATUSES = [ ("trial", "Trial"), ("active", "Active"), ("suspended", "Suspended"), ("cancelled", "Cancelled"), ] CONSULTANT_BILLING_CYCLES = [ ("manual", "Manual / Not Billed"), ("monthly", "Monthly"), ("quarterly", "Quarterly"), ("yearly", "Yearly"), ] CONSULTANT_ONBOARDING_STATUSES = [ ("draft", "Draft"), ("invited", "Invited"), ("active", "Active"), ("approved", "Approved"), ("suspended", "Suspended"), ("inactive", "Inactive"), ] def normalise_text(value: str | None) -> str | None: text = (value or "").strip() return text or None def _first_branch_id_for_tenant(db: Session, tenant_id: int) -> int | None: return db.execute( select(Branch.id).where(Branch.tenant_id == tenant_id, Branch.is_active.is_(True)).order_by(Branch.id.asc()) ).scalar_one_or_none() def _consultant_role(db: Session) -> Role | None: return db.execute(select(Role).where(Role.name == "Consultant", Role.is_active.is_(True))).scalar_one_or_none() def _user_has_role(db: Session, *, user_id: int, role_id: int) -> bool: return db.execute( select(UserRole.id).where(UserRole.user_id == user_id, UserRole.role_id == role_id) ).scalar_one_or_none() is not None def ensure_consultant_login_user( db: Session, *, tenant_id: int, branch_id: int | None, email: str, full_name: str, password: str | None, invite_user: bool, ) -> tuple[User, str | None]: """Create or update a portal login for a consultant and assign Consultant role. Returns (user, invite_url). Existing users are not overwritten except that the Consultant role is added if missing and basic login flags are enabled. """ login_email = (email or "").strip().lower() if not login_email: raise ValueError("Login email is required to create consultant login.") resolved_branch_id = branch_id or _first_branch_id_for_tenant(db, tenant_id) if not resolved_branch_id: raise ValueError("A branch is required to create consultant login user.") user = db.execute(select(User).where(User.email == login_email)).scalar_one_or_none() if user and int(user.tenant_id) != int(tenant_id): raise ValueError("A user with this login email already exists in another tenant.") if user is None: temp_password = (password or "").strip() or secrets.token_urlsafe(12) user = User( email=login_email, full_name=(full_name or login_email).strip(), password_hash=hash_password(temp_password), tenant_id=tenant_id, branch_id=resolved_branch_id, is_active=True, allow_login=True, is_locked=False, deleted_at=None, must_change_password=True, password_changed_at_utc=None, ) db.add(user) db.flush() else: user.full_name = (full_name or user.full_name or login_email).strip() user.branch_id = resolved_branch_id user.is_active = True user.allow_login = True user.is_locked = False if password and password.strip(): user.password_hash = hash_password(password.strip()) user.must_change_password = True role = _consultant_role(db) if not role: raise ValueError("Consultant role is not available. Run startup/permission seeding first.") if not _user_has_role(db, user_id=int(user.id), role_id=int(role.id)): db.add(UserRole(user_id=int(user.id), role_id=int(role.id))) invite_url = None if invite_user: db.flush() token = issue_invite_token(db, user) invite_url = f"/invite/accept?token={token}" return user, invite_url def list_consultant_role_users(db: Session, *, tenant_id: int, branch_id: int | None = None) -> list[User]: query = ( select(User) .join(UserRole, UserRole.user_id == User.id) .join(Role, Role.id == UserRole.role_id) .where(User.tenant_id == tenant_id, User.is_active.is_(True), Role.name == "Consultant") ) if branch_id: query = query.where(or_(User.branch_id == branch_id, User.branch_id.is_(None))) return db.execute(query.order_by(User.full_name.asc(), User.email.asc())).scalars().all() def list_consultants( db: Session, *, tenant_id: int, branch_id: int | None = None, q: str = "", include_inactive: bool = False, ) -> list[ConsultantProfile]: query = ( select(ConsultantProfile) .options(selectinload(ConsultantProfile.user)) .where(ConsultantProfile.tenant_id == tenant_id) ) if branch_id: query = query.where(or_(ConsultantProfile.branch_id == branch_id, ConsultantProfile.branch_id.is_(None))) if not include_inactive: query = query.where(ConsultantProfile.is_active.is_(True)) if q.strip(): term = f"%{q.strip()}%" query = query.where( or_( ConsultantProfile.contact_person.ilike(term), ConsultantProfile.firm_name.ilike(term), ConsultantProfile.email.ilike(term), ConsultantProfile.mobile.ilike(term), ConsultantProfile.specialisation.ilike(term), ) ) return db.execute(query.order_by(ConsultantProfile.contact_person.asc())).scalars().all() def get_consultant(db: Session, *, tenant_id: int, consultant_id: int, branch_id: int | None = None) -> ConsultantProfile | None: query = ( select(ConsultantProfile) .options(selectinload(ConsultantProfile.user), selectinload(ConsultantProfile.links)) .where(ConsultantProfile.id == consultant_id, ConsultantProfile.tenant_id == tenant_id) ) if branch_id: query = query.where(or_(ConsultantProfile.branch_id == branch_id, ConsultantProfile.branch_id.is_(None))) return db.execute(query).scalar_one_or_none() def get_consultant_by_user(db: Session, *, tenant_id: int, user_id: int) -> ConsultantProfile | None: return db.execute( select(ConsultantProfile) .options(selectinload(ConsultantProfile.user)) .where( ConsultantProfile.tenant_id == tenant_id, ConsultantProfile.user_id == user_id, ConsultantProfile.is_active.is_(True), ) ).scalar_one_or_none() def create_or_update_consultant(db: Session, *, payload: dict, user_id: int, consultant: ConsultantProfile | None = None) -> ConsultantProfile: if consultant is None: consultant = ConsultantProfile( tenant_id=payload["tenant_id"], branch_id=payload.get("branch_id"), user_id=payload.get("user_id"), contact_person=payload["contact_person"], created_by_user_id=user_id, ) db.add(consultant) consultant.branch_id = payload.get("branch_id") consultant.user_id = payload.get("user_id") consultant.consultant_type = payload.get("consultant_type") or "external_consultant" consultant.firm_name = normalise_text(payload.get("firm_name")) consultant.contact_person = payload["contact_person"].strip() consultant.email = normalise_text(payload.get("email")) consultant.mobile = normalise_text(payload.get("mobile")) consultant.specialisation = normalise_text(payload.get("specialisation")) consultant.gstin = normalise_text(payload.get("gstin")) consultant.pan = normalise_text(payload.get("pan")) consultant.address = normalise_text(payload.get("address")) consultant.status = payload.get("status") or "active" consultant.onboarding_status = payload.get("onboarding_status") or "approved" consultant.is_platform_partner = bool(payload.get("is_platform_partner")) consultant.is_franchise_partner = bool(payload.get("is_franchise_partner")) consultant.is_saas_customer = bool(payload.get("is_saas_customer")) consultant.is_active = bool(payload.get("is_active", True)) consultant.remarks = normalise_text(payload.get("remarks")) consultant.updated_by_user_id = user_id return consultant def update_consultant_own_profile(db: Session, *, consultant: ConsultantProfile, payload: dict, user_id: int) -> ConsultantProfile: consultant.firm_name = normalise_text(payload.get("firm_name")) consultant.contact_person = (payload.get("contact_person") or consultant.contact_person or "").strip() consultant.email = normalise_text(payload.get("email")) consultant.mobile = normalise_text(payload.get("mobile")) consultant.specialisation = normalise_text(payload.get("specialisation")) consultant.gstin = normalise_text(payload.get("gstin")) consultant.pan = normalise_text(payload.get("pan")) consultant.address = normalise_text(payload.get("address")) consultant.updated_by_user_id = user_id return consultant def list_clients_available_for_link( db: Session, *, tenant_id: int, branch_id: int | None = None, partner_user_id: int | None = None, ) -> list[Client]: query = select(Client).where(Client.tenant_id == tenant_id, Client.is_archived.is_(False), Client.is_active.is_(True)) if branch_id: query = query.where(Client.branch_id == branch_id) if partner_user_id: query = query.where(Client.partner_id == partner_user_id) return db.execute(query.order_by(Client.client_name.asc())).scalars().all() def list_client_links_for_consultant(db: Session, *, consultant_id: int) -> list[ClientConsultantLink]: return db.execute( select(ClientConsultantLink) .options(selectinload(ClientConsultantLink.client), selectinload(ClientConsultantLink.service_catalogue)) .where(ClientConsultantLink.consultant_id == consultant_id) .order_by(ClientConsultantLink.is_active.desc(), ClientConsultantLink.id.desc()) ).scalars().all() def consultant_linked_clients_page( db: Session, *, consultant: ConsultantProfile, q: str = "", client_group_id: int | None = None, status: str = "active", page: int = 1, per_page: int = 25, sort_by: str = "client_name", sort_order: str = "asc", ) -> dict: """Return a compact, searchable and paginated list of linked firm clients. Link permission columns remain unchanged and continue to govern the existing service, due-date, communication and document workflows. They are deliberately not exposed in the client-list UI. """ q = (q or "").strip() status = (status or "").strip().lower() per_page = min(max(int(per_page or 25), 1), 100) page = max(int(page or 1), 1) stmt = ( select(ClientConsultantLink, Client, ClientGroup) .join(Client, Client.id == ClientConsultantLink.client_id) .outerjoin(ClientGroup, ClientGroup.id == Client.client_group_id) .where( ClientConsultantLink.tenant_id == consultant.tenant_id, ClientConsultantLink.consultant_id == consultant.id, ClientConsultantLink.is_active.is_(True), Client.tenant_id == consultant.tenant_id, Client.is_archived.is_(False), ) ) if status: stmt = stmt.where(Client.status == status) if client_group_id: stmt = stmt.where(Client.client_group_id == int(client_group_id)) if q: like = f"%{q}%" stmt = stmt.where( or_( Client.client_code.ilike(like), Client.client_name.ilike(like), Client.trade_name.ilike(like), Client.pan.ilike(like), Client.gstin.ilike(like), ClientGroup.group_name.ilike(like), ClientGroup.group_code.ilike(like), ) ) total = int(db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one() or 0) pages = max(ceil(total / per_page), 1) page = min(page, pages) sort_columns = { "client_code": Client.client_code, "client_name": Client.client_name, "pan": Client.pan, "client_group": ClientGroup.group_name, "status": Client.status, } sort_column = sort_columns.get(sort_by, Client.client_name) ordering = sort_column.desc() if (sort_order or "asc").lower() == "desc" else sort_column.asc() rows = [] result = db.execute( stmt.order_by(ordering, Client.id.asc()) .offset((page - 1) * per_page) .limit(per_page) ).all() for link, client, group in result: rows.append({ "link": link, "client": client, "client_group_id": getattr(group, "id", None), "client_group_code": getattr(group, "group_code", None), "client_group_name": getattr(group, "group_name", None), "relationship_label": (link.relationship_type or "accounts_consultant").replace("_", " ").title(), }) groups_stmt = ( select(ClientGroup.id, ClientGroup.group_code, ClientGroup.group_name) .join(Client, Client.client_group_id == ClientGroup.id) .join(ClientConsultantLink, ClientConsultantLink.client_id == Client.id) .where( ClientConsultantLink.tenant_id == consultant.tenant_id, ClientConsultantLink.consultant_id == consultant.id, ClientConsultantLink.is_active.is_(True), Client.tenant_id == consultant.tenant_id, Client.is_archived.is_(False), ClientGroup.is_active.is_(True), ) .distinct() .order_by(ClientGroup.group_name.asc()) ) groups = [ {"id": int(group_id), "code": code, "name": name} for group_id, code, name in db.execute(groups_stmt).all() ] return { "rows": rows, "groups": groups, "meta": {"total": total, "page": page, "per_page": per_page, "pages": pages}, "filters": { "q": q, "client_group_id": client_group_id, "status": status, "sort_by": sort_by if sort_by in sort_columns else "client_name", "sort_order": "desc" if (sort_order or "asc").lower() == "desc" else "asc", }, } def link_client_to_consultant( db: Session, *, tenant_id: int, consultant_id: int, client: Client, relationship_type: str, is_primary: bool, can_view_client: bool, can_view_services: bool, can_view_due_dates: bool, can_view_communications: bool, remarks: str | None, user_id: int, ) -> ClientConsultantLink: link = db.execute( select(ClientConsultantLink).where( ClientConsultantLink.tenant_id == tenant_id, ClientConsultantLink.client_id == client.id, ClientConsultantLink.consultant_id == consultant_id, ) ).scalar_one_or_none() if link is None: link = ClientConsultantLink( tenant_id=tenant_id, branch_id=client.branch_id, client_id=client.id, consultant_id=consultant_id, created_by_user_id=user_id, ) db.add(link) link.relationship_type = relationship_type or "accounts_consultant" link.is_primary = bool(is_primary) link.can_view_client = bool(can_view_client) link.can_view_services = bool(can_view_services) link.can_view_due_dates = bool(can_view_due_dates) link.can_view_communications = bool(can_view_communications) link.is_active = True link.remarks = normalise_text(remarks) link.updated_by_user_id = user_id return link def set_link_active(db: Session, *, tenant_id: int, link_id: int, active: bool, user_id: int) -> ClientConsultantLink | None: link = db.execute( select(ClientConsultantLink).where(ClientConsultantLink.id == link_id, ClientConsultantLink.tenant_id == tenant_id) ).scalar_one_or_none() if link: link.is_active = active link.updated_by_user_id = user_id return link def linked_client_ids_for_consultant(db: Session, *, consultant_id: int, require_communications: bool = False) -> list[int]: query = select(ClientConsultantLink.client_id).where( ClientConsultantLink.consultant_id == consultant_id, ClientConsultantLink.is_active.is_(True), ) if require_communications: query = query.where(ClientConsultantLink.can_view_communications.is_(True)) return [int(x) for x in db.execute(query).scalars().all()] def consultant_dashboard_payload(db: Session, *, consultant: ConsultantProfile) -> dict: """Build the consultant portal dashboard payload. This function is intentionally read-only. It does not create or update any consultant records. It only aggregates already available data from consultant links, task communications, service requests, conversion requests and the consultant workspace. """ links = list_client_links_for_consultant(db, consultant_id=consultant.id) active_links = [link for link in links if link.is_active] communication_client_ids = [int(link.client_id) for link in active_links if link.can_view_communications] due_date_client_ids = [int(link.client_id) for link in active_links if link.can_view_due_dates] comments: list[tuple[ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue]] = [] if communication_client_ids: comments = db.execute( select(ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue) .join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id) .join(ClientServiceSubscription, ClientServiceSubscription.id == ServiceTaskComment.subscription_id) .join(Client, Client.id == ClientServiceTaskInstance.client_id) .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceTaskInstance.service_catalogue_id) .where( ServiceTaskComment.tenant_id == consultant.tenant_id, ServiceTaskComment.visibility == "consultant", ServiceTaskComment.is_deleted.is_(False), ClientServiceTaskInstance.client_id.in_(communication_client_ids), ClientServiceTaskInstance.tenant_id == consultant.tenant_id, ClientServiceTaskInstance.is_active.is_(True), ) .order_by(ServiceTaskComment.created_at_utc.desc(), ServiceTaskComment.id.desc()) .limit(50) ).all() today = date.today() next_30_days = today + timedelta(days=30) upcoming_due = 0 overdue = 0 due_items: list[tuple[ClientServiceSubscription, Client, ServiceCatalogue]] = [] overdue_items: list[tuple[ClientServiceSubscription, Client, ServiceCatalogue]] = [] if due_date_client_ids: due_query = ( select(ClientServiceSubscription, Client, ServiceCatalogue) .join(Client, Client.id == ClientServiceSubscription.client_id) .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceSubscription.service_catalogue_id) .where( ClientServiceSubscription.tenant_id == consultant.tenant_id, ClientServiceSubscription.client_id.in_(due_date_client_ids), ClientServiceSubscription.is_active.is_(True), ClientServiceSubscription.current_due_date.is_not(None), ) ) due_items = db.execute( due_query.where( ClientServiceSubscription.current_due_date >= today, ClientServiceSubscription.current_due_date <= next_30_days, ) .order_by(ClientServiceSubscription.current_due_date.asc(), Client.client_name.asc()) .limit(8) ).all() overdue_items = db.execute( due_query.where( ClientServiceSubscription.current_due_date < today, ClientServiceSubscription.status != "completed", ) .order_by(ClientServiceSubscription.current_due_date.asc(), Client.client_name.asc()) .limit(8) ).all() upcoming_due = len(due_items) overdue = len(overdue_items) managed_clients = list_consultant_managed_clients( db, tenant_id=consultant.tenant_id, consultant_id=consultant.id, include_inactive=False, ) managed_stats = consultant_managed_clients_stats(db, consultant=consultant) workspace_summary = consultant_workspace_summary(db, consultant=consultant) service_requests = list_consultant_service_requests(db, consultant=consultant) open_service_requests = [r for r in service_requests if r.status in {"submitted", "under_review", "accepted"}] pending_service_requests = [r for r in service_requests if r.status in {"submitted", "under_review"}] conversion_requests = [ r for r in managed_clients if getattr(r, "conversion_status", "not_requested") in {"requested", "under_review", "approved", "rejected"} ] pending_conversions = [r for r in conversion_requests if r.conversion_status in {"requested", "under_review"}] converted_clients = [r for r in managed_clients if getattr(r, "linked_firm_client_id", None)] consultant_user_id = int(getattr(consultant, "user_id", 0) or 0) firm_messages = [row for row in comments if int(getattr(row[0], "created_by_user_id", 0) or 0) != consultant_user_id] consultant_replies = [row for row in comments if int(getattr(row[0], "created_by_user_id", 0) or 0) == consultant_user_id] workspace = workspace_summary.get("workspace") workspace_alerts: list[str] = [] remaining = workspace_summary.get("managed_clients_remaining") limit = workspace_summary.get("managed_clients_limit") usage_percent = int(workspace_summary.get("usage_percent") or 0) if workspace is None: workspace_alerts.append("Workspace setup is pending.") elif not getattr(workspace, "is_active", True): workspace_alerts.append("Workspace is inactive.") elif getattr(workspace, "subscription_status", "") in {"suspended", "cancelled"}: workspace_alerts.append(f"Subscription status is {workspace.subscription_status.replace('_', ' ').title()}.") if limit and remaining is not None and remaining <= 0: workspace_alerts.append("Managed client limit has been reached.") elif limit and usage_percent >= 80: workspace_alerts.append("Managed client usage is above 80% of the workspace limit.") return { "links": links, "active_links": active_links, "comments": comments[:8], "recent_firm_messages": firm_messages[:6], "recent_consultant_replies": consultant_replies[:6], "managed_clients": managed_clients[:8], "workspace_summary": workspace_summary, "workspace": workspace, "workspace_alerts": workspace_alerts, "service_requests": service_requests[:6], "open_service_requests": open_service_requests[:6], "pending_service_requests": pending_service_requests[:6], "conversion_requests": conversion_requests[:6], "pending_conversions": pending_conversions[:6], "due_items": due_items, "overdue_items": overdue_items, "stats": { "linked_clients": len(active_links), "communication_enabled_clients": len(communication_client_ids), "due_date_enabled_clients": len(due_date_client_ids), "pending_clarifications": len(firm_messages), "consultant_replies": len(consultant_replies), "upcoming_due": int(upcoming_due or 0), "overdue": int(overdue or 0), "open_service_requests": len(open_service_requests), "pending_service_requests": len(pending_service_requests), "pending_conversions": len(pending_conversions), "converted_clients": len(converted_clients), **managed_stats, }, } def _next_workspace_code(db: Session, *, tenant_id: int) -> str: count = db.execute( select(func.count(ConsultantWorkspace.id)).where(ConsultantWorkspace.tenant_id == tenant_id) ).scalar_one() return f"CW-{int(count or 0) + 1:04d}" def get_workspace_by_consultant( db: Session, *, tenant_id: int, consultant_id: int, ) -> ConsultantWorkspace | None: return db.execute( select(ConsultantWorkspace).where( ConsultantWorkspace.tenant_id == tenant_id, ConsultantWorkspace.consultant_id == consultant_id, ) ).scalar_one_or_none() def ensure_consultant_workspace( db: Session, *, consultant: ConsultantProfile, user_id: int | None = None, ) -> ConsultantWorkspace: workspace = get_workspace_by_consultant(db, tenant_id=consultant.tenant_id, consultant_id=consultant.id) if workspace: return workspace workspace = ConsultantWorkspace( tenant_id=consultant.tenant_id, branch_id=consultant.branch_id, consultant_id=consultant.id, workspace_code=_next_workspace_code(db, tenant_id=consultant.tenant_id), workspace_name=consultant.firm_name or consultant.contact_person, workspace_type="franchise_partner" if consultant.is_franchise_partner else "consultant_saas", plan_code="franchise" if consultant.is_franchise_partner else "starter", subscription_status="active" if consultant.is_saas_customer or consultant.is_franchise_partner else "trial", billing_cycle="manual", max_managed_clients=100 if consultant.is_franchise_partner else 25, max_user_accounts=5 if consultant.is_franchise_partner else 1, allow_client_portal=False, allow_firm_referrals=True, allow_service_marketplace=bool(consultant.is_franchise_partner or consultant.is_platform_partner), is_active=True, created_by_user_id=user_id, updated_by_user_id=user_id, ) db.add(workspace) return workspace def parse_optional_date(value: str | None) -> date | None: if not value: return None text = value.strip() if not text: return None for fmt in ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y"): try: return datetime.strptime(text, fmt).date() except ValueError: continue return None def update_consultant_workspace( db: Session, *, workspace: ConsultantWorkspace, payload: dict, user_id: int, admin_mode: bool = False, ) -> ConsultantWorkspace: """Update consultant workspace. Normal consultant portal users may update only safe self-service fields (workspace name and remarks). Commercial controls such as plan, billing, subscription status, client/user limits and feature switches remain firm/admin controlled unless admin_mode=True is explicitly used by a future internal route. """ workspace.workspace_name = normalise_text(payload.get("workspace_name")) or workspace.workspace_name workspace.remarks = normalise_text(payload.get("remarks")) if admin_mode: workspace.workspace_type = payload.get("workspace_type") or workspace.workspace_type workspace.plan_code = payload.get("plan_code") or workspace.plan_code workspace.billing_cycle = payload.get("billing_cycle") or workspace.billing_cycle workspace.subscription_status = payload.get("subscription_status") or workspace.subscription_status workspace.subscription_start_date = payload.get("subscription_start_date") workspace.subscription_end_date = payload.get("subscription_end_date") workspace.max_managed_clients = int(payload.get("max_managed_clients") or workspace.max_managed_clients or 25) workspace.max_user_accounts = int(payload.get("max_user_accounts") or workspace.max_user_accounts or 1) workspace.allow_client_portal = bool(payload.get("allow_client_portal")) workspace.allow_firm_referrals = bool(payload.get("allow_firm_referrals")) workspace.allow_service_marketplace = bool(payload.get("allow_service_marketplace")) workspace.is_active = bool(payload.get("is_active", True)) workspace.updated_by_user_id = user_id return workspace def consultant_workspace_summary(db: Session, *, consultant: ConsultantProfile) -> dict: workspace = get_workspace_by_consultant(db, tenant_id=consultant.tenant_id, consultant_id=consultant.id) managed_total = db.execute( select(func.count(ConsultantManagedClient.id)).where( ConsultantManagedClient.tenant_id == consultant.tenant_id, ConsultantManagedClient.consultant_id == consultant.id, ConsultantManagedClient.is_active.is_(True), ) ).scalar_one() if not workspace: return { "workspace": None, "managed_clients_used": int(managed_total or 0), "managed_clients_limit": None, "managed_clients_remaining": None, "usage_percent": 0, } limit = max(int(workspace.max_managed_clients or 0), 0) used = int(managed_total or 0) remaining = max(limit - used, 0) if limit else None usage_percent = int(round((used / limit) * 100)) if limit else 0 return { "workspace": workspace, "managed_clients_used": used, "managed_clients_limit": limit, "managed_clients_remaining": remaining, "usage_percent": usage_percent, } def consultant_can_add_managed_client(db: Session, *, consultant: ConsultantProfile) -> tuple[bool, dict]: """Return whether the consultant can add one more managed client under workspace limits.""" summary = consultant_workspace_summary(db, consultant=consultant) limit = summary.get("managed_clients_limit") used = int(summary.get("managed_clients_used") or 0) if limit is None or int(limit or 0) <= 0: return True, summary return used < int(limit), summary def _next_consultant_client_code(db: Session, *, tenant_id: int, consultant_id: int) -> str: count = db.execute( select(func.count(ConsultantManagedClient.id)).where( ConsultantManagedClient.tenant_id == tenant_id, ConsultantManagedClient.consultant_id == consultant_id, ) ).scalar_one() return f"CMC-{consultant_id}-{int(count or 0) + 1:04d}" def list_consultant_managed_clients( db: Session, *, tenant_id: int, consultant_id: int, q: str = "", status: str = "", include_inactive: bool = False, ) -> list[ConsultantManagedClient]: query = select(ConsultantManagedClient).where( ConsultantManagedClient.tenant_id == tenant_id, ConsultantManagedClient.consultant_id == consultant_id, ) if not include_inactive: query = query.where(ConsultantManagedClient.is_active.is_(True)) if status.strip(): query = query.where(ConsultantManagedClient.status == status.strip()) if q.strip(): term = f"%{q.strip()}%" query = query.where( or_( ConsultantManagedClient.client_name.ilike(term), ConsultantManagedClient.trade_name.ilike(term), ConsultantManagedClient.client_code.ilike(term), ConsultantManagedClient.pan.ilike(term), ConsultantManagedClient.gstin.ilike(term), ConsultantManagedClient.email.ilike(term), ConsultantManagedClient.mobile.ilike(term), ) ) return db.execute(query.order_by(ConsultantManagedClient.client_name.asc())).scalars().all() def get_consultant_managed_client( db: Session, *, tenant_id: int, consultant_id: int, managed_client_id: int, ) -> ConsultantManagedClient | None: return db.execute( select(ConsultantManagedClient).where( ConsultantManagedClient.id == managed_client_id, ConsultantManagedClient.tenant_id == tenant_id, ConsultantManagedClient.consultant_id == consultant_id, ) ).scalar_one_or_none() def create_or_update_managed_client( db: Session, *, consultant: ConsultantProfile, payload: dict, user_id: int, managed_client: ConsultantManagedClient | None = None, ) -> ConsultantManagedClient: if managed_client is None: can_add, summary = consultant_can_add_managed_client(db, consultant=consultant) if not can_add: raise ValueError( "Managed client limit reached for this consultant workspace " f"({summary.get('managed_clients_used')}/{summary.get('managed_clients_limit')})." ) managed_client = ConsultantManagedClient( tenant_id=consultant.tenant_id, branch_id=consultant.branch_id, consultant_id=consultant.id, created_by_user_id=user_id, ) db.add(managed_client) managed_client.client_code = normalise_text(payload.get("client_code")) or managed_client.client_code or _next_consultant_client_code( db, tenant_id=consultant.tenant_id, consultant_id=consultant.id ) managed_client.client_name = (payload.get("client_name") or "").strip() managed_client.trade_name = normalise_text(payload.get("trade_name")) managed_client.client_type = payload.get("client_type") or "Other" managed_client.pan = normalise_text(payload.get("pan")) managed_client.gstin = normalise_text(payload.get("gstin")) managed_client.tan = normalise_text(payload.get("tan")) managed_client.contact_person_name = normalise_text(payload.get("contact_person_name")) managed_client.mobile = normalise_text(payload.get("mobile")) managed_client.email = normalise_text(payload.get("email")) managed_client.address_line_1 = normalise_text(payload.get("address_line_1")) managed_client.address_line_2 = normalise_text(payload.get("address_line_2")) managed_client.city = normalise_text(payload.get("city")) managed_client.state = normalise_text(payload.get("state")) managed_client.pincode = normalise_text(payload.get("pincode")) managed_client.country = normalise_text(payload.get("country")) or "India" managed_client.service_interest = normalise_text(payload.get("service_interest")) managed_client.relationship_stage = payload.get("relationship_stage") or "managed" managed_client.status = payload.get("status") or "active" managed_client.is_active = bool(payload.get("is_active", True)) managed_client.notes = normalise_text(payload.get("notes")) managed_client.updated_by_user_id = user_id return managed_client def consultant_managed_clients_stats(db: Session, *, consultant: ConsultantProfile) -> dict: rows = list_consultant_managed_clients( db, tenant_id=consultant.tenant_id, consultant_id=consultant.id, include_inactive=True, ) return { "managed_clients": len([r for r in rows if r.is_active]), "prospects": len([r for r in rows if r.relationship_stage == "lead" and r.is_active]), "referred_to_firm": len([r for r in rows if r.relationship_stage == "referred_to_firm" and r.is_active]), } CONSULTANT_SERVICE_REQUEST_STATUSES = [ ("submitted", "Submitted"), ("under_review", "Under Review"), ("accepted", "Accepted"), ("rejected", "Rejected"), ("converted", "Converted to Engagement"), ("closed", "Closed"), ] CONSULTANT_SERVICE_REQUEST_PRIORITIES = [ ("low", "Low"), ("normal", "Normal"), ("high", "High"), ("urgent", "Urgent"), ] def _allowed_consultant_client_ids(db: Session, *, consultant: ConsultantProfile, require_communications: bool = True) -> list[int]: query = select(ClientConsultantLink.client_id).where( ClientConsultantLink.tenant_id == consultant.tenant_id, ClientConsultantLink.consultant_id == consultant.id, ClientConsultantLink.is_active.is_(True), ) if require_communications: query = query.where(ClientConsultantLink.can_view_communications.is_(True)) return [int(x) for x in db.execute(query).scalars().all()] def list_consultant_visible_communications( db: Session, *, consultant: ConsultantProfile, q: str = "", limit: int = 100, ) -> list[tuple[ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue]]: client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True) if not client_ids: return [] safe_limit = max(1, min(int(limit or 100), 200)) query = ( select(ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue) .join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id) .join(ClientServiceSubscription, ClientServiceSubscription.id == ServiceTaskComment.subscription_id) .join(Client, Client.id == ClientServiceTaskInstance.client_id) .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceTaskInstance.service_catalogue_id) .where( ServiceTaskComment.tenant_id == consultant.tenant_id, ServiceTaskComment.visibility == "consultant", ServiceTaskComment.is_deleted.is_(False), ClientServiceTaskInstance.client_id.in_(client_ids), ClientServiceTaskInstance.tenant_id == consultant.tenant_id, ClientServiceTaskInstance.is_active.is_(True), ) ) if q.strip(): term = f"%{q.strip()}%" query = query.where( or_( Client.client_name.ilike(term), Client.client_code.ilike(term), ServiceCatalogue.service_name.ilike(term), ServiceCatalogue.service_code.ilike(term), ClientServiceTaskInstance.task_name.ilike(term), ServiceTaskComment.message.ilike(term), ) ) return db.execute(query.order_by(ServiceTaskComment.created_at_utc.desc(), ServiceTaskComment.id.desc()).limit(safe_limit)).all() def get_consultant_visible_communication( db: Session, *, consultant: ConsultantProfile, comment_id: int, ) -> tuple[ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue] | None: client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True) if not client_ids: return None return db.execute( select(ServiceTaskComment, ClientServiceTaskInstance, ClientServiceSubscription, Client, ServiceCatalogue) .join(ClientServiceTaskInstance, ClientServiceTaskInstance.id == ServiceTaskComment.task_instance_id) .join(ClientServiceSubscription, ClientServiceSubscription.id == ServiceTaskComment.subscription_id) .join(Client, Client.id == ClientServiceTaskInstance.client_id) .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceTaskInstance.service_catalogue_id) .where( ServiceTaskComment.id == comment_id, ServiceTaskComment.tenant_id == consultant.tenant_id, ServiceTaskComment.visibility == "consultant", ServiceTaskComment.is_deleted.is_(False), ClientServiceTaskInstance.client_id.in_(client_ids), ClientServiceTaskInstance.tenant_id == consultant.tenant_id, ClientServiceTaskInstance.is_active.is_(True), ) ).first() def list_consultant_task_timeline( db: Session, *, consultant: ConsultantProfile, task_id: int, ) -> list[ServiceTaskComment]: client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True) if not client_ids: return [] task = db.execute( select(ClientServiceTaskInstance).where( ClientServiceTaskInstance.id == task_id, ClientServiceTaskInstance.tenant_id == consultant.tenant_id, ClientServiceTaskInstance.client_id.in_(client_ids), ClientServiceTaskInstance.is_active.is_(True), ) ).scalar_one_or_none() if not task: return [] return db.execute( select(ServiceTaskComment) .options(selectinload(ServiceTaskComment.created_by)) .where( ServiceTaskComment.tenant_id == consultant.tenant_id, ServiceTaskComment.task_instance_id == task_id, ServiceTaskComment.visibility == "consultant", ServiceTaskComment.is_deleted.is_(False), ) .order_by(ServiceTaskComment.created_at_utc.asc(), ServiceTaskComment.id.asc()) ).scalars().all() def add_consultant_task_reply( db: Session, *, consultant: ConsultantProfile, task: ClientServiceTaskInstance, message: str, user_id: int, ) -> ServiceTaskComment | None: client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=True) if int(task.client_id) not in client_ids: return None if getattr(task, "is_locked", False) or getattr(getattr(task, "subscription", None), "is_locked", False): return None clean_message = (message or "").strip() if not clean_message: return None row = ServiceTaskComment( tenant_id=task.tenant_id, branch_id=task.branch_id, subscription_id=task.subscription_id, task_instance_id=task.id, comment_type="consultant_clarification", visibility="consultant", message=clean_message, created_by_user_id=user_id, ) db.add(row) task.updated_by_user_id = user_id return row def _next_service_request_no(db: Session, *, tenant_id: int) -> str: count = db.execute( select(func.count(ConsultantServiceRequest.id)).where(ConsultantServiceRequest.tenant_id == tenant_id) ).scalar_one() return f"CSR-{int(count or 0) + 1:05d}" def list_consultant_requestable_services(db: Session) -> list[ServiceCatalogue]: flagged = db.execute( select(ServiceCatalogue) .where(ServiceCatalogue.is_active.is_(True), ServiceCatalogue.is_consultant_requestable.is_(True)) .order_by(ServiceCatalogue.service_name.asc()) ).scalars().all() if flagged: return flagged return db.execute( select(ServiceCatalogue) .where(ServiceCatalogue.is_active.is_(True)) .order_by(ServiceCatalogue.service_name.asc()) ).scalars().all() def _valid_managed_client_for_consultant(db: Session, *, consultant: ConsultantProfile, managed_client_id: int | None) -> ConsultantManagedClient | None: if not managed_client_id: return None return get_consultant_managed_client( db, tenant_id=consultant.tenant_id, consultant_id=consultant.id, managed_client_id=int(managed_client_id), ) def _valid_firm_client_for_consultant(db: Session, *, consultant: ConsultantProfile, firm_client_id: int | None) -> Client | None: if not firm_client_id: return None allowed_client_ids = _allowed_consultant_client_ids(db, consultant=consultant, require_communications=False) if int(firm_client_id) not in allowed_client_ids: return None return db.execute( select(Client).where( Client.id == int(firm_client_id), Client.tenant_id == consultant.tenant_id, Client.is_archived.is_(False), ) ).scalar_one_or_none() def create_consultant_service_request( db: Session, *, consultant: ConsultantProfile, payload: dict, user_id: int, ) -> ConsultantServiceRequest: service_id = payload.get("service_catalogue_id") service = None if service_id: service = db.execute( select(ServiceCatalogue).where(ServiceCatalogue.id == int(service_id), ServiceCatalogue.is_active.is_(True)) ).scalar_one_or_none() managed_client = _valid_managed_client_for_consultant( db, consultant=consultant, managed_client_id=payload.get("managed_client_id") ) firm_client = _valid_firm_client_for_consultant( db, consultant=consultant, firm_client_id=payload.get("firm_client_id") ) if not managed_client and not firm_client: raise ValueError("Select either one managed client or one linked firm client for the service request.") requested_service_name = normalise_text(payload.get("requested_service_name")) or (service.service_name if service else None) if not requested_service_name: raise ValueError("Service name is required.") subject = normalise_text(payload.get("subject")) or requested_service_name row = ConsultantServiceRequest( tenant_id=consultant.tenant_id, branch_id=consultant.branch_id, consultant_id=consultant.id, managed_client_id=managed_client.id if managed_client else None, firm_client_id=firm_client.id if firm_client else None, service_catalogue_id=service.id if service else None, request_no=_next_service_request_no(db, tenant_id=consultant.tenant_id), request_type=payload.get("request_type") or "service_request", status="submitted", priority=payload.get("priority") or "normal", requested_service_name=requested_service_name, requested_due_date=payload.get("requested_due_date"), subject=subject, description=normalise_text(payload.get("description")), consultant_notes=normalise_text(payload.get("consultant_notes")), created_by_user_id=user_id, updated_by_user_id=user_id, is_active=True, ) db.add(row) return row def list_consultant_service_requests( db: Session, *, consultant: ConsultantProfile, status: str = "", ) -> list[ConsultantServiceRequest]: query = ( select(ConsultantServiceRequest) .options( selectinload(ConsultantServiceRequest.managed_client), selectinload(ConsultantServiceRequest.firm_client), selectinload(ConsultantServiceRequest.service_catalogue), selectinload(ConsultantServiceRequest.reviewed_by), ) .where( ConsultantServiceRequest.tenant_id == consultant.tenant_id, ConsultantServiceRequest.consultant_id == consultant.id, ConsultantServiceRequest.is_active.is_(True), ) ) if status.strip(): query = query.where(ConsultantServiceRequest.status == status.strip()) return db.execute(query.order_by(ConsultantServiceRequest.created_at_utc.desc(), ConsultantServiceRequest.id.desc())).scalars().all() def get_consultant_service_request( db: Session, *, tenant_id: int, request_id: int, consultant_id: int | None = None, ) -> ConsultantServiceRequest | None: query = ( select(ConsultantServiceRequest) .options( selectinload(ConsultantServiceRequest.consultant), selectinload(ConsultantServiceRequest.managed_client), selectinload(ConsultantServiceRequest.firm_client), selectinload(ConsultantServiceRequest.service_catalogue), selectinload(ConsultantServiceRequest.created_by), selectinload(ConsultantServiceRequest.reviewed_by), ) .where(ConsultantServiceRequest.id == request_id, ConsultantServiceRequest.tenant_id == tenant_id) ) if consultant_id: query = query.where(ConsultantServiceRequest.consultant_id == consultant_id) return db.execute(query).scalar_one_or_none() def list_all_consultant_service_requests( db: Session, *, tenant_id: int, branch_id: int | None = None, status: str = "", ) -> list[ConsultantServiceRequest]: query = ( select(ConsultantServiceRequest) .options( selectinload(ConsultantServiceRequest.consultant), selectinload(ConsultantServiceRequest.managed_client), selectinload(ConsultantServiceRequest.firm_client), selectinload(ConsultantServiceRequest.service_catalogue), ) .where(ConsultantServiceRequest.tenant_id == tenant_id, ConsultantServiceRequest.is_active.is_(True)) ) if branch_id: query = query.where(ConsultantServiceRequest.branch_id == branch_id) if status.strip(): query = query.where(ConsultantServiceRequest.status == status.strip()) return db.execute(query.order_by(ConsultantServiceRequest.created_at_utc.desc(), ConsultantServiceRequest.id.desc())).scalars().all() def update_consultant_service_request_status( db: Session, *, request: ConsultantServiceRequest, status: str, firm_response: str | None, user_id: int, ) -> ConsultantServiceRequest: allowed = {code for code, _label in CONSULTANT_SERVICE_REQUEST_STATUSES} clean_status = (status or "under_review").strip() if clean_status not in allowed: clean_status = "under_review" request.status = clean_status request.firm_response = normalise_text(firm_response) request.reviewed_by_user_id = user_id request.reviewed_at_utc = datetime.now(timezone.utc) request.updated_by_user_id = user_id return request # --------------------------------------------------------------------------- # Phase 5A.7 / 5A.8 helpers # Consultant-managed client conversion + workspace limit enforcement # --------------------------------------------------------------------------- def request_managed_client_conversion( db: Session, *, consultant: ConsultantProfile, managed_client: ConsultantManagedClient, notes: str | None, user_id: int, ) -> ConsultantManagedClient: if managed_client.tenant_id != consultant.tenant_id or managed_client.consultant_id != consultant.id: raise ValueError("Managed client is not available for this consultant.") if managed_client.linked_firm_client_id: raise ValueError("This managed client is already linked to a firm client.") if managed_client.conversion_status in {"requested", "under_review"}: raise ValueError("Conversion request is already pending review.") managed_client.conversion_status = "requested" managed_client.relationship_stage = "conversion_requested" managed_client.conversion_requested_at_utc = datetime.now(timezone.utc) managed_client.conversion_requested_by_user_id = user_id managed_client.conversion_notes = normalise_text(notes) managed_client.updated_by_user_id = user_id return managed_client def list_consultant_conversion_requests( db: Session, *, tenant_id: int, branch_id: int | None = None, status: str = "", ) -> list[ConsultantManagedClient]: query = ( select(ConsultantManagedClient) .options( selectinload(ConsultantManagedClient.consultant), selectinload(ConsultantManagedClient.linked_firm_client), ) .where( ConsultantManagedClient.tenant_id == tenant_id, ConsultantManagedClient.conversion_status != "not_requested", ) ) if branch_id: query = query.where(ConsultantManagedClient.branch_id == branch_id) if status.strip(): query = query.where(ConsultantManagedClient.conversion_status == status.strip()) return db.execute( query.order_by( ConsultantManagedClient.conversion_requested_at_utc.desc().nullslast(), ConsultantManagedClient.id.desc(), ) ).scalars().all() def get_client_conversion_request( db: Session, *, tenant_id: int, managed_client_id: int, branch_id: int | None = None, ) -> ConsultantManagedClient | None: query = ( select(ConsultantManagedClient) .options( selectinload(ConsultantManagedClient.consultant), selectinload(ConsultantManagedClient.linked_firm_client), ) .where( ConsultantManagedClient.id == managed_client_id, ConsultantManagedClient.tenant_id == tenant_id, ConsultantManagedClient.conversion_status != "not_requested", ) ) if branch_id: query = query.where(ConsultantManagedClient.branch_id == branch_id) return db.execute(query).scalar_one_or_none() def _next_converted_client_code(db: Session, *, tenant_id: int, source_code: str | None) -> str: base = (normalise_text(source_code) or "CONSULTANT").replace(" ", "-").upper()[:35] candidate = f"FC-{base}" existing = db.execute(select(Client.id).where(Client.tenant_id == tenant_id, Client.client_code == candidate)).first() if not existing: return candidate count = db.execute(select(func.count(Client.id)).where(Client.tenant_id == tenant_id)).scalar_one() return f"FC-{base}-{int(count or 0) + 1:04d}"[:50] def approve_managed_client_conversion( db: Session, *, managed_client: ConsultantManagedClient, user_id: int, partner_user_id: int | None = None, client_code: str | None = None, firm_notes: str | None = None, ) -> Client: if managed_client.linked_firm_client_id: existing_client = db.get(Client, int(managed_client.linked_firm_client_id)) if existing_client: return existing_client if managed_client.conversion_status not in {"requested", "under_review", "rejected"}: raise ValueError("Only requested/under-review conversion records can be approved.") if not managed_client.branch_id: raise ValueError("Managed client has no branch. Set branch before approving conversion.") pan = normalise_text(managed_client.pan) gstin = normalise_text(managed_client.gstin) if pan: duplicate_pan = db.execute( select(Client).where(Client.tenant_id == managed_client.tenant_id, Client.pan == pan, Client.is_archived.is_(False)) ).scalar_one_or_none() if duplicate_pan: raise ValueError(f"A firm client with PAN {pan} already exists: {duplicate_pan.client_name}.") if gstin: duplicate_gstin = db.execute( select(Client).where(Client.tenant_id == managed_client.tenant_id, Client.gstin == gstin, Client.is_archived.is_(False)) ).scalar_one_or_none() if duplicate_gstin: raise ValueError(f"A firm client with GSTIN {gstin} already exists: {duplicate_gstin.client_name}.") clean_code = normalise_text(client_code) or _next_converted_client_code( db, tenant_id=managed_client.tenant_id, source_code=managed_client.client_code ) duplicate_code = db.execute( select(Client).where(Client.tenant_id == managed_client.tenant_id, Client.client_code == clean_code) ).scalar_one_or_none() if duplicate_code: raise ValueError(f"Client code {clean_code} already exists.") client = Client( tenant_id=managed_client.tenant_id, branch_id=int(managed_client.branch_id), partner_id=partner_user_id, engagement_mode="hybrid", client_code=clean_code, client_name=managed_client.client_name, trade_name=managed_client.trade_name, client_type=managed_client.client_type or "Other", pan=pan, gstin=gstin, tan=normalise_text(managed_client.tan), contact_person_name=managed_client.contact_person_name, mobile=managed_client.mobile, email=managed_client.email, address_line_1=managed_client.address_line_1, address_line_2=managed_client.address_line_2, city=managed_client.city, state=managed_client.state, pincode=managed_client.pincode, country=managed_client.country or "India", status="active", client_category="Consultant Referral", notes=(managed_client.notes or "") + ("\n\nConverted from consultant-managed client."), is_active=True, is_archived=False, created_by_user_id=user_id, updated_by_user_id=user_id, ) db.add(client) db.flush() managed_client.linked_firm_client_id = client.id managed_client.conversion_status = "approved" managed_client.relationship_stage = "converted_to_firm_client" managed_client.conversion_reviewed_at_utc = datetime.now(timezone.utc) managed_client.conversion_reviewed_by_user_id = user_id managed_client.conversion_firm_notes = normalise_text(firm_notes) managed_client.updated_by_user_id = user_id link_client_to_consultant( db, tenant_id=managed_client.tenant_id, consultant_id=managed_client.consultant_id, client=client, relationship_type="audit_coordination", is_primary=True, can_view_client=True, can_view_services=True, can_view_due_dates=True, can_view_communications=True, remarks="Auto-linked after consultant-managed client conversion.", user_id=user_id, ) db.add( ClientAuditLog( client_id=client.id, tenant_id=client.tenant_id, branch_id=client.branch_id, actor_user_id=user_id, action="converted_from_consultant_client", summary="Client created from consultant-managed client conversion request.", payload_json={ "consultant_id": managed_client.consultant_id, "managed_client_id": managed_client.id, "partner_id": partner_user_id, }, ) ) return client def mark_conversion_under_review( db: Session, *, managed_client: ConsultantManagedClient, user_id: int, firm_notes: str | None = None, ) -> ConsultantManagedClient: if managed_client.conversion_status != "requested": raise ValueError("Only requested conversions can be marked under review.") managed_client.conversion_status = "under_review" managed_client.conversion_reviewed_by_user_id = user_id managed_client.conversion_reviewed_at_utc = datetime.now(timezone.utc) managed_client.conversion_firm_notes = normalise_text(firm_notes) managed_client.updated_by_user_id = user_id return managed_client def reject_managed_client_conversion( db: Session, *, managed_client: ConsultantManagedClient, user_id: int, firm_notes: str | None, ) -> ConsultantManagedClient: if managed_client.conversion_status not in {"requested", "under_review"}: raise ValueError("Only pending conversion requests can be rejected.") managed_client.conversion_status = "rejected" managed_client.relationship_stage = "managed" managed_client.conversion_reviewed_by_user_id = user_id managed_client.conversion_reviewed_at_utc = datetime.now(timezone.utc) managed_client.conversion_firm_notes = normalise_text(firm_notes) managed_client.updated_by_user_id = user_id return managed_client def get_primary_client_consultant_link(db: Session, *, tenant_id: int, client_id: int) -> ClientConsultantLink | None: return db.execute( select(ClientConsultantLink) .where( ClientConsultantLink.tenant_id == tenant_id, ClientConsultantLink.client_id == client_id, ClientConsultantLink.is_primary.is_(True), ClientConsultantLink.is_active.is_(True), ) .order_by(ClientConsultantLink.id.asc()) ).scalars().first() def sync_primary_client_consultant_link( db: Session, *, client: Client, consultant_id: int | None, actor_user_id: int ) -> ClientConsultantLink | None: """Synchronise the client's unrestricted primary consultant link without deleting history.""" links = db.execute( select(ClientConsultantLink).where( ClientConsultantLink.tenant_id == client.tenant_id, ClientConsultantLink.client_id == client.id, ) ).scalars().all() selected = None for link in links: if consultant_id and int(link.consultant_id) == int(consultant_id) and link.service_catalogue_id is None: selected = link elif link.is_primary: link.is_primary = False link.updated_by_user_id = actor_user_id if consultant_id is None: db.flush() return None consultant = get_consultant(db, tenant_id=int(client.tenant_id), consultant_id=int(consultant_id)) if not consultant or not consultant.is_active: raise ValueError("Selected primary consultant is not active in this firm.") if selected is None: selected = ClientConsultantLink( tenant_id=client.tenant_id, branch_id=client.branch_id, client_id=client.id, consultant_id=consultant.id, service_catalogue_id=None, relationship_type="accounts_consultant", is_primary=True, is_active=True, can_view_client=True, can_view_services=True, can_view_due_dates=True, can_view_communications=True, can_view_engagements=True, can_view_task_status=True, can_view_assignee=True, can_view_document_requests=True, can_upload_documents=True, can_reply_to_clarifications=True, can_view_filing_details=True, can_view_final_documents=True, can_view_permanent_documents=False, can_receive_notifications=True, can_act_for_client=True, effective_from=getattr(client, "referral_date", None) or date.today(), created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id, ) db.add(selected) else: selected.branch_id = client.branch_id selected.is_primary = True selected.is_active = True selected.updated_by_user_id = actor_user_id db.flush() return selected def get_client_consultant_summary(db: Session, *, tenant_id: int, client_id: int) -> dict: primary = get_primary_client_consultant_link(db, tenant_id=tenant_id, client_id=client_id) referred_id = db.execute(select(Client.referred_by_consultant_id).where(Client.id == client_id)).scalar_one_or_none() ids = {int(x) for x in (getattr(primary, "consultant_id", None), referred_id) if x} profiles = db.execute(select(ConsultantProfile).where(ConsultantProfile.id.in_(ids))).scalars().all() if ids else [] by_id = {int(x.id): x for x in profiles} return { "primary": by_id.get(int(primary.consultant_id)) if primary else None, "referred_by": by_id.get(int(referred_id)) if referred_id else None, }