import requests
import json
import threading
from tqdm import tqdm
from functools import lru_cache
import queue
q = queue.Queue()
COOKIES = {
'buvid3': 'A9D9FD41-2F2F-F3CF-97C5-542B29E0987660281infoc',
'b_nut': '1668477260',
'i-wanna-go-back': '-1',
'b_ut': '7',
'_uuid': 'D2EC4D61-E4BE-3C64-38102-9FE1988447C161515infoc',
'buvid4': '75898C1F-9A03-059F-BABE-A245AC20E09D38213-022060821-VGouk5e+pYiR45pgchAoBw%3D%3D',
'buvid_fp': '8882ffafd2e8f96a039dd6cdf729ad22',
'CURRENT_FNVAL': '4048',
'rpdid': "|(u))umm)~JJ0J'uY~|)kk)YJ",
'CURRENT_QUALITY': '32',
}
HEADERS = {
'authority': 'api.bilibili.com',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'accept-language': 'zh-CN,zh;q=0.9',
'cache-control': 'max-age=0',
'sec-ch-ua': '"Not?A_Brand";v="8", "Chromium";v="108", "Google Chrome";v="108"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36',
}
def get_score(type, media_id=4315402, ps=20, sort=0, cursor='', **kwargs):
params = {
'media_id': media_id,
'ps': ps,
'sort': sort,
'cursor': cursor,
**kwargs,
}
scores = 0
# @lru_cache()
def fetch_data(params_tuple):
params = dict(params_tuple)
response = requests.get(f'https://api.bilibili.com/pgc/review/{type}/list', params=params, cookies=COOKIES, headers=HEADERS)
data = json.loads(response.text)
return data['data']['list'], data['data']['next']
total = 0
pbar = tqdm(total=total)
while True:
reviews, cursor = fetch_data(tuple(sorted(params.items())))
if not reviews:
break
total += len(reviews)
for review in reviews:
scores += review["score"]
params['cursor'] = cursor
pbar.update(len(reviews))
pbar.close()
return scores, total
def get_short_scores(q):
short_scores, short_total = get_score('short')
q.put((short_scores, short_total))
def get_long_scores(q):
long_scores, long_total = get_score('long')
q.put((long_scores, long_total))
def test_get_score():
# short_scores, short_total = get_score('short')
# long_scores, long_total = get_score('long')
# avg = (short_scores + long_scores) / (short_total + long_total)
# print(avg)
thread1 = threading.Thread(target=get_short_scores, args=(q,))
thread2 = threading.Thread(target=get_long_scores, args=(q,))
thread1.start()
thread2.start()
thread1.join()
thread2.join()
short_scores, short_total = q.get()
long_scores, long_total = q.get()
scores = short_scores + long_scores
total = short_total + long_total
avg = scores / total
print("真实评分：", avg)
test_get_score()
bilibili:控分成功
发表于 2023-1-5 12:51
