from __future__ import annotations from dataclasses import dataclass import html from pathlib import Path import re import zipfile from django.core.management.base import BaseCommand, CommandError from django.db import transaction from xml.etree import ElementTree as ET from apps.accounts.models import User from apps.business.distributor.models import DistributorID from apps.business.distributor.services.distributor_service import create_distributor_id from apps.business.orders.services.order_service import ( create_pending_payment_order, fulfill_pending_payment_order, ) from apps.business.products.models import Product try: from openpyxl import Workbook except ImportError: # pragma: no cover Workbook = None @dataclass class RunStats: processed_rows: int = 0 purchased_count: int = 0 reused_distributor_count: int = 0 missing_user_count: int = 0 missing_sponsor_count: int = 0 failed_count: int = 0 class Command(BaseCommand): help = ( "Create and approve cash-payment purchases for legacy users using TWM Coins 365, " "generate distributor IDs, and write distributor IDs to column J in the workbook." ) def add_arguments(self, parser): parser.add_argument("--file", required=True, help="Absolute path to legacy workbook (.xlsx)") parser.add_argument("--sheet", default=None, help="Optional sheet name. Defaults to active sheet.") parser.add_argument( "--product-name", default="TWM Coins 365", help="Product name to purchase for each user (default: TWM Coins 365).", ) parser.add_argument( "--admin-mobile", default="8668192080", help="Admin mobile to use as reviewer for payment approval.", ) parser.add_argument( "--output", default=None, help="Optional output workbook path. Defaults to overwrite --file.", ) parser.add_argument( "--dry-run", action="store_true", default=False, help="Preview actions without creating orders/distributor IDs.", ) parser.add_argument( "--start-row", type=int, default=2, help="First worksheet row to process (default: 2).", ) parser.add_argument( "--max-rows", type=int, default=None, help="Optional maximum number of worksheet rows to process.", ) @staticmethod def _clean_string(value) -> str: if value is None: return "" return str(value).strip() @staticmethod def _clean_phone(value) -> str: raw = Command._clean_string(value) if not raw: return "" if raw.endswith(".0") and raw.replace(".", "", 1).isdigit(): raw = raw[:-2] digits = "".join(ch for ch in raw if ch.isdigit()) return digits if digits else raw @staticmethod def _read_zip_text(zf: zipfile.ZipFile, name: str) -> str: original_update_crc = zipfile.ZipExtFile._update_crc zipfile.ZipExtFile._update_crc = lambda self, newdata: None try: return zf.read(name).decode("utf-8", errors="ignore") finally: zipfile.ZipExtFile._update_crc = original_update_crc @staticmethod def _attr_value(attrs: str, name: str) -> str: match = re.search(rf'{name}="([^"]*)"', attrs) return match.group(1) if match else "" def _resolve_sheet_path(self, zf: zipfile.ZipFile, sheet_name: str | None) -> str: workbook_xml = self._read_zip_text(zf, "xl/workbook.xml") workbook_rels = self._read_zip_text(zf, "xl/_rels/workbook.xml.rels") workbook_root = ET.fromstring(workbook_xml) rels_root = ET.fromstring(workbook_rels) namespaces = { "main": "http://schemas.openxmlformats.org/spreadsheetml/2006/main", "rel": "http://schemas.openxmlformats.org/package/2006/relationships", } requested_rid = None first_rid = None for sheet in workbook_root.findall("main:sheets/main:sheet", namespaces): rid = sheet.attrib.get("{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id") if first_rid is None: first_rid = rid if sheet_name and sheet.attrib.get("name") == sheet_name: requested_rid = rid break rid = requested_rid or first_rid if not rid: raise CommandError("Could not resolve a worksheet in the workbook.") target = None for rel in rels_root.findall("rel:Relationship", namespaces): if rel.attrib.get("Id") == rid: target = rel.attrib.get("Target") break if not target: raise CommandError("Could not resolve worksheet target from workbook relationships.") if target.startswith("/"): target = target.lstrip("/") if not target.startswith("xl/"): target = f"xl/{target}" return target def _load_shared_strings(self, zf: zipfile.ZipFile) -> list[str]: if "xl/sharedStrings.xml" not in zf.namelist(): return [] shared_xml = self._read_zip_text(zf, "xl/sharedStrings.xml") try: root = ET.fromstring(shared_xml) except ET.ParseError: return [] namespace = {"main": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"} values: list[str] = [] for item in root.findall("main:si", namespace): texts = [node.text or "" for node in item.findall(".//main:t", namespace)] values.append("".join(texts)) return values def _extract_cell_value(self, cell_xml: str, shared_strings: list[str]) -> tuple[str, str]: attrs_match = re.match(r"]*)>", cell_xml) attrs = attrs_match.group(1) if attrs_match else "" ref = self._attr_value(attrs, "r") cell_type = self._attr_value(attrs, "t") if cell_type == "inlineStr": match = re.search(r"]*>(.*?)", cell_xml, flags=re.S) return ref, html.unescape(match.group(1)) if match else "" if cell_type == "s": match = re.search(r"(.*?)", cell_xml, flags=re.S) if not match: return ref, "" index_text = html.unescape(match.group(1)).strip() if index_text.isdigit(): index = int(index_text) if 0 <= index < len(shared_strings): return ref, shared_strings[index] return ref, index_text match = re.search(r"(.*?)", cell_xml, flags=re.S) return ref, html.unescape(match.group(1)) if match else "" def _iter_sheet_rows(self, sheet_xml: str, shared_strings: list[str]): for row_match in re.finditer(r']*r="(\d+)"[^>]*>(.*?)', sheet_xml, flags=re.S): row_num = int(row_match.group(1)) row_xml = row_match.group(2) row_values: dict[str, str] = {} for cell_match in re.finditer(r']*>.*?', row_xml, flags=re.S): ref, value = self._extract_cell_value(cell_match.group(0), shared_strings) if ref: column = re.match(r"([A-Z]+)", ref) row_values[column.group(1) if column else ref] = value yield row_num, row_values @staticmethod def _col_to_index(col: str) -> int: idx = 0 for ch in col: idx = idx * 26 + (ord(ch) - ord('A') + 1) return idx @staticmethod def _index_to_col(index: int) -> str: out = "" while index > 0: index, rem = divmod(index - 1, 26) out = chr(ord('A') + rem) + out return out def _resolve_admin_user(self, admin_mobile: str) -> User: admin = User.objects.filter(mobile=admin_mobile).first() if admin and (getattr(admin, "is_staff", False) or str(getattr(admin, "role", "")).upper() == "ADMIN"): return admin fallback = User.objects.filter(is_staff=True).order_by("date_joined").first() if fallback: return fallback raise CommandError("No admin user available for payment approval.") def _resolve_sponsor_distributor( self, referred_by_value: str, sponsor_value: str, referral_to_row: dict[str, int], rows_by_number: dict[int, dict[str, str]], distributor_by_referral_code: dict[str, DistributorID], product: Product, admin_root_distributor: DistributorID | None, ) -> tuple[DistributorID | None, str]: referred_by_value = self._clean_string(referred_by_value) sponsor_value = self._clean_string(sponsor_value) if referred_by_value: dist = distributor_by_referral_code.get(referred_by_value) if dist: return dist, "referred_by_map" sponsor_row = referral_to_row.get(referred_by_value) if sponsor_row: sponsor_row_values = rows_by_number.get(sponsor_row, {}) sponsor_dist_code = self._clean_string(sponsor_row_values.get("J")) if sponsor_dist_code: dist = DistributorID.objects.filter(distributor_code=sponsor_dist_code, product=product).first() if dist: return dist, "row_j_distributor_code" sponsor_user = User.objects.filter(referral_code=referred_by_value).first() if sponsor_user: dist = DistributorID.objects.filter(user=sponsor_user, product=product).order_by("-created_at").first() if dist: return dist, "direct_referred_by_code" if sponsor_value.upper() in {"ADMIN", "COMPANY", "ROOT"} and admin_root_distributor: return admin_root_distributor, "admin_root" # Fallback: sponsor value itself may be referral code. if sponsor_value: sponsor_user = User.objects.filter(referral_code=sponsor_value).first() if sponsor_user: dist = DistributorID.objects.filter(user=sponsor_user, product=product).order_by("-created_at").first() if dist: return dist, "direct_referral_code" # Fallback: sponsor value may be mobile. sponsor_mobile = self._clean_phone(sponsor_value) if sponsor_mobile: sponsor_user = User.objects.filter(mobile=sponsor_mobile).first() if sponsor_user: dist = DistributorID.objects.filter(user=sponsor_user, product=product).order_by("-created_at").first() if dist: return dist, "direct_mobile" if not referred_by_value and not sponsor_value: return None, "no_sponsor" return None, "missing" def _save_workbook(self, rows_by_number: dict[int, dict[str, str]], output_path: Path): if Workbook is None: raise CommandError("openpyxl is required to write the output workbook.") max_col_idx = self._col_to_index("J") for values in rows_by_number.values(): for col in values.keys(): max_col_idx = max(max_col_idx, self._col_to_index(col)) wb = Workbook() ws = wb.active for row_num in sorted(rows_by_number.keys()): values = rows_by_number[row_num] for col_idx in range(1, max_col_idx + 1): col = self._index_to_col(col_idx) ws.cell(row=row_num, column=col_idx).value = values.get(col, "") output_path.parent.mkdir(parents=True, exist_ok=True) wb.save(str(output_path)) def handle(self, *args, **options): file_path = Path(options["file"]) sheet_name = options.get("sheet") product_name = self._clean_string(options.get("product_name")) admin_mobile = self._clean_phone(options.get("admin_mobile")) dry_run = bool(options.get("dry_run")) start_row = max(2, int(options.get("start_row") or 2)) max_rows = options.get("max_rows") output_raw = options.get("output") output_path = Path(output_raw) if output_raw else file_path if not file_path.exists(): raise CommandError(f"File not found: {file_path}") if file_path.suffix.lower() != ".xlsx": raise CommandError("Only .xlsx files are supported.") product = Product.objects.filter(name=product_name).first() if not product: raise CommandError(f"Product not found: {product_name}") admin_user = self._resolve_admin_user(admin_mobile) admin_root_distributor = DistributorID.objects.filter(user=admin_user, product=product).order_by("created_at").first() if not admin_root_distributor and not dry_run: admin_root_distributor = create_distributor_id(admin_user, product, None) self.stdout.write( self.style.WARNING( f"Created admin root distributor for sponsor fallback: {admin_root_distributor.distributor_code}" ) ) with zipfile.ZipFile(file_path) as zf: sheet_path = self._resolve_sheet_path(zf, sheet_name) sheet_xml = self._read_zip_text(zf, sheet_path) shared_strings = self._load_shared_strings(zf) complete_rows = list(self._iter_sheet_rows(sheet_xml, shared_strings)) if not complete_rows: raise CommandError("No readable rows were found in workbook sheet.") rows_by_number = {row_num: dict(row_values) for row_num, row_values in complete_rows} for row_num, row_values in rows_by_number.items(): if row_num >= 2: row_values["J"] = "" referral_to_row: dict[str, int] = {} for row_num, row_values in complete_rows: referral_code = self._clean_string(row_values.get("E")) if referral_code and referral_code not in referral_to_row: referral_to_row[referral_code] = row_num stats = RunStats() distributor_by_referral_code: dict[str, DistributorID] = {} self.stdout.write( f"Starting legacy purchase activation for product '{product.name}' " f"with reviewer '{admin_user.mobile or admin_user.email or admin_user.id}'." ) for row_num in sorted(rows_by_number.keys()): if row_num < start_row: continue if max_rows is not None and stats.processed_rows >= max_rows: break row_values = rows_by_number[row_num] stats.processed_rows += 1 if stats.processed_rows % 100 == 0: self.stdout.write(f"Progress: processed {stats.processed_rows} rows...") legacy_id = self._clean_string(row_values.get("C")) referral_code = self._clean_string(row_values.get("E")) phone = self._clean_phone(row_values.get("G")) sponsor_value = self._clean_string(row_values.get("H")) referred_by_value = self._clean_string(row_values.get("I")) if not phone: stats.missing_user_count += 1 continue user = User.objects.filter(mobile=phone).first() if not user: stats.missing_user_count += 1 continue sponsor_distributor, sponsor_source = self._resolve_sponsor_distributor( referred_by_value=referred_by_value, sponsor_value=sponsor_value, referral_to_row=referral_to_row, rows_by_number=rows_by_number, distributor_by_referral_code=distributor_by_referral_code, product=product, admin_root_distributor=admin_root_distributor, ) if (referred_by_value or sponsor_value) and sponsor_distributor is None: stats.missing_sponsor_count += 1 continue client_reference_id = f"LEGACY365-R2-{legacy_id or phone}-{row_num}" try: if dry_run: row_values["J"] = "" continue with transaction.atomic(): order = create_pending_payment_order( user=user, product=product, sponsor_distributor=sponsor_distributor, client_reference_id=client_reference_id, payment_method="cash_payment", qr_reference="", amount=product.base_cost, ) if order.status == "PAYMENT_REVIEW": order, distributor, _ = fulfill_pending_payment_order(order, reviewed_by=admin_user) else: distributor = order.distributor if not distributor: raise Exception("Distributor not created during approval flow.") row_values["J"] = distributor.distributor_code stats.purchased_count += 1 if referral_code: distributor_by_referral_code[referral_code] = distributor self.stdout.write( f"Row {row_num}: {phone} -> {distributor.distributor_code} " f"(order={order.reference_id}, sponsor={sponsor_source})" ) except Exception as exc: stats.failed_count += 1 self.stdout.write(self.style.WARNING(f"Row {row_num} failed for {phone}: {exc}")) if not dry_run: self._save_workbook(rows_by_number, output_path) self.stdout.write(self.style.SUCCESS("Legacy purchase activation completed.")) self.stdout.write(f"Processed rows: {stats.processed_rows}") self.stdout.write(f"Purchased+approved: {stats.purchased_count}") self.stdout.write(f"Reused existing distributor: {stats.reused_distributor_count}") self.stdout.write(f"Missing user: {stats.missing_user_count}") self.stdout.write(f"Missing sponsor mapping: {stats.missing_sponsor_count}") self.stdout.write(f"Failed rows: {stats.failed_count}") self.stdout.write(f"Dry run: {dry_run}") self.stdout.write(f"Saved workbook: {output_path}")