#!/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 sql_quote(value: Optional[str]) -> str: if value is None: return "NULL" v = str(value) v = v.replace("\\", "\\\\").replace("'", "''") return f"'{v}'" @dataclass class UserRow: didar_user_id: str display_name: str first_name: str last_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 find_user_id(users: List[UserRow], target_display: str) -> str: t = norm_name(target_display) for u in users: if norm_name(u.display_name) == t: return u.didar_user_id if norm_name((u.first_name + " " + u.last_name).strip()) == t: return u.didar_user_id # fallback: match last name for u in users: if norm_name(u.last_name) == t: return u.didar_user_id return "" def main() -> int: root = Path(__file__).resolve().parent.parent sql_dump_path = root / "hanafivi_didash (1).sql" xlsx_path = root / "معاملات بلا تکلیف 23 آذر 1404.xlsx" out_dir = root / "output" out_dir.mkdir(parents=True, exist_ok=True) out_sql_path = out_dir / "patch_transfer_and_blataklif_deals_23azar1404.sql" sql_dump = sql_dump_path.read_text(encoding="utf-8", errors="ignore") users = parse_users_from_sql_dump(sql_dump) janan_id = find_user_id(users, "جانان ظهوری") sanaz_id = find_user_id(users, "ساناز پارسا") soroush_id = find_user_id(users, "سروش حیدریان") if not janan_id: raise SystemExit("Could not find Janan Zohouri didar_user_id in SQL dump users.") if not sanaz_id: raise SystemExit("Could not find Sanaz Parsa didar_user_id in SQL dump users.") if not soroush_id: raise SystemExit("Could not find Soroush Heydarian didar_user_id in SQL dump users.") 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) if h} if "کد دیدار مشتری" not in idx: raise SystemExit("Missing header: کد دیدار مشتری") # NOTE: This Excel column contains the Didar "code" (numeric), not didar_contact_id (GUID). didar_codes: List[str] = [] for r in range(2, ws.max_row + 1): v = ws.cell(r, idx["کد دیدار مشتری"]).value if v in (None, ""): continue s = str(v).strip().replace(".0", "") if s: didar_codes.append(s) # de-dupe preserve order seen = set() didar_codes = [x for x in didar_codes if not (x in seen or seen.add(x))] lines: List[str] = [] lines.append("-- Generated by scripts/generate_patch_transfer_and_blataklif_deals_23azar1404.py") lines.append(f"-- After import_excel_23azar1404.sql") 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("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("-- Ensure lead pool display name") lines.append(f"UPDATE `users` SET `display_name`='لید استخر', `username`='لید استخر' WHERE `didar_user_id`={sql_quote(janan_id)};") lines.append("") lines.append("-- Transfer leads owned by Sanaz Parsa and Soroush Heydarian to lead pool (Janan), and store previous owner name/id") lines.append( "UPDATE `persons` p " "JOIN `users` u ON u.`didar_user_id` = p.`owner_didar_id` " "SET " "p.`previous_owner_id` = p.`owner_didar_id`, " "p.`previous_owner_name` = u.`display_name`, " f"p.`owner_didar_id` = {sql_quote(janan_id)}, " "p.`last_sync` = NOW() " f"WHERE p.`owner_didar_id` IN ({sql_quote(sanaz_id)}, {sql_quote(soroush_id)});" ) lines.append("") lines.append("-- Create temporary list for Blatakalif deals (from Excel)") lines.append("-- Source column is Didar contact CODE (persons.code), not persons.didar_contact_id") lines.append("CREATE TEMPORARY TABLE IF NOT EXISTS `tmp_blataklif_codes` (`code` varchar(255) PRIMARY KEY) ENGINE=Memory;") lines.append("TRUNCATE TABLE `tmp_blataklif_codes`;") batch_size = 800 for i in range(0, len(didar_codes), batch_size): chunk = didar_codes[i : i + batch_size] values = ", ".join(f"({sql_quote(code)})" for code in chunk) lines.append(f"INSERT IGNORE INTO `tmp_blataklif_codes` (`code`) VALUES {values};") lines.append("") lines.append("-- Insert ONE pending 'بلاتکلیف' deal per contact if no other pending 'بلاتکلیف' deal exists for that contact") lines.append( "INSERT INTO `deals` " "(`didar_deal_id`,`owner_didar_id`,`contact_didar_id`,`title`,`status`,`pipeline_stage`,`register_time`,`last_sync`,`requested_services`,`city`,`job_title`) " "SELECT " "CONCAT('excel_blataklif_23azar1404_', p.`code`) AS didar_deal_id, " "p.`owner_didar_id`, " "p.`didar_contact_id`, " "CONCAT('معامله بلاتکلیف | ', COALESCE(NULLIF(p.`mobile_phone`,''), p.`didar_contact_id`)) AS title, " "'Pending' AS status, " "'بلاتکلیف' AS pipeline_stage, " "COALESCE(p.`register_time`, NOW()) AS register_time, " "NOW() AS last_sync, " "p.`requested_services`, " "p.`city`, " "p.`job_title` " "FROM `persons` p " "JOIN `tmp_blataklif_codes` t ON t.`code` = p.`code` " "WHERE NOT EXISTS (" " SELECT 1 FROM `deals` d " " WHERE d.`contact_didar_id` = p.`didar_contact_id` AND d.`status`='Pending' AND d.`pipeline_stage`='بلاتکلیف'" ");" ) lines.append("") lines.append("-- Diagnostics") lines.append( "SELECT COUNT(*) AS transferred_to_lead_pool " f"FROM `persons` WHERE `owner_didar_id`={sql_quote(janan_id)} AND `previous_owner_id` IN ({sql_quote(sanaz_id)}, {sql_quote(soroush_id)});" ) lines.append( "SELECT COUNT(*) AS created_blataklif_deals " "FROM `deals` WHERE `didar_deal_id` LIKE 'excel_blataklif_23azar1404_%';" ) lines.append("") lines.append("COMMIT;") out_sql_path.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Wrote: {out_sql_path} (codes: {len(didar_codes)})") return 0 if __name__ == "__main__": raise SystemExit(main())