q792602257/XiguaLiveDanmakuHelper
Archived
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

重新修正了B站无法上传多P的问题

Browse Source
This commit is contained in:
Jerry Yan 2022-02-01 23:32:41 +08:00
parent b1f45ee90d
commit 1feed57b4e
13 changed files with 1199 additions and 557 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
Common.py
View File

@ -6,12 +6,12 @@ import psutil
from api import XiGuaLiveApi
import json
import threading
from bilibili import *
from bilibili import Bilibili, VideoPart
# 默认设置
config = {
# 录像的主播名
"l_u": "永恒de草薙",
"l_u": "97621754276",
"b_u": "自己的B站账号",
"b_p": "自己的B站密码",
# 标题及预留时间位置
@ -302,18 +302,14 @@ def loginBilibili(force=False):
global b
if getTimeDelta(datetime.now(), loginTime) < 86400 * 10 and not force:
return False
if os.path.exists('cookie'):
try:
with open('cookie', 'r', encoding='utf8') as f:
_cookie = f.readline().strip()
b = Bilibili(_cookie)
loginTime = datetime.now()
appendOperation("Cookie 登录")
return True
except Exception as e:
appendError(e)
appendOperation("Cookie 登录失败")
return False
try:
b.login()
loginTime = datetime.now()
return True
except Exception as e:
appendError(e)
appendOperation("登录失败")
return False
else:
appendOperation("设置了不上传，所以不会登陆")
@ -373,7 +369,7 @@ def uploadVideo(name):
loginBilibili()
doClean()
if forceNotUpload is False:
b.preUpload(VideoPart(name, os.path.basename(name)))
b.preUpload(VideoPart(title=name, path=os.path.basename(name)))
else:
appendUploadStatus("设置了不上传，所以[{}]不会上传了".format(name))
if not forceNotEncode:
@ -403,9 +399,9 @@ def encodeVideo(name):
_new_name = os.path.splitext(name)[0] + ".mp4"
_code = os.system(config["enc"].format(f=name, t=_new_name))
if _code != 0:
Common.appendError("Encode {} with Non-Zero Return.".format(name))
appendError("Encode {} with Non-Zero Return.".format(name))
return False
Common.modifyLastEncodeStatus("Encode >{}< Finished".format(name))
modifyLastEncodeStatus("Encode >{}< Finished".format(name))
uploadQueue.put(_new_name)

0
access_token Normal file
View File

497
bilibili.py
View File

@ -1,151 +1,19 @@
# coding=utf-8
import os
import re
import json as JSON
import Common
import rsa
import math
import base64
import hashlib
import requests
from urllib import parse
from requests.adapters import HTTPAdapter
from urllib3 import Retry
class VideoPart:
def __init__(self, path, title='', desc=''):
self.path = path
self.title = title
self.desc = desc
from bilibiliuploader import core, VideoPart
class Bilibili:
def __init__(self, cookie=None):
self.files = []
self.videos = []
self.session = requests.session()
self.session.keep_alive = False
if cookie:
self.session.headers["cookie"] = cookie
self.csrf = re.search('bili_jct=(.*?);', cookie).group(1)
self.mid = re.search('DedeUserID=(.*?);', cookie).group(1)
self.session.headers['Accept'] = 'application/json, text/javascript, */*; q=0.01'
self.session.headers['Referer'] = 'https://space.bilibili.com/{mid}/#!/'.format(mid=self.mid)
# session.headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36'
# session.headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8'
def __init__(self):
self.access_token = ""
self.session_id = ""
self.user_id = ""
self.parts = []
def login(self, user, pwd):
"""
:param user: username
:type user: str
:param pwd: password
:type pwd: str
:return: if success return True
else raise Exception
"""
APPKEY = '4409e2ce8ffd12b8'
ACTIONKEY = 'appkey'
BUILD = 101800
DEVICE = 'android_tv_yst'
MOBI_APP = 'android_tv_yst'
PLATFORM = 'android'
APPSECRET = '59b43e04ad6965f34319062b478f83dd'
def md5(s):
h = hashlib.md5()
h.update(s.encode('utf-8'))
return h.hexdigest()
def sign(s):
"""
:return: return sign
"""
return md5(s + APPSECRET)
def signed_body(body):
"""
:return: body which be added sign
"""
if isinstance(body, str):
return body + '&sign=' + sign(body)
elif isinstance(body, dict):
ls = []
for k, v in body.items():
ls.append(k + '=' + v)
body['sign'] = sign('&'.join(ls))
return body
def getkey():
"""
:return: hash, key
"""
r = self.session.post(
'https://passport.bilibili.com/api/oauth2/getKey',
signed_body({'appkey': APPKEY}),
)
# {"ts":1544152439,"code":0,"data":{"hash":"99c7573759582e0b","key":"-----BEGIN PUBLIC----- -----END PUBLIC KEY-----\n"}}
json = r.json()
data = json['data']
return data['hash'], data['key']
def access_token_2_cookie(access_token):
r = self.session.get(
'https://passport.bilibili.com/api/login/sso?' + \
signed_body(
'access_key={access_token}&appkey={appkey}&gourl=https%3A%2F%2Faccount.bilibili.com%2Faccount%2Fhome'
.format(access_token=access_token, appkey=APPKEY),
),
allow_redirects=False,
)
return r.cookies.get_dict(domain=".bilibili.com")
self.session.headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8'
h, k = getkey()
pwd = base64.b64encode(
rsa.encrypt(
(h + pwd).encode('utf-8'),
rsa.PublicKey.load_pkcs1_openssl_pem(k.encode())
)
)
user = parse.quote_plus(user)
pwd = parse.quote_plus(pwd)
r = self.session.post(
'https://passport.snm0516.aisee.tv/api/tv/login',
signed_body(
'appkey={appkey}&build={build}&captcha=&channel=master&'
'guid=XYEBAA3E54D502E37BD606F0589A356902FCF&mobi_app={mobi_app}&'
'password={password}&platform={platform}&token=5598158bcd8511e2&ts=0&username={username}'
.format(appkey=APPKEY, build=BUILD, platform=PLATFORM, mobi_app=MOBI_APP, username=user, password=pwd)),
)
json = r.json()
if json['code'] == -105:
# need captcha
raise Exception('TODO: login with captcha')
if json['code'] != 0:
raise Exception(r.text)
access_token = json['data']['token_info']['access_token']
cookie_dict = access_token_2_cookie(access_token)
cookie = '; '.join(
'%s=%s' % (k, v)
for k, v in cookie_dict.items()
)
self.session.headers["cookie"] = cookie
self.csrf = re.search('bili_jct=(.*?)(;|$)', cookie).group(1)
self.mid = re.search('DedeUserID=(.*?)(;|$)', cookie).group(1)
self.session.headers['Accept'] = 'application/json, text/javascript, */*; q=0.01'
self.session.headers['Referer'] = 'https://space.bilibili.com/{mid}/#!/'.format(mid=self.mid)
return True
def login(self):
from Common import appendOperation
with open("access_token", "r") as f:
self.access_token = f.read(64).strip()
self.session_id, self.user_id, expires = core.login_by_access_token(self.access_token)
appendOperation("B站登录，UID【{}】，过期时间【{}】".format(self.user_id, expires))
def upload(self,
parts,
@ -187,125 +55,21 @@ class Bilibili:
:param parts: e.g. VideoPart('part path', 'part title', 'part desc'), or [VideoPart(...), VideoPart(...)]
:type parts: VideoPart or list<VideoPart>
"""
self.session.headers['Content-Type'] = 'application/json; charset=utf-8'
from Common import appendUploadStatus, modifyLastUploadStatus
if not isinstance(parts, list):
parts = [parts]
# retry by status
retries = Retry(
total=max_retry,
backoff_factor=1,
status_forcelist=(504, ),
)
self.session.mount('https://', HTTPAdapter(max_retries=retries))
self.session.mount('http://', HTTPAdapter(max_retries=retries))
#
def log_status(video_part: VideoPart, chunks_index: int, chunks_num: int):
modifyLastUploadStatus("Uploading >{}< @ {:.2f}%".format(video_part.path, 100.0 * chunks_index / chunks_num))
for part in parts:
filepath = part.path
filename = os.path.basename(filepath)
filesize = os.path.getsize(filepath)
Common.appendUploadStatus("Upload >{}< Started".format(filepath))
self.files.append(part)
r = self.session.get('https://member.bilibili.com/preupload?'
'os=upos&upcdn=ws&name={name}&size={size}&r=upos&profile=ugcupos%2Fyb&ssl=0'
.format(name=parse.quote_plus(filename), size=filesize))
"""return example
{
"upos_uri": "upos://ugc/i181012ws18x52mti3gg0h33chn3tyhp.mp4",
"biz_id": 58993125,
"endpoint": "//upos-hz-upcdnws.acgvideo.com",
"endpoints": [
"//upos-hz-upcdnws.acgvideo.com",
"//upos-hz-upcdntx.acgvideo.com"
],
"chunk_retry_delay": 3,
"chunk_retry": 200,
"chunk_size": 4194304,
"threads": 2,
"timeout": 900,
"auth": "os=upos&cdn=upcdnws&uid=&net_state=4&device=&build=&os_version=&ak=×tamp=&sign=",
"OK": 1
}
"""
json = r.json()
upos_uri = json['upos_uri']
endpoint = json['endpoint']
auth = json['auth']
biz_id = json['biz_id']
chunk_size = json['chunk_size']
self.session.headers['X-Upos-Auth'] = auth # add auth header
r = self.session.post('https:{}/{}?uploads&output=json'.format(endpoint, upos_uri.replace('upos://', '')))
# {"upload_id":"72eb747b9650b8c7995fdb0efbdc2bb6","key":"\/i181012ws2wg1tb7tjzswk2voxrwlk1u.mp4","OK":1,"bucket":"ugc"}
json = r.json()
upload_id = json['upload_id']
with open(filepath, 'rb') as f:
chunks_num = math.ceil(filesize / chunk_size)
chunks_index = 0
chunks_data = f.read(chunk_size)
Common.modifyLastUploadStatus(
"Uploading >{}< @ {:.2f}%".format(filepath, 100.0 * chunks_index / chunks_num))
while True:
if not chunks_data:
break
def upload_chunk():
r = self.session.put('https:{endpoint}/{upos_uri}?'
'partNumber={part_number}&uploadId={upload_id}&chunk={chunk}&chunks={chunks}&size={size}&start={start}&end={end}&total={total}'
.format(endpoint=endpoint,
upos_uri=upos_uri.replace('upos://', ''),
part_number=chunks_index + 1, # starts with 1
upload_id=upload_id,
chunk=chunks_index,
chunks=chunks_num,
size=len(chunks_data),
start=chunks_index * chunk_size,
end=chunks_index * chunk_size + len(chunks_data),
total=filesize,
),
chunks_data,
)
return r
def retry_upload_chunk():
"""return :class:`Response` if upload success, else return None."""
for i in range(max_retry):
r = upload_chunk()
if r.status_code == 200:
return r
Common.modifyLastUploadStatus(
"Uploading >{}< @ {:.2f}% RETRY[{}]".format(filepath, 100.0 * chunks_index / chunks_num, max_retry))
return None
r = retry_upload_chunk()
if r:
Common.modifyLastUploadStatus(
"Uploading >{}< @ {:.2f}%".format(filepath, 100.0 * chunks_index / chunks_num))
else:
Common.modifyLastUploadStatus(
"Uploading >{}< FAILED @ {:.2f}%".format(filepath, 100.0 * chunks_index / chunks_num))
continue
chunks_data = f.read(chunk_size)
chunks_index += 1 # start with 0
# NOT DELETE! Refer to https://github.com/comwrg/bilibiliupload/issues/15#issuecomment-424379769
self.session.post('https:{endpoint}/{upos_uri}?'
'output=json&name={name}&profile=ugcupos%2Fyb&uploadId={upload_id}&biz_id={biz_id}'
.format(endpoint=endpoint,
upos_uri=upos_uri.replace('upos://', ''),
name=filename,
upload_id=upload_id,
biz_id=biz_id,
),
{"parts": [{"partNumber": i, "eTag": "etag"} for i in range(1, chunks_num + 1)]},
)
self.videos.append({'filename': upos_uri.replace('upos://ugc/', '').split('.')[0],
'title': part.title,
'desc': part.desc})
Common.modifyLastUploadStatus("Upload >{}< Finished".format(filepath))
__f = open("uploaded.json", "w")
JSON.dump(self.videos, __f)
__f.close()
appendUploadStatus("Start Uploading >{}<".format(part.path))
while True:
status = core.upload_video_part(self.access_token, self.session_id, self.user_id, part, max_retry, cb=log_status)
if status:
break
# 上传完毕
modifyLastUploadStatus("Upload >{}< Finished；【{}】".format(part.path, part.server_file_name))
self.parts.append(part)
def finishUpload(self,
title,
@ -332,209 +96,26 @@ class Bilibili:
:type cover: str
:param no_reprint: (optional) 0=可以转载, 1=禁止转载(default)
:type no_reprint: int
:param copyright: (optional) 0=转载的, 1=自制的(default)
:type copyright: int
"""
if len(self.videos) == 0:
from Common import appendUploadStatus, modifyLastUploadStatus, appendError
if len(self.parts) == 0:
return
Common.appendUploadStatus("[{}]投稿中，请稍后".format(title))
self.session.headers['Content-Type'] = 'application/json; charset=utf-8'
appendUploadStatus("[{}]投稿中，请稍后".format(title))
copyright = 2 if source else 1
r = self.session.post('https://member.bilibili.com/x/vu/web/add?csrf=' + self.csrf,
json={
"copyright": copyright,
"source": source,
"title": title,
"tid": tid,
"tag": ','.join(tag),
"no_reprint": no_reprint,
"desc": desc,
"cover": cover,
"mission_id": 0,
"order_id": 0,
"videos": self.videos}
)
Common.modifyLastUploadStatus("[{}] Published | Result : {}".format(title, r.text))
try:
avid, bvid = core.upload(self.access_token, self.session_id, self.user_id, self.parts, copyright,
title=title, tid=tid, tag=','.join(tag), desc=desc, source=source, cover=cover, no_reprint=no_reprint)
modifyLastUploadStatus("[{}]投稿成功；AVID【{}】，BVID【{}】".format(title, avid, bvid))
self.clear()
except Exception as e:
modifyLastUploadStatus("[{}]投稿失败".format(title))
appendError(e)
def reloadFromPrevious(self):
if os.path.exists("uploaded.json"):
__f = open("uploaded.json", "r")
try:
self.videos = JSON.load(__f)
Common.appendUploadStatus("RELOAD SUCCESS")
except:
Common.appendUploadStatus("RELOAD Failed")
self.videos = []
__f.close()
os.remove("uploaded.json")
else:
Common.appendUploadStatus("RELOAD Failed")
self.videos = []
...
def clear(self):
self.files.clear()
self.videos.clear()
if (os.path.exists("uploaded.json")):
os.remove("uploaded.json")
def appendUpload(self,
aid,
parts,
title="",
tid="",
tag="",
desc="",
source='',
cover='',
no_reprint=1,
):
"""
:param aid: just aid
:type aid: int
:param parts: e.g. VideoPart('part path', 'part title', 'part desc'), or [VideoPart(...), VideoPart(...)]
:type parts: VideoPart or list<VideoPart>
:param title: video's title
:type title: str
:param tid: video type, see: https://member.bilibili.com/x/web/archive/pre
or https://github.com/uupers/BiliSpider/wiki/%E8%A7%86%E9%A2%91%E5%88%86%E5%8C%BA%E5%AF%B9%E5%BA%94%E8%A1%A8
:type tid: int
:param tag: video's tag
:type tag: list<str>
:param desc: video's description
:type desc: str
:param source: (optional) 转载地址
:type source: str
:param cover: (optional) cover's URL, use method *cover_up* to get
:type cover: str
:param no_reprint: (optional) 0=可以转载, 1=禁止转载(default)
:type no_reprint: int
"""
self.session.headers['Content-Type'] = 'application/json; charset=utf-8'
p = self.session.get("https://member.bilibili.com/x/web/archive/view?aid={}&history=".format(aid))
j = p.json()
if len(self.videos) == 0:
for i in j['data']['videos']:
self.videos.append({'filename': i['filename'],
'title': i["title"],
'desc': i["desc"]})
if (title == ""): title = j["data"]["archive"]['title']
if (tag == ""): tag = j["data"]["archive"]['tag']
if (no_reprint == ""): no_reprint = j["data"]["archive"]['no_reprint']
if (desc == ""): desc = j["data"]["archive"]['desc']
if (source == ""): source = j["data"]["archive"]['source']
if (tid == ""): tid = j["data"]["archive"]['tid']
self.preUpload(parts)
self.editUpload(aid, title, tid, tag, desc, source, cover, no_reprint)
def editUpload(self,
aid,
title,
tid,
tag,
desc,
source='',
cover='',
no_reprint=1,
):
"""
:param aid: just aid
:type aid: int
:param parts: e.g. VideoPart('part path', 'part title', 'part desc'), or [VideoPart(...), VideoPart(...)]
:type parts: VideoPart or list<VideoPart>
:param title: video's title
:type title: str
:param tid: video type, see: https://member.bilibili.com/x/web/archive/pre
or https://github.com/uupers/BiliSpider/wiki/%E8%A7%86%E9%A2%91%E5%88%86%E5%8C%BA%E5%AF%B9%E5%BA%94%E8%A1%A8
:type tid: int
:param tag: video's tag
:type tag: list<str>
:param desc: video's description
:type desc: str
:param source: (optional) 转载地址
:type source: str
:param cover: (optional) cover's URL, use method *cover_up* to get
:type cover: str
:param no_reprint: (optional) 0=可以转载, 1=禁止转载(default)
:type no_reprint: int
"""
copyright = 2 if source else 1
r = self.session.post('https://member.bilibili.com/x/vu/web/edit?csrf=' + self.csrf,
json={
"aid": aid,
"copyright": copyright,
"source": source,
"title": title,
"tid": tid,
"tag": ','.join(tag),
"no_reprint": no_reprint,
"desc": desc,
"cover": cover,
"mission_id": 0,
"order_id": 0,
"videos": self.videos}
)
print(r.text)
def addChannel(self, name, intro=''):
"""
:param name: channel's name
:type name: str
:param intro: channel's introduction
:type intro: str
"""
r = self.session.post(
url='https://space.bilibili.com/ajax/channel/addChannel',
data={
'name': name,
'intro': intro,
'aids': '',
'csrf': self.csrf,
},
# name=123&intro=123&aids=&csrf=565d7ed17cef2cc8ad054210c4e64324&_=1497077610768
)
# return
# {"status":true,"data":{"cid":"15812"}}
print(r.json())
def channel_addVideo(self, cid, aids):
"""
:param cid: channel's id
:type cid: int
:param aids: videos' id
:type aids: list<int>
"""
r = self.session.post(
url='https://space.bilibili.com/ajax/channel/addVideo',
data={
'aids': '%2C'.join(aids),
'cid': cid,
'csrf': self.csrf
}
# aids=9953555%2C9872953&cid=15814&csrf=565d7ed17cef2cc8ad054210c4e64324&_=1497079332679
)
print(r.json())
def cover_up(self, img):
"""
:param img: img path or stream
:type img: str or BufferedReader
:return: img URL
"""
if isinstance(img, str):
f = open(img, 'rb')
else:
f = img
r = self.session.post(
url='https://member.bilibili.com/x/vu/web/cover/up',
data={
'cover': b'data:image/jpeg;base64,' + (base64.b64encode(f.read())),
'csrf': self.csrf,
}
)
# print(r.text)
# {"code":0,"data":{"url":"http://i0.hdslb.com/bfs/archive/67db4a6eae398c309244e74f6e85ae8d813bd7c9.jpg"},"message":"","ttl":1}
return r.json()['data']['url']
self.parts = []

4
bilibiliuploader/README.md Normal file
View File

@ -0,0 +1,4 @@
修改自
https://github.com/FortuneDayssss/BilibiliUploader/
LICENSE：GPL

4
bilibiliuploader/__init__.py Normal file
View File

@ -0,0 +1,4 @@
from .bilibiliuploader import BilibiliUploader
from .core import VideoPart
__version__ = '0.0.6'

118
bilibiliuploader/bilibiliuploader.py Normal file
View File

@ -0,0 +1,118 @@
import bilibiliuploader.core as core
from bilibiliuploader.util import cipher
import json
class BilibiliUploader():
def __init__(self):
self.access_token = None
self.refresh_token = None
self.sid = None
self.mid = None
def login(self, username, password):
code, self.access_token, self.refresh_token, self.sid, self.mid, _ = core.login(username, password)
if code != 0: # success
print("login fail, error code = {}".format(code))
def login_by_access_token(self, access_token, refresh_token=None):
self.access_token = access_token
self.refresh_token = refresh_token
self.sid, self.mid, _ = core.login_by_access_token(access_token)
def login_by_access_token_file(self, file_name):
with open(file_name, "r") as f:
login_data = json.loads(f.read())
self.access_token = login_data["access_token"]
self.refresh_token = login_data["refresh_token"]
self.sid, self.mid, _ = core.login_by_access_token(self.access_token)
def save_login_data(self, file_name=None):
login_data = json.dumps(
{
"access_token": self.access_token,
"refresh_token": self.refresh_token
}
)
try:
with open(file_name, "w+") as f:
f.write(login_data)
finally:
return login_data
def upload(self,
parts,
copyright: int,
title: str,
tid: int,
tag: str,
desc: str,
source: str = '',
cover: str = '',
no_reprint: int = 0,
open_elec: int = 1,
max_retry: int = 5,
thread_pool_workers: int = 1):
return core.upload(self.access_token,
self.sid,
self.mid,
parts,
copyright,
title,
tid,
tag,
desc,
source,
cover,
no_reprint,
open_elec,
max_retry,
thread_pool_workers)
def edit(self,
avid=None,
bvid=None,
parts=None,
insert_index=None,
copyright=None,
title=None,
tid=None,
tag=None,
desc=None,
source=None,
cover=None,
no_reprint=None,
open_elec=None,
max_retry: int = 5,
thread_pool_workers: int = 1):
if not avid and not bvid:
print("please provide avid or bvid")
return None, None
if not avid:
avid = cipher.bv2av(bvid)
if not isinstance(parts, list):
parts = [parts]
if type(avid) is str:
avid = int(avid)
core.edit_videos(
self.access_token,
self.sid,
self.mid,
avid,
bvid,
parts,
insert_index,
copyright,
title,
tid,
tag,
desc,
source,
cover,
no_reprint,
open_elec,
max_retry,
thread_pool_workers
)

790
bilibiliuploader/core.py Normal file
View File

@ -0,0 +1,790 @@
import requests
from datetime import datetime
from bilibiliuploader.util import cipher as cipher
from urllib import parse
import os
import math
import hashlib
from bilibiliuploader.util.retry import Retry
from concurrent.futures import ThreadPoolExecutor, as_completed
import base64
# From PC ugc_assisstant
# APPKEY = 'aae92bc66f3edfab'
# APPSECRET = 'af125a0d5279fd576c1b4418a3e8276d'
APPKEY = '1d8b6e7d45233436'
APPSECRET = '560c52ccd288fed045859ed18bffd973'
LOGIN_APPKEY = '783bbb7264451d82'
# upload chunk size = 2MB
CHUNK_SIZE = 2 * 1024 * 1024
# captcha
CAPTCHA_RECOGNIZE_URL = "NOT SUPPORT"
class VideoPart:
"""
Video Part of a post.
每个对象代表一个分P
Attributes:
path: file path in local file system.
title: title of the video part.
desc: description of the video part.
server_file_name: file name in bilibili server. generated by pre-upload API.
"""
def __init__(self, path, title='', desc='', server_file_name=None):
self.path = path
self.title = title
self.desc = desc
self.server_file_name = server_file_name
def __repr__(self):
return '<{clazz}, path: {path}, title: {title}, desc: {desc}, server_file_name:{server_file_name}>' \
.format(clazz=self.__class__.__name__,
path=self.path,
title=self.title,
desc=self.desc,
server_file_name=self.server_file_name)
def get_key_old(sid=None, jsessionid=None):
"""
get public key, hash and session id for login.
Args:
sid: session id. only for captcha login.
jsessionid: j-session id. only for captcha login.
Returns:
hash: salt for password encryption.
pubkey: rsa public key for password encryption.
sid: session id.
"""
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': "application/json, text/javascript, */*; q=0.01"
}
post_data = {
'appkey': APPKEY,
'platform': "pc",
'ts': str(int(datetime.now().timestamp()))
}
post_data['sign'] = cipher.sign_dict(post_data, APPSECRET)
cookie = {}
if sid:
cookie['sid'] = sid
if jsessionid:
cookie['JSESSIONID'] = jsessionid
r = requests.post(
# "https://passport.bilibili.com/api/oauth2/getKey",
"https://passport.bilibili.com/x/passport-login/web/key",
headers=headers,
data=post_data,
cookies=cookie
)
print(r.content.decode())
r_data = r.json()['data']
if sid:
return r_data['hash'], r_data['key'], sid
return r_data['hash'], r_data['key'], r.cookies['sid']
def get_key():
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': "application/json, text/javascript, */*; q=0.01"
}
params_data = {
'appkey': LOGIN_APPKEY,
# 'ts': str(int(datetime.now().timestamp()))
}
params_data['sign'] = cipher.login_sign_dict_bin(params_data)
r = requests.get(
"https://passport.bilibili.com/x/passport-login/web/key",
headers=headers,
params=params_data
)
r_data = r.json()['data']
return r_data['hash'], r_data['key'], ''
def get_capcha(sid):
headers = {
'User-Agent': '',
'Accept-Encoding': 'gzip,deflate',
}
params = {
'appkey': APPKEY,
'platform': 'pc',
'ts': str(int(datetime.now().timestamp()))
}
params['sign'] = cipher.sign_dict(params, APPSECRET)
r = requests.get(
"https://passport.bilibili.com/captcha",
headers=headers,
params=params,
cookies={
'sid': sid
}
)
print(r.status_code)
capcha_data = r.content
return r.cookies['JSESSIONID'], capcha_data
def recognize_captcha(img: bytes):
img_base64 = str(base64.b64encode(img), encoding='utf-8')
r = requests.post(
url=CAPTCHA_RECOGNIZE_URL,
data={'image': img_base64}
)
return r.content.decode()
def login(username, password):
"""
bilibili login.
Args:
username: plain text username for bilibili.
password: plain text password for bilibili.
Returns:
code: login response code (0: success, -105: captcha error, ...).
access_token: token for further operation.
refresh_token: token for refresh access_token.
sid: session id.
mid: member id.
expires_in: access token expire time (30 days)
"""
hash, pubkey, sid = get_key()
encrypted_password = cipher.encrypt_login_password(password, hash, pubkey)
url_encoded_username = parse.quote_plus(username)
url_encoded_password = parse.quote_plus(encrypted_password)
post_data = {
'appkey': LOGIN_APPKEY,
'password': url_encoded_password,
'ts': str(int(datetime.now().timestamp())),
'username': url_encoded_username
}
post_data['sign'] = cipher.login_sign_dict_bin(post_data)
# avoid multiple url parse
post_data['username'] = username
post_data['password'] = encrypted_password
headers = {
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'User-Agent': '',
'Accept-Encoding': 'gzip,deflate',
}
r = requests.post(
# "https://passport.bilibili.com/api/v3/oauth2/login",
"https://passport.bilibili.com/x/passport-login/oauth2/login",
headers=headers,
data=post_data,
)
response = r.json()
response_code = response['code']
if response_code == 0:
login_data = response['data']['token_info']
return response_code, login_data['access_token'], login_data['refresh_token'], sid, login_data['mid'], \
login_data["expires_in"]
elif response_code == -105: # captcha error, retry=5
retry_cnt = 5
while response_code == -105 and retry_cnt > 0:
response_code, access_token, refresh_token, sid, mid, expire_in = login_captcha(username, password, sid)
if response_code == 0:
return response_code, access_token, refresh_token, sid, mid, expire_in
retry_cnt -= 1
# other error code
return response_code, None, None, sid, None, None
def login_captcha(username, password, sid):
"""
bilibili login with captcha.
depend on captcha recognize service, please do not use this as first choice.
Args:
username: plain text username for bilibili.
password: plain text password for bilibili.
sid: session id
Returns:
code: login response code (0: success, -105: captcha error, ...).
access_token: token for further operation.
refresh_token: token for refresh access_token.
sid: session id.
mid: member id.
expires_in: access token expire time (30 days)
"""
jsessionid, captcha_img = get_capcha(sid)
captcha_str = recognize_captcha(captcha_img)
hash, pubkey, sid = get_key(sid, jsessionid)
encrypted_password = cipher.encrypt_login_password(password, hash, pubkey)
url_encoded_username = parse.quote_plus(username)
url_encoded_password = parse.quote_plus(encrypted_password)
post_data = {
'appkey': APPKEY,
'captcha': captcha_str,
'password': url_encoded_password,
'platform': "pc",
'ts': str(int(datetime.now().timestamp())),
'username': url_encoded_username
}
post_data['sign'] = cipher.sign_dict(post_data, APPSECRET)
# avoid multiple url parse
post_data['username'] = username
post_data['password'] = encrypted_password
post_data['captcha'] = captcha_str
headers = {
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'User-Agent': '',
'Accept-Encoding': 'gzip,deflate',
}
r = requests.post(
"https://passport.bilibili.com/api/oauth2/login",
headers=headers,
data=post_data,
cookies={
'JSESSIONID': jsessionid,
'sid': sid
}
)
response = r.json()
if response['code'] == 0:
login_data = response['data']
return response['code'], login_data['access_token'], login_data['refresh_token'], sid, login_data['mid'], \
login_data["expires_in"]
else:
return response['code'], None, None, sid, None, None
def login_by_access_token(access_token):
"""
bilibili access token login.
Args:
access_token: Bilibili access token got by previous username/password login.
Returns:
sid: session id.
mid: member id.
expires_in: access token expire time
"""
headers = {
'Connection': 'keep-alive',
'Accept-Encoding': 'gzip,deflate',
'Host': 'passport.bilibili.com',
'User-Agent': '',
}
login_params = {
'appkey': APPKEY,
'access_token': access_token,
'platform': "pc",
'ts': str(int(datetime.now().timestamp())),
}
login_params['sign'] = cipher.sign_dict(login_params, APPSECRET)
r = requests.get(
url="https://passport.bilibili.com/api/oauth2/info",
headers=headers,
params=login_params
)
login_data = r.json()['data']
return r.cookies['sid'], login_data['mid'], login_data["expires_in"]
def upload_cover(access_token, sid, cover_file_path):
with open(cover_file_path, "rb") as f:
cover_pic = f.read()
headers = {
'Connection': 'keep-alive',
'Host': 'member.bilibili.com',
'Accept-Encoding': 'gzip,deflate',
'User-Agent': '',
}
params = {
"access_key": access_token,
}
params["sign"] = cipher.sign_dict(params, APPSECRET)
files = {
'file': ("cover.png", cover_pic, "Content-Type: image/png"),
}
r = requests.post(
"http://member.bilibili.com/x/vu/client/cover/up",
headers=headers,
params=params,
files=files,
cookies={
'sid': sid
},
verify=False,
)
return r.json()["data"]["url"]
def upload_chunk(upload_url, server_file_name, local_file_name, chunk_data, chunk_size, chunk_id, chunk_total_num):
"""
upload video chunk.
Args:
upload_url: upload url by pre_upload api.
server_file_name: file name on server by pre_upload api.
local_file_name: video file name in local fs.
chunk_data: binary data of video chunk.
chunk_size: default of ugc_assisstant is 2M.
chunk_id: chunk number.
chunk_total_num: total chunk number.
Returns:
True: upload chunk success.
False: upload chunk fail.
"""
print("chunk{}/{}".format(chunk_id, chunk_total_num))
print("filename: {}".format(local_file_name))
files = {
'version': (None, '2.0.0.1054'),
'filesize': (None, chunk_size),
'chunk': (None, chunk_id),
'chunks': (None, chunk_total_num),
'md5': (None, cipher.md5_bytes(chunk_data)),
'file': (local_file_name, chunk_data, 'application/octet-stream')
}
r = requests.post(
url=upload_url,
files=files,
cookies={
'PHPSESSID': server_file_name
},
)
print(r.status_code)
print(r.content)
if r.status_code == 200 and r.json()['OK'] == 1:
return True
else:
return False
def upload_video_part(access_token, sid, mid, video_part: VideoPart, max_retry=5, cb=None):
"""
upload a video file.
Args:
access_token: access token generated by login api.
sid: session id.
mid: member id.
video_part: local video file data.
max_retry: max retry number for each chunk.
cb: 回调
Returns:
status: success or fail.
server_file_name: server file name by pre_upload api.
"""
if cb is None:
cb = lambda f, c, t: None
if not isinstance(video_part, VideoPart):
return False
if video_part.server_file_name is not None:
return True
headers = {
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'User-Agent': '',
'Accept-Encoding': 'gzip,deflate',
}
r = requests.get(
"http://member.bilibili.com/preupload?access_key={}&mid={}&profile=ugcfr%2Fpc3".format(access_token, mid),
headers=headers,
cookies={
'sid': sid
},
verify=False,
)
pre_upload_data = r.json()
upload_url = pre_upload_data['url']
complete_upload_url = pre_upload_data['complete']
server_file_name = pre_upload_data['filename']
local_file_name = video_part.path
file_size = os.path.getsize(local_file_name)
chunk_total_num = int(math.ceil(file_size / CHUNK_SIZE))
file_hash = hashlib.md5()
with open(local_file_name, 'rb') as f:
for chunk_id in range(0, chunk_total_num):
chunk_data = f.read(CHUNK_SIZE)
cb(video_part, chunk_id, chunk_total_num)
status = Retry(max_retry=max_retry, success_return_value=True).run(
upload_chunk,
upload_url,
server_file_name,
os.path.basename(local_file_name),
chunk_data,
CHUNK_SIZE,
chunk_id,
chunk_total_num
)
if not status:
return False
file_hash.update(chunk_data)
print(file_hash.hexdigest())
# complete upload
post_data = {
'chunks': chunk_total_num,
'filesize': file_size,
'md5': file_hash.hexdigest(),
'name': os.path.basename(local_file_name),
'version': '2.0.0.1054',
}
r = requests.post(
url=complete_upload_url,
data=post_data,
headers=headers,
)
print(r.status_code)
print(r.content)
video_part.server_file_name = server_file_name
return True
def upload(access_token,
sid,
mid,
parts,
copyright: int,
title: str,
tid: int,
tag: str,
desc: str,
source: str = '',
cover: str = '',
no_reprint: int = 0,
open_elec: int = 1,
max_retry: int = 5,
thread_pool_workers: int = 1):
"""
upload video.
Args:
access_token: oauth2 access token.
sid: session id.
mid: member id.
parts: VideoPart list.
copyright: 原创/转载.
title: 投稿标题.
tid: 分区id.
tag: 标签.
desc: 投稿简介.
source: 转载地址.
cover: 封面图片文件路径.
no_reprint: 可否转载.
open_elec: 充电.
max_retry: max retry time for each chunk.
thread_pool_workers: max upload threads.
Returns:
(aid, bvid)
aid: av号
bvid: bv号
"""
if not isinstance(parts, list):
parts = [parts]
status = True
with ThreadPoolExecutor(max_workers=thread_pool_workers) as tpe:
t_list = []
for video_part in parts:
print("upload {} added in pool".format(video_part.title))
t_obj = tpe.submit(upload_video_part, access_token, sid, mid, video_part, max_retry)
t_obj.video_part = video_part
t_list.append(t_obj)
for t_obj in as_completed(t_list):
status = status and t_obj.result()
print("video part {} finished, status: {}".format(t_obj.video_part.title, t_obj.result()))
if not status:
print("upload failed")
return None, None
# cover
if os.path.isfile(cover):
try:
cover = upload_cover(access_token, sid, cover)
except:
cover = ''
else:
cover = ''
# submit
headers = {
'Connection': 'keep-alive',
'Content-Type': 'application/json',
'User-Agent': '',
}
post_data = {
'build': 1054,
'copyright': copyright,
'cover': cover,
'desc': desc,
'no_reprint': no_reprint,
'open_elec': open_elec,
'source': source,
'tag': tag,
'tid': tid,
'title': title,
'videos': []
}
for video_part in parts:
post_data['videos'].append({
"desc": video_part.desc,
"filename": video_part.server_file_name,
"title": video_part.title
})
params = {
'access_key': access_token,
}
params['sign'] = cipher.sign_dict(params, APPSECRET)
r = requests.post(
url="http://member.bilibili.com/x/vu/client/add",
params=params,
headers=headers,
verify=False,
cookies={
'sid': sid
},
json=post_data,
)
print("submit")
print(r.status_code)
print(r.content.decode())
data = r.json()["data"]
return data["aid"], data["bvid"]
def get_post_data(access_token, sid, avid):
headers = {
'Connection': 'keep-alive',
'Host': 'member.bilibili.com',
'Accept-Encoding': 'gzip,deflate',
'User-Agent': '',
}
params = {
"access_key": access_token,
"aid": avid,
"build": "1054"
}
params["sign"] = cipher.sign_dict(params, APPSECRET)
r = requests.get(
url="http://member.bilibili.com/x/client/archive/view",
headers=headers,
params=params,
cookies={
'sid': sid
}
)
return r.json()["data"]
def edit_videos(
access_token,
sid,
mid,
avid=None,
bvid=None,
parts=None,
insert_index=None,
copyright=None,
title=None,
tid=None,
tag=None,
desc=None,
source=None,
cover=None,
no_reprint=None,
open_elec=None,
max_retry: int = 5,
thread_pool_workers: int = 1):
"""
insert videos into existed post.
Args:
access_token: oauth2 access token.
sid: session id.
mid: member id.
avid: av number,
bvid: bv string,
parts: VideoPart list.
insert_index: new video index.
copyright: 原创/转载.
title: 投稿标题.
tid: 分区id.
tag: 标签.
desc: 投稿简介.
source: 转载地址.
cover: cover url.
no_reprint: 可否转载.
open_elec: 充电.
max_retry: max retry time for each chunk.
thread_pool_workers: max upload threads.
Returns:
(aid, bvid)
aid: av号
bvid: bv号
"""
if not avid and not bvid:
print("please provide avid or bvid")
return None, None
if not avid:
avid = cipher.bv2av(bvid)
if not isinstance(parts, list):
parts = [parts]
if type(avid) is str:
avid = int(avid)
post_video_data = get_post_data(access_token, sid, avid)
status = True
with ThreadPoolExecutor(max_workers=thread_pool_workers) as tpe:
t_list = []
for video_part in parts:
print("upload {} added in pool".format(video_part.title))
t_obj = tpe.submit(upload_video_part, access_token, sid, mid, video_part, max_retry)
t_obj.video_part = video_part
t_list.append(t_obj)
for t_obj in as_completed(t_list):
status = status and t_obj.result()
print("video part {} finished, status: {}".format(t_obj.video_part.title, t_obj.result()))
if not status:
print("upload failed")
return None, None
headers = {
'Connection': 'keep-alive',
'Content-Type': 'application/json',
'User-Agent': '',
}
submit_data = {
'aid': avid,
'build': 1054,
'copyright': post_video_data["archive"]["copyright"],
'cover': post_video_data["archive"]["cover"],
'desc': post_video_data["archive"]["desc"],
'no_reprint': post_video_data["archive"]["no_reprint"],
'open_elec': post_video_data["archive_elec"]["state"], # open_elec not tested
'source': post_video_data["archive"]["source"],
'tag': post_video_data["archive"]["tag"],
'tid': post_video_data["archive"]["tid"],
'title': post_video_data["archive"]["title"],
'videos': post_video_data["videos"]
}
# cover
if os.path.isfile(cover):
try:
cover = upload_cover(access_token, sid, cover)
except:
cover = ''
else:
cover = ''
# edit archive data
if copyright:
submit_data["copyright"] = copyright
if title:
submit_data["title"] = title
if tid:
submit_data["tid"] = tid
if tag:
submit_data["tag"] = tag
if desc:
submit_data["desc"] = desc
if source:
submit_data["source"] = source
if cover:
submit_data["cover"] = cover
if no_reprint:
submit_data["no_reprint"] = no_reprint
if open_elec:
submit_data["open_elec"] = open_elec
if type(insert_index) is int:
for i, video_part in enumerate(parts):
submit_data['videos'].insert(insert_index + i, {
"desc": video_part.desc,
"filename": video_part.server_file_name,
"title": video_part.title
})
elif insert_index is None:
for video_part in parts:
submit_data['videos'].append({
"desc": video_part.desc,
"filename": video_part.server_file_name,
"title": video_part.title
})
else:
print("wrong insert index")
return None, None
params = {
'access_key': access_token,
}
params['sign'] = cipher.sign_dict(params, APPSECRET)
r = requests.post(
url="http://member.bilibili.com/x/vu/client/edit",
params=params,
headers=headers,
verify=False,
cookies={
'sid': sid
},
json=submit_data,
)
print("edit submit")
print(r.status_code)
print(r.content.decode())
data = r.json()["data"]
return data["aid"], data["bvid"]

1
bilibiliuploader/util/__init__.py Normal file
View File

@ -0,0 +1 @@
from .cipher import *

119
bilibiliuploader/util/cipher.py Normal file
View File

@ -0,0 +1,119 @@
import hashlib
import rsa
import base64
import subprocess
import platform
import os.path
def md5(data: str):
"""
generate md5 hash of utf-8 encoded string.
"""
return hashlib.md5(data.encode("utf-8")).hexdigest()
def md5_bytes(data: bytes):
"""
generate md5 hash of binary.
"""
return hashlib.md5(data).hexdigest()
def sign_str(data: str, app_secret: str):
"""
sign a string of request parameters
Args:
data: string of request parameters, must be sorted by key before input.
app_secret: a secret string coupled with app_key.
Returns:
A hash string. len=32
"""
return md5(data + app_secret)
def sign_dict(data: dict, app_secret: str):
"""
sign a dictionary of request parameters
Args:
data: dictionary of request parameters.
app_secret: a secret string coupled with app_key.
Returns:
A hash string. len=32
"""
data_str = []
keys = list(data.keys())
keys.sort()
for key in keys:
data_str.append("{}={}".format(key, data[key]))
data_str = "&".join(data_str)
data_str = data_str + app_secret
return md5(data_str)
def login_sign_dict_bin(data: dict):
data_str = []
keys = list(data.keys())
keys.sort()
for key in keys:
data_str.append("{}={}".format(key, data[key]))
data_str = "&".join(data_str)
package_directory = os.path.dirname(os.path.abspath(__file__))
if platform.system().lower() == 'windows':
print(data_str)
print(subprocess.Popen([os.path.join(package_directory, "sign.exe"), data_str], stdout=subprocess.PIPE).communicate()[0].decode().strip())
return subprocess.Popen([os.path.join(package_directory, "sign.exe"), data_str], stdout=subprocess.PIPE).communicate()[0].decode().strip()
if platform.system().lower() == 'linux':
return subprocess.Popen([os.path.join(package_directory, "sign.out"), data_str], stdout=subprocess.PIPE).communicate()[0].decode().strip()
raise Exception("Operating System is not supported.")
def encrypt_login_password(password, hash, pubkey):
"""
encrypt password for login api.
Args:
password: plain text of user password.
hash: hash provided by /api/oauth2/getKey.
pubkey: public key provided by /api/oauth2/getKey.
Returns:
An encrypted cipher of password.
"""
return base64.b64encode(rsa.encrypt(
(hash + password).encode('utf-8'),
rsa.PublicKey.load_pkcs1_openssl_pem(pubkey.encode()),
))
def av2bv(av: int):
table = 'fZodR9XQDSUm21yCkr6zBqiveYah8bt4xsWpHnJE7jL5VG3guMTKNPAwcF'
tr = {}
for i in range(58):
tr[table[i]] = i
s = [11, 10, 3, 8, 4, 6]
xor = 177451812
add = 8728348608
av = (av ^ xor) + add
r = list('BV1 4 1 7 ')
for i in range(6):
r[s[i]] = table[av // 58 ** i % 58]
return ''.join(r)
def bv2av(bv: str):
table = 'fZodR9XQDSUm21yCkr6zBqiveYah8bt4xsWpHnJE7jL5VG3guMTKNPAwcF'
tr = {}
for i in range(58):
tr[table[i]] = i
s = [11, 10, 3, 8, 4, 6]
xor = 177451812
add = 8728348608
r = 0
for i in range(6):
r += tr[bv[s[i]]] * 58 ** i
return (r - add) ^ xor

18
bilibiliuploader/util/retry.py Normal file
View File

@ -0,0 +1,18 @@
class Retry:
def __init__(self, max_retry, success_return_value):
self.max_retry = max_retry
self.success_return_value = success_return_value
def run(self, func, *args, **kwargs):
status = False
for i in range(0, self.max_retry):
try:
return_value = func(*args, **kwargs)
except Exception:
return_value = not self.success_return_value
if return_value == self.success_return_value:
status = True
break
return status

BIN
bilibiliuploader/util/sign.exe Normal file
View File

Binary file not shown.

BIN
bilibiliuploader/util/sign.out Normal file
View File

Binary file not shown.

11
requirements.txt Normal file
View File

@ -0,0 +1,11 @@
psutil>=5.9.0
certifi>=2020.4.5.1
chardet>=3.0.4
idna>=2.9
pyasn1>=0.4.8
requests>=2.23.0
rsa>=4.0
urllib3>=1.25.9
flask>=2.0.2
flask_cors>=3.0.10
protobuf>=3.19.4