from __future__ import annotations import io import json import re from dataclasses import dataclass from typing import Any from openpyxl import Workbook, load_workbook from sqlalchemy import select from sqlalchemy.orm import Session from app.modules.client_identity.service import ensure_identity, normalize_pan, placeholder_email from app.modules.client_groups.service import get_group_by_code, resolve_or_create_group from app.modules.clients import repository from app.modules.clients.models import Client from app.modules.clients.schemas import ClientCreate from app.modules.clients.service import create_client_service from app.modules.consultants.models import ConsultantProfile from app.modules.consultants.service import get_consultant from app.modules.core.iam.models import User TEMPLATE_COLUMNS = [ "uploader_user_id", "firm_tenant_id", "partner_user_id", "referred_by_consultant_id", "primary_consultant_id", "referral_date", "referral_reference", "referral_status", "communication_routing_mode", "branch_id", "client_group_code", "client_group_name", "group_type", "group_relationship", "is_group_head", "client_code", "client_name", "client_type", "engagement_mode", "email", "portal_password", "portal_password_confirm", "mobile", "pan", "gstin", "tan", "cin_llpin", "msme_no", "iec_code", "contact_person_name", "contact_person_designation", "alternate_mobile", "alternate_email", "address_line_1", "address_line_2", "city", "state", "pincode", "country", "client_category", "risk_category", "onboarding_date", "closing_date", "notes", "status", "gst_applicable", "income_tax_applicable", "tds_applicable", "roc_applicable", "audit_applicable", "pf_applicable", "esi_applicable", "professional_tax_applicable", "payroll_applicable", "msme_applicable", "import_export_applicable", ] BOOL_FIELDS = { "is_group_head", "gst_applicable", "income_tax_applicable", "tds_applicable", "roc_applicable", "audit_applicable", "pf_applicable", "esi_applicable", "professional_tax_applicable", "payroll_applicable", "msme_applicable", "import_export_applicable", } PAN_RE = re.compile(r"^[A-Z]{5}[0-9]{4}[A-Z]$") VALID_ROUTING_MODES = { "client_direct", "consultant_primary", "client_and_consultant", "firm_only", } VALID_REFERRAL_STATUSES = {"active", "inactive", "ended", "pending"} @dataclass class ImportPreview: valid_rows: list[dict] errors: list[dict] total_rows: int def _clean(value: Any) -> str | None: if value is None: return None text = str(value).strip() return text or None def _to_bool(value: Any) -> bool: return str(value or "").strip().lower() in {"1", "true", "yes", "y", "on"} def _to_int(value: Any) -> int | None: text = _clean(value) if not text: return None try: return int(float(text)) except (TypeError, ValueError): return None def _normalise_email(value: Any) -> str | None: text = _clean(value) return text.lower() if text else None def _normalise_pan_for_preview(value: Any) -> str | None: text = re.sub(r"\s+", "", str(value or "").upper()) return text or None def _row_dict(ws, row_idx: int) -> dict[str, Any]: headers = [str(cell.value).strip() if cell.value is not None else "" for cell in ws[1]] values = [cell.value for cell in ws[row_idx]] return { headers[index]: values[index] if index < len(values) else None for index in range(len(headers)) if headers[index] } def _consultant_for_id( db: Session, *, tenant_id: int, value: Any, field_name: str, messages: list[str], ) -> int | None: text = _clean(value) if not text: return None consultant_id = _to_int(text) if not consultant_id: messages.append(f"{field_name} must contain a valid consultant profile id.") return None consultant = get_consultant(db, tenant_id=tenant_id, consultant_id=consultant_id) if not consultant or not consultant.is_active: messages.append(f"{field_name} must be an active consultant profile id in the same firm.") return None return int(consultant.id) def build_client_import_template_bytes(*, current_user, tenant_id: int, partner_id: int | None) -> bytes: workbook = Workbook() worksheet = workbook.active worksheet.title = "clients_import" worksheet.append(TEMPLATE_COLUMNS) sample_values = { "uploader_user_id": current_user.id, "firm_tenant_id": tenant_id, "partner_user_id": partner_id or current_user.id, "referred_by_consultant_id": "", "primary_consultant_id": "", "referral_date": "", "referral_reference": "", "referral_status": "active", "communication_routing_mode": "client_and_consultant", "branch_id": getattr(current_user, "branch_id", "") or "", "client_code": "CLT-001", "client_name": "Sample Client", "client_type": "Other", "engagement_mode": "internal_managed", "email": "client@example.com", "portal_password": "ChangeMe@123", "portal_password_confirm": "ChangeMe@123", "mobile": "9876543210", "pan": "ABCDE1234F", "gstin": "", "tan": "", "cin_llpin": "", "msme_no": "", "iec_code": "", "contact_person_name": "Client Contact", "contact_person_designation": "Proprietor", "alternate_mobile": "", "alternate_email": "", "address_line_1": "Address line 1", "address_line_2": "", "city": "Chennai", "state": "Tamil Nadu", "pincode": "600001", "country": "India", "client_category": "", "risk_category": "", "onboarding_date": "", "closing_date": "", "notes": "", "status": "active", "gst_applicable": "yes", "income_tax_applicable": "yes", "tds_applicable": "no", "roc_applicable": "no", "audit_applicable": "no", "pf_applicable": "no", "esi_applicable": "no", "professional_tax_applicable": "no", "payroll_applicable": "no", "msme_applicable": "no", "import_export_applicable": "no", } worksheet.append([sample_values.get(column, "") for column in TEMPLATE_COLUMNS]) worksheet.freeze_panes = "A2" worksheet.auto_filter.ref = worksheet.dimensions instructions = workbook.create_sheet("instructions") instructions.append(["Field", "Notes"]) notes = { "uploader_user_id": "Must match the logged-in uploader user id exactly.", "firm_tenant_id": "Must match the active firm/tenant context.", "partner_user_id": "Must be an active Partner user in the same firm.", "referred_by_consultant_id": "Optional active consultant profile id who introduced the client.", "primary_consultant_id": "Optional active consultant profile id for the operational client link.", "client_group_code": "Optional tenant-unique group code. Existing group is matched by this code.", "client_group_name": "Required only when creating a new group during import.", "group_type": "Family, Business Group, Promoter Group, Trust Group, Common Management, or Other.", "group_relationship": "Optional relationship such as Spouse, HUF, Company, Trust, or Related Concern.", "is_group_head": "yes/no. Only one group head is recommended per group.", "referral_date": "Optional referral date supported by the existing client schema.", "referral_reference": "Optional referral or source reference.", "referral_status": "active, inactive, ended, or pending.", "communication_routing_mode": "client_direct, consultant_primary, client_and_consultant, or firm_only.", "branch_id": "Optional. If blank, uploader or partner branch is used.", "pan": "Required for Phase 6 PAN identity; must be unique within the firm and valid, e.g. ABCDE1234F.", "email": "Optional only when PAN is provided. A non-deliverable @invalid.local placeholder is generated when blank.", "portal_password": "Retained. Required for import and must contain at least 8 characters.", "portal_password_confirm": "Retained. Must exactly match portal_password.", } for field, note in notes.items(): instructions.append([field, note]) output = io.BytesIO() workbook.save(output) return output.getvalue() def build_preview( db: Session, *, current_user, scope, role_names: set[str], upload_bytes: bytes, ) -> ImportPreview: workbook = load_workbook(io.BytesIO(upload_bytes), data_only=True) worksheet = workbook[workbook.sheetnames[0]] headers = [str(cell.value).strip() if cell.value is not None else "" for cell in worksheet[1]] missing = [column for column in TEMPLATE_COLUMNS if column not in headers] if missing: return ImportPreview( valid_rows=[], errors=[{"row_number": 1, "messages": [f'Missing required columns: {", ".join(missing)}']}], total_rows=0, ) valid_rows: list[dict] = [] errors: list[dict] = [] active_tenant_id = int(scope.tenant_id) active_branch_id = int(scope.branch_id or getattr(current_user, "branch_id", 0) or 0) seen_client_codes: dict[str, int] = {} seen_pans: dict[str, int] = {} seen_emails: dict[str, int] = {} for row_idx in range(2, worksheet.max_row + 1): raw = _row_dict(worksheet, row_idx) if not any(value not in (None, "") for value in raw.values()): continue messages: list[str] = [] cleaned = { key: (_to_bool(value) if key in BOOL_FIELDS else _clean(value)) for key, value in raw.items() } uploader_user_id = _to_int(cleaned.get("uploader_user_id")) or 0 firm_tenant_id = _to_int(cleaned.get("firm_tenant_id")) or 0 partner_user_id = _to_int(cleaned.get("partner_user_id")) or 0 branch_id = _to_int(cleaned.get("branch_id")) or 0 if uploader_user_id != int(current_user.id): messages.append("uploader_user_id must match the currently logged-in user id.") if firm_tenant_id != active_tenant_id: messages.append("firm_tenant_id must match the active firm/tenant context of the uploader.") partner = ( repository.get_partner_for_tenant( db, partner_user_id=partner_user_id, tenant_id=firm_tenant_id, ) if partner_user_id else None ) if not partner: messages.append("partner_user_id must belong to an active Partner user in the same firm.") if "partner" in {str(name).lower() for name in role_names} and partner_user_id != int(current_user.id): messages.append("Partner uploader can import only for their own partner_user_id.") if branch_id: branch = repository.get_branch(db, branch_id) if not branch or int(branch.tenant_id) != firm_tenant_id: messages.append("branch_id must belong to the same firm/tenant.") else: branch_id = int( getattr(partner, "branch_id", None) or active_branch_id or getattr(current_user, "branch_id", 0) or 0 ) if not branch_id: messages.append("branch_id is required when uploader and partner have no branch mapped.") referred_id = _consultant_for_id( db, tenant_id=firm_tenant_id, value=cleaned.get("referred_by_consultant_id"), field_name="referred_by_consultant_id", messages=messages, ) primary_id = _consultant_for_id( db, tenant_id=firm_tenant_id, value=cleaned.get("primary_consultant_id"), field_name="primary_consultant_id", messages=messages, ) routing_mode = (cleaned.get("communication_routing_mode") or "client_and_consultant").lower() if routing_mode not in VALID_ROUTING_MODES: messages.append("communication_routing_mode is invalid.") referral_status = (cleaned.get("referral_status") or "active").lower() if referral_status not in VALID_REFERRAL_STATUSES: messages.append("referral_status must be active, inactive, ended, or pending.") client_code = (cleaned.get("client_code") or "").strip() if client_code: code_key = client_code.upper() if code_key in seen_client_codes: messages.append(f"Duplicate client_code in upload; first used on row {seen_client_codes[code_key]}.") else: seen_client_codes[code_key] = row_idx pan = _normalise_pan_for_preview(cleaned.get("pan")) if not pan: messages.append("PAN is required for Phase 6 client identity and PAN login.") elif not PAN_RE.fullmatch(pan): messages.append("PAN must be a valid 10-character value such as ABCDE1234F.") else: if pan in seen_pans: messages.append(f"Duplicate PAN in upload; first used on row {seen_pans[pan]}.") else: seen_pans[pan] = row_idx existing_pan = repository.get_client_by_pan(db, tenant_id=firm_tenant_id, pan=pan) if existing_pan: messages.append( f"PAN already exists for client {existing_pan.client_code} - {existing_pan.client_name}; existing records are never overwritten by import." ) email = _normalise_email(cleaned.get("email")) if not email and pan and PAN_RE.fullmatch(pan): email = placeholder_email(firm_tenant_id, pan) if not email: messages.append("Email is required when a valid PAN is not available for placeholder generation.") else: if email in seen_emails: messages.append(f"Duplicate email in upload; first used on row {seen_emails[email]}.") else: seen_emails[email] = row_idx existing_user = db.execute(select(User).where(User.email == email)).scalar_one_or_none() if existing_user: messages.append("Email is already used by another ERP login.") password = cleaned.get("portal_password") or "" password_confirm = cleaned.get("portal_password_confirm") or "" if not password: messages.append("portal_password is required for imported clients.") elif len(password) < 8: messages.append("portal_password must contain at least 8 characters.") if password != password_confirm: messages.append("portal_password and portal_password_confirm must match.") payload = { "tenant_id": firm_tenant_id, "branch_id": branch_id, "partner_id": partner_user_id or None, "client_group_id": None, "group_relationship": cleaned.get("group_relationship"), "is_group_head": bool(cleaned.get("is_group_head")), "referred_by_consultant_id": referred_id, "primary_consultant_id": primary_id, "referral_date": cleaned.get("referral_date"), "referral_reference": cleaned.get("referral_reference"), "referral_status": referral_status, "communication_routing_mode": routing_mode, "engagement_mode": cleaned.get("engagement_mode") or "internal_managed", "client_code": client_code, "client_name": cleaned.get("client_name") or "", "trade_name": None, "client_type": cleaned.get("client_type") or "Other", "pan": pan, "gstin": cleaned.get("gstin"), "tan": cleaned.get("tan"), "cin_llpin": cleaned.get("cin_llpin"), "msme_no": cleaned.get("msme_no"), "iec_code": cleaned.get("iec_code"), "contact_person_name": cleaned.get("contact_person_name"), "contact_person_designation": cleaned.get("contact_person_designation"), "mobile": cleaned.get("mobile"), "alternate_mobile": cleaned.get("alternate_mobile"), "email": email, "alternate_email": _normalise_email(cleaned.get("alternate_email")), "address_line_1": cleaned.get("address_line_1"), "address_line_2": cleaned.get("address_line_2"), "city": cleaned.get("city"), "state": cleaned.get("state"), "pincode": cleaned.get("pincode"), "country": cleaned.get("country") or "India", "status": cleaned.get("status") or "active", "client_category": cleaned.get("client_category"), "risk_category": cleaned.get("risk_category"), "onboarding_date": cleaned.get("onboarding_date"), "closing_date": cleaned.get("closing_date"), "notes": cleaned.get("notes"), "gst_applicable": bool(cleaned.get("gst_applicable")), "income_tax_applicable": bool(cleaned.get("income_tax_applicable")), "tds_applicable": bool(cleaned.get("tds_applicable")), "roc_applicable": bool(cleaned.get("roc_applicable")), "audit_applicable": bool(cleaned.get("audit_applicable")), "pf_applicable": bool(cleaned.get("pf_applicable")), "esi_applicable": bool(cleaned.get("esi_applicable")), "professional_tax_applicable": bool(cleaned.get("professional_tax_applicable")), "payroll_applicable": bool(cleaned.get("payroll_applicable")), "msme_applicable": bool(cleaned.get("msme_applicable")), "import_export_applicable": bool(cleaned.get("import_export_applicable")), } try: ClientCreate(**payload) except Exception as exc: messages.append(str(exc)) if messages: errors.append({"row_number": row_idx, "messages": messages, "row": cleaned}) continue valid_rows.append( { "row_number": row_idx, "tenant_id": firm_tenant_id, "branch_id": branch_id, "partner_id": partner_user_id, "client_group_code": cleaned.get("client_group_code"), "client_group_name": cleaned.get("client_group_name"), "group_type": cleaned.get("group_type") or "Family", "client_payload": payload, "portal_password": password, "portal_password_confirm": password_confirm, "email_was_placeholder": email.endswith("@invalid.local"), } ) return ImportPreview( valid_rows=valid_rows, errors=errors, total_rows=len(valid_rows) + len(errors), ) def serialize_preview_rows(valid_rows: list[dict]) -> str: return json.dumps(valid_rows, default=str) def deserialize_preview_rows(raw: str) -> list[dict]: rows = json.loads(raw or "[]") return rows if isinstance(rows, list) else [] def commit_import( db: Session, *, current_user, scope, current_user_roles: list[str], preview_rows: list[dict], ) -> dict: created: list[dict] = [] failures: list[dict] = [] for item in preview_rows: try: payload = dict(item["client_payload"]) pan = normalize_pan(payload.get("pan")) payload["pan"] = pan payload["email"] = ( _normalise_email(payload.get("email")) or placeholder_email(int(payload["tenant_id"]), pan) ) existing_pan = repository.get_client_by_pan( db, tenant_id=int(payload["tenant_id"]), pan=pan, ) if existing_pan: raise ValueError( f"PAN already exists for client {existing_pan.client_code} - {existing_pan.client_name}." ) existing_user = db.execute( select(User).where(User.email == payload["email"]) ).scalar_one_or_none() if existing_user: raise ValueError("Email is already used by another ERP login.") group = resolve_or_create_group( db, tenant_id=int(payload["tenant_id"]), actor_user_id=current_user.id, group_code=item.get("client_group_code"), group_name=item.get("client_group_name"), group_type=item.get("group_type"), ) payload["client_group_id"] = group.id if group else None if group and payload.get("is_group_head"): existing_head = db.execute(select(Client).where(Client.tenant_id == int(payload["tenant_id"]), Client.client_group_id == group.id, Client.is_group_head.is_(True))).scalar_one_or_none() if existing_head: raise ValueError(f"Group {group.group_code} already has group head {existing_head.client_code} - {existing_head.client_name}.") data = ClientCreate(**payload) row = create_client_service( db, data=data, actor_user_id=current_user.id, scope=scope, current_user_roles=current_user_roles, portal_password=item.get("portal_password"), portal_password_confirm=item.get("portal_password_confirm"), ) identity = ensure_identity( db, client=row, actor_user_id=current_user.id, requested_email=row.email, ) if identity.status == "not_invited" and row.portal_user_id: identity.status = "active" identity.activated_at_utc = identity.activated_at_utc or identity.updated_at_utc db.commit() created.append( { "id": row.id, "client_code": row.client_code, "client_name": row.client_name, "pan": row.pan, "portal_user_id": row.portal_user_id, "identity_id": identity.id, "login_email": identity.login_email, "email_is_placeholder": bool(identity.email_is_placeholder), } ) except Exception as exc: db.rollback() failures.append( { "row_number": item.get("row_number"), "message": str(getattr(exc, "detail", exc)), } ) return {"created": created, "failures": failures}