#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ MDP Termination Tool ==================== 自动批量处理用户账号的 CRM termination 使用方法: python mdp_termination.py --excel AdTermination.xlsx python mdp_termination.py --employee-id 162727 python mdp_termination.py --config """ import argparse import asyncio import csv import json import logging import os import sys import subprocess import tempfile from datetime import datetime from pathlib import Path from typing import Dict, List, Optional try: import openpyxl from openpyxl import load_workbook import requests except ImportError as e: print(f"错误: 缺少必需的 Python 库: {e}") print("请运行: pip install -r requirements.txt") sys.exit(1) # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.StreamHandler(sys.stdout)] ) logger = logging.getLogger(__name__) # ============================================================================= # DPAPI 密码解密模块 # ============================================================================= class PasswordManager: """密码管理器 - 处理加密和解密""" @staticmethod def decrypt_password(password_file: str) -> str: """解密 DPAPI 加密的密码文件""" if not os.path.exists(password_file): raise FileNotFoundError(f"密码文件不存在: {password_file}") # 首先尝试使用 dpapi 库 try: import dpapi with open(password_file, 'rb') as f: encrypted_bytes = f.read() decrypted_bytes = dpapi.decrypt(encrypted_bytes) return decrypted_bytes.decode('utf-8') except ImportError: # dpapi 库不可用，使用 ctypes pass except Exception: # dpapi 解密失败，尝试 ctypes pass # 使用 ctypes 备选方案 try: import ctypes from ctypes import wintypes crypt32 = ctypes.windll.crypt32 class DATA_BLOB(ctypes.Structure): _fields_ = [ ("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte)) ] with open(password_file, 'rb') as f: encrypted_bytes = f.read() # 创建字节数组 buffer_type = ctypes.c_byte * len(encrypted_bytes) buffer = buffer_type.from_buffer_copy(encrypted_bytes) input_blob = DATA_BLOB() input_blob.cbData = len(encrypted_bytes) input_blob.pbData = buffer output_blob = DATA_BLOB() result = crypt32.CryptUnprotectData( ctypes.byref(input_blob), None, None, None, None, 0 ) if not result: raise ValueError("DPAPI 解密失败") # 提取解密字节 decrypted_buffer = (ctypes.c_byte * output_blob.cbData)() ctypes.memmove(decrypted_buffer, output_blob.pbData, output_blob.cbData) decrypted_bytes = bytes(decrypted_buffer) ctypes.windll.kernel32.LocalFree(output_blob.pbData) return decrypted_bytes.decode('utf-8') except Exception as e: raise ValueError(f"密码解密失败: {e}") @staticmethod def encrypt_password(password: str, output_file: str): """加密密码并保存到文件""" # 首先尝试使用 dpapi 库 try: import dpapi password_bytes = password.encode('utf-8') encrypted_bytes = dpapi.encrypt(password_bytes) os.makedirs(os.path.dirname(output_file) or '.', exist_ok=True) with open(output_file, 'wb') as f: f.write(encrypted_bytes) logger.info(f"密码已加密保存到: {output_file}") return except ImportError: # dpapi 库不可用，使用 ctypes pass except Exception: # dpapi 加密失败，尝试 ctypes pass # 使用 ctypes 备选方案 try: import ctypes from ctypes import wintypes crypt32 = ctypes.windll.crypt32 class DATA_BLOB(ctypes.Structure): _fields_ = [ ("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte)) ] password_bytes = password.encode('utf-8') # 创建字节数组 buffer_type = ctypes.c_byte * len(password_bytes) buffer = buffer_type.from_buffer_copy(password_bytes) input_blob = DATA_BLOB() input_blob.cbData = len(password_bytes) input_blob.pbData = buffer output_blob = DATA_BLOB() result = crypt32.CryptProtectData( ctypes.byref(input_blob), None, None, None, None, 0, ctypes.byref(output_blob) ) if not result: raise ValueError("DPAPI 加密失败") # 提取加密字节 encrypted_buffer = (ctypes.c_byte * output_blob.cbData)() ctypes.memmove(encrypted_buffer, output_blob.pbData, output_blob.cbData) encrypted_bytes = bytes(encrypted_buffer) os.makedirs(os.path.dirname(output_file) or '.', exist_ok=True) with open(output_file, 'wb') as f: f.write(encrypted_bytes) ctypes.windll.kernel32.LocalFree(output_blob.pbData) logger.info(f"密码已加密保存到: {output_file}") except Exception as e: raise ValueError(f"密码加密失败: {e}") # ============================================================================= # AD 查询模块 # ============================================================================= class ADQueryError(Exception): """AD 查询错误""" pass def query_ad_by_employee_id(employee_id: str, domain: str = "macausjm-glp.com") -> Dict: """通过 EmployeeId 查询 AD 用户""" if not employee_id or not str(employee_id).strip(): raise ADQueryError("EmployeeId 不能为空") employee_id = str(employee_id).strip() ps_command = [ "powershell", "-Command", f"Import-Module ActiveDirectory; " f"Get-ADUser -Filter \"EmployeeId -eq '{employee_id}'\" " f"-Properties DisplayName, SamAccountName, UserPrincipalName, EmployeeId, DistinguishedName " f"| ConvertTo-Json" ] try: logger.info(f"查询 AD: EmployeeId={employee_id}") result = subprocess.run( ps_command, capture_output=True, text=True, timeout=30, encoding='utf-8' ) if result.stderr and "Cannot find" in result.stderr: raise ADQueryError(f"用户不存在: EmployeeId='{employee_id}'") if not result.stdout.strip(): raise ADQueryError(f"用户不存在: EmployeeId='{employee_id}'") try: user_data = json.loads(result.stdout) except json.JSONDecodeError: raise ADQueryError(f"AD 查询失败: EmployeeId='{employee_id}'") if not user_data or not isinstance(user_data, dict): raise ADQueryError(f"用户不存在: EmployeeId='{employee_id}'") display_name = user_data.get('DisplayName') if not display_name: raise ADQueryError(f"DisplayName 为空: EmployeeId='{employee_id}'") user_info = { 'DisplayName': display_name, 'SamAccountName': user_data.get('SamAccountName', ''), 'UserPrincipalName': user_data.get('UserPrincipalName', ''), 'EmployeeId': employee_id, 'DistinguishedName': user_data.get('DistinguishedName', '') } logger.info(f"找到用户: {display_name}") return user_info except subprocess.TimeoutExpired: raise ADQueryError(f"AD 查询超时: EmployeeId='{employee_id}'") except FileNotFoundError: raise ADQueryError("PowerShell 未找到") except ADQueryError: raise except Exception as e: raise ADQueryError(f"AD 查询错误: {e}") # ============================================================================= # Playwright 登录模块 # ============================================================================= class CRMLoginError(Exception): """CRM 登录错误""" pass async def login_to_crm(url: str, username: str, password: str, headless: bool = False, incognito: bool = False) -> Dict: """使用 Playwright 自动登录 CRM""" try: from playwright.async_api import async_playwright except ImportError: raise CRMLoginError("Playwright 未安装。请运行: pip install playwright") async with async_playwright() as p: browser = None context = None try: mode_desc = "headless" if headless else "visible" if incognito: mode_desc += " + incognito" logger.info(f"启动浏览器 ({mode_desc})...") launch_options = {"headless": headless} if incognito: temp_dir = tempfile.mkdtemp(prefix="playwright_incognito_") launch_options["args"] = [ f"--user-data-dir={temp_dir}", "--incognito" ] # 使用系统 Chrome 浏览器（如果可用） # 如果要使用 Playwright 下载的 Chromium，将 channel="chrome" 改为 channel=None browser = await p.chromium.launch(**launch_options, channel="chrome") context_options = { "viewport": {"width": 1280, "height": 720}, "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } context = await browser.new_context(**context_options) page = await context.new_page() login_url = f"{url.rstrip('/')}/MDP/" logger.info(f"访问: {login_url}") await page.goto(login_url, timeout=30000, wait_until='domcontentloaded') await asyncio.sleep(2) current_url = page.url if 'main.aspx' not in current_url: # 填写登录表单 username_selectors = [ 'input[name="username"]', 'input[type="text"]', 'input[id*="user"]' ] username_field = None for selector in username_selectors: try: username_field = await page.wait_for_selector(selector, timeout=2000) if username_field: break except: continue if not username_field: raise CRMLoginError("找不到用户名输入框") await username_field.fill(username) logger.info(f"用户名: {username}") password_selectors = [ 'input[name="password"]', 'input[type="password"]', 'input[id*="pass"]' ] password_field = None for selector in password_selectors: try: password_field = await page.wait_for_selector(selector, timeout=2000) if password_field: break except: continue if not password_field: raise CRMLoginError("找不到密码输入框") await password_field.fill(password) logger.info("密码已输入") submit_selectors = [ 'button[type="submit"]', 'input[type="submit"]' ] submit_button = None for selector in submit_selectors: try: submit_button = await page.wait_for_selector(selector, timeout=2000) if submit_button: break except: continue if submit_button: await submit_button.click() else: await password_field.press('Enter') logger.info("等待登录完成...") try: await page.wait_for_url("**/main.aspx**", timeout=30000) except: pass await asyncio.sleep(2) cookies = await context.cookies() logger.info(f"获取到 {len(cookies)} 个 cookies") req_client_id = None org_id = None for cookie in cookies: if cookie['name'] == 'ReqClientId': req_client_id = cookie['value'] elif cookie['name'] == 'orgId': org_id = cookie['value'] if not req_client_id or not org_id: raise CRMLoginError("无法提取 ReqClientId 或 orgId") logger.info("CRM 登录成功") return { 'ReqClientId': req_client_id, 'orgId': org_id } except CRMLoginError: raise except Exception as e: raise CRMLoginError(f"登录错误: {e}") finally: if context: await context.close() if browser: await browser.close() # ============================================================================= # CRM API 模块 # ============================================================================= class CRMAPIError(Exception): """CRM API 错误""" pass class CRMClient: """CRM API 客户端""" def __init__(self, base_url: str, cookies: Dict, timeout: int = 30): self.base_url = base_url.rstrip('/') self.cookies = cookies self.timeout = timeout self.api_endpoint = f"{self.base_url}/MDP/AppWebServices/InlineEditWebService.asmx" self.headers = { 'Accept': 'application/xml, text/xml, */*; q=0.01', 'Content-Type': 'text/xml; charset=UTF-8', 'SOAPAction': 'http://schemas.microsoft.com/xrm/2011/Contracts/Services/IOrganizationService/Execute', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } def _build_soap_envelope(self, body: str) -> str: return f''' {body} ''' def _execute_request(self, soap_body: str) -> str: envelope = self._build_soap_envelope(soap_body) response = requests.post( self.api_endpoint, data=envelope.encode('utf-8'), headers=self.headers, cookies=self.cookies, timeout=self.timeout ) response.raise_for_status() return response.text def get_display_name_variants(self, display_name: str) -> List[str]: """生成 DisplayName 变体（处理 -d 后缀）""" display_name = display_name.strip() variants = [display_name] if display_name.endswith("-d"): base_name = display_name[:-2].strip() if base_name and base_name != display_name: variants.append(base_name) else: variants.append(display_name + "-d") return variants def find_user_by_display_name(self, display_name: str) -> Optional[Dict]: """在 CRM 中查找用户""" soap_body = f''' Query systemuserid false systemuser 11 fullname Equal {display_name} false systemuserid domainname fullname ito_staffno RetrieveMultiple ''' try: response_xml = self._execute_request(soap_body) if '' in response_xml: return None import re guid_match = re.search(r'([A-F0-9-]{36})', response_xml, re.IGNORECASE) if not guid_match: return None systemuserid = guid_match.group(1) domainname_match = re.search(r'([^<]+)', response_xml) staffno_match = re.search(r'([^<]+)', response_xml) return { 'systemuserid': systemuserid, 'domainname': domainname_match.group(1) if domainname_match else '', 'fullname': display_name, 'ito_staffno': staffno_match.group(1) if staffno_match else '' } except Exception: return None def find_user_with_fallback(self, display_name: str, employee_id: str) -> Dict: """查找用户（支持 DisplayName 变体匹配）""" variants = self.get_display_name_variants(display_name) logger.info(f"在 CRM 搜索: {display_name}") logger.info(f" 变体: {variants}") for idx, variant in enumerate(variants, 1): logger.info(f" 尝试 {idx}/{len(variants)}: '{variant}'") user_info = self.find_user_by_display_name(variant) if user_info: crm_staffno = user_info.get('ito_staffno', '') if not crm_staffno: logger.warning(f" ✗ ito_staffno 为空") continue if crm_staffno != employee_id: logger.warning(f" ✗ EmployeeId 不匹配: 期望 '{employee_id}', CRM 有 '{crm_staffno}'") continue logger.info(f" ✓ 找到: {variant} (CRM ID: {user_info['systemuserid']})") if variant != display_name: logger.info(f" ℹ 使用变体 '{variant}' 而非原始 '{display_name}'") return user_info logger.warning(f" ✗ 未找到: '{variant}'") raise CRMAPIError( f"用户不存在: {display_name}\n" f" 尝试的变体: {variants}\n" f" EmployeeId: {employee_id}" ) def disable_user(self, systemuserid: str) -> bool: """禁用 CRM 用户""" soap_body = f''' EntityMoniker {systemuserid} systemuser State 1 Status -1 SetState ''' try: logger.info(f"禁用 CRM 用户: {systemuserid}") response_xml = self._execute_request(soap_body) if '' in response_xml: raise CRMAPIError("禁用用户失败") logger.info(f"成功禁用用户: {systemuserid}") return True except Exception as e: if isinstance(e, CRMAPIError): raise raise CRMAPIError(f"禁用用户错误: {e}") def disable_user_with_retry(self, systemuserid: str, max_retries: int = 3) -> tuple: """禁用用户（支持重试）""" import time for attempt in range(1, max_retries + 1): try: self.disable_user(systemuserid) return True, f"成功 (尝试 {attempt}/{max_retries})" except CRMAPIError as e: if attempt < max_retries: wait_time = 2 ** attempt logger.info(f"等待 {wait_time}s 后重试...") time.sleep(wait_time) else: return False, f"失败 ({max_retries} 次尝试后): {e}" # ============================================================================= # 主程序逻辑 # ============================================================================= class TerminationResult: """终止结果""" def __init__(self, employee_id: str): self.employee_id = employee_id self.display_name = "" self.status = "pending" self.message = "" self.crm_user_id = "" self.timestamp = datetime.now().isoformat() def load_config(config_file: str = "config.json") -> Dict: """加载配置文件""" if not os.path.exists(config_file): return {} with open(config_file, 'r', encoding='utf-8') as f: return json.load(f) def save_config(config: Dict, config_file: str = "config.json"): """保存配置文件""" with open(config_file, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) def read_excel_employee_ids(file_path: str) -> List[str]: """从 Excel 读取 EmployeeID""" if not os.path.exists(file_path): raise FileNotFoundError(f"文件不存在: {file_path}") workbook = load_workbook(file_path, read_only=True) sheet = workbook.active employee_ids = [] # 查找 EmployeeID 列 headers = [] for cell in sheet[1]: headers.append(cell.value) employee_id_col = None for idx, header in enumerate(headers): if header and str(header).strip().upper() == "EMPLOYEEID": employee_id_col = idx + 1 break if not employee_id_col: employee_id_col = 1 # 读取数据 for row in sheet.iter_rows(min_row=2, min_col=employee_id_col, max_col=employee_id_col): cell_value = row[0].value if cell_value: employee_ids.append(str(cell_value).strip()) workbook.close() logger.info(f"从 Excel 读取了 {len(employee_ids)} 个 EmployeeID") return employee_ids def generate_report(results: List[TerminationResult], report_path: str): """生成报告""" os.makedirs(report_path, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") report_file = os.path.join(report_path, f"Termination_Report_{timestamp}.csv") with open(report_file, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(['EmployeeID', 'DisplayName', 'CRMUserId', 'Status', 'Message', 'Timestamp']) for result in results: writer.writerow([ result.employee_id, result.display_name, result.crm_user_id, result.status, result.message, result.timestamp ]) logger.info(f"报告已生成: {report_file}") total = len(results) success = sum(1 for r in results if r.status == "success") failed = sum(1 for r in results if r.status == "failed") print("\n" + "="*60) print("终止摘要") print("="*60) print(f"总数: {total}") print(f"成功: {success}") print(f"失败: {failed}") print("="*60) print(f"\n报告保存到: {report_file}") async def terminate_user(employee_id: str, ad_domain: str, crm_client: CRMClient, max_retries: int = 3, use_fuzzy_match: bool = True) -> TerminationResult: """终止单个用户""" result = TerminationResult(employee_id) try: print(f"\n{'='*60}") print(f"处理: {employee_id}") print('='*60) # 步骤 1: 查询 AD print(f"[1/3] 查询 AD...") try: user_info = query_ad_by_employee_id(employee_id, ad_domain) display_name = user_info['DisplayName'] result.display_name = display_name print(f" 找到: {display_name}") except ADQueryError as e: result.status = "failed" result.message = f"AD 查询失败: {e}" print(f" 错误: {e}") return result # 步骤 2: 查询 CRM print(f"[2/3] 查询 CRM...") try: if use_fuzzy_match: print(f" 模式: 模糊匹配（尝试变体）") else: print(f" 模式: 精确匹配") user_data = crm_client.find_user_with_fallback(display_name, employee_id) systemuserid = user_data['systemuserid'] result.crm_user_id = systemuserid print(f" 找到: CRM ID = {systemuserid}") except CRMAPIError as e: result.status = "failed" result.message = f"CRM 查询失败: {e}" print(f" 错误: {e}") return result # 步骤 3: 禁用用户 print(f"[3/3] 禁用用户...") success, message = crm_client.disable_user_with_retry(systemuserid, max_retries) if success: result.status = "success" result.message = message print(f" 成功: {message}") else: result.status = "failed" result.message = message print(f" 失败: {message}") return result except Exception as e: result.status = "failed" result.message = f"未预期的错误: {e}" logger.error(f"错误: {e}", exc_info=True) return result async def main(): """主函数""" # 解析命令行参数 parser = argparse.ArgumentParser( description='MDP Termination Tool - 批量用户终止工具', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: python mdp_termination.py --config python mdp_termination.py --excel AdTermination.xlsx python mdp_termination.py --employee-id 162727 python mdp_termination.py --excel AdTermination.xlsx --what-if """ ) parser.add_argument('--config', action='store_true', help='配置向导') parser.add_argument('--employee-id', type=str, help='单个 Employee ID') parser.add_argument('--excel', type=str, help='Excel 文件') parser.add_argument('--config-file', type=str, default='config.json', help='配置文件路径') parser.add_argument('--report-path', type=str, default='./reports', help='报告输出目录') parser.add_argument('--headless', action='store_true', help='无头模式（不显示浏览器）') parser.add_argument('--incognito', action='store_true', help='隐身模式') parser.add_argument('--what-if', action='store_true', help='模拟模式（不实际禁用）') parser.add_argument('--domain', type=str, help='AD 域名') parser.add_argument('--max-retries', type=int, help='最大重试次数') parser.add_argument('--no-fuzzy-match', action='store_true', help='禁用模糊匹配') args = parser.parse_args() # ==================== 配置向导 ==================== if args.config: print("\n" + "="*60) print(" MDP Termination Tool - 配置向导") print("="*60 + "\n") config = load_config(args.config_file) print("请输入以下配置信息:\n") # CRM URL crm_url = config.get('crm_url', 'https://internalcrm.macausjm-glp.com') crm_url_input = input(f"CRM URL [默认: {crm_url}]: ").strip() if crm_url_input: crm_url = crm_url_input # AD 域名 ad_domain = config.get('ad_domain', 'macausjm-glp.com') ad_domain_input = input(f"AD 域名 [默认: {ad_domain}]: ").strip() if ad_domain_input: ad_domain = ad_domain_input # CRM 用户名 crm_username = config.get('crm_username', '') if not crm_username: crm_username = "DOMAIN\\username" print(f"\nCRM 用户名格式: DOMAIN\\username") crm_username_input = input(f"CRM 用户名 [默认: {crm_username}]: ").strip() if crm_username_input: crm_username = crm_username_input # 密码 import getpass print("\nCRM 密码:") password = getpass.getpass("密码: ") password_confirm = getpass.getpass("确认密码: ") if password != password_confirm: print("\n错误: 两次密码输入不一致") sys.exit(1) # 加密并保存密码 password_file = "password.enc" try: print(f"\n加密密码中...") PasswordManager.encrypt_password(password, password_file) print(f"✓ 密码已加密保存到: {password_file}") except Exception as e: print(f"\n错误: {e}") sys.exit(1) # 保存配置 config['crm_url'] = crm_url config['ad_domain'] = ad_domain config['crm_username'] = crm_username config['password_file'] = password_file save_config(config, args.config_file) print(f"✓ 配置已保存到: {args.config_file}") print("\n" + "="*60) print("配置完成!") print("="*60) print("\n下一步:") print(" 1. 运行测试: python mdp_termination.py --employee-id 000 --what-if") print(" 2. 处理用户: python mdp_termination.py --excel AdTermination.xlsx") print() return # ==================== 验证参数 ==================== if not args.employee_id and not args.excel: parser.error("需要指定 --employee-id 或 --excel") if args.employee_id and args.excel: parser.error("不能同时指定 --employee-id 和 --excel") # ==================== 加载配置 ==================== config = load_config(args.config_file) if not config: print(f"\n错误: 配置文件不存在或为空: {args.config_file}") print("请先运行: python mdp_termination.py --config") sys.exit(1) crm_url = config.get('crm_url', 'https://internalcrm.macausjm-glp.com') crm_username = config.get('crm_username', '') password_file = config.get('password_file', 'password.enc') ad_domain = args.domain or config.get('ad_domain', 'macausjm-glp.com') max_retries = args.max_retries or config.get('max_retries', 3) if not crm_username: print("\n错误: 配置中缺少 CRM 用户名") print("请运行: python mdp_termination.py --config") sys.exit(1) # ==================== What-If 模式 ==================== if args.what_if: print("\n" + "!"*60) print("WHAT-IF 模式: 不会实际禁用用户") print("!"*60 + "\n") if args.excel: employee_ids = read_excel_employee_ids(args.excel) print(f"将处理 {len(employee_ids)} 个用户:") for eid in employee_ids[:10]: # 只显示前10个 print(f" - {eid}") if len(employee_ids) > 10: print(f" ... 还有 {len(employee_ids) - 10} 个") else: print(f"将处理用户: {args.employee_id}") print("\n模拟模式完成，未做任何修改") return # ==================== 解密密码 ==================== try: password = PasswordManager.decrypt_password(password_file) logger.info(f"密码已加载: {password_file}") except Exception as e: print(f"\n错误: 无法解密密码文件 - {e}") print("请确保密码文件在当前计算机上生成") print("或运行: python mdp_termination.py --config") sys.exit(1) # ==================== 登录 CRM ==================== print("\n" + "="*60) print("CRM 登录") print("="*60) print(f"\nURL: {crm_url}") print(f"用户名: {crm_username}") print(f"无头模式: {args.headless}") print(f"隐身模式: {args.incognito}\n") try: cookies = await login_to_crm( crm_url, crm_username, password, headless=args.headless, incognito=args.incognito ) print("\n✓ 登录成功\n") except CRMLoginError as e: print(f"\n错误: CRM 登录失败 - {e}") sys.exit(1) # ==================== 创建 CRM 客户端 ==================== crm_client = CRMClient(crm_url, cookies) # ==================== 获取 EmployeeID 列表 ==================== employee_ids = [] if args.employee_id: employee_ids = [args.employee_id] else: print("\n" + "="*60) print("读取 Excel 文件") print("="*60) employee_ids = read_excel_employee_ids(args.excel) if not employee_ids: print("\n错误: 没有 EmployeeID 可处理") sys.exit(1) print(f"\n总共: {len(employee_ids)} 个用户\n") # ==================== 处理用户 ==================== print("="*60) print("处理用户") print("="*60) results = [] for idx, employee_id in enumerate(employee_ids, 1): print(f"\n[{idx}/{len(employee_ids)}]") result = await terminate_user( employee_id, ad_domain, crm_client, max_retries, use_fuzzy_match=not args.no_fuzzy_match ) results.append(result) # 失败时停止 if result.status == "failed": logger.error(f"处理失败，停止执行") print("\n" + "!"*60) print("因错误停止") print("!"*60) break # ==================== 生成报告 ==================== print("\n" + "="*60) print("生成报告") print("="*60) generate_report(results, args.report_path) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n\n用户中断") sys.exit(1) except Exception as e: logger.error(f"未预期的错误: {e}", exc_info=True) print(f"\n错误: {e}") sys.exit(1)