企业微信加好友工具批量，可导入手机号自动添加，python脚本分享哈~

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
企业微信批量加好友工具 v2.1
功能：
1. 支持CSV/Excel导入手机号
2. 自动登录企业微信PC客户端
3. 模拟人工操作批量添加好友
4. 完善的日志记录和异常处理
"""

import os
import time
import csv
import random
import logging
from typing import List
import pyautogui
import pyperclip
import openpyxl
from pynput import keyboard
from datetime import datetime

# 配置日志系统
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('wecom_add.log'),
        logging.StreamHandler()
    ]
)

class WeComBatchAdder:
    def __init__(self):
        self.delay_range = (0.5, 1.5)  # 操作延迟区间(秒)
        self.max_retry = 3             # 最大重试次数
        self.screen_width, self.screen_height = pyautogui.size()
        self.wecom_position = None     # 企业微信窗口位置
        self.running = True            # 运行状态标志
        
        # 监听退出快捷键
        self.listener = keyboard.Listener(on_press=self.on_key_press)
        self.listener.start()
    
    def on_key_press(self, key):
        """监听ESC键退出程序"""
        if key == keyboard.Key.esc:
            logging.warning("用户手动终止程序")
            self.running = False
    
    def random_delay(self):
        """随机延迟模拟人工操作"""
        delay = random.uniform(*self.delay_range)
        time.sleep(delay)
    
    def locate_wecom(self):
        """定位企业微信窗口位置"""
        try:
            wecom_icon = pyautogui.locateOnScreen('wecom_icon.png', confidence=0.8)
            if wecom_icon:
                self.wecom_position = (
                    wecom_icon.left + wecom_icon.width//2,
                    wecom_icon.top + wecom_icon.height//2
                )
                logging.info(f"定位到企业微信图标位置: {self.wecom_position}")
                return True
            return False
        except Exception as e:
            logging.error(f"定位企业微信失败: {str(e)}")
            return False
    
    def activate_wecom(self):
        """激活企业微信窗口"""
        if not self.locate_wecom():
            raise Exception("无法定位企业微信窗口")
        
        pyautogui.click(*self.wecom_position)
        self.random_delay()
        pyautogui.hotkey('ctrl', 'f')  # 打开搜索框
        self.random_delay()
    
    def search_and_add(self, phone: str):
        """搜索并添加指定手机号"""
        if not self.running:
            raise KeyboardInterrupt()
        
        logging.info(f"正在处理手机号: {phone}")
        
        # 输入手机号
        pyperclip.copy(phone)
        self.random_delay()
        pyautogui.hotkey('ctrl', 'v')
        self.random_delay()
        
        # 等待搜索结果
        time.sleep(2)
        
        # 检测是否找到用户
        try:
            user_pos = pyautogui.locateOnScreen('user_icon.png', confidence=0.7)
            if not user_pos:
                logging.warning(f"未找到用户: {phone}")
                pyautogui.press('esc')  # 关闭搜索
                self.random_delay()
                return False
            
            # 点击用户
            pyautogui.click(user_pos.left + user_pos.width//2, user_pos.top + user_pos.height//2)
            self.random_delay()
            
            # 点击添加按钮
            add_btn = pyautogui.locateOnScreen('add_btn.png', confidence=0.8)
            if not add_btn:
                logging.warning(f"添加按钮未找到: {phone}")
                return False
                
            pyautogui.click(add_btn.left + add_btn.width//2, add_btn.top + add_btn.height//2)
            self.random_delay()
            
            # 处理验证消息
            verify_btn = pyautogui.locateOnScreen('verify_btn.png', confidence=0.8)
            if verify_btn:
                pyautogui.click(verify_btn.left + verify_btn.width//2, verify_btn.top + verify_btn.height//2)
                self.random_delay()
                pyautogui.typewrite(f"您好，我是{os.getenv('USERNAME', '默认名称')}")
                self.random_delay()
                pyautogui.press('enter')
                self.random_delay()
            
            logging.info(f"成功发送好友请求: {phone}")
            return True
            
        except Exception as e:
            logging.error(f"添加好友时出错: {phone} - {str(e)}")
            return False
        finally:
            pyautogui.press('esc')  # 确保关闭所有弹窗
            self.random_delay()
    
    def load_phones_from_csv(self, filepath: str) -> List[str]:
        """从CSV文件加载手机号"""
        phones = []
        try:
            with open(filepath, mode='r', encoding='utf-8-sig') as f:
                reader = csv.reader(f)
                for row in reader:
                    if row and row[0].isdigit() and len(row[0]) == 11:
                        phones.append(row[0])
            return phones
        except Exception as e:
            logging.error(f"读取CSV文件失败: {str(e)}")
            raise
    
    def load_phones_from_excel(self, filepath: str, sheet_name: str = None) -> List[str]:
        """从Excel文件加载手机号"""
        phones = []
        try:
            wb = openpyxl.load_workbook(filepath)
            sheet = wb[sheet_name] if sheet_name else wb.active
            
            for row in sheet.iter_rows(values_only=True):
                if row and str(row[0]).isdigit() and len(str(row[0])) == 11:
                    phones.append(str(row[0]))
            return phones
        except Exception as e:
            logging.error(f"读取Excel文件失败: {str(e)}")
            raise
    
    def run(self, phone_file: str, file_type: str = 'csv'):
        """主运行方法"""
        start_time = datetime.now()
        success_count = 0
        total_count = 0
        
        try:
            # 加载手机号
            if file_type.lower() == 'csv':
                phones = self.load_phones_from_csv(phone_file)
            else:
                phones = self.load_phones_from_excel(phone_file)
            
            total_count = len(phones)
            if total_count == 0:
                logging.error("未找到有效的手机号")
                return
            
            logging.info(f"共加载 {total_count} 个手机号，开始处理...")
            
            # 激活企业微信
            self.activate_wecom()
            
            # 批量处理
            for idx, phone in enumerate(phones, 1):
                if not self.running:
                    break
                    
                logging.info(f"进度: {idx}/{total_count} ({idx/total_count:.1%})")
                
                retry = 0
                while retry < self.max_retry:
                    try:
                        if self.search_and_add(phone):
                            success_count += 1
                            break
                        retry += 1
                    except KeyboardInterrupt:
                        self.running = False
                        break
                    except Exception as e:
                        logging.error(f"处理失败: {phone} - {str(e)}")
                        retry += 1
                        if retry < self.max_retry:
                            time.sleep(5)  # 重试前等待
                
                # 随机间隔防止频繁操作
                time.sleep(random.uniform(2, 5))
            
        except Exception as e:
            logging.error(f"程序运行异常: {str(e)}")
        finally:
            elapsed = datetime.now() - start_time
            logging.info(
                f"处理完成！成功: {success_count}, 失败: {total_count - success_count}, "
                f"耗时: {elapsed.total_seconds():.1f}秒"
            )


if __name__ == '__main__':
    # 使用示例
    adder = WeComBatchAdder()
    
    # 从CSV文件导入
    # adder.run('phones.csv', 'csv')
    
    # 从Excel文件导入
    adder.run('phones.xlsx', 'excel')