Created
December 27, 2018 08:50
-
-
Save Foair/515230b5b995b51a6f0b427a365b8544 to your computer and use it in GitHub Desktop.
Revisions
-
Foair created this gist
Dec 27, 2018 .There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -0,0 +1,251 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- import asyncio import base64 import hashlib import json import random import sys import time from urllib.parse import quote_plus import requests import rsa class Authenticate(object): def __init__(self, config): """通过配置初始化配置信息""" self.username = config['auth']['user'] self.password = config['auth']['password'] self.headers = { 'User-Agent': config['browser']['ua'], 'Accept-Language': config['browser']['lang'] } self.config = config self.cookies = {} self.access_key = '' self.csrf = '' self.mid = '' @staticmethod def calc_sign(string): """计算加盐后字符串的 MD5""" string += '560c52ccd288fed045859ed18bffd973' md5 = hashlib.md5() md5.update(string.encode()) sign = md5.hexdigest() return sign def __get_pwd(self): """获得 RSA 公钥，并对用户名和密码进行处理""" url = 'https://passport.bilibili.com/api/oauth2/getKey' params = { 'appkey': self.config['mobile']['key'], 'sign': self.calc_sign('appkey=%s' % self.config['mobile']['key']) } response = requests.post(url, data=params) value = response.json()['data'] # 获得 RSA 公钥 key = value['key'] salt = value['hash'] # 从 RSA 证书字符串加载公钥 pubkey = rsa.PublicKey.load_pkcs1_openssl_pem(key.encode()) # 使用公钥对加盐后的密码进行加密并进行 Base64 编码 password = base64.b64encode(rsa.encrypt((salt + self.password).encode(), pubkey)) # 对结果进行 URL 编码 return quote_plus(self.username), quote_plus(password) def login(self): """进行登录操作""" # 接收经过处理后的用户名和密码 user, password = self.__get_pwd() # 生成签名 query = 'appkey=1d8b6e7d45233436&password=%s&username=%s' % (password, user) sign = self.calc_sign(query) # 构造载荷 payload = query + '&sign=' + sign headers = { 'Content-Type': self.config['contentType']['form'] } # 发送登录的请求 url = 'https://passport.bilibili.com/api/v2/oauth2/login' response = requests.post(url, data=payload, headers=headers) # noinspection PyBroadException try: data = response.json()['data'] cookies = data['cookie_info']['cookies'] # 从 JSON 数组构造 Cookie 字典 cookies_dict = {} for cookie in cookies: cookies_dict[cookie['name']] = cookie['value'] # 获得 Cookie、CSRF token、用户 ID 和 access_key self.cookies = cookies_dict self.csrf = cookies_dict['bili_jct'] self.mid = cookies_dict['DedeUserID'] self.access_key = data['token_info']['access_token'] except Exception: print('登录失败') sys.exit(1) class Bilibili(Authenticate): def __init__(self, *args): # 初始化父类并进行登录 super().__init__(*args) self.login() async def do_watch(self): """模拟观看视频""" # noinspection PyBroadException try: # 获得视频列表 video_list = await self.get_videos() # 从视频列表中随机选出一个视频 aid = random.choice(video_list) # 通过视频 av 号获得 cid cid = await self.get_cid(aid) # 执行视频观看任务 await self.watch(aid, cid) print('观看视频完成') except Exception: print('无法完成观看视频') async def do_share(self): """分享视频控制""" # noinspection PyBroadException try: await self.share() print('分享任务完成') except Exception: print('无法完成分享') @staticmethod def current_time(): """返回当前的时间戳字符串，精确到毫秒""" return str(int(time.time() * 1000)) async def get_subscribed(self): """获得 50 个关注 UP 主""" url = 'https://api.bilibili.com/x/relation/followings?vmid=%s&ps=50&order=desc' % self.mid headers = { 'User-Agent': self.config['browser']['ua'], 'Accept-Language': self.config['browser']['lang'] } response = requests.get(url, headers=headers, cookies=self.cookies) # 返回 UP 主们的用户 ID return [user['mid'] for user in response.json()['data']['list']] async def get_videos(self): """从最近关注的 50 UP 主获得视频列表""" subscribed_50_ups = await self.get_subscribed() # 从 50 个 UP 主中随机选取一个 mid = random.choice(subscribed_50_ups) # 获得视频 av 号的列表 url = 'https://space.bilibili.com/ajax/member/getSubmitVideos?mid=%s&pagesize=100&tid=0' % mid vlist = requests.get(url).json()['data']['vlist'] video_list = [v['aid'] for v in vlist] # 返回视频 av 号的列表 return video_list async def share(self): """模拟分享视频""" # 随机选择一个视频 video_list = await self.get_videos() aid = random.choice(video_list) # 对参数进行构造和签名 ts = self.current_time() params = 'access_key={}&aid={}&appkey=1d8b6e7d45233436&build=5260003&from=7&mobi_app=android' \ '&platform=android&ts={}'.format(self.access_key, aid, ts) sign = self.calc_sign(params) data = {'access_key': self.access_key, 'aid': aid, 'appkey': '1d8b6e7d45233436', 'build': '5260003', 'from': '7', 'mobi_app': 'android', 'platform': 'android', 'ts': ts, 'sign': sign} url = 'https://app.bilibili.com/x/v2/view/share/add' headers = { 'User-Agent': self.config['mobile']['ua'], 'Cookie': 'sid=8wfvu7i7' } response = requests.post(url, headers=headers, data=data).json() print('分享视频', response) @staticmethod async def get_cid(aid): """获得 cid""" url = 'https://www.bilibili.com/widget/getPageList?aid=%s' % aid response = requests.get(url) cid = response.json()[0]['cid'] return cid async def watch(self, aid, cid): """向服务端发送心跳信息，模拟观看视频""" url = 'https://api.bilibili.com/x/report/web/heartbeat' headers = { 'User-Agent': self.config['browser']['ua'], 'Referer': 'https://www.bilibili.com/video/av%s' % aid } data = { 'aid': aid, 'cid': cid, 'mid': self.mid, 'csrf': self.csrf, 'played_time': '0', 'realtime': '0', 'start_ts': self.current_time(), 'type': '3', 'dt': '2', 'play_type': '1' } response = requests.post(url, headers=headers, data=data, cookies=self.cookies) print('观看视频', response.text) def query_reward(self): """查询今日任务的完成情况""" url = 'https://account.bilibili.com/home/reward' headers = { 'Referer': 'https://account.bilibili.com/account/home', 'User-Agent': self.config['browser']['ua'] } data = requests.get(url, headers=headers, cookies=self.cookies).json()['data'] login = data['login'] watch = data['watch_av'] share = data['share_av'] coins = data['coins_av'] # 是否登录、是否观看视频、是否分享视频、投币数量 print(login, watch, share, coins) if __name__ == '__main__': # 从 JSON 文件中读取配置 f = open('config.json', encoding='utf_8') config_json = json.load(f) f.close() bilibili = Bilibili(config_json) # 自定义任务：分享、观看视频 task = [ bilibili.do_share(), bilibili.do_watch() ] # 获得一个事件循环队列 loop = asyncio.get_event_loop() # 等待任务完成 loop.run_until_complete(asyncio.wait(task)) # 查询奖励 bilibili.query_reward() This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -0,0 +1 @@ {"browser": {"ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36", "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8", "lang": "zh-CN,zh;q=0.9,en;q=0.8", "ae": "gzip, deflate, br"}, "mobile": {"ua": "Mozilla/5.0 BiliDroid/5.26.3 ([email protected])", "key": "1d8b6e7d45233436"}, "auth": {"user": "", "password": ""}, "meta": {"name": "\u6d41\u4e91\u6d6e\u661f"}, "contentType": {"json": "application/json", "form": "application/x-www-form-urlencoded", "formData": "multipart/form-data"}}
Foair
created
this gist