记一次微信公众号批量投票，主要记录思路，具体的公众号已打码。
主线思路：
1.分析投票所需数据（判断 是通过ip 还是 需要注册账号后才可以进行投票）
2.批量注册第三方平台的账号（若拥有大量可用微信号，则可用微信号直接替代此步）
3.利用已注册的账号进行遍历，随机投票，每日循环。
实施步骤：
1.使用 mitmproxy 监测 投票过程（亦可用fiddler进行抓包），分析并记录 必要请求及所需数据。
上图中 红色字体所标记数据为 用户数据，
Referer 中涉及到 被投票用户的 work_id 和 library_id ，可以通过分享页面的url获取
data 中涉及到 被投票用户的work_id 和 投票者的token ，投票者的token 是每个用户特有的识别码，故需要进行注册账号
注：请求所需的 代理ip 下方有介绍，主要是通过实时爬取 站大爷的可用ip 来实现。
2. 注册该公众号平台的账号，并使用 selenium 模拟chrome访问，获取控件并填写信息。
图中信息均为必填信息，
昵称， 现在各种平台比较多，可以比较快捷的获取大量用户名，或者自己编辑
密码、确认密码 更不用说，是比较容易填写的，初看比较困难的是如何获取到大量的手机号，以及能得到其验证码。
因为国家针对sim卡实行实名制，所以正规渠道比较少。在网上搜索后发现，
①. 有很多的接码平台可以实现此类功能，但是大多要付费、还有很多api要了解，乍一看比较麻烦，就没有使用这种办法。
②. 还发现有一些机构个人会在网上免费公开少量的手机号，可以在线接收短信注册码验证码（例 点击查看），类似的网站如果需要可以自行在网上搜索。此方法的确可以使用，但是实操时发现很多手机号发送验证码后不能及时接收到，即同一时间有很多用户使用，导致运营商延迟较大。所以也没使用这种办法。
③. 最后因为找不到手机号，就 按了F12 ，这下才发现，原来每次点击 验证码后，都会响应到一个json串，即验证码，于是得来全不费工夫。最后自己随机生成手机号，并获取响应到的json串，以此达到 获取验证码的效果。至此，批量注册账号的问题得以解决了。
注册完账号后，发现 用户的token值 并不会显示出来，
于是在网页的js中开始检索， 发现 通过 get_user_token() 方法可以直接获取当前用户的token，在F12 的 console 中调试后发现可行。就通过此方法获取的 token 值。
3. 获取大量 实时免费的高匿代理ip。
通过前两步，已经可以实现注册账号、进行投票了，但没有高匿代理ip，极易被人察觉，所以这一步就是想办法获取代理ip了。
初步调研了目前主流的免费的代理ip网站，最终选择了站大爷，因为每个小时都更新，而且使用公众号的主站测试了一下，发现大部分是可以成功访问的。
项目中使用的：
数据库：SSDB，详细信息可参见git
代理ip池： proxy_pool，详细信息可参见git
为了尽量少的影响他人服务器，我的请求速度写的极慢，按需修改。
代码部分：
1，投票
import json
import random
import time
import pyssdb
import requests
from selenium import webdriver
class vote_wechat(object):
"""
投票部分
先 请求 upload_visit 上传服务器数据，返回值为result: 1 说明被服务器认可
然后 请求add_praise 上传token ，服务器返回成功
requests.get("http://192.168.1.7:5010/get/").content.decode('utf-8')
此行的意思是 请求局域网内的代理ip库，从中随机取一个可用ip
可以根据自己的配置情况自行修改
"""
def __init__(self,
token="******",
proxy={'http': requests.get("http://192.168.1.7:5010/get/").content.decode('utf-8')}):
# 默认微信用户 - 小号
self.token = token
# 默认代理ip - 随机选取
self.proxy = proxy
# 被投票人信息
work_id = "******"
library_id = "******"
self.url = 'http://www.y***k.net/wh/api/activity/fruit/add_praise'
Agent = [
'Mozilla/5.0 (iPhone; CPU iPhone OS 12_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/16A366 MicroMessenger/6.7.3(0x16070321) NetType/WIFI Language/zh_CN',
'Mozilla/5.0 (Linux; Android 7.0; Redmi Note 4X Build/NRD90M; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044304 Mobile Safari/537.36 MicroMessenger/6.7.2.1340(0x2607023A) NetType/WIFI Language/zh_CN',
'Mozilla/5.0 (Linux; Android 7.1.1; vivo X20A Build/NMF26X; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044304 Mobile Safari/537.36 MicroMessenger/6.7.2.1340(0x2607023A) NetType/4G Language/zh_CN',
'Mozilla/5.0 (Linux; Android 7.1.1; OD105 Build/NMF26F; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044208 Mobile Safari/537.36 wxwork/2.4.9 MicroMessenger/6.3.22 NetType/4G Language/zh',
'Mozilla/5.0 (Linux; Android 7.1.1; OPPO R11st Build/NMF26X; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044304 Mobile Safari/537.36 MicroMessenger/6.7.2.1340(0x2607023A) NetType/WIFI Language/zh_CN',
'Mozilla/5.0 (Linux; Android 8.0; SM-G9500 Build/R16NW; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044208 Mobile Safari/537.36 MicroMessenger/6.7.2.1340(0x2607023A) NetType/WIFI Language/zh_CN',
'Mozilla/5.0 (Linux; Android 8.0; DUK-AL20 Build/HUAWEIDUK-AL20; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044353 Mobile Safari/537.36 MicroMessenger/6.7.3.1360(0x26070333) NetType/WIFI Language/zh_CN Process/tools',
'Mozilla/5.0 (Linux; Android 8.0; MHA-AL00 Build/HUAWEIMHA-AL00; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044304 Mobile Safari/537.36 MicroMessenger/6.7.3.1360(0x26070333) NetType/NON_NETWORK Language/zh_CN Process/tools',
'Mozilla/5.0 (Linux; Android 8.1; EML-AL00 Build/HUAWEIEML-AL00; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/53.0.2785.143 Crosswalk/24.53.595.0 XWEB/358 MMWEBSDK/23 Mobile Safari/537.36 MicroMessenger/6.7.2.1340(0x2607023A) NetType/4G Language/zh_CN',
'Mozilla/5.0 (Linux; Android 8.1; PAR-AL00 Build/HUAWEIPAR-AL00; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044304 Mobile Safari/537.36 MicroMessenger/6.7.3.1360(0x26070333) NetType/WIFI Language/zh_CN Process/tools',
]
self.Agent = Agent[random.randint(0, len(Agent) - 1)]
self.headers = {
"Host": "www.y***k.net",
"Proxy-Connection": "keep-alive",
"Content-Length": "101",
"Accept": "application/json, text/javascript, */*; q=0.01",
"Origin": "http://www.y***k.net",
"X-Requested-With": "XMLHttpRequest",
'User-Agent': self.Agent,
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Referer": "http://www.y***k.net/wh/web/picture.html?page=picture&work_id={work_id}&from=singlemessage&library_id={library_id}&is_weixin=1&login_version=1&is_login=1".
format(work_id=work_id, library_id=library_id),
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "zh-CN,en-US;q=0.9"}
self.data = {
"issign": "1",
"mac": "",
"work_id": work_id,
"token": token}
def get_proxy(self):
return requests.get("http://192.168.1.7:5010/get/").content.decode('utf-8')
def run(self):
i = 0
while i < 3:
try:
# 投票前先请求分享页面，增加浏览数
url_read = 'http://www.y***k.net/wh/api/asset/info/?workID=******&mac=&token='
hreader_read = {
"Host": "www.y***k.net",
"Proxy-Connection": "keep-alive",
"Accept": "application/json, text/javascript, */*; q=0.01",
"Origin": "http://www.y***k.net",
"X-Requested-With": "XMLHttpRequest",
'User-Agent': self.Agent,
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "zh-CN,en-US;q=0.9"}
for j in range(random.randint(2, 5)):
# 请求分享页，增加浏览量
req_read = requests.get(url_read, headers=hreader_read, proxies=self.proxy)
time.sleep(random.random() * 10)
req = requests.post(self.url, data=self.data, headers=self.headers, proxies=self.proxy)
req_json = json.loads(req.content)
if req.status_code == 200 and 'praise_times' in req_json:
print('\r累计{}票，本账号今日已投{}票'.format(req_json['praise_times'], req_json['has_vote']))
i += 1
time.sleep(random.randint(2, 5))
else:
print(req_json['err_msg'])
except Exception as e:
# self.proxy = {'http': self.get_proxy()}
print('遇到问题{}，跳过1次'.format(e))
i += 1
time.sleep(1)
2，注册账号
class register_user(object):
"""
自动注册一个账号(用手机号注册账号，获取每个账号的token,一个账号需要1分钟左右时间创建)
格式：
ip 账号 token
小号 ******
大号 ******
:return:账号所用ip/手机号/token
"""
def __init__(self, proxy=requests.get("http://192.168.1.7:5010/get/").content.decode('utf-8')):
self.info_list = []
self.token_ = 'null'
self.proxy = proxy
self.agent = vote_wechat().Agent
self.nickname = self.nickname()[random.randint(0, 500)]
self.tele_num = self.random_(17)
self.regest_url = 'http://www.y***k.net/wh/web/mreg.html'
options = webdriver.ChromeOptions()
options.add_argument('--headless')
options.add_argument("disable-infobars")
options.add_argument('User-Agent={}'.format(self.agent))
options.add_argument('--proxy-server={0}'.format(self.proxy))
self.chrome = webdriver.Chrome(chrome_options=options)
self.chrome.set_page_load_timeout(20)
self.chrome.get(self.regest_url)
def nickname(self):
# 从SSDB数据库取昵称(根据自身数据情况进行替换)
c = pyssdb.Client('192.168.1.7')
nick = []
user_list = c.hkeys('User_name_icon', '', '', 500)
for i in user_list:
rc = i.decode('utf-8').split('|,|')[1]
nick.append(rc)
c.disconnect()
return nick
def random_(self, up, num=9):
"""
随机生成手机号码
:param up: 手机号前缀，默认2位
:param num:默认9位数
:return:默认11位随机数字
"""
nex = []
for i in range(num):
the_o = random.randint(0, 9)
nex.append(str(the_o))
return str(up) + ''.join(nex)
def send_sms(self):
# 发送验证码请求
send_sms_url = 'http://www.y***k.net/wh/api/common/send_validate_sms/'
header_send_sms = {
"Accept": "application/json, text/javascript, */*; q=0.01",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "zh-CN,zh;q=0.9",
"Content-Length": "32",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Host": "www.y***k.net",
"Origin": "http://www.y***k.net",
"Proxy-Connection": "keep-alive",
"Referer": "http://www.y***k.net/wh/web/mreg.html",
"User-Agent": self.agent,
"X-Requested-With": "XMLHttpRequest",
}
data_send_sms = {
'phone': self.tele_num,
'name': self.nickname,
}
proxy_send_sms = {
'http': self.proxy
}
send_sms_req = requests.post(send_sms_url, headers=header_send_sms, data=data_send_sms, proxies=proxy_send_sms)
v_code = json.loads(send_sms_req.content)['v_code']
return v_code
def get_proxy(self):
return requests.get("http://192.168.1.7:5010/get/").content.decode('utf-8')
def edit_info(self):
print('\r正在浏览器编辑中......', end='')
# 输入昵称、手机号、密码、确认密码，等待10秒后，输入验证码
nickname_input = self.chrome.find_element_by_id("doc-ipt-1")
nickname_input.send_keys(self.nickname)
phone_input = self.chrome.find_element_by_id("doc-ipt-2")
phone_input.send_keys(self.tele_num)
psd_input = self.chrome.find_element_by_id("doc-ipt-3")
psd_input.send_keys(self.tele_num)
psd_again_input = self.chrome.find_element_by_id("doc-ipt-4")
psd_again_input.send_keys(self.tele_num)
time.sleep(10)
# 输入验证码
print('\r发送验证码中......', end='')
verification_input = self.chrome.find_element_by_id("doc-ipt-5")
verification_input.send_keys(self.send_sms())
def run(self):
self.edit_info()
# 随机等待15-50秒后，提交信息
time.sleep(random.randint(15, 50))
btn = self.chrome.find_element_by_id('confirm_btn')
btn.click()
# 执行js,获取token值
js = 'document.getElementById("doc-ipt-1").className = get_user_token()'
count = 0
while self.token_ == 'null' or count > 10:
count += 1
print('\r“token_” key is :', self.token_, end='')
time.sleep(1)
self.chrome.execute_script(js)
self.token_ = self.chrome.find_element_by_id("doc-ipt-1").get_attribute('class')
print('\r{} {} {}'.format(self.proxy, self.tele_num, self.token_))
self.info_list.extend([self.proxy, self.tele_num, self.token_])
return self.info_list
3，代理ip写入，检查是否为高匿、是否可访问
# 检查url是否可以通过代理ip访问
class check_url(object):
"""
检查url是否可以通过代理ip访问
case:
check = check_url('http://httpbin.org/ip')
while not check.check_ip():
http_str = check.refresh()
if check.proxies['http'] == http_str:
break
check.proxies['http'] = http_str
"""
def __init__(self, url, ip_proxy=None):
self.url = url
if ip_proxy:
self.proxies = {
'http': ip_proxy}
else:
self.proxies = {
'http': requests.get("http://192.168.1.7:5010/get/").content.decode('utf-8')}
def check_ip(self):
# 能访问返回True
try:
# 超过20秒的代理就不要了
r = requests.get(self.url, proxies=self.proxies, timeout=10, verify=False)
if r.status_code == 200:
# print(r.content)
print('{}可以正常访问'.format(self.proxies['http']))
return True
except Exception as e:
print('\r{}---->{}---更换ip中'.format(self.proxies['http'], e), end='')
time.sleep(0.1)
def refresh(self):
return requests.get("http://192.168.1.7:5010/get/").content.decode('utf-8')
# 从站大爷ip网站中取最新ip，写入文件
def write_zdy_ip():
# 写入文件位置为：E:\ip3.txt
def now_time(time_str="%Y-%m-%d-%H-", replace=['年', '月', '日', '时'], replace_str='-'):
time_stamp = time.localtime(time.time())
str_now_time = str(time_stamp.tm_year) + replace_str + str(time_stamp.tm_mon) + replace_str + \
str(time_stamp.tm_mday) + replace_str + str(time_stamp.tm_hour) + replace_str
# str_now_time = time.strftime(time_str, time_stamp) # 含0
show_now_time = str_now_time
for item in replace:
# 将 中文 转化成 unicode
show_now_time = show_now_time.replace(replace_str, item, 1)
item = item.encode('unicode_escape').decode()
str_now_time = str_now_time.replace(replace_str, item, 1)
print('当前时间：', show_now_time)
return str_now_time
zdy_url = 'http://ip.zdaye.com/dayProxy.html'
headers = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3",
"Cookie": "acw_tc=781bad0815554276156886732e5dc4c8fa7bc5f5f7d49920ab817374586a8b; ASPSESSIONIDQSCADTTD=CPLINKOCBKENHNPBEHDCMAKJ; __51cke__=; Hm_lvt_8fd158bb3e69c43ab5dd05882cf0b234=1555427618; ASPSESSIONIDACRDDQQA=GMPHOIPCLFKBFGCPFMBDFAPM; __tins__16949115=%7B%22sid%22%3A%201555427617018%2C%20%22vd%22%3A%204%2C%20%22expires%22%3A%201555429600579%7D; __51laig__=4; Hm_lpvt_8fd158bb3e69c43ab5dd05882cf0b234=1555427801",
"Host": "ip.zdaye.com",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.103 Safari/537.36",
}
referer = {'Referer': 'http://ip.zdaye.com/dayProxy.html'}
res = requests.get(zdy_url, headers=headers)
if res.status_code == 200:
soup = BeautifulSoup(res.content, "html.parser")
panel = soup.find('div', class_='panel-body')
panel = str(panel)
time_str = now_time()
pattern = r'.+href="(.+)">{}.+'.format(time_str)
pattern_ = r'.+href="(.+)">{}.+'.format('2019\u5e74')
if len(re.findall(pattern, panel)) < 1:
href_day = re.findall(pattern_, panel)[0]
url_day = 'http://ip.zdaye.com' + re.findall(pattern_, panel)[0]
else:
href_day = re.findall(pattern, panel)[0]
url_day = 'http://ip.zdaye.com' + re.findall(pattern, panel)[0]
options = webdriver.ChromeOptions()
options.add_argument('--headless')
options.add_argument("disable-infobars")
chrome_day = webdriver.Chrome(chrome_options=options)
chrome_day.get(zdy_url)
hre = chrome_day.find_elements_by_css_selector('a[href="{}"]'.format(href_day))[0]
hre.click()
chrome_day.refresh()
text = chrome_day.find_elements_by_css_selector('.cont')[0].text
with open('E:\ip3.txt', 'w', encoding='utf-8') as f:
f.write(text)
print('ip已写入...', end='')
# 从站大爷将ip放置txt文件(E:\ip3.txt)中，开始执行
ip_proxy = []
userful_ip = []
with open('E:\ip3.txt', 'r', encoding='utf-8') as f:
for item in f:
ip = item.split('@')[0]
type = re.findall(r'.+\[(.+)\].+', item)[0]
if type == '高匿':
ip_proxy.append(ip)
print('共{}个高匿ip，现在开始检查可用性...'.format(len(ip_proxy)))
# 检查高匿ip是否可以访问目标网站
for i in range(len(ip_proxy)):
check = check_url('http://www.y***k.net', ip_proxy[i])
if not check.check_ip():
print('{}不可用'.format(ip_proxy[i]))
else:
userful_ip.append(ip_proxy[i])
print('共{}个可用ip，可用ip集合：'.format(len(userful_ip)), userful_ip)
return userful_ip
4，主函数
# 批量 新建账号
def new_start(userful_ip):
for i in range(len(userful_ip)):
try:
print('第{}次开始运行...'.format(i + 1))
info_list = register_user(proxy=userful_ip[i]).run()
info_str = info_list[0] + '|' + info_list[1] + '|' + info_list[2] + '\n'
with open(r"E:\users.txt", 'a') as f:
f.write(info_str)
print('已成功写入第{}个用户,响应等待中...'.format(i + 1))
time.sleep(random.randint(30, 600))
except Exception as e:
print('Error:', e)
# 批量 投票
def wechat_start():
info_ip, token, ip_token = [], [], []
# with open(r'E:\users.txt', 'r') as f:
with open(r'E:\users_repeat2.txt', 'r') as f:
for item in f:
item_list = item.split('|')
info_ip.append(item_list)
for info in info_ip:
if info[0] not in ip_token:
ip_token.append(info[0])
token.append(info[2].strip())
# else:
# info__str = info[0] + '|' + info[1] + '|' + info[2]
# with open(r"E:\users_repeat2.txt", 'a') as f:
# f.write(info__str)
print('可用用户：', len(ip_token))
# time.sleep(100)
# 10个用户一组进行投票，每个人间隔3~6分钟，每组间隔15~30分钟
# 创建一个10的列表
down = [i * 10 for i in range(1, 20)]
# 将所有用户投票
for i in range(len(token)):
if i+1 < 1:
continue
for itemm in down:
if i == itemm:
time.sleep(random.randint(60 * 15, 60 * 30))
print('正在使用第%s个用户投票' % str(i+1))
wechat_yhl = vote_wechat(token=token[i], proxy={'http': ip_token[i]})
wechat_yhl.run()
time.sleep(random.randint(60 * 3, 60 * 6))
# 0.获得最新ip
userful_ip = write_zdy_ip()
# 1. 注册
new_start(userful_ip)
# 2. 遍历 账号文件夹去投票
wechat_start()
while True:
# 无限注册账号
try:
new_start(write_zdy_ip())
time.sleep(60 * 90)
except Exception:
print('遇到问题，重试中...')
time.sleep(10)