下载地址：https://www.pan38.com/yun/share.php?code=JCnzE   提取密码：1133

企业微信机器人实现了自动回复消息和自动同意好友请求功能。主程序使用多线程处理消息轮询和好友请求处理，配置文件可以自定义回复规则。使用时需要替换config.py中的企业微信API凭证。

import requests
import json
import time
from typing import Dict, Any
from threading import Thread

class WeChatBot:
    def __init__(self, corp_id: str, corp_secret: str, agent_id: int):
        self.corp_id = corp_id
        self.corp_secret = corp_secret
        self.agent_id = agent_id
        self.access_token = ""
        self.token_expire = 0
        self.base_url = "https://qyapi.weixin.qq.com/cgi-bin"
        self._get_token()
        
    def _get_token(self) -> None:
        url = f"{self.base_url}/gettoken?corpid={self.corp_id}&corpsecret={self.corp_secret}"
        response = requests.get(url)
        data = response.json()
        if data["errcode"] == 0:
            self.access_token = data["access_token"]
            self.token_expire = time.time() + data["expires_in"] - 300
        else:
            raise Exception(f"获取token失败: {data['errmsg']}")
            
    def _check_token(self) -> None:
        if time.time() >= self.token_expire:
            self._get_token()
            
    def get_unread_messages(self) -> list:
        self._check_token()
        url = f"{self.base_url}/message/get?access_token={self.access_token}"
        response = requests.post(url, json={"agentid": self.agent_id})
        return response.json().get("item", [])
        
    def auto_reply(self, message: Dict[str, Any]) -> None:
        if message["msgtype"] == "text":
            content = message["text"]["content"]
            reply = self._generate_reply(content)
            self._send_message(message["fromusername"], reply)
            
    def _generate_reply(self, content: str) -> str:
        # 这里可以接入NLP模型实现智能回复
        if "你好" in content:
            return "您好，我是自动回复助手，有什么可以帮您？"
        elif "谢谢" in content:
            return "不客气，这是我应该做的！"
        else:
            return "已收到您的消息，稍后会有人工客服为您服务。"
            
    def _send_message(self, user_id: str, content: str) -> None:
        self._check_token()
        url = f"{self.base_url}/message/send?access_token={self.access_token}"
        data = {
            "touser": user_id,
            "msgtype": "text",
            "agentid": self.agent_id,
            "text": {"content": content},
            "safe": 0
        }
        requests.post(url, json=data)
        
    def auto_accept_friend(self) -> None:
        self._check_token()
        url = f"{self.base_url}/externalcontact/get_follow_user_list?access_token={self.access_token}"
        response = requests.get(url)
        follow_users = response.json().get("follow_user", [])
        
        for user in follow_users:
            url = f"{self.base_url}/externalcontact/get_unassigned_list?access_token={self.access_token}"
            data = {"page_id": 0, "page_size": 1000}
            response = requests.post(url, json=data)
            pending_requests = response.json().get("info", [])
            
            for req in pending_requests:
                self._accept_request(req["external_userid"], user["userid"])
                
    def _accept_request(self, external_id: str, user_id: str) -> None:
        url = f"{self.base_url}/externalcontact/transfer?access_token={self.access_token}"
        data = {
            "external_userid": external_id,
            "handover_userid": user_id,
            "takeover_userid": user_id
        }
        requests.post(url, json=data)
        
    def run(self) -> None:
        def message_loop():
            while True:
                messages = self.get_unread_messages()
                for msg in messages:
                    self.auto_reply(msg)
                time.sleep(5)
                
        def friend_loop():
            while True:
                self.auto_accept_friend()
                time.sleep(60)
                
        Thread(target=message_loop).start()
        Thread(target=friend_loop).start()

    "corp_id": "your_corp_id",
    "corp_secret": "your_corp_secret",
    "agent_id": 1000002,
    "auto_reply_rules": {
        "你好": "您好，我是自动回复助手",
        "谢谢": "不客气",
        "default": "已收到您的消息"
    }
}

from wechat_bot import WeChatBot
import config
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

def main():
    try:
        bot = WeChatBot(
            config.corp_id,
            config.corp_secret,
            config.agent_id
        )
        bot.run()
        logging.info("企业微信机器人已启动")
        while True:
            time.sleep(3600)
    except Exception as e:
        logging.error(f"程序异常: {str(e)}")

if __name__ == "__main__":
    main()