A R R R Associates
282 lines
11 KiB
Python
282 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.clients import repository
|
|
from app.modules.clients.schemas import ClientCreate
|
|
from app.modules.clients.service import create_client_service
|
|
|
|
TEMPLATE_COLUMNS = [
|
|
"uploader_user_id",
|
|
"firm_tenant_id",
|
|
"partner_user_id",
|
|
"branch_id",
|
|
"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 = {
|
|
"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",
|
|
}
|
|
|
|
@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
|
|
txt = str(value).strip()
|
|
return txt or None
|
|
|
|
|
|
def _to_bool(value: Any) -> bool:
|
|
txt = str(value or '').strip().lower()
|
|
return txt 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(['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]:
|
|
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]}
|
|
|
|
|
|
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]
|
|
if missing:
|
|
return ImportPreview(valid_rows=[], errors=[{'row_number': 1, 'messages': [f'Missing required columns: {", ".join(missing)}']}], total_rows=0)
|
|
|
|
valid_rows = []
|
|
errors = []
|
|
active_tenant_id = int(scope.tenant_id)
|
|
active_branch_id = int(scope.branch_id or getattr(current_user, 'branch_id', 0) or 0)
|
|
|
|
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()):
|
|
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
|
|
|
|
if uploader_user_id != int(current_user.id):
|
|
msgs.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
|
|
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.')
|
|
|
|
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.')
|
|
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:
|
|
msgs.append('branch_id is required when uploader and partner have no branch mapped.')
|
|
|
|
payload = {
|
|
'tenant_id': firm_tenant_id,
|
|
'branch_id': branch_id,
|
|
'partner_id': partner_user_id or None,
|
|
'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,
|
|
}
|
|
|
|
try:
|
|
ClientCreate(**payload)
|
|
except Exception as exc:
|
|
msgs.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})
|
|
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'),
|
|
})
|
|
|
|
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 = []
|
|
failures = []
|
|
for item in preview_rows:
|
|
try:
|
|
data = ClientCreate(**item['client_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'),
|
|
)
|
|
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}
|