下载地址:https://www.pan38.com/yun/share.php?code=JCnzE 提取密码:1133
这个实现包含两个主要模块:用户ID采集和私信发送。采集模块通过解析贴吧页面获取用户ID,发送模块模拟用户操作发送私信。使用时需要注意:1.需要有效的百度cookie 2.发送频率不宜过高 3.内容应符合贴吧规定。
import requests
from bs4 import BeautifulSoup
import re
import time
import random
class TiebaUserSpider:
def init(self, tieba_name):
self.tieba_name = tieba_name
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
self.user_list = set()
def get_page_users(self, page_num):
url = f"https://tieba.baidu.com/f?kw={self.tieba_name}&pn={(page_num-1)*50}"
try:
response = requests.get(url, headers=self.headers)
soup = BeautifulSoup(response.text, 'html.parser')
user_links = soup.find_all('a', {'class': 'p_author_name'})
for link in user_links:
username = link.text.strip()
if username not in self.user_list:
self.user_list.add(username)
print(f"第{page_num}页采集完成,共采集到{len(self.user_list)}个用户")
time.sleep(random.uniform(1, 3))
except Exception as e:
print(f"采集第{page_num}页时出错: {str(e)}")
def run(self, start_page=1, end_page=10):
for page in range(start_page, end_page+1):
self.get_page_users(page)
with open(f'{self.tieba_name}_users.txt', 'w', encoding='utf-8') as f:
for user in self.user_list:
f.write(user + '\n')
print(f"采集完成,共采集到{len(self.user_list)}个用户,已保存到{self.tieba_name}_users.txt")
if name == 'main':
spider = TiebaUserSpider('python')
spider.run(1, 5)
requests
import time
import random
from fake_useragent import UserAgent
class TiebaMessageSender:
def init(self, cookie):
self.cookie = cookie
self.ua = UserAgent()
self.headers = {
'User-Agent': self.ua.random,
'Cookie': self.cookie,
'Referer': 'https://tieba.baidu.com/',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
}
def send_message(self, to_user, content):
url = "https://tieba.baidu.com/msg/send"
data = {
'to_user': to_user,
'content': content,
'anonymous': '0',
'is_comment': '0'
}
try:
response = requests.post(url, headers=self.headers, data=data)
result = response.json()
if result.get('no') == 0:
print(f"发送给{to_user}成功")
return True
else:
print(f"发送给{to_user}失败: {result.get('error')}")
return False
except Exception as e:
print(f"发送给{to_user}时出错: {str(e)}")
return False
def batch_send(self, user_list_file, message, delay=5, max_retry=3):
success_count = 0
fail_count = 0
with open(user_list_file, 'r', encoding='utf-8') as f:
users = [line.strip() for line in f if line.strip()]
for i, user in enumerate(users):
retry = 0
while retry < max_retry:
if self.send_message(user, message):
success_count += 1
break
else:
retry += 1
if retry == max_retry:
fail_count += 1
if i < len(users) - 1:
sleep_time = random.uniform(delay-1, delay+1)
print(f"等待{sleep_time:.1f}秒后继续...")
time.sleep(sleep_time)
print(f"发送完成,成功{success_count}条,失败{fail_count}条")
if name == 'main':
# 需要替换为有效的百度cookie
cookie = "BDUSS=your_bduss_here;"
sender = TiebaMessageSender(cookie)
sender.batch_send('python_users.txt', '您好,这是一条测试消息', delay=10)