Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Align client import with PAN identity and consultant linkage

Browse Source
This commit is contained in:
A R R R Associates
2026-07-23 22:52:51 +05:30
parent 76039d915d
commit 5e6584702c
1 changed files with 436 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/clients/import_service.py
+436 -168
View File
@@ -2,17 +2,23 @@ from __future__ import annotations
import io import io
import json import json
import re
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime
from typing import Any from typing import Any
from openpyxl import Workbook, load_workbook from openpyxl import Workbook, load_workbook
from sqlalchemy import select
from sqlalchemy.orm import Session 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 import repository
from app.modules.clients.models import Client
from app.modules.clients.schemas import ClientCreate from app.modules.clients.schemas import ClientCreate
from app.modules.clients.service import create_client_service 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.consultants.service import get_consultant
from app.modules.core.iam.models import User
TEMPLATE_COLUMNS = [ TEMPLATE_COLUMNS = [
"uploader_user_id", "uploader_user_id",
@@ -69,11 +75,29 @@ TEMPLATE_COLUMNS = [
] ]
BOOL_FIELDS = { BOOL_FIELDS = {
"gst_applicable", "income_tax_applicable", "tds_applicable", "roc_applicable", "gst_applicable",
"audit_applicable", "pf_applicable", "esi_applicable", "professional_tax_applicable", "income_tax_applicable",
"payroll_applicable", "msme_applicable", "import_export_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 @dataclass
class ImportPreview: class ImportPreview:
valid_rows: list[dict] valid_rows: list[dict]
@@ -84,198 +108,380 @@ class ImportPreview:
def _clean(value: Any) -> str | None: def _clean(value: Any) -> str | None:
if value is None: if value is None:
return None return None
txt = str(value).strip() text = str(value).strip()
return txt or None return text or None
def _to_bool(value: Any) -> bool: def _to_bool(value: Any) -> bool:
txt = str(value or '').strip().lower() return str(value or "").strip().lower() in {"1", "true", "yes", "y", "on"}
return txt in {'1','true','yes','y','on'}
def build_client_import_template_bytes(*, current_user, tenant_id: int, partner_id: int | None) -> bytes: def _to_int(value: Any) -> int | None:
wb = Workbook() text = _clean(value)
ws = wb.active if not text:
ws.title = 'clients_import' return None
ws.append(TEMPLATE_COLUMNS) try:
sample = [ return int(float(text))
current_user.id, tenant_id, partner_id or current_user.id, getattr(current_user, 'branch_id', '') or '', except (TypeError, ValueError):
'CLT-001', 'Sample Client', 'Other', 'internal_managed', 'client@example.com', 'ChangeMe@123', 'ChangeMe@123', return None
'9876543210', '', '', '', '', '', '', 'Client Contact', 'Proprietor', '', '', 'Address line 1', '', 'Chennai', 'Tamil Nadu', '600001', 'India', '', '', '', '', '', 'active',
'yes', 'yes', 'no', 'no', 'no', 'no', 'no', 'no', 'no', 'no', 'no'
] def _normalise_email(value: Any) -> str | None:
ws.append(sample) text = _clean(value)
ref = wb.create_sheet('instructions') return text.lower() if text else None
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.']) def _normalise_pan_for_preview(value: Any) -> str | None:
ref.append(['partner_user_id', 'Must be an active Partner user mapped to the same firm.']) text = re.sub(r"\s+", "", str(value or "").upper())
ref.append(['branch_id', 'Optional. If blank, uploader branch or partner branch will be used.']) return text or None
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 _row_dict(ws, row_idx: int) -> dict[str, Any]: 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]] headers = [str(cell.value).strip() if cell.value is not None else "" for cell in ws[1]]
values = [c.value for c in ws[row_idx]] values = [cell.value for cell in ws[row_idx]]
return {headers[i]: values[i] if i < len(values) else None for i in range(len(headers)) if headers[i]} 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: def _consultant_for_id(
wb = load_workbook(io.BytesIO(upload_bytes), data_only=True) db: Session,
ws = wb[wb.sheetnames[0]] *,
headers = [str(c.value).strip() if c.value is not None else '' for c in ws[1]] tenant_id: int,
missing = [c for c in TEMPLATE_COLUMNS if c not in headers] 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: 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 = [] valid_rows: list[dict] = []
errors = [] errors: list[dict] = []
active_tenant_id = int(scope.tenant_id) 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): for row_idx in range(2, worksheet.max_row + 1):
raw = _row_dict(ws, row_idx) raw = _row_dict(worksheet, row_idx)
if not any(v not in (None, '') for v in raw.values()): if not any(value not in (None, "") for value in raw.values()):
continue continue
msgs: list[str] = []
cleaned = {k: (_to_bool(v) if k in BOOL_FIELDS else _clean(v)) for k, v in raw.items()}
try: messages: list[str] = []
uploader_user_id = int(cleaned.get('uploader_user_id') or 0) cleaned = {
except Exception: key: (_to_bool(value) if key in BOOL_FIELDS else _clean(value))
uploader_user_id = 0 for key, value in raw.items()
try: }
firm_tenant_id = int(cleaned.get('firm_tenant_id') or 0)
except Exception: uploader_user_id = _to_int(cleaned.get("uploader_user_id")) or 0
firm_tenant_id = 0 firm_tenant_id = _to_int(cleaned.get("firm_tenant_id")) or 0
try: partner_user_id = _to_int(cleaned.get("partner_user_id")) or 0
partner_user_id = int(cleaned.get('partner_user_id') or 0) branch_id = _to_int(cleaned.get("branch_id")) or 0
except Exception:
partner_user_id = 0
try:
branch_id = int(cleaned.get('branch_id') or 0)
except Exception:
branch_id = 0
if uploader_user_id != int(current_user.id): 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: if firm_tenant_id != active_tenant_id:
msgs.append('firm_tenant_id must match the active firm/tenant context of the uploader.') 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
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: if not partner:
msgs.append('partner_user_id must belong to an active Partner user in the same firm.') messages.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): if "partner" in {str(name).lower() for name 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 uploader can import only for their own partner_user_id.")
if branch_id: if branch_id:
branch = repository.get_branch(db, branch_id) branch = repository.get_branch(db, branch_id)
if not branch or int(branch.tenant_id) != firm_tenant_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: 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: 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"): referred_id = _consultant_for_id(
raw_consultant_id = cleaned.get(consultant_field) db,
if raw_consultant_id: tenant_id=firm_tenant_id,
try: value=cleaned.get("referred_by_consultant_id"),
consultant_id = int(raw_consultant_id) field_name="referred_by_consultant_id",
except Exception: messages=messages,
consultant_id = 0 )
consultant = get_consultant(db, tenant_id=firm_tenant_id, consultant_id=consultant_id) if consultant_id else None primary_id = _consultant_for_id(
if not consultant or not consultant.is_active: db,
msgs.append(f"{consultant_field} must be an active consultant profile id in the same firm.") 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 = { payload = {
'tenant_id': firm_tenant_id, "tenant_id": firm_tenant_id,
'branch_id': branch_id, "branch_id": branch_id,
'partner_id': partner_user_id or None, "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, "referred_by_consultant_id": referred_id,
'primary_consultant_id': int(cleaned.get('primary_consultant_id')) if cleaned.get('primary_consultant_id') else None, "primary_consultant_id": primary_id,
'referral_date': cleaned.get('referral_date'), "referral_date": cleaned.get("referral_date"),
'referral_reference': cleaned.get('referral_reference'), "referral_reference": cleaned.get("referral_reference"),
'referral_status': cleaned.get('referral_status') or 'active', "referral_status": referral_status,
'communication_routing_mode': cleaned.get('communication_routing_mode') or 'client_and_consultant', "communication_routing_mode": routing_mode,
'engagement_mode': cleaned.get('engagement_mode') or 'internal_managed', "engagement_mode": cleaned.get("engagement_mode") or "internal_managed",
'client_code': cleaned.get('client_code') or '', "client_code": client_code,
'client_name': cleaned.get('client_name') or '', "client_name": cleaned.get("client_name") or "",
'trade_name': None, "trade_name": None,
'client_type': cleaned.get('client_type') or 'Other', "client_type": cleaned.get("client_type") or "Other",
'pan': cleaned.get('pan'), "pan": pan,
'gstin': cleaned.get('gstin'), "gstin": cleaned.get("gstin"),
'tan': cleaned.get('tan'), "tan": cleaned.get("tan"),
'cin_llpin': cleaned.get('cin_llpin'), "cin_llpin": cleaned.get("cin_llpin"),
'msme_no': cleaned.get('msme_no'), "msme_no": cleaned.get("msme_no"),
'iec_code': cleaned.get('iec_code'), "iec_code": cleaned.get("iec_code"),
'contact_person_name': cleaned.get('contact_person_name'), "contact_person_name": cleaned.get("contact_person_name"),
'contact_person_designation': cleaned.get('contact_person_designation'), "contact_person_designation": cleaned.get("contact_person_designation"),
'mobile': cleaned.get('mobile'), "mobile": cleaned.get("mobile"),
'alternate_mobile': cleaned.get('alternate_mobile'), "alternate_mobile": cleaned.get("alternate_mobile"),
'email': cleaned.get('email'), "email": email,
'alternate_email': cleaned.get('alternate_email'), "alternate_email": _normalise_email(cleaned.get("alternate_email")),
'address_line_1': cleaned.get('address_line_1'), "address_line_1": cleaned.get("address_line_1"),
'address_line_2': cleaned.get('address_line_2'), "address_line_2": cleaned.get("address_line_2"),
'city': cleaned.get('city'), "city": cleaned.get("city"),
'state': cleaned.get('state'), "state": cleaned.get("state"),
'pincode': cleaned.get('pincode'), "pincode": cleaned.get("pincode"),
'country': cleaned.get('country') or 'India', "country": cleaned.get("country") or "India",
'status': cleaned.get('status') or 'active', "status": cleaned.get("status") or "active",
'client_category': cleaned.get('client_category'), "client_category": cleaned.get("client_category"),
'risk_category': cleaned.get('risk_category'), "risk_category": cleaned.get("risk_category"),
'onboarding_date': cleaned.get('onboarding_date'), "onboarding_date": cleaned.get("onboarding_date"),
'closing_date': cleaned.get('closing_date'), "closing_date": cleaned.get("closing_date"),
'notes': cleaned.get('notes'), "notes": cleaned.get("notes"),
'gst_applicable': cleaned.get('gst_applicable') or False, "gst_applicable": bool(cleaned.get("gst_applicable")),
'income_tax_applicable': cleaned.get('income_tax_applicable') or False, "income_tax_applicable": bool(cleaned.get("income_tax_applicable")),
'tds_applicable': cleaned.get('tds_applicable') or False, "tds_applicable": bool(cleaned.get("tds_applicable")),
'roc_applicable': cleaned.get('roc_applicable') or False, "roc_applicable": bool(cleaned.get("roc_applicable")),
'audit_applicable': cleaned.get('audit_applicable') or False, "audit_applicable": bool(cleaned.get("audit_applicable")),
'pf_applicable': cleaned.get('pf_applicable') or False, "pf_applicable": bool(cleaned.get("pf_applicable")),
'esi_applicable': cleaned.get('esi_applicable') or False, "esi_applicable": bool(cleaned.get("esi_applicable")),
'professional_tax_applicable': cleaned.get('professional_tax_applicable') or False, "professional_tax_applicable": bool(cleaned.get("professional_tax_applicable")),
'payroll_applicable': cleaned.get('payroll_applicable') or False, "payroll_applicable": bool(cleaned.get("payroll_applicable")),
'msme_applicable': cleaned.get('msme_applicable') or False, "msme_applicable": bool(cleaned.get("msme_applicable")),
'import_export_applicable': cleaned.get('import_export_applicable') or False, "import_export_applicable": bool(cleaned.get("import_export_applicable")),
} }
try: try:
ClientCreate(**payload) ClientCreate(**payload)
except Exception as exc: except Exception as exc:
msgs.append(str(exc)) messages.append(str(exc))
if not cleaned.get('portal_password'): if messages:
msgs.append('portal_password is required for imported clients.') errors.append({"row_number": row_idx, "messages": messages, "row": cleaned})
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})
continue continue
valid_rows.append({ valid_rows.append(
'row_number': row_idx, {
'tenant_id': firm_tenant_id, "row_number": row_idx,
'branch_id': branch_id, "tenant_id": firm_tenant_id,
'partner_id': partner_user_id, "branch_id": branch_id,
'client_payload': payload, "partner_id": partner_user_id,
'portal_password': cleaned.get('portal_password'), "client_payload": payload,
'portal_password_confirm': cleaned.get('portal_password_confirm'), "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: 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]: 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 [] 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: def commit_import(
created = [] db: Session,
failures = [] *,
current_user,
scope,
current_user_roles: list[str],
preview_rows: list[dict],
) -> dict:
created: list[dict] = []
failures: list[dict] = []
for item in preview_rows: for item in preview_rows:
try: 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( row = create_client_service(
db, db,
data=data, data=data,
actor_user_id=current_user.id, actor_user_id=current_user.id,
scope=scope, scope=scope,
current_user_roles=current_user_roles, current_user_roles=current_user_roles,
portal_password=item.get('portal_password'), portal_password=item.get("portal_password"),
portal_password_confirm=item.get('portal_password_confirm'), 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: except Exception as exc:
failures.append({'row_number': item.get('row_number'), 'message': str(getattr(exc, 'detail', exc))}) db.rollback()
return {'created': created, 'failures': failures} failures.append(
{
"row_number": item.get("row_number"),
"message": str(getattr(exc, "detail", exc)),
}
)
return {"created": created, "failures": failures}