Python开发微信公众号之批量投票刷票记录

2019年5月25日07:59:30 发表评论 1,947 views

微信公众号批量投票,主要记录思路,具体的公众号已打码。

主线思路:

1.分析投票所需数据(判断  是通过ip  还是 需要注册账号后才可以进行投票)

2.批量注册第三方平台的账号(若拥有大量可用微信号,则可用微信号直接替代此步)

3.利用已注册的账号进行遍历,随机投票,每日循环。

 

实施步骤:

1.使用  mitmproxy 监测 投票过程(亦可用fiddler进行抓包),分析并记录 必要请求及所需数据。

Python开发微信公众号之批量投票刷票记录

上图中  红色字体所标记数据为 用户数据,

Referer 中涉及到  被投票用户的 work_id 和 library_id  ,可以通过分享页面的url获取

data 中涉及到 被投票用户的work_id 和 投票者的token ,投票者的token 是每个用户特有的识别码,故需要进行注册账号

注:请求所需的 代理ip 下方有介绍,主要是通过实时爬取 站大爷的可用ip 来实现。

 

2. 注册该公众号平台的账号,并使用 selenium 模拟chrome访问,获取控件并填写信息。

Python开发微信公众号之批量投票刷票记录

图中信息均为必填信息,

昵称, 现在各种平台比较多,可以比较快捷的获取大量用户名,或者自己编辑

密码、确认密码  更不用说,是比较容易填写的,初看比较困难的是如何获取到大量的手机号,以及能得到其验证码。

因为国家针对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)

---------------------
作者:dianepure
来源:CSDN
原文:https://blog.csdn.net/dianepure/article/details/89676437
版权声明:本文为博主原创文章,转载请附上博文链接!

发表评论

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen: