初始提交
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
try:
|
||||
from Provider import SessionProvider, SongStatus, AccountType
|
||||
except:
|
||||
from Extensions.Provider import SessionProvider, SongStatus, AccountType
|
||||
import httpx
|
||||
from functools import lru_cache
|
||||
import re
|
||||
from enum import Enum
|
||||
from bilibili_api import user, login_func, sync, login, Credential
|
||||
from pyzbar import pyzbar
|
||||
from PIL import Image
|
||||
import qrcode
|
||||
import time
|
||||
|
||||
class BilibiliSession(SessionProvider):
|
||||
sessionKey = None
|
||||
session = None
|
||||
providerName = "Bilibili"
|
||||
URLKeyword = ["b23.tv", "bilibili.com", "bilibili.tv"]
|
||||
accountType = AccountType.ACCOUNT_NOLOGIN
|
||||
|
||||
def login(self) -> dict:
|
||||
status, self.session = login_func.check_qrcode_events(self.sessionKey)
|
||||
print(status)
|
||||
if status == login_func.QrCodeLoginEvents.SCAN:
|
||||
return {"status": "400", "msg": "请扫描二维码"}, 400
|
||||
elif status == login_func.QrCodeLoginEvents.CONF:
|
||||
return {"status": "400", "msg": "请在手机上确认登录"}, 400
|
||||
elif status == login_func.QrCodeLoginEvents.TIMEOUT:
|
||||
return {"status": "400", "msg": "二维码已过期，请重新获取"}, 400
|
||||
elif status == login_func.QrCodeLoginEvents.DONE:
|
||||
self.status.cache_clear()
|
||||
print("success!")
|
||||
return {"status": "200", "msg": "登录成功"}, 200
|
||||
else:
|
||||
return {"status": "400", "msg": "未知错误"}, 400
|
||||
|
||||
@lru_cache(maxsize=8)
|
||||
def status(self) -> dict:
|
||||
return sync(user.get_self_info(self.session))
|
||||
|
||||
def userStatus(self) -> dict:
|
||||
return self.status()
|
||||
|
||||
def logout(self) -> dict:
|
||||
return {"status": 201, "msg": "无法登出，直接换帐号就好了"}, 201
|
||||
|
||||
def qrcode(self):
|
||||
_session = login.update_qrcode_data()
|
||||
loginURL = _session["url"]
|
||||
self.sessionKey = _session["qrcode_key"]
|
||||
qrcode.make(loginURL).save("/tmp/qrcode.png")
|
||||
return loginURL
|
||||
|
||||
if __name__ == "__main__":
|
||||
bisess = BilibiliSession()
|
||||
print(bisess.qrcode())
|
||||
while(True):
|
||||
if bisess.login()[1] == 200:
|
||||
break
|
||||
time.sleep(2)
|
||||
print(bisess.userStatus())
|
||||
@@ -0,0 +1,160 @@
|
||||
from pyncm import apis as ncm
|
||||
import pyncm
|
||||
try:
|
||||
from Provider import SessionProvider, SongStatus, AccountType
|
||||
except:
|
||||
from Extensions.Provider import SessionProvider, SongStatus, AccountType
|
||||
import httpx
|
||||
from functools import lru_cache
|
||||
import re
|
||||
from enum import Enum
|
||||
|
||||
class NeteaseSession(SessionProvider):
|
||||
session = None
|
||||
providerName = "Netease"
|
||||
URLKeyword = ["music.163.com", "163cn.tv", "网易云", "netease"]
|
||||
accountType = AccountType.ACCOUNT_NOLOGIN
|
||||
|
||||
def updateAccountType(self) -> AccountType:
|
||||
response = ncm.login.GetCurrentLoginStatus()["account"]
|
||||
if not response:
|
||||
self.accountType = AccountType.ACCOUNT_NOLOGIN
|
||||
return
|
||||
if response["paidFee"]:
|
||||
self.accountType = AccountType.ACCOUNT_VIP
|
||||
else:
|
||||
self.accountType = AccountType.ACCOUNT_USER
|
||||
|
||||
def login(self) -> dict:
|
||||
self.status.cache_clear()
|
||||
status = ncm.login.LoginQrcodeCheck(self.session)
|
||||
if status["code"] == 803:
|
||||
ncm.login.WriteLoginInfo(ncm.login.GetCurrentLoginStatus())
|
||||
self.updateAccountType()
|
||||
return {"status": 200, "msg": _("Netease account logged in.")}, 200
|
||||
else:
|
||||
return {"status": 408, "msg": _("Netease account login failed.")}, 408
|
||||
|
||||
def qrcode(self):
|
||||
self.session = ncm.login.LoginQrcodeUnikey()["unikey"]
|
||||
loginURL = f"https://music.163.com/login?codekey={self.session}"
|
||||
#return self._qrcode(loginURL)
|
||||
return loginURL
|
||||
|
||||
def logout(self) -> dict:
|
||||
self.status.cache_clear()
|
||||
ncm.login.LoginLogout()
|
||||
self.accountType = AccountType.ACCOUNT_NOLOGIN
|
||||
return {"status": 200, "msg": _("Netease account logged out.")}, 200
|
||||
|
||||
@lru_cache(maxsize=8)
|
||||
def status(self) -> dict:
|
||||
return ncm.login.GetCurrentLoginStatus()
|
||||
|
||||
def userStatus(self) -> dict:
|
||||
try:
|
||||
status = self.status()
|
||||
if self.accountType == AccountType.ACCOUNT_NOLOGIN:
|
||||
return {"status": 200, "accountType": self.accountType, "username": "未登录"}
|
||||
if self.accountType == AccountType.ACCOUNT_ANONYMOUS:
|
||||
return {"status": 200, "accountType": self.accountType, "username": "匿名用户"}
|
||||
return {"status": 200, "accountType": self.accountType, "username": status["profile"]["nickname"]}
|
||||
except Exception as e:
|
||||
print(e)
|
||||
return {"status": 500, "msg": _("Get status failed, internal server error!")}
|
||||
|
||||
def loginAnonymous(self) -> dict:
|
||||
self.status.cache_clear()
|
||||
try:
|
||||
ncm.login.LoginViaAnonymousAccount()
|
||||
self.accountType = AccountType.ACCOUNT_ANONYMOUS
|
||||
return {"status": 200, "msg": _("Netease anonymous account logged in.")}, 200
|
||||
except:
|
||||
return {"status": 408, "msg": _("Netease anonymous account login failed, timed out.")}, 408
|
||||
|
||||
def checkURL(self, content: str) -> bool:
|
||||
for i in self.URLKeyword:
|
||||
if i in content:
|
||||
return True
|
||||
return False
|
||||
|
||||
def parseURL(self, content: str) -> int or None:
|
||||
uid = None
|
||||
try:
|
||||
if "music.163.com" in content:
|
||||
uid = re.search(r"song\?id=[0-9]+",content).group()
|
||||
uid = int(uid[8:])
|
||||
elif "网易云" in content:
|
||||
uid = re.search(r"song\?id=[0-9]+",content).group()
|
||||
uid = int(uid[8:])
|
||||
elif "163cn.tv" in content:
|
||||
uid = httpx.get(content).headers.get("location")
|
||||
uid = re.search(r"song\?id=[0-9]+",uid).group()
|
||||
uid = int(uid[8:])
|
||||
elif type(eval(content)) == type(1):
|
||||
uid = eval(content)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
return None
|
||||
return uid
|
||||
|
||||
def checkSong(self, content: int) -> SongStatus:
|
||||
response = self._getSongStatus(content)
|
||||
try:
|
||||
if len(response["songs"]) == 0:
|
||||
return SongStatus.SONG_NOTEXIST
|
||||
if response["songs"][0]["fee"] == 1:
|
||||
return SongStatus.SONG_ISVIP
|
||||
return SongStatus.SONG_EXIST
|
||||
except:
|
||||
return SongStatus.SONG_NOTEXIST
|
||||
return SongStatus.SONG_NOTEXIST
|
||||
|
||||
def getSongURL(self, content: int) -> str:
|
||||
response = ncm.track.GetTrackAudio(content)
|
||||
return response["data"][0]["url"]
|
||||
|
||||
def getSongImage(self, content: int) -> str:
|
||||
response = self._getSongStatus(content)
|
||||
return response["songs"][0]["al"]["picUrl"]
|
||||
|
||||
def getSongName(self, content: int) -> str:
|
||||
response = self._getSongStatus(content)
|
||||
name = response["songs"][0]["name"] + "-"
|
||||
for i in response["songs"][0]["ar"]:
|
||||
name += i["name"]
|
||||
name += "/"
|
||||
return name[:-1]
|
||||
|
||||
def getSongLyric(self, content: int) -> str:
|
||||
#print(self._getSongStatus(content))
|
||||
response = ncm.track.GetTrackLyrics(content)
|
||||
return response['lrc']['lyric']
|
||||
|
||||
def dumpSession(self) -> str:
|
||||
return pyncm.DumpSessionAsString(pyncm.GetCurrentSession())
|
||||
|
||||
def restoreSession(self, session: str):
|
||||
session = pyncm.LoadSessionFromString(session)
|
||||
pyncm.SetCurrentSession(session)
|
||||
|
||||
@lru_cache(maxsize=256)
|
||||
def _getSongStatus(self, content: int) -> dict:
|
||||
response = ncm.track.GetTrackDetail(content)
|
||||
return response
|
||||
|
||||
if __name__ == "__main__":
|
||||
ncmsess = NeteaseSession()
|
||||
print(ncmsess.parseURL("分享心华的单曲《一人行者》: https://y.music.163.com/m/song?id=526116050&userid=1578177131&dlt=0846 (来自@网易云音乐)"))
|
||||
# print(ncmsess.loginAnonymous())
|
||||
# print(ncmsess.dumpSession())
|
||||
# print(ncmsess.checkSong(1828241226))
|
||||
print(ncmsess.getSongLyric(30953009))
|
||||
# print(ncmsess.checkSong(2014291207))
|
||||
# print(ncmsess.checkSong(1901371647))
|
||||
# print(ncmsess.checkSong(1828241226))
|
||||
# print(ncmsess.getSongName(1828241226))
|
||||
# print(ncmsess.getSongImage(1828241226))
|
||||
# print(ncmsess.getSongURL(1828241226))
|
||||
# print(ncmsess.status())
|
||||
# print(ncmsess._getSongStatus.cache_info())
|
||||
@@ -0,0 +1,65 @@
|
||||
import qrcode
|
||||
from enum import Enum, StrEnum
|
||||
from abc import abstractmethod, ABC
|
||||
|
||||
class SongStatus(Enum):
|
||||
SONG_EXIST = 1
|
||||
SONG_NOTEXIST = 0
|
||||
SONG_NOTAVAIL = 2
|
||||
SONG_ISVIP = 3
|
||||
|
||||
class AccountType(StrEnum):
|
||||
ACCOUNT_NOLOGIN = "未登录"
|
||||
ACCOUNT_ANONYMOUS = "匿名用户"
|
||||
ACCOUNT_USER = "普通用户"
|
||||
ACCOUNT_VIP = "VIP用户"
|
||||
ACCOUNT_COOKIE = "Cookie登录"
|
||||
def __json__(self):
|
||||
return self.value
|
||||
|
||||
class SessionProvider():
|
||||
tempPath = None
|
||||
session = None
|
||||
|
||||
@abstractmethod
|
||||
def login(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def logout(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def qrcode(self):
|
||||
pass
|
||||
|
||||
def _qrcode(self, content):
|
||||
raise DeprecationWarning("Please use string instead.")
|
||||
if type(content)!=type("string"):
|
||||
raise ValueError("QRCode content should be string")
|
||||
img = qrcode.make(content)
|
||||
return img
|
||||
|
||||
@abstractmethod
|
||||
def status(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def loginAnonymous(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def checkSong(self, content: int):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def parseURL(self, content: str):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def dumpSession(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def restoreSession(self, session: str):
|
||||
pass
|
||||
@@ -0,0 +1,99 @@
|
||||
try:
|
||||
from Provider import SessionProvider, SongStatus, AccountType
|
||||
except:
|
||||
from Extensions.Provider import SessionProvider, SongStatus, AccountType
|
||||
from enum import Enum
|
||||
import httpx
|
||||
import re
|
||||
from functools import lru_cache
|
||||
|
||||
try:
|
||||
print(_)
|
||||
except:
|
||||
_ = lambda x: x
|
||||
|
||||
# 次奥，要手写API了
|
||||
class QQMusicSession(SessionProvider):
|
||||
session = None
|
||||
providerName = "QQMusic"
|
||||
URL_Keyword = ["y.qq.com"]
|
||||
accountType = AccountType.ACCOUNT_NOLOGIN
|
||||
cookies = {}
|
||||
def __init__(self):
|
||||
print(_("WARN: This extension is deprecated due to the change of APIkey of y.qq.com!"))
|
||||
|
||||
def loginAnonymous(self):
|
||||
return {"status": 501, "msg": _("QQMusic does not support anonymous login.")}, 501
|
||||
|
||||
def loginCredential(self, content: str, key: str):
|
||||
cookies = content
|
||||
return {"status": 200, "msg": _("Cookies updated.")}
|
||||
|
||||
def login(self) -> dict:
|
||||
if cookies:
|
||||
self.accountType = AccountType.ACCOUNT_COOKIE
|
||||
return {"status": 200, "msg": _("QQMusic login succeeded.")}
|
||||
else:
|
||||
return {"status": 401, "msg": _("Cookie required!")}, 401
|
||||
|
||||
def logout(self):
|
||||
self.accountType = AccountType.ACCOUNT_NOLOGIN
|
||||
self.cookies = ""
|
||||
return {"status": 200, "msg": _("QQMusic logout succeeded.")}
|
||||
|
||||
def parseURL(self, content: str) -> str or None:
|
||||
mid = None
|
||||
try:
|
||||
if "fcgi-bin/u?__" in content:
|
||||
content = re.search(r"https://.*", content).group()
|
||||
print(content)
|
||||
mid = httpx.get(content).headers.get("location")
|
||||
mid = re.search(r"songmid=(.+?)&", mid).group(1)
|
||||
elif "ryqq/songDetail" in content:
|
||||
mid = re.search(r"/[^/]+$", content).group()[1:]
|
||||
except Exception as e:
|
||||
print(e)
|
||||
return None
|
||||
return mid
|
||||
|
||||
def checkURL(self, content: str) -> bool:
|
||||
for i in self.URLKeyword:
|
||||
if i in content:
|
||||
return True
|
||||
return False
|
||||
|
||||
def getSongURL(self, content: str) -> str:
|
||||
payload = r"https://u.y.qq.com/cgi-bin/musicu.fcg?format=json&data={%22req_0%22:{%22module%22:%22vkey.GetVkeyServer%22,%22method%22:%22CgiGetVkey%22,%22param%22:{%22filename%22:[%22PREFIXSONGMIDSONGMID.SUFFIX%22],%22guid%22:%2210000%22,%22songmid%22:[%22SONGMID%22],%22songtype%22:[0],%22uin%22:%220%22,%22loginflag%22:1,%22platform%22:%2220%22}},%22loginUin%22:%220%22,%22comm%22:{%22uin%22:%220%22,%22format%22:%22json%22,%22ct%22:24,%22cv%22:0}}"
|
||||
payload = payload.replace("PREFIX", "M500")
|
||||
payload = payload.replace("SONGMID", content)
|
||||
payload = payload.replace("SUFFIX", "mp3")
|
||||
data = httpx.get(payload, cookies=self.cookies)
|
||||
print(payload)
|
||||
print(data.json())
|
||||
print(a:=data.json()["req_0"]["data"]["midurlinfo"][0]["purl"])
|
||||
print(b:=data.json()["req_0"]["data"]["sip"][0])
|
||||
return b+a
|
||||
|
||||
@lru_cache(maxsize=256)
|
||||
def _getSongStatus(self, content:str) -> dict:
|
||||
payload = r"https://u.y.qq.com/cgi-bin/musicu.fcg?format=json&data={%22req%22:{%22module%22:%22music.pf_song_detail_svr%22,%22method%22:%22get_song_detail_yqq%22,%22param%22:{%22songmid%22:%22SONGMID%22}},%22loginUin%22:%220%22,%22comm%22:{%22uin%22:%220%22,%22format%22:%22json%22,%22ct%22:24,%22cv%22:0}}"
|
||||
payload = payload.replace("SONGMID", content)
|
||||
data = httpx.get(payload,cookies=self.cookies )
|
||||
print(data.json())
|
||||
|
||||
def dumpSession(self) -> str:
|
||||
return self.cookies
|
||||
|
||||
def restoreSession(self, session: str):
|
||||
self.cookies = session
|
||||
self.login()
|
||||
|
||||
if __name__ == "__main__":
|
||||
cookies = "pac_uid=1_1599172402; iip=0; _qimei_fingerprint=a60e880d499fb09dfa3fc2e7b67f5cac; _qimei_q36=; _qimei_h38=5261d3497d310405f78852440900000df17a1e; RK=vx0t0LFmeo; ptcz=ad36d5ca2a0e7a6b7250c853a187f263446dfb2530128858c1c61c5fcc10875c; pgv_pvid=2920262853; eas_sid=d14770F6p8X5a1M6V2z6X8H9W1; qq_domain_video_guid_verify=c6c04001d0488e55; o_cookie=1599172402; suid=ek168787168502135101; _qimei_uuid42=18401162425100258014e7193532ca71547ede2d09; vversion_name=8.2.95; pgv_info=ssid=s4743344675; uin=1599172402; fqm_pvqid=d34c53e5-031d-4762-b0e1-10a3b13013b6; fqm_sessionid=12a505d3-2ba4-4fd8-b025-6d773e4153e6; ts_refer=cn.bing.com/; ts_uid=2582842080; ts_last=y.qq.com/n/ryqq/songDetail/002vbWwe3ZmffT; login_type=1; psrf_qqaccess_token=2BFFE211C6375E16CAE328C65066052A; qm_keyst=Q_H_L_63k3Naa-qgZKZwiv-Q6TaqMCUYqy0nrKuDW6p2NbmOYb0AYcnh1ttAqA9Az9tpu2D1FCrAkeJFQ; wxunionid=; euin=oK4qNK6lowvzoc**; psrf_qqrefresh_token=B75229616609EF6988510371BEF3EFA2; psrf_qqunionid=9A90D76C7D610B8DF336DC5F19EC8773; qqmusic_key=Q_H_L_63k3Naa-qgZKZwiv-Q6TaqMCUYqy0nrKuDW6p2NbmOYb0AYcnh1ttAqA9Az9tpu2D1FCrAkeJFQ; tmeLoginType=2; music_ignore_pskey=202306271436Hn@vBj; psrf_musickey_createtime=1714623437; wxrefresh_token=; psrf_access_token_expiresAt=1722399437; psrf_qqopenid=027F5F42CADC5D52F710B2936B81B0C5; wxopenid="
|
||||
qmsess = QQMusicSession()
|
||||
smid = qmsess.parseURL("https://c6.y.qq.com/base/fcgi-bin/u?__=yCNzlNSNbaWD")
|
||||
smid = qmsess.parseURL("https://y.qq.com/n/ryqq/songDetail/000lKV2G0PEkc3")
|
||||
print(qmsess.loginCredential(cookies, ""))
|
||||
print(qmsess.login())
|
||||
print(qmsess.getSongURL(smid))
|
||||
#print(qmsess._getSongStatus(smid))
|
||||
Reference in New Issue
Block a user