#coding:utf-8
'''
@author: ZainCheung
@LastEditors: ZainCheung
@description:網易云音樂全自動每日打卡云函式版
@Date: 2020-06-25 14:28:48
@LastEditTime: 2020-09-01 18:20:00
'''
from configparser import ConfigParser
from threading import Timer
import requests
import random
import hashlib
import datetime
import time
import json
import logging
import math
logger = logging.getLogger()
grade = [10,40,70,130,200,400,1000,3000,8000,20000]
api = ''
class Task(object):
'''
物件的建構式
'''
def __init__(self, uin, pwd, sckey, countrycode=86):
self.uin = uin
self.pwd = pwd
self.countrycode = countrycode
self.sckey = sckey
'''
帶上用戶的cookie去發送資料
url:完整的URL路徑
postJson:要以post方式發送的資料
回傳response
'''
def getResponse(self, url, postJson):
response = requests.post(url, data=https://bbs.csdn.net/topics/postJson, headers={'Content-Type':'application/x-www-form-urlencoded'},cookies=self.cookies)
return response
'''
登錄
'''
def login(self):
data = {"uin":self.uin,"pwd":self.pwd,"countrycode":self.countrycode,"r":random.random()}
if '@' in self.uin:
url = api + '?do=email'
else:
url = api + '?do=login'
response = requests.post(url, data=https://bbs.csdn.net/topics/data, headers={'Content-Type':'application/x-www-form-urlencoded'})
code = json.loads(response.text)['code']
self.name = json.loads(response.text)['profile']['nickname']
self.uid = json.loads(response.text)['account']['id']
if code==200:
self.error = ''
else:
self.error = '登錄失敗，請檢查賬號'
self.cookies = response.cookies.get_dict()
self.log('登錄成功')
'''
每日簽到
'''
def sign(self):
url = api + '?do=sign'
response = self.getResponse(url, {"r":random.random()})
data = json.loads(response.text)
if data['code'] == 200:
self.log('簽到成功')
else:
self.log('重復簽到')
'''
每日打卡300首歌
'''
def daka(self):
url = api + '?do=daka'
response = self.getResponse(url, {"r":random.random()})
self.log(response.text)
'''
查詢用戶詳情
'''
def detail(self):
url = api + '?do=detail'
data = {"uid":self.uid, "r":random.random()}
response = self.getResponse(url, data)
data = json.loads(response.text)
self.level = data['level']
self.listenSongs = data['listenSongs']
self.log('獲取用戶詳情成功')
'''
Server推送
'''
def server(self):
if self.sckey == '':
return
url = 'https://sc.ftqq.com/' + self.sckey + '.send'
self.diyText() # 構造發送內容
response = requests.get(url,params={"text":self.title, "desp":self.content})
data = json.loads(response.text)
if data['errno'] == 0:
self.log('用戶:' + self.name + ' Server醬推送成功')
else:
self.log('用戶:' + self.name + ' Server醬推送失敗,請檢查sckey是否正確')
'''
自定義要推送到微信的內容
title:訊息的標題
content:訊息的內容,支持MarkDown格式
'''
def diyText(self):
# today = datetime.date.today()
# kaoyan_day = datetime.date(2020,12,21) #2021考研黨的末日
# date = (kaoyan_day - today).days
for count in grade:
if self.level < 10:
if self.listenSongs < 20000:
if self.listenSongs < count:
self.tip = '還需聽歌' + str(count-self.listenSongs) + '首即可升級'
break
else:
self.tip = '你已經聽夠20000首歌曲,如果登錄天數達到800天即可滿級'
else:
self.tip = '恭喜你已經滿級!'
if self.error == '':
state = ("- 目前已完成簽到\n"
"- 今日共打卡" + str(self.dakanum) + "次\n"
"- 今日共播放" + str(self.dakaSongs) + "首歌\n"
"- 還需要打卡" + str(self.day) +"天")
self.title = ("網易云今日打卡" + str(self.dakaSongs) + "首，已播放" + str(self.listenSongs) + "首")
else:
state = self.error
self.title = '網易云聽歌任務出現問題！'
self.content = (
"------\n"
"#### 賬戶資訊\n"
"- 用戶名稱：" + str(self.name) + "\n"
"- 當前等級：" + str(self.level) + "級\n"
"- 累計播放：" + str(self.listenSongs) + "首\n"
"- 升級提示：" + self.tip + "\n\n"
"------\n"
"#### 任務狀態\n" + str(state) + "\n\n"
"------\n"
"#### 打卡日志\n" + self.dakaSongs_list + "\n\n")
'''
列印日志
'''
def log(self, text):
time_stamp = datetime.datetime.now()
print(time_stamp.strftime('%Y.%m.%d-%H:%M:%S') + ' ' + str(text))
self.time =time_stamp.strftime('%H:%M:%S')
self.list.append("- [" + self.time + "] " + str(text) + "\n\n")
'''
開始執行
'''
def start(self):
try:
self.list = []
self.list.append("- 初始化完成\n\n")
self.login()
self.sign()
self.detail()
counter = self.listenSongs
for i in range(1,10):
self.daka()
#self.log('用戶:' + self.name + ' 第' + str(i) + '次打卡成功,即將休眠10秒')
self.log('第' + str(i) + '次打卡成功,即將休眠10秒')
time.sleep(10)
self.dakanum = i
self.detail()
self.dakaSongs = self.listenSongs - counter
self.log('今日已打卡播放' + str(self.dakaSongs) + '首')
if self.dakaSongs == 300:
break
if self.listenSongs >= 20000:
self.day = 0
else:
self.day = math.ceil((20000 - self.listenSongs)/300)
self.list.append("- 打卡結束，訊息推送\n\n")
self.dakaSongs_list = ''.join(self.list)
self.server()
except:
self.log('用戶任務執行中斷,請檢查賬號密碼是否正確')
else:
self.log('用戶:' + self.name + ' 今日任務已完成')
'''
初始化：讀取配置,組態檔為init.config
回傳字典型別的配置物件
'''
def init():
global api # 初始化時設定api
config = ConfigParser()
config.read('init.config', encoding='UTF-8-sig')
uin = config['token']['account']
pwd = config['token']['password']
countrycode = config['token']['countrycode']
api = config['setting']['api']
md5Switch = config.getboolean('setting','md5Switch')
peopleSwitch = config.getboolean('setting','peopleSwitch')
sckey = config['setting']['sckey']
logger.info('組態檔讀取完畢')
conf = {
'uin': uin,
'pwd': pwd,
'countrycode': countrycode,
'api': api,
'md5Switch': md5Switch,
'peopleSwitch':peopleSwitch,
'sckey':sckey
}
return conf
'''
MD5加密
str:待加密字符
回傳加密后的字符
'''
def md5(str):
hl = hashlib.md5()
hl.update(str.encode(encoding='utf-8'))
return hl.hexdigest()
'''
加載Json檔案
jsonPath:json檔案的名字,例如account.json
'''
def loadJson(jsonPath):
with open(jsonPath,encoding='utf-8') as f:
account = json.load(f)
return account
'''
檢查api
'''
def check():
url = api + '?do=check'
respones = requests.get(url)
if respones.status_code == 200:
logger.info('api測驗正常')
else:
logger.error('api測驗例外')
'''
任務池
'''
def taskPool():
config = init()
check() # 每天對api做一次檢查
if config['peopleSwitch'] is True:
logger.info('多人開關已打開,即將執行進行多人任務')
account = loadJson("account.json")
for man in account:
logger.info('賬號: ' + man['account'] + ' 開始執行\n========================================')
task = Task(man['account'], man['password'], man['sckey'])
task.start()
time.sleep(10)
logger.info('所有賬號已全部完成任務,服務進入休眠中,等待明天重新啟動')
else :
logger.info('賬號: ' + config['uin'] + ' 開始執行\n========================================')
if config['md5Switch'] is True:
logger.info('MD5開關已打開,即將開始為你加密,密碼不會上傳至服務器,請知悉')
config['pwd'] = md5(config['pwd'])
task = Task(config['uin'], config['pwd'], config['sckey'], config['countrycode'])
task.start()
'''
程式的入口
'''
def main(event,content):
taskPool()
吧
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/252277.html
上一篇：求助Python資料庫問題