#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 简化版本 - 专注于正确提取 systemuserid 和调用 API """ import os import sys # 强制立即输出 sys.stdout.reconfigure(line_buffering=True) sys.stderr.reconfigure(line_buffering=True) print("="*60) print(" MDP Termination - 简化 API 版本") print("="*60) print() # 导入 print(">>> 导入模块...") from playwright.sync_api import sync_playwright import json import subprocess import time from openpyxl import load_workbook from datetime import datetime import csv print("[OK] 所有模块已导入") print() # 加载配置 print(">>> 加载配置...") config_file = "config.json" with open(config_file, "r", encoding="utf-8") as f: config = json.load(f) crm_url = config.get('crm', {}).get('url') or config.get('crm_url') username = sys.argv[2] if len(sys.argv) > 2 else None if not username: username = config.get('crm', {}).get('username') or config.get('crm_username') excel_file = sys.argv[1] if len(sys.argv) > 1 else None if not os.path.exists(excel_file): print(f"[ERROR] Excel 文件不存在: {excel_file}") sys.exit(1) print(f"[OK] 配置已加载") print(f" CRM URL: {crm_url}") print(f" 用户名: {username}") print(f" Excel 文件: {excel_file}") print() # 读取 Excel print(f">>> 读取 Excel 文件...") try: wb = load_workbook(excel_file, read_only=True) ws = wb.active employee_data = [] for row in ws.iter_rows(min_row=2, values_only=True): if len(row) >= 2 and row[1]: emp_id = str(row[1]).strip() if emp_id: employee_data.append({'employee_id': emp_id}) wb.close() employee_ids = [d['employee_id'] for d in employee_data] print(f"[OK] 找到 {len(employee_ids)} 个 EmployeeId") if len(employee_ids) <= 10: print(f" 列表: {', '.join(employee_ids)}") except Exception as e: print(f"[ERROR] 读取 Excel 失败: {e}") sys.exit(1) print() def query_ad(employee_id): """查询 AD""" try: ps_command = [ "powershell", "-Command", f"Import-Module ActiveDirectory; Get-ADUser -Filter \"EmployeeId -eq '{employee_id}'\" -Properties DisplayName, SamAccountName | ConvertTo-Json" ] result = subprocess.run(ps_command, capture_output=True, text=True, timeout=30, encoding='utf-8') if result.stderr and "Cannot find" in result.stderr: return None, "用户不存在" if not result.stdout.strip(): return None, "用户不存在" user_data = json.loads(result.stdout) if not user_data or not isinstance(user_data, dict): return None, "用户不存在" display_name = user_data.get('DisplayName') # 移除 -d 后缀 if display_name and display_name.endswith('-d'): display_name = display_name[:-2].strip() print(f" [INFO] 使用不含 -d 的名称") return display_name, None except Exception as e: return None, f"AD 查询失败: {str(e)}" def extract_systemuserid_from_url(url): """从 URL 提取 systemuserid""" try: print(f" [INFO] 从 URL 提取 systemuserid") print(f" URL: {url}") # 解码 URL from urllib.parse import unquote decoded_url = unquote(url) print(f" 解码后 URL: {decoded_url[:150]}...") import re # 方法 1: 查找带连字符的 GUID (标准格式: 8-4-4-4-12) guid_pattern_with_hyphens = r'id=([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})' match = re.search(guid_pattern_with_hyphens, decoded_url, re.IGNORECASE) if match: systemuserid = match.group(1) print(f" [OK] 提取成功（带连字符）: {systemuserid}") return systemuserid # 方法 2: 查找不带连字符的 GUID (32 位十六进制) guid_pattern_no_hyphens = r'id=([0-9a-f]{32})' match = re.search(guid_pattern_no_hyphens, decoded_url, re.IGNORECASE) if match: guid_no_hyphens = match.group(1) # 转换为带连字符的格式: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx systemuserid = f"{guid_no_hyphens[0:8]}-{guid_no_hyphens[8:12]}-{guid_no_hyphens[12:16]}-{guid_no_hyphens[16:20]}-{guid_no_hyphens[20:32]}" print(f" [OK] 提取成功（转换后）: {systemuserid}") return systemuserid # 方法 3: 在整个 URL 中查找任何 32 位十六进制 GUID any_guid_pattern = r'([0-9a-f]{32})' guids = re.findall(any_guid_pattern, decoded_url, re.IGNORECASE) if guids: print(f" [INFO] 找到 {len(guids)} 个 GUID:") for i, guid in enumerate(guids[:5]): # 只显示前5个 print(f" {i+1}. {guid}") # 转换第一个并使用 guid_no_hyphens = guids[0] systemuserid = f"{guid_no_hyphens[0:8]}-{guid_no_hyphens[8:12]}-{guid_no_hyphens[12:16]}-{guid_no_hyphens[16:20]}-{guid_no_hyphens[20:32]}" print(f" [OK] 使用第一个 GUID（转换后）: {systemuserid}") return systemuserid return None except Exception as e: import traceback print(f" [ERROR] 提取失败: {e}") traceback.print_exc() return None def disable_user_via_api(systemuserid, page): """通过页面中的 JavaScript 调用 API 禁用用户""" try: print(f" 调用 CRM API 禁用用户...") print(f" SystemUserId: {systemuserid}") soap_url = f"{crm_url.rstrip('/')}/MDP/XRMServices/2011/Organization.svc/web" soap_body = f''' 9.0 EntityMoniker {systemuserid} systemuser State 1 Status -1 MaintainLegacyAppServerBehavior true SetState ''' print(f" 发送 SOAP 请求（使用浏览器 fetch）...") # 在浏览器中执行 fetch，这样会自动包含所有 cookies js_code = f''' async () => {{ try {{ const response = await fetch("{soap_url}", {{ method: "POST", headers: {{ "Accept": "application/xml, text/xml, */*; q=0.01", "ClientAppName": "WebClient", "ClientAppVersion": "9.0", "Content-Type": "text/xml; charset=UTF-8", "SOAPAction": "http://schemas.microsoft.com/xrm/2011/Contracts/Services/IOrganizationService/Execute", "X-Requested-With": "XMLHttpRequest" }}, body: {repr(soap_body)} }}); const status = response.status; const text = await response.text(); return {{ success: status === 200, status, responseText: text }}; }} catch (error) {{ return {{ success: false, error: error.toString() }}; }} }} ''' result = page.evaluate(js_code) print(f" 响应状态码: {result.get('status', 'N/A')}") if result.get('success') and result.get('status') == 200: response_text = result.get('responseText', '') if 'ErrorResponse' in response_text: print(f" [ERROR] API 返回错误") print(f" 响应: {response_text[:500]}") return False, "API 返回错误" # 检查是否成功 if 'Success' in response_text or 'ExecuteResult' in response_text: print(f" [OK] API 调用成功") return True, "用户已禁用" else: print(f" [WARNING] 未知响应") print(f" 响应: {response_text[:500]}") return False, "未知响应" else: error_msg = result.get('error', f"HTTP 错误: {result.get('status', 'unknown')}") print(f" [ERROR] {error_msg}") return False, error_msg except Exception as e: import traceback print(f" [ERROR] API 调用失败: {e}") traceback.print_exc() return False, f"API 调用失败: {str(e)}" # 主程序 try: print(">>> 初始化浏览器...") with sync_playwright() as p: print(" [1/2] 启动浏览器...") browser = p.chromium.launch( headless=False, channel="msedge", slow_mo=100, args=['--incognito', '--disable-blink-features=AutomationControlled'] ) context = browser.new_context( ignore_https_errors=True, accept_downloads=False, storage_state=None, ) page = context.new_page() print(" [2/2] 访问 CRM...") login_url = f"{crm_url.rstrip('/')}/MDP/" page.goto(login_url, wait_until="domcontentloaded", timeout=30000) time.sleep(3) current_url = page.url print(f" 当前 URL: {current_url}") if 'errorhandler.aspx' in current_url: print() print(" [WARNING] 检测到错误页面") print(" >> 请在浏览器中手动登录或刷新") print(" >> 按 Ctrl+C 继续执行...") try: time.sleep(30) except KeyboardInterrupt: print() print(" [INFO] 继续执行...") print() print("="*60) print(" 开始处理用户...") print("="*60) print() # 处理所有用户 results = [] for idx, employee_id in enumerate(employee_ids, 1): print(f"\n>>> [{idx}/{len(employee_ids)}] 处理 EmployeeId: {employee_id}") print("-"*60) result = { 'employee_id': employee_id, 'display_name': '', 'systemuserid': '', 'status': 'pending', 'message': '', 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } try: # 1. 查询 AD print(f" 步骤 1/3: 查询 AD") display_name, error = query_ad(employee_id) if error: print(f" [ERROR] {error}") result['status'] = 'failed' result['message'] = error results.append(result) continue print(f" [OK] DisplayName: {display_name}") result['display_name'] = display_name # 2. 获取 systemuserid print(f" 步骤 2/3: 获取 systemuserid") # 搜索用户 search_url = f"{crm_url.rstrip('/')}/MDP/multientityquickfind/multientityquickfind.aspx?text={display_name}&option=0&pagemode=iframe" print(f" 搜索 URL: {search_url}") page.goto(search_url, wait_until="domcontentloaded", timeout=30000) time.sleep(5) # 监听新窗口 page_state = {'new_page': None, 'created': False} def handle_new_page(new_p): page_state['new_page'] = new_p page_state['created'] = True context.on('page', handle_new_page) # 点击用户卡片 print(f" 点击用户卡片...") try: elements = page.query_selector_all('span#attribone') for elem in elements: text = elem.inner_text() or "" if display_name.lower() in text.lower(): elem.click(timeout=5000) break except Exception as e: print(f" [WARNING] 点击失败: {e}") # 等待新窗口 print(f" 等待新窗口...") time.sleep(5) if page_state['created'] and page_state['new_page']: print(f" [OK] 新窗口已打开") # 获取新窗口 URL new_url = page_state['new_page'].url print(f" 新窗口 URL: {new_url}") # 提取 systemuserid systemuserid = extract_systemuserid_from_url(new_url) if systemuserid: result['systemuserid'] = systemuserid print(f" [OK] SystemUserId: {systemuserid}") # 关闭新窗口 try: page_state['new_page'].close() time.sleep(1) except: pass # 3. 禁用用户 print(f" 步骤 3/3: 禁用用户") success, message = disable_user_via_api(systemuserid, page) if success: result['status'] = 'success' result['message'] = message print(f" [OK] {message}") else: result['status'] = 'failed' result['message'] = message print(f" [ERROR] {message}") else: result['status'] = 'failed' result['message'] = "无法获取 systemuserid" print(f" [ERROR] 无法获取 systemuserid") # 关闭新窗口 try: page_state['new_page'].close() except: pass else: result['status'] = 'failed' result['message'] = "未检测到新窗口打开" print(f" [ERROR] 未检测到新窗口打开") results.append(result) # 每处理完一个用户，暂停 if idx < len(employee_ids): print(f"\n 等待 2 秒后处理下一个...") time.sleep(2) except Exception as e: import traceback print(f" [ERROR] 处理失败: {e}") traceback.print_exc() result['status'] = 'failed' result['message'] = f"异常: {str(e)}" results.append(result) # 生成报告 print() print("="*60) print(" 所有用户处理完成！生成报告...") print("="*60) print() timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') report_file = f"Termination_Report_{timestamp}.csv" with open(report_file, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f) writer.writerow(['EmployeeId', 'DisplayName', 'SystemUserId', 'Status', 'Message', 'Timestamp']) for result in results: writer.writerow([ result['employee_id'], result['display_name'], result['systemuserid'], result['status'], result['message'], result['timestamp'] ]) print(f"[OK] 报告已生成: {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() print("统计:") print(f" 总数: {total}") print(f" 成功: {success}") print(f" 失败: {failed}") if failed > 0: print() print("失败的用户:") for r in results: if r['status'] == 'failed': print(f" - {r['employee_id']}: {r['message']}") print() print("浏览器将在 3 秒后关闭...") time.sleep(3) context.close() browser.close() except KeyboardInterrupt: print("\n[INFO] 用户中断") except Exception as e: print(f"\n[ERROR] 错误: {e}") import traceback traceback.print_exc() print() print("="*60) print(" 程序结束") print("="*60) print()