From 5e6584702c58031f682d6f060420dbf133999ce0 Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Thu, 23 Jul 2026 22:52:51 +0530 Subject: [PATCH] Align client import with PAN identity and consultant linkage --- app/modules/clients/import_service.py | 604 +++++++++++++++++++------- 1 file changed, 436 insertions(+), 168 deletions(-) diff --git a/app/modules/clients/import_service.py b/app/modules/clients/import_service.py index f5481b3..f842812 100644 --- a/app/modules/clients/import_service.py +++ b/app/modules/clients/import_service.py @@ -2,17 +2,23 @@ from __future__ import annotations import io import json +import re from dataclasses import dataclass -from datetime import datetime 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.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", @@ -69,11 +75,29 @@ TEMPLATE_COLUMNS = [ ] BOOL_FIELDS = { - "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", + "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] @@ -84,198 +108,380 @@ class ImportPreview: def _clean(value: Any) -> str | None: if value is None: return None - txt = str(value).strip() - return txt or None + text = str(value).strip() + return text or None def _to_bool(value: Any) -> bool: - txt = str(value or '').strip().lower() - return txt in {'1','true','yes','y','on'} + return str(value or "").strip().lower() in {"1", "true", "yes", "y", "on"} -def build_client_import_template_bytes(*, current_user, tenant_id: int, partner_id: int | None) -> bytes: - wb = Workbook() - ws = wb.active - ws.title = 'clients_import' - ws.append(TEMPLATE_COLUMNS) - sample = [ - current_user.id, tenant_id, partner_id or current_user.id, getattr(current_user, 'branch_id', '') or '', - 'CLT-001', 'Sample Client', 'Other', 'internal_managed', 'client@example.com', 'ChangeMe@123', 'ChangeMe@123', - '9876543210', '', '', '', '', '', '', 'Client Contact', 'Proprietor', '', '', 'Address line 1', '', 'Chennai', 'Tamil Nadu', '600001', 'India', '', '', '', '', '', 'active', - 'yes', 'yes', 'no', 'no', 'no', 'no', 'no', 'no', 'no', 'no', 'no' - ] - ws.append(sample) - ref = wb.create_sheet('instructions') - ref.append(['Field', 'Notes']) - ref.append(['uploader_user_id', 'Must match the logged-in uploader user id exactly.']) - ref.append(['firm_tenant_id', 'Must match the active firm/tenant context of the upload.']) - ref.append(['partner_user_id', 'Must be an active Partner user mapped to the same firm.']) - ref.append(['branch_id', 'Optional. If blank, uploader branch or partner branch will be used.']) - ref.append(['referred_by_consultant_id', 'Optional active consultant profile id who introduced the client.']) - ref.append(['primary_consultant_id', 'Optional active consultant profile id for the operational client link.']) - ref.append(['communication_routing_mode', 'client_direct, consultant_primary, client_and_consultant, or firm_only.']) - ref.append(['email', 'Used as the client frontend login email.']) - ref.append(['portal_password', 'Minimum 8 characters.']) - ref.append(['portal_password_confirm', 'Must match portal_password.']) - bio = io.BytesIO() - wb.save(bio) - return bio.getvalue() +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(c.value).strip() if c.value is not None else '' for c in ws[1]] - values = [c.value for c in ws[row_idx]] - return {headers[i]: values[i] if i < len(values) else None for i in range(len(headers)) if headers[i]} + 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 build_preview(db: Session, *, current_user, scope, role_names: set[str], upload_bytes: bytes) -> ImportPreview: - wb = load_workbook(io.BytesIO(upload_bytes), data_only=True) - ws = wb[wb.sheetnames[0]] - headers = [str(c.value).strip() if c.value is not None else '' for c in ws[1]] - missing = [c for c in TEMPLATE_COLUMNS if c not in headers] +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.", + "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) + return ImportPreview( + valid_rows=[], + errors=[{"row_number": 1, "messages": [f'Missing required columns: {", ".join(missing)}']}], + total_rows=0, + ) - valid_rows = [] - errors = [] + 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) + 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, ws.max_row + 1): - raw = _row_dict(ws, row_idx) - if not any(v not in (None, '') for v in raw.values()): + 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 - msgs: list[str] = [] - cleaned = {k: (_to_bool(v) if k in BOOL_FIELDS else _clean(v)) for k, v in raw.items()} - try: - uploader_user_id = int(cleaned.get('uploader_user_id') or 0) - except Exception: - uploader_user_id = 0 - try: - firm_tenant_id = int(cleaned.get('firm_tenant_id') or 0) - except Exception: - firm_tenant_id = 0 - try: - partner_user_id = int(cleaned.get('partner_user_id') or 0) - except Exception: - partner_user_id = 0 - try: - branch_id = int(cleaned.get('branch_id') or 0) - except Exception: - branch_id = 0 + 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): - msgs.append('uploader_user_id must match the currently logged-in user id.') + messages.append("uploader_user_id must match the currently logged-in user id.") if firm_tenant_id != active_tenant_id: - msgs.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 + 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: - msgs.append('partner_user_id must belong to an active Partner user in the same firm.') - if 'partner' in role_names and partner_user_id != int(current_user.id): - msgs.append('Partner uploader can import only for their own partner_user_id.') + 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: - msgs.append('branch_id must belong to the same firm/tenant.') + 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) + 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: - msgs.append('branch_id is required when uploader and partner have no branch mapped.') + messages.append("branch_id is required when uploader and partner have no branch mapped.") - for consultant_field in ("referred_by_consultant_id", "primary_consultant_id"): - raw_consultant_id = cleaned.get(consultant_field) - if raw_consultant_id: - try: - consultant_id = int(raw_consultant_id) - except Exception: - consultant_id = 0 - consultant = get_consultant(db, tenant_id=firm_tenant_id, consultant_id=consultant_id) if consultant_id else None - if not consultant or not consultant.is_active: - msgs.append(f"{consultant_field} must be an active consultant profile id in the same firm.") + 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, - 'referred_by_consultant_id': int(cleaned.get('referred_by_consultant_id')) if cleaned.get('referred_by_consultant_id') else None, - 'primary_consultant_id': int(cleaned.get('primary_consultant_id')) if cleaned.get('primary_consultant_id') else None, - 'referral_date': cleaned.get('referral_date'), - 'referral_reference': cleaned.get('referral_reference'), - 'referral_status': cleaned.get('referral_status') or 'active', - 'communication_routing_mode': cleaned.get('communication_routing_mode') or 'client_and_consultant', - 'engagement_mode': cleaned.get('engagement_mode') or 'internal_managed', - 'client_code': cleaned.get('client_code') or '', - 'client_name': cleaned.get('client_name') or '', - 'trade_name': None, - 'client_type': cleaned.get('client_type') or 'Other', - 'pan': cleaned.get('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': cleaned.get('email'), - 'alternate_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': cleaned.get('gst_applicable') or False, - 'income_tax_applicable': cleaned.get('income_tax_applicable') or False, - 'tds_applicable': cleaned.get('tds_applicable') or False, - 'roc_applicable': cleaned.get('roc_applicable') or False, - 'audit_applicable': cleaned.get('audit_applicable') or False, - 'pf_applicable': cleaned.get('pf_applicable') or False, - 'esi_applicable': cleaned.get('esi_applicable') or False, - 'professional_tax_applicable': cleaned.get('professional_tax_applicable') or False, - 'payroll_applicable': cleaned.get('payroll_applicable') or False, - 'msme_applicable': cleaned.get('msme_applicable') or False, - 'import_export_applicable': cleaned.get('import_export_applicable') or False, + "tenant_id": firm_tenant_id, + "branch_id": branch_id, + "partner_id": partner_user_id or None, + "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: - msgs.append(str(exc)) + messages.append(str(exc)) - if not cleaned.get('portal_password'): - msgs.append('portal_password is required for imported clients.') - if cleaned.get('portal_password') != cleaned.get('portal_password_confirm'): - msgs.append('portal_password and portal_password_confirm must match.') - - # intra-file duplicate client codes - if any(v.get('client_code') == payload['client_code'] and v.get('tenant_id') == firm_tenant_id for v in valid_rows): - msgs.append('Duplicate client_code found within the same upload file.') - - if msgs: - errors.append({'row_number': row_idx, 'messages': msgs, 'row': cleaned}) + 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_payload': payload, - 'portal_password': cleaned.get('portal_password'), - 'portal_password_confirm': cleaned.get('portal_password_confirm'), - }) + valid_rows.append( + { + "row_number": row_idx, + "tenant_id": firm_tenant_id, + "branch_id": branch_id, + "partner_id": partner_user_id, + "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)) + return ImportPreview( + valid_rows=valid_rows, + errors=errors, + total_rows=len(valid_rows) + len(errors), + ) def serialize_preview_rows(valid_rows: list[dict]) -> str: @@ -283,26 +489,88 @@ def serialize_preview_rows(valid_rows: list[dict]) -> str: def deserialize_preview_rows(raw: str) -> list[dict]: - rows = json.loads(raw or '[]') + 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 = [] - failures = [] +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: - data = ClientCreate(**item['client_payload']) + 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.") + + 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'), + 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), + } ) - created.append({'id': row.id, 'client_code': row.client_code, 'client_name': row.client_name}) except Exception as exc: - failures.append({'row_number': item.get('row_number'), 'message': str(getattr(exc, 'detail', exc))}) - return {'created': created, 'failures': failures} + db.rollback() + failures.append( + { + "row_number": item.get("row_number"), + "message": str(getattr(exc, "detail", exc)), + } + ) + + return {"created": created, "failures": failures}