微信公众号 / 小程序百万级 OpenID 自动化获取工具

import requestsimport jsonimport timeimport osfrom datetime import datetime
class OpenIDCollector:    def __init__(self, app_id, app_secret):        self.app_id = app_id        self.app_secret = app_secret        self.access_token = None        self.token_expire_time = 0        self.base_dir = "openid_data"        self.progress_file = os.path.join(self.base_dir, "progress.json")        self.initialize_environment()            def initialize_environment(self):        """创建存储目录并加载进度"""        os.makedirs(self.base_dir, exist_ok=True)        self.load_progress()            def load_progress(self):        """加载进度文件"""        self.progress = {            "last_next_openid": "",            "total_collected": 0,            "last_token_time": 0,            "call_count": 0,            "last_call_time": 0        }                if os.path.exists(self.progress_file):            try:                with open(self.progress_file, 'r') as f:                    self.progress = json.load(f)            except:                print("进度文件损坏，已重置")        def save_progress(self):        """保存进度文件"""        with open(self.progress_file, 'w') as f:            json.dump(self.progress, f, indent=2)        def get_access_token(self):        """获取或刷新Access Token"""        now = time.time()        # 检查token是否有效（提前5分钟刷新）        if self.access_token and now < self.progress["last_token_time"] + 6600:            return self.access_token                url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={self.app_id}&secret={self.app_secret}"        response = requests.get(url)        data = response.json()                if 'access_token' in data:            self.access_token = data['access_token']            self.progress["last_token_time"] = now            print(f"获取新Access Token: {self.access_token[:10]}...")            return self.access_token        else:            raise Exception(f"获取Access Token失败: {data}")        def rate_limit_control(self):        """API调用频率控制"""        now = time.time()        elapsed = now - self.progress["last_call_time"]                # 每天最多100次调用        if self.progress["call_count"] >= 100:            # 计算到第二天0点的剩余时间            tomorrow = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)            tomorrow = tomorrow.timestamp() + 86400            wait_seconds = tomorrow - now            print(f"今日调用已达上限，等待 {wait_seconds/3600:.2f} 小时")            time.sleep(wait_seconds)            self.progress["call_count"] = 0                # 每分钟最多5次调用（间隔12秒以上）        if elapsed < 12:            wait_time = 12 - elapsed            print(f"频率控制: 等待 {wait_time:.2f} 秒")            time.sleep(wait_time)                self.progress["call_count"] += 1        self.progress["last_call_time"] = time.time()        def get_user_list(self, next_openid=""):        """获取用户OpenID列表"""        self.rate_limit_control()        token = self.get_access_token()        url = f"https://api.weixin.qq.com/cgi-bin/user/get?access_token={token}&next_openid={next_openid}"        response = requests.get(url)        return response.json()        def save_openids(self, openids, batch_num):        """保存OpenID到分区文件"""        partition = batch_num // 10  # 每10万条一个分区        partition_dir = os.path.join(self.base_dir, f"partition_{partition:02d}")        os.makedirs(partition_dir, exist_ok=True)                filename = os.path.join(partition_dir, f"batch_{batch_num:03d}.json")        with open(filename, 'w') as f:            json.dump(openids, f)                print(f"已保存批次 {batch_num}: {len(openids)} 条OpenID")        def collect_openids(self, target_count=1000000):        """主收集函数"""        batch_num = 0        next_openid = self.progress["last_next_openid"]                while self.progress["total_collected"] < target_count:            batch_num += 1            print(f"\n处理批次 #{batch_num} (总计: {self.progress['total_collected']})")                        try:                data = self.get_user_list(next_openid)                                if 'data' in data and 'openid' in data['data']:                    openids = data['data']['openid']                    count = len(openids)                                        if count > 0:                        self.save_openids(openids, batch_num)                        self.progress["total_collected"] += count                        self.progress["last_next_openid"] = data.get('next_openid', '')                        next_openid = self.progress["last_next_openid"]                                        # 检查是否还有更多数据                    if not next_openid or count == 0:                        print("已获取所有用户OpenID")                        break                else:                    print(f"API返回异常: {data}")                    if data.get('errcode') == 40001:  # token失效                        self.access_token = None            except Exception as e:                print(f"发生错误: {str(e)}")                        # 每批次后保存进度            self.save_progress()                print(f"\n任务完成! 共获取 {self.progress['total_collected']} 条OpenID")
# 使用示例if __name__ == "__main__":    # 从环境变量或配置文件中获取这些值    APP_ID = "YOUR_APP_ID"    APP_SECRET = "YOUR_APP_SECRET"        collector = OpenIDCollector(APP_ID, APP_SECRET)    collector.collect_openids(target_count=1000000)

python wechat_openid_collector.py

openid_data/├── progress.json          # 进度记录文件├── partition_00/          # 分区目录 (0-99,999条)│   ├── batch_001.json     # 第1批次 (10,000条)│   ├── batch_002.json     # 第2批次│   └── ...├── partition_01/          # 分区目录 (100,000-199,999条)└── ...

# 分布式处理示例（概念代码）if is_master_node:    # 主节点分配任务    for partition in range(0, 100, 10):        assign_task(partition, partition+9)else:    # 工作节点处理指定分区    process_partition(start_part, end_part)

import gzip
def save_compressed(openids, filename):    with gzip.open(filename + '.gz', 'wt') as f:        json.dump(openids, f)

import smtplibfrom email.mime.text import MIMEText
def send_alert(message):    msg = MIMEText(message)    msg['Subject'] = 'OpenID收集异常'    msg['From'] = 'monitor@example.com'    msg['To'] = 'admin@example.com'        with smtplib.SMTP('smtp.example.com') as server:        server.send_message(msg)

   100万条OpenID获取时间 ≈ 100次调用 × 15秒间隔 ≈ 25分钟   + 每日上限等待（如中途达到上限）≈ 24小时