#!/usr/bin/env python3 # -*- coding: utf-8 -*- import re from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import openpyxl def norm_name(value: Any) -> str: if value is None: return "" s = str(value).strip() s = s.replace("ي", "ی").replace("ك", "ک") s = re.sub(r"\s+", " ", s) s = s.replace("خانم", "").replace("آقای", "") return s.strip() def phone_to_str(value: Any) -> str: if value is None: return "" s = str(value).strip() if not s: return "" # Excel numbers can be float-like s = s.replace(".0", "") digits = re.sub(r"\D+", "", s) return digits def sql_quote(value: Optional[str]) -> str: if value is None: return "NULL" v = str(value) v = v.replace("\\", "\\\\").replace("'", "''") return f"'{v}'" # Jalali -> Gregorian conversion (no external deps) # Source: adaptation of well-known jalaali conversion algorithm. def jalali_to_gregorian(jy: int, jm: int, jd: int) -> Tuple[int, int, int]: jy += 1595 days = -355668 + (365 * jy) + ((jy // 33) * 8) + (((jy % 33) + 3) // 4) + jd if jm < 7: days += (jm - 1) * 31 else: days += ((jm - 7) * 30) + 186 gy = 400 * (days // 146097) days %= 146097 if days > 36524: gy += 100 * ((days - 1) // 36524) days = (days - 1) % 36524 if days >= 365: days += 1 gy += 4 * (days // 1461) days %= 1461 if days > 365: gy += (days - 1) // 365 days = (days - 1) % 365 gd = days + 1 leap = (gy % 4 == 0 and gy % 100 != 0) or (gy % 400 == 0) g_month_days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31] gm = 1 for md in g_month_days: if gd <= md: break gd -= md gm += 1 return gy, gm, gd def parse_jalali_date(date_value: Any) -> Optional[Tuple[int, int, int, str]]: if date_value is None: return None s = str(date_value).strip() if not s: return None m = re.match(r"^\s*(\d{4})/(\d{1,2})/(\d{1,2})\s*$", s) if not m: return None jy, jm, jd = int(m.group(1)), int(m.group(2)), int(m.group(3)) gy, gm, gd = jalali_to_gregorian(jy, jm, jd) return gy, gm, gd, s @dataclass class UserRow: didar_user_id: str first_name: str last_name: str display_name: str def parse_users_from_sql_dump(sql_text: str) -> List[UserRow]: m = re.search(r"INSERT INTO `users` .*? VALUES\s*(.*?);", sql_text, re.S) if not m: return [] values = m.group(1) rows: List[str] = [] buf = "" depth = 0 for ch in values: if ch == "(": depth += 1 if depth > 0: buf += ch if ch == ")" and depth > 0: depth -= 1 if depth == 0: rows.append(buf) buf = "" def unq(s: str) -> str: s = s.strip() if s.upper() == "NULL": return "" if s.startswith("'") and s.endswith("'"): return s[1:-1] return s users: List[UserRow] = [] for row in rows: inner = row[1:-1] fields: List[str] = [] cur = "" inq = False i = 0 while i < len(inner): ch = inner[i] if ch == "'" and (i == 0 or inner[i - 1] != "\\"): inq = not inq cur += ch elif ch == "," and not inq: fields.append(cur.strip()) cur = "" else: cur += ch i += 1 fields.append(cur.strip()) if len(fields) < 6: continue users.append( UserRow( didar_user_id=unq(fields[1]), first_name=unq(fields[3]), last_name=unq(fields[4]), display_name=unq(fields[5]), ) ) return users def build_owner_map(users: List[UserRow]) -> Dict[str, str]: mapping: Dict[str, str] = {} for u in users: keys = [ u.display_name, (u.first_name + " " + u.last_name).strip(), u.last_name, u.first_name, ] for k in keys: nk = norm_name(k) if nk and nk not in mapping: mapping[nk] = u.didar_user_id return mapping def build_lead_source_map() -> Dict[str, str]: # Persian label -> internal code used by UI return { "سایت": "website", "تلگرام": "telegram", "اینستاگرام": "instagram", "واتساپ": "whatsapp", "کتاب": "book", "کمپین": "campaign", "ایونت": "event", "پشتیبانی": "support", "تماس ورودی": "incoming_call", "مراجعه حضوری": "visit", "معرف": "referral", "سایر": "other", } def build_content_topic_map() -> Dict[str, str]: # Based on assets/js/app.js contentTopics labels return { "هوش خودرو": "car_intelligence", "کارشناسی حضوری": "in_person_inspection", "کسب و کار": "business", "مینی دوره هوش خودرو": "mini_course", "کانال VIP": "vip_channel", "بودجه بندی": "budgeting", "نمایشگاه آنلاین": "online_exhibition", "مشاوره با استاد": "consulting", "دستگاه کارشناسی": "inspection_device", "تست هوش خودرو": "car_intelligence_test", "تست تیپ شخصیت مالی": "financial_personality", "معامله گر هوشمند": "smart_trader", "آموزش دیتیلینگ": "detailing_training", "دیوار خودرو": "divar_car", "تست کارشناسی": "inspection_test", "کتاب هوش خودرو": "book", # Excel sometimes uses shorter variants "دیوارخودرو": "divar_car", "مشاوره": "consulting", } def main() -> int: root = Path(__file__).resolve().parent.parent xlsx_path = root / "1 کد لید موفق و ناموفق 23 آذر 1404.xlsx" sql_dump_path = root / "hanafivi_didash (1).sql" out_dir = root / "output" out_dir.mkdir(parents=True, exist_ok=True) out_sql_path = out_dir / "import_excel_23azar1404.sql" wb = openpyxl.load_workbook(xlsx_path, data_only=True) ws = wb.active header = [str((ws.cell(1, c).value or "")).strip() for c in range(1, ws.max_column + 1)] idx = {h: i + 1 for i, h in enumerate(header)} fixed_cols = { "ردیف", "کد دیدار مشتری", "نام مشتری", "نام خانوادگی مشتری", "تلفن همراه اصلی", "تلفن همراه سوم", "تلفن همراه چهارم", "تلفن همراه دوم", "مسئول قبلی", "مسئول فعلی", "تاریخ ثبت لید", "شهر", "شغل", "شیوه آشنایی", "جزئیات شیوه آشنایی", "موضوع محتوا", "موضعیت فروش", } product_cols = [(i + 1, h) for i, h in enumerate(header) if h and h not in fixed_cols] products = [h.strip() for _, h in product_cols if h.strip()] sql_dump = sql_dump_path.read_text(encoding="utf-8", errors="ignore") users = parse_users_from_sql_dump(sql_dump) owner_map = build_owner_map(users) # Identify Janan didar_user_id janan_id = "" for u in users: if norm_name(u.display_name) == norm_name("جانان ظهوری") or norm_name(u.last_name) == norm_name("ظهوری"): if u.didar_user_id: janan_id = u.didar_user_id break if not janan_id: raise SystemExit("Could not find Janan Zohouri in users from SQL dump.") lead_source_map = build_lead_source_map() content_topic_map = build_content_topic_map() required_headers = [ "کد دیدار مشتری", "نام مشتری", "نام خانوادگی مشتری", "تلفن همراه اصلی", "تلفن همراه دوم", "تلفن همراه سوم", "تلفن همراه چهارم", "مسئول فعلی", "مسئول قبلی", "تاریخ ثبت لید", "موضعیت فروش", ] missing = [h for h in required_headers if h not in idx] if missing: raise SystemExit(f"Missing required headers: {missing}") # Generate SQL lines: List[str] = [] lines.append("-- Generated by scripts/generate_import_from_excel_23azar1404.py") lines.append(f"-- Source Excel: {xlsx_path.name}") lines.append(f"-- Generated at: {datetime.now().isoformat(sep=' ', timespec='seconds')}") lines.append("SET NAMES utf8mb4;") lines.append("SET sql_mode = 'NO_AUTO_VALUE_ON_ZERO';") lines.append("START TRANSACTION;") lines.append("") lines.append("-- Ensure column exists for unknown previous owner names") lines.append("ALTER TABLE `persons` ADD COLUMN IF NOT EXISTS `previous_owner_name` varchar(255) DEFAULT NULL;") lines.append("") lines.append("-- Rename Janan user display name to 'لید استخر' for UI") lines.append(f"UPDATE `users` SET `display_name`='لید استخر', `username`='لید استخر' WHERE `didar_user_id`={sql_quote(janan_id)};") lines.append("") lines.append("-- Upsert product catalog (from Excel columns)") for p in products: lines.append(f"INSERT IGNORE INTO `products` (`name`) VALUES ({sql_quote(p)});") lines.append("") # Upsert persons cols = [ "didar_contact_id", "owner_didar_id", "previous_owner_id", "previous_owner_name", "code", "first_name", "last_name", "mobile_phone", "secondary_mobile_phone", "mobile_phone_3", "mobile_phone_4", "contact_type", "city", "job_title", "acquaintance_source", "acquaintance_detail", "content_topic", "sale_status", "register_time_jalali", "register_time", "customer_products", "requested_services", "last_sync", ] lines.append("-- Upsert persons (persons)") batch_size = 500 values_batch: List[str] = [] def flush_batch() -> None: nonlocal values_batch if not values_batch: return insert = "INSERT INTO `persons` (" + ", ".join(f"`{c}`" for c in cols) + ") VALUES\n" insert += ",\n".join(values_batch) insert += "\nON DUPLICATE KEY UPDATE\n" insert += ",\n".join( [ "`owner_didar_id`=VALUES(`owner_didar_id`)", "`previous_owner_id`=VALUES(`previous_owner_id`)", "`previous_owner_name`=VALUES(`previous_owner_name`)", "`code`=VALUES(`code`)", "`first_name`=VALUES(`first_name`)", "`last_name`=VALUES(`last_name`)", "`mobile_phone`=VALUES(`mobile_phone`)", "`secondary_mobile_phone`=VALUES(`secondary_mobile_phone`)", "`mobile_phone_3`=VALUES(`mobile_phone_3`)", "`mobile_phone_4`=VALUES(`mobile_phone_4`)", "`contact_type`=VALUES(`contact_type`)", "`city`=VALUES(`city`)", "`job_title`=VALUES(`job_title`)", "`acquaintance_source`=VALUES(`acquaintance_source`)", "`acquaintance_detail`=VALUES(`acquaintance_detail`)", "`content_topic`=VALUES(`content_topic`)", "`sale_status`=VALUES(`sale_status`)", "`register_time_jalali`=VALUES(`register_time_jalali`)", "`register_time`=VALUES(`register_time`)", "`customer_products`=VALUES(`customer_products`)", "`requested_services`=VALUES(`requested_services`)", "`last_sync`=VALUES(`last_sync`)", ] ) insert += ";\n" lines.append(insert) values_batch = [] # iterate rows for r in range(2, ws.max_row + 1): didar_contact_id = ws.cell(r, idx["کد دیدار مشتری"]).value if didar_contact_id in (None, ""): continue didar_contact_id_s = str(didar_contact_id).strip().replace(".0", "") first_name = (ws.cell(r, idx["نام مشتری"]).value or "") last_name = (ws.cell(r, idx["نام خانوادگی مشتری"]).value or "") first_name_s = str(first_name).strip() if first_name is not None else "" last_name_s = str(last_name).strip() if last_name is not None else "" mobile1 = phone_to_str(ws.cell(r, idx["تلفن همراه اصلی"]).value) mobile2 = phone_to_str(ws.cell(r, idx["تلفن همراه دوم"]).value) mobile3 = phone_to_str(ws.cell(r, idx["تلفن همراه سوم"]).value) mobile4 = phone_to_str(ws.cell(r, idx["تلفن همراه چهارم"]).value) cur_owner_name = norm_name(ws.cell(r, idx["مسئول فعلی"]).value) prev_owner_name = norm_name(ws.cell(r, idx["مسئول قبلی"]).value) # Apply special mapping: "لید استخر" => Janan id if norm_name(cur_owner_name) == norm_name("لید استخر"): owner_id = janan_id else: owner_id = owner_map.get(norm_name(cur_owner_name), "") prev_owner_id = owner_map.get(norm_name(prev_owner_name), "") prev_owner_name_store = "" if prev_owner_name and not prev_owner_id: prev_owner_name_store = prev_owner_name city = (ws.cell(r, idx.get("شهر", 0)).value or "") if idx.get("شهر") else "" job = (ws.cell(r, idx.get("شغل", 0)).value or "") if idx.get("شغل") else "" city_s = str(city).strip() if city is not None else "" job_s = str(job).strip() if job is not None else "" source_raw = norm_name(ws.cell(r, idx.get("شیوه آشنایی", 0)).value) if idx.get("شیوه آشنایی") else "" detail_raw = norm_name(ws.cell(r, idx.get("جزئیات شیوه آشنایی", 0)).value) if idx.get("جزئیات شیوه آشنایی") else "" topic_raw = norm_name(ws.cell(r, idx.get("موضوع محتوا", 0)).value) if idx.get("موضوع محتوا") else "" source = lead_source_map.get(source_raw, source_raw) detail = detail_raw topic = content_topic_map.get(topic_raw, topic_raw) sale_raw = norm_name(ws.cell(r, idx["موضعیت فروش"]).value) sale_status = 1 if sale_raw == "موفق" else 0 date_raw = ws.cell(r, idx["تاریخ ثبت لید"]).value parsed = parse_jalali_date(date_raw) reg_jalali = parsed[3] if parsed else (str(date_raw).strip() if date_raw else "") reg_time = None if parsed: gy, gm, gd, _ = parsed reg_time = f"{gy:04d}-{gm:02d}-{gd:02d} 00:00:00" selected_products: List[str] = [] for cidx, pname in product_cols: v = ws.cell(r, cidx).value if v in (None, "", 0): continue selected_products.append(pname.strip()) products_csv = ",".join(dict.fromkeys([p for p in selected_products if p])) # preserve order + dedupe now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") row_values = [ sql_quote(didar_contact_id_s), sql_quote(owner_id) if owner_id else "NULL", sql_quote(prev_owner_id) if prev_owner_id else "NULL", sql_quote(prev_owner_name_store) if prev_owner_name_store else "NULL", sql_quote(didar_contact_id_s), # code sql_quote(first_name_s) if first_name_s else "NULL", sql_quote(last_name_s) if last_name_s else "NULL", sql_quote(mobile1) if mobile1 else "NULL", sql_quote(mobile2) if mobile2 else "NULL", sql_quote(mobile3) if mobile3 else "NULL", sql_quote(mobile4) if mobile4 else "NULL", sql_quote("Lead"), sql_quote(city_s) if city_s else "NULL", sql_quote(job_s) if job_s else "NULL", sql_quote(source) if source else "NULL", sql_quote(detail) if detail else "NULL", sql_quote(topic) if topic else "NULL", str(int(sale_status)), sql_quote(reg_jalali) if reg_jalali else "NULL", sql_quote(reg_time) if reg_time else "NULL", sql_quote(products_csv) if products_csv else "NULL", sql_quote(products_csv) if products_csv else "NULL", sql_quote(now), ] values_batch.append("(" + ", ".join(row_values) + ")") if len(values_batch) >= batch_size: flush_batch() flush_batch() lines.append("COMMIT;") out_sql_path.write_text("\n".join(lines), encoding="utf-8") print(f"Wrote: {out_sql_path}") return 0 if __name__ == "__main__": raise SystemExit(main())