下载地址:https://www.pan38.com/yun/share.php?code=JCnzE 提取密码:1133
这个贴吧私信群发工具包含三个主要文件:主程序、配置文件和入口文件。主程序实现了登录、获取好友列表、发送私信等核心功能,并加入了防封策略。使用时请遵守贴吧规则,合理设置发送频率和数量。
import requests
import time
import random
import json
from bs4 import BeautifulSoup
import re
import os
import hashlib
from fake_useragent import UserAgent
class TiebaSpammer:
def init(self, username, password):
self.session = requests.Session()
self.ua = UserAgent()
self.username = username
self.password = password
self.cookies = None
self.headers = {
'User-Agent': self.ua.random,
'Referer': 'https://tieba.baidu.com/',
'Host': 'tieba.baidu.com',
'Connection': 'keep-alive'
}
self.login_url = 'https://passport.baidu.com/v2/api/?login'
self.message_url = 'https://tieba.baidu.com/mo/q/newmoindex'
self.token_pattern = re.compile(r"name='token' value='(.*?)'")
self.captcha_url = 'https://passport.baidu.com/cgi-bin/genimage?'
def login(self):
# 获取登录token
login_page = self.session.get('https://passport.baidu.com/v2/api/?getapi&tpl=mn&apiver=v3')
token_match = self.token_pattern.search(login_page.text)
if not token_match:
raise Exception("获取token失败")
token = token_match.group(1)
# 构造登录参数
login_data = {
'username': self.username,
'password': self.password,
'token': token,
'tpl': 'mn',
'apiver': 'v3',
'tt': str(int(time.time() * 1000)),
'codestring': '',
'isPhone': 'false',
'staticpage': 'https://passport.baidu.com/static/passpc-account/html/v3Jump.html',
'loginType': '1',
'callback': 'parent.bd__pcbs__ra48vi'
}
# 发送登录请求
response = self.session.post(self.login_url, data=login_data, headers=self.headers)
if 'err_no=0' in response.text:
print("登录成功")
self.cookies = response.cookies.get_dict()
return True
else:
print("登录失败:", response.text)
return False
def get_friends_list(self, page=1):
friends = []
params = {
'pn': page,
'type': 'friend'
}
response = self.session.get('https://tieba.baidu.com/mo/q/contact/list', params=params, headers=self.headers)
soup = BeautifulSoup(response.text, 'html.parser')
items = soup.find_all('div', class_='list_item')
for item in items:
name = item.find('span', class_='name').text.strip()
uid = item.find('input', {'name': 'uid'})['value']
friends.append({'name': name, 'uid': uid})
return friends
def send_private_message(self, uid, content):
data = {
'content': content,
'to_uid': uid,
'from_uid': '',
'fid': '',
'tid': '',
'type': '1',
'anonymous': '0',
'is_comment': '0'
}
response = self.session.post(self.message_url, data=data, headers=self.headers)
if response.json().get('err_no') == 0:
print(f"发送成功: {uid}")
return True
else:
print(f"发送失败: {response.text}")
return False
def batch_send_messages(self, uids, content, delay=5, max_per_day=50):
sent_count = 0
for uid in uids:
if sent_count >= max_per_day:
print("达到每日发送上限")
break
if self.send_private_message(uid, content):
sent_count += 1
sleep_time = delay + random.uniform(-1, 1)
print(f"等待 {sleep_time:.2f} 秒后继续...")
time.sleep(sleep_time)
def anti_ban_strategy(self):
# 随机更换User-Agent
self.headers['User-Agent'] = self.ua.random
# 随机延迟
time.sleep(random.uniform(1, 3))
# 随机访问其他页面
random_page = random.choice(['https://tieba.baidu.com/f?kw=%E7%99%BE%E5%BA%A6',
'https://tieba.baidu.com/f?kw=python'])
self.session.get(random_page, headers=self.headers)
def run(self, message_content, max_friends=100):
if not self.login():
return
all_friends = []
page = 1
while len(all_friends) < max_friends:
friends = self.get_friends_list(page)
if not friends:
break
all_friends.extend(friends)
page += 1
time.sleep(2)
print(f"获取到 {len(all_friends)} 个好友")
uids = [friend['uid'] for friend in all_friends[:max_friends]]
self.batch_send_messages(uids, message_content)