"""وارد کردن گروهی مشتری از فایل اکسل، با گزارش دقیق و پیشرفت زنده (اجرا در ترد پس‌زمینه).""" import io import os import re import threading import time import uuid from flask import current_app from app.extensions import db PENDING_MAX_AGE_SECONDS = 24 * 60 * 60 # فایل‌های موقتی که وارد نشدن (انصراف/بستن صفحه) بعد از یک روز پاک می‌شن TEMPLATE_HEADERS = [ "نام", "نام خانوادگی", "شهر", "تلفن ثابت", "شماره موبایل", "جنسیت (آقا/خانم)", "تاریخ تولد (شمسی، مثال 1370/06/15)", "کد مشتری", "معرف", "نحوه آشنایی", "اینستاگرام", "وضعیت (جدید/فعال/غیرفعال/خواب)", "آدرس", "توضیحات", ] GENDER_MAP = {"آقا": "male", "خانم": "female"} STATUS_MAP = {"جدید": "new", "فعال": "active", "غیرفعال": "inactive", "خواب": "sleeping"} def build_template_excel() -> io.BytesIO: from openpyxl import Workbook from openpyxl.styles import Font, Alignment, PatternFill wb = Workbook() ws = wb.active ws.title = "قالب مشتریان" ws.sheet_view.rightToLeft = True ws.append(TEMPLATE_HEADERS) header_fill = PatternFill(start_color="D4AF37", end_color="D4AF37", fill_type="solid") for cell in ws[1]: cell.font = Font(bold=True) cell.fill = header_fill cell.alignment = Alignment(horizontal="center") ws.append(["مریم", "کریمی", "تهران", "", "09121234567", "خانم", "1372/03/10", "", "", "اینستاگرام", "جدید", "", ""]) for col in ws.columns: ws.column_dimensions[col[0].column_letter].width = 22 buffer = io.BytesIO() wb.save(buffer) buffer.seek(0) return buffer def _cell(row, idx): v = row[idx].value if idx < len(row) else None if v is None: return "" return str(v).strip() def _tmp_import_dir(): path = current_app.instance_path and os.path.join(current_app.instance_path, "tmp_imports") os.makedirs(path, exist_ok=True) return path def is_valid_token(token) -> bool: """توکن فقط همون شناسه‌ی هگزی که خودمون ساختیم؛ وگرنه مقداری مثل ../../x می‌تونست به فایل‌های دیگه‌ی سرور برسه.""" return isinstance(token, str) and re.fullmatch(r"[0-9a-f]{32}", token) is not None def _cleanup_stale_pending(directory: str): now = time.time() for name in os.listdir(directory): path = os.path.join(directory, name) try: if name.endswith(".xlsx") and now - os.path.getmtime(path) > PENDING_MAX_AGE_SECONDS: os.remove(path) except OSError: pass def save_pending_import(file_bytes: bytes) -> str: """فایل آپلودشده رو موقتاً روی دیسک نگه می‌داره تا بعد از تأیید کاربر (تصمیم درباره تکراری‌ها) واقعاً وارد بشه.""" directory = _tmp_import_dir() _cleanup_stale_pending(directory) token = uuid.uuid4().hex path = os.path.join(directory, f"{token}.xlsx") with open(path, "wb") as f: f.write(file_bytes) return token def load_pending_import(token: str) -> bytes | None: if not is_valid_token(token): return None path = os.path.join(_tmp_import_dir(), f"{token}.xlsx") if not os.path.isfile(path): return None with open(path, "rb") as f: return f.read() def delete_pending_import(token: str): if not is_valid_token(token): return path = os.path.join(_tmp_import_dir(), f"{token}.xlsx") if os.path.isfile(path): try: os.remove(path) except OSError: pass def find_duplicate_conflicts(file_bytes: bytes) -> list[dict]: """ردیف‌هایی که شماره تلفن (ثابت یا موبایل) از قبل توی دیتابیس هست رو برمی‌گردونه.""" from app.models.customer import Customer from app.services.customer_service import normalize_phone from openpyxl import load_workbook try: wb = load_workbook(io.BytesIO(file_bytes), read_only=True, data_only=True) ws = wb.active rows = list(ws.iter_rows(min_row=2)) except Exception: return [] conflicts = [] for i, row in enumerate(rows, start=2): first_name = _cell(row, 0) last_name = _cell(row, 1) raw_landline = _cell(row, 3) raw_mobile = _cell(row, 4) landline = normalize_phone(raw_landline) mobile = normalize_phone(raw_mobile) if not first_name or not last_name or (not landline and not mobile): continue existing = None matched_field = None if mobile: existing = Customer.query.filter(Customer.mobile_phone.in_({mobile, raw_mobile}), Customer.is_deleted.is_(False)).first() matched_field = "mobile" if not existing and landline: existing = Customer.query.filter(Customer.landline_phone.in_({landline, raw_landline}), Customer.is_deleted.is_(False)).first() matched_field = "landline" if existing: conflicts.append({ "row": i, "new_name": f"{first_name} {last_name}", "phone": mobile if matched_field == "mobile" else landline, "existing_customer_id": existing.id, "existing_name": f"{existing.first_name} {existing.last_name}", }) return conflicts def start_import_job(file_bytes: bytes, user_id: int, skip_rows=None) -> int: """جاب رو می‌سازه و پردازش رو در ترد پس‌زمینه شروع می‌کنه. شناسه جاب رو برمی‌گردونه.""" from app.models.import_job import ImportJob job = ImportJob(user_id=user_id, status="running") db.session.add(job) db.session.commit() job_id = job.id app_obj = current_app._get_current_object() thread = threading.Thread( target=_process_import, args=(app_obj, job_id, file_bytes, user_id, skip_rows or set()), daemon=True ) thread.start() return job_id def _process_import(app, job_id, file_bytes, user_id, skip_rows=frozenset()): with app.app_context(): from app.models.import_job import ImportJob from app.models.customer import Customer from app.services.jalali import jalali_str_to_gregorian_date from app.services.customer_service import next_customer_code, normalize_phone, refresh_statuses from app.services.activity_logger import log_action from app.models.user import User job = db.session.get(ImportJob, job_id) try: from openpyxl import load_workbook wb = load_workbook(io.BytesIO(file_bytes), read_only=True, data_only=True) ws = wb.active rows = list(ws.iter_rows(min_row=2)) except Exception as exc: job.status = "error" job.error_message = f"خواندن فایل اکسل ممکن نشد: {exc}" db.session.commit() return job.total_rows = len(rows) db.session.commit() notes = [] # {"row": i, "reason": "...", "type": "skipped"|"warning"} created = 0 user = db.session.get(User, user_id) for i, row in enumerate(rows, start=2): first_name = _cell(row, 0) last_name = _cell(row, 1) if not first_name or not last_name: notes.append({"row": i, "reason": "نام یا نام خانوادگی خالی است — این ردیف رد شد", "type": "skipped"}) elif i in skip_rows: notes.append({"row": i, "reason": "به دلیل تکراری بودن شماره تلفن، طبق انتخاب شما رد شد", "type": "skipped"}) else: city = _cell(row, 2) or None landline = normalize_phone(_cell(row, 3)) mobile = normalize_phone(_cell(row, 4)) gender = GENDER_MAP.get(_cell(row, 5), None) birth_date = None birth_raw = _cell(row, 6) if birth_raw: try: birth_date = jalali_str_to_gregorian_date(birth_raw) except Exception: notes.append({"row": i, "reason": f"تاریخ تولد «{birth_raw}» قابل تشخیص نبود — بدون تاریخ تولد ثبت شد", "type": "warning"}) customer_code = _cell(row, 7) or None if customer_code and Customer.query.filter_by(customer_code=customer_code).first(): notes.append({"row": i, "reason": f"کد مشتری «{customer_code}» تکراری بود — کد جدید ساخته شد", "type": "warning"}) customer_code = next_customer_code() referrer_text = _cell(row, 8) or None how_found = _cell(row, 9) or None instagram_id = _cell(row, 10) or None # فقط «غیرفعال» دستی نگه داشته می‌شه؛ بقیه‌ی وضعیت‌ها خودکار از روی تماس و خرید حساب می‌شن status = "inactive" if STATUS_MAP.get(_cell(row, 11)) == "inactive" else "new" address = _cell(row, 12) or None description = _cell(row, 13) or None customer = Customer( first_name=first_name, last_name=last_name, city=city, landline_phone=landline, mobile_phone=mobile, gender=gender, birth_date=birth_date, customer_code=customer_code, referrer_text=referrer_text, how_found=how_found, instagram_id=instagram_id, status=status, address=address, description=description, created_by=user_id, ) db.session.add(customer) created += 1 job.processed_rows = i - 1 job.created_count = created job.skipped_count = sum(1 for n in notes if n["type"] == "skipped") db.session.commit() job.status = "done" job.errors = notes db.session.commit() refresh_statuses() log_action(user, "create", "customer_import", job_id, {"created": created, "skipped": len(notes), "total": len(rows)})