""" bilibili_api.audio_uploader 音频上传 """ import asyncio import os import json import time from enum import Enum from typing import List, Union, Optional from dataclasses import dataclass, field from . import user from .utils.upos import UposFile, UposFileUploader from .utils.utils import get_api, raise_for_statement from .utils.picture import Picture from .utils.AsyncEvent import AsyncEvent from .utils.credential import Credential from .exceptions.ApiException import ApiException from .utils.network import Api, get_session, HEADERS from .exceptions.NetworkException import NetworkException from enum import Enum _API = get_api("audio_uploader") class SongCategories: class ContentType(Enum): # cr_type """ 内容类型 + MUSIC: 音乐 + AUDIO_PROGRAM: 有声节目 """ MUSIC = 1 AUDIO_PROGRAM = 2 class CreationType(Enum): """ 创作类型 + ORIGINAL: 原创 + COVER: 翻唱/翻奏 + REMIX: 改编/remix """ ORIGINAL = 1 COVER = 2 REMIX = 48 class SongType(Enum): """ 声音类型 + HUMAN_SINGING: 人声演唱 + VOCALOID: VOCALOID歌手 + HUMAN_GHOST: 人力鬼畜 + PURE_MUSIC: 纯音乐/演奏 """ HUMAN_SINGING = 3 VOCALOID = 4 HUMAN_GHOST = 5 PURE_MUSIC = 6 class Style(Enum): """ 音乐风格 + POP: 流行 + ANCIENT: 古风 + ROCK: 摇滚 + FOLK: 民谣 + ELECTRONIC: 电子 + DANCE: 舞曲 + RAP: 说唱 + LIGHT_MUSIC: 轻音乐 + ACAPELLA: 阿卡贝拉 + JAZZ: 爵士 + COUNTRY: 乡村 + RNB_SOUL: R&B/Soul + CLASSICAL: 古典 + ETHNIC: 民族 + BRITISH: 英伦 + METAL: 金属 + PUNK: 朋克 + BLUES: 蓝调 + REGGAE: 雷鬼 + WORLD_MUSIC: 世界音乐 + LATIN: 拉丁 + ALTERNATIVE_INDEPENDENT: 另类/独立 + NEW_AGE: New Age + POST_ROCK: 后摇 + BOSSA_NOVA: Bossa Nova """ POP = 7 ANCIENT = 8 ROCK = 9 FOLK = 10 ELECTRONIC = 11 DANCE = 12 RAP = 13 LIGHT_MUSIC = 14 ACAPELLA = 15 JAZZ = 16 COUNTRY = 17 RNB_SOUL = 18 CLASSICAL = 19 ETHNIC = 20 BRITISH = 21 METAL = 22 PUNK = 23 BLUES = 24 REGGAE = 25 WORLD_MUSIC = 26 LATIN = 27 ALTERNATIVE_INDEPENDENT = 28 NEW_AGE = 29 POST_ROCK = 30 BOSSA_NOVA = 31 class Language(Enum): """ 语言 + CHINESE: 华语 + JAPANESE: 日语 + ENGLISH: 英语 + KOREAN: 韩语 + CANTONESE: 粤语 + OTHER_LANGUAGES: 其他语种 """ CHINESE = 32 JAPANESE = 33 ENGLISH = 34 KOREAN = 35 CANTONESE = 36 OTHER_LANGUAGES = 37 class Theme(Enum): """ 主题 + ANIMATION: 动画 + GAME: 游戏 + FILM_AND_TELEVISION: 影视 + INTERNET_SONG: 网络歌曲 + SECOND_CREATION: 同人 + IDOL: 偶像 """ ANIMATION = 38 GAME = 39 FILM_AND_TELEVISION = 40 INTERNET_SONG = 41 SECOND_CREATION = 42 IDOL = 43 class AudioType(Enum): """ 有声节目类型 + RADIO_DRAMA: 广播剧 + AUDIO_STORY: 有声故事 + OTHER: 其他 """ RADIO_DRAMA = 44 AUDIO_STORY = 45 OTHER = 47 class CompilationCategories: class SongType(Enum): """ 声音类型 + HUMAN_SINGING: 人声演唱 + VOCALOID_SINGER: VOCALOID歌手 + HUMAN_KICHUKU: 人力鬼畜 + PURE_MUSIC: 纯音乐 """ HUMAN_SINGING = 102 VOCALOID_SINGER = 103 HUMAN_KICHUKU = 104 PURE_MUSIC = 105 class CreationType(Enum): """ 创作类型 + ORIGINAL: 原创 + COVER: 翻唱/翻奏 + REMIX: 改编/remix """ ORIGINAL = 106 COVER = 107 REMIX = 108 class Language(Enum): """ 语种 + CHINESE: 中文 + JAPANESE: 日语 + ENGLISH: 英语 + KOREAN: 韩语 + CANTONESE: 粤语 + OTHER_LANGUAGES: 其他语种 """ CHINESE = 109 JAPANESE = 110 ENGLISH = 111 KOREAN = 112 CANTONESE = 113 OTHER_LANGUAGES = 114 class Theme(Enum): """ 主题来源 + GAME: 游戏 + ANIMATION: 动画 + FILM_AND_TELEVISION: 影视 + NETWORK_SONG: 网络歌曲 + DERIVATIVE_WORK: 同人 + IDOL: 偶像 """ ANIMATION = 115 GAME = 116 FILM_AND_TELEVISION = 117 NETWORK_SONG = 118 DERIVATIVE_WORK = 119 IDOL = 120 class Style(Enum): """ 风格 + POP: 流行 + ANCIENT_STYLE: 古风 + ROCK: 摇滚 + FOLK_SONG: 民谣 + ELECTRONIC: 电子 + DANCE_MUSIC: 舞曲 + RAP: 说唱 + LIGHT_MUSIC: 轻音乐 + A_CAPPELLA: 阿卡贝拉 + JAZZ: 爵士 + COUNTRY_MUSIC: 乡村 + R_AND_B: R&B/Soul + CLASSICAL: 古典 + CLASSICAL: 古典 + ETHNIC: 民族 + BRITISH: 英伦 + METAL: 金属 + PUNK: 朋克 + BLUES: 蓝调 + REGGAE: 雷鬼 + WORLD_MUSIC: 世界音乐 + LATIN: 拉丁 + ALTERNATIVE: 另类/独立 + NEW_AGE: New Age + POST_ROCK: 后摇 + BOSSA_NOVA: Bossa Nova """ POP = 121 ANCIENT_STYLE = 122 ROCK = 123 FOLK_SONG = 124 ELECTRONIC = 125 DANCE_MUSIC = 126 RAP = 127 LIGHT_MUSIC = 128 A_CAPPELLA = 129 JAZZ = 130 COUNTRY_MUSIC = 131 R_AND_B = 132 CLASSICAL = 133 ETHNIC = 134 BRITISH = 135 METAL = 136 PUNK = 137 BLUES = 138 REGGAE = 139 WORLD_MUSIC = 140 LATIN = 141 ALTERNATIVE = 142 NEW_AGE = 143 POST_ROCK = 144 BOSSA_NOVA = 145 class ContentType(Enum): """ 内容类型 + MUSIC: 音乐 + AUDIO_PROGRAM: 有声节目 """ MUSIC = 146 AUDIO_PROGRAM = 147 class AudioType(Enum): """ 声音类型 + RADIO_DRAMA: 广播剧 + AUDIO_STORY: 有声故事 + ASMR: ASMR + OTHER: 其他 """ RADIO_DRAMA = 148 AUDIO_STORY = 149 ASMR = 150 OTHER = 151 class AudioUploaderEvents(Enum): """ 上传事件枚举 Events: + PREUPLOAD 获取上传信息 + PREUPLOAD_FAILED 获取上传信息失败 + PRE_CHUNK 上传分块前 + AFTER_CHUNK 上传分块后 + CHUNK_FAILED 区块上传失败 + PRE_COVER 上传封面前 + AFTER_COVER 上传封面后 + COVER_FAILED 上传封面失败 + PRE_SUBMIT 提交音频前 + SUBMIT_FAILED 提交音频失败 + AFTER_SUBMIT 提交音频后 + COMPLETED 完成上传 + ABORTED 用户中止 + FAILED 上传失败 """ PREUPLOAD = "PREUPLOAD" PREUPLOAD_FAILED = "PREUPLOAD_FAILED" PRE_CHUNK = "PRE_CHUNK" AFTER_CHUNK = "AFTER_CHUNK" CHUNK_FAILED = "CHUNK_FAILED" PRE_COVER = "PRE_COVER" AFTER_COVER = "AFTER_COVER" COVER_FAILED = "COVER_FAILED" PRE_SUBMIT = "PRE_SUBMIT" SUBMIT_FAILED = "SUBMIT_FAILED" AFTER_SUBMIT = "AFTER_SUBMIT" COMPLETED = "COMPLETE" ABORTED = "ABORTED" FAILED = "FAILED" @dataclass class AuthorInfo: name: str uid: int = 0 @dataclass class SongMeta: """ content_type (SongCategories.ContentType): 内容类型 song_type (Union[SongCategories.SongType, SongCategories.AudioType]): 歌曲类型 creation_type (SongCategories.CreationType): 创作类型 language (Optional[SongCategories.Language]): 语言类型 theme (Optional[SongCategories.Theme]): 主题来源 style (Optional[SongCategories.Style]): 风格类型 singer (List[AuthorInfo]): 歌手 player (Optional[List[AuthorInfo]]): 演奏 sound_source (Optional[List[AuthorInfo]]): 音源 tuning (Optional[List[AuthorInfo]]): 调音 lyricist (Optional[List[AuthorInfo]]): 作词 arranger (List[AuthorInfo]): 编曲 composer (Optional[List[AuthorInfo]]): 作曲 mixer (Optional[str]): 混音 cover_maker (Optional[List[AuthorInfo]]): 封面制作者 instrument (Optional[List[str]]): 乐器 origin_url (Optional[str]): 原曲链接 origin_title (Optional[str]): 原曲标题 title (str): 标题 cover (Optional[Picture]): 封面 description (Optional[str]): 描述 tags (Union[List[str], str]): 标签 aid (Optional[int]): 视频 aid cid (Optional[int]): 视频 cid tid (Optional[int]): 视频 tid compilation_id (Optional[int]): 合辑 ID lrc (Optional[str]): 歌词 """ title: str desc: str tags: Union[List[str], str] content_type: SongCategories.ContentType song_type: Union[SongCategories.SongType, SongCategories.AudioType] creation_type: SongCategories.CreationType language: Optional[SongCategories.Language] = None theme: Optional[SongCategories.Theme] = None style: Optional[SongCategories.Style] = None singer: Optional[List[AuthorInfo]] = field(default_factory=list) player: Optional[List[AuthorInfo]] = field(default_factory=list) sound_source: Optional[List[AuthorInfo]] = field(default_factory=list) tuning: Optional[List[AuthorInfo]] = field(default_factory=list) lyricist: Optional[List[AuthorInfo]] = field(default_factory=list) arranger: Optional[List[AuthorInfo]] = field(default_factory=list) composer: Optional[List[AuthorInfo]] = field(default_factory=list) mixer: Optional[List[AuthorInfo]] = field(default_factory=list) cover_maker: Optional[List[AuthorInfo]] = field(default_factory=list) instrument: Optional[List[str]] = field(default_factory=list) origin_url: Optional[str] = None origin_title: Optional[str] = None cover: Optional[Picture] = None aid: Optional[int] = None cid: Optional[int] = None tid: Optional[int] = None lrc: Optional[str] = None compilation_id: Optional[int] = None is_bgm: bool = True class AudioUploader(AsyncEvent): """ 音频上传 """ __song_id: int __upos_file: UposFile __task: asyncio.Task def _check_meta(self): raise_for_statement(self.meta.content_type is not None) raise_for_statement(self.meta.song_type is not None) raise_for_statement(self.meta.cover is not None and isinstance(self.meta.cover, str)) if self.meta.content_type == SongCategories.ContentType.MUSIC: raise_for_statement(self.meta.creation_type is not None) raise_for_statement(self.meta.song_type is not None) raise_for_statement(self.meta.language is not None) if self.meta.song_type == SongCategories.SongType.HUMAN_SINGING: raise_for_statement(self.meta.singer is not None) raise_for_statement(self.meta.language is not None) elif self.meta.song_type in [ SongCategories.SongType.VOCALOID, SongCategories.SongType.HUMAN_GHOST, ]: raise_for_statement(self.meta.sound_source is not None) raise_for_statement(self.meta.language is not None) raise_for_statement(self.meta.tuning is not None) elif self.meta.song_type == SongCategories.SongType.PURE_MUSIC: raise_for_statement(self.meta.player is not None) if isinstance(self.meta.tags, str): self.meta.tags = self.meta.tags.split(",") raise_for_statement(len(self.meta.tags != 0)) raise_for_statement(self.meta.title is not None) raise_for_statement(self.meta.cover is not None) raise_for_statement(self.meta.desc is not None) def __init__(self, path: str, meta: SongMeta, credential: Credential): """ 初始化 Args: path (str): 文件路径 meta (AudioMeta): 元数据 credential (Credential): 账号信息 """ super().__init__() self.path = path self.meta = meta self.credential = credential self.__upos_file = UposFile(path) async def _preupload(self) -> dict: """ 分 P 上传初始化 Returns: dict: 初始化信息 """ self.dispatch(AudioUploaderEvents.PREUPLOAD.value, {"song": self.meta}) api = _API["preupload"] # 首先获取音频文件预检信息 session = get_session() resp = await session.get( api["url"], params={ "profile": "uga/bup", "name": os.path.basename(self.path), "size": self.__upos_file.size, "r": "upos", "ssl": "0", "version": "2.6.0", "build": 2060400, }, cookies=self.credential.get_cookies(), headers=HEADERS.copy(), ) if resp.status_code >= 400: self.dispatch( AudioUploaderEvents.PREUPLOAD_FAILED.value, {"song": self.meta} ) raise NetworkException(resp.status_code, resp.reason_phrase) preupload = resp.json() if preupload["OK"] != 1: self.dispatch( AudioUploaderEvents.PREUPLOAD_FAILED.value, {"song": self.meta} ) raise ApiException(json.dumps(preupload)) url = f'https:{preupload["endpoint"]}/{preupload["upos_uri"].removeprefix("upos://")}' headers = HEADERS.copy() headers["x-upos-auth"] = preupload["auth"] # 获取 upload_id resp = await session.post( url, headers=headers, params={ "uploads": "", "output": "json", "profile": "uga/bup", "filesize": self.__upos_file.size, "partsize": preupload["chunk_size"], "biz_id": preupload["biz_id"], }, ) if resp.status_code >= 400: self.dispatch( AudioUploaderEvents.PREUPLOAD_FAILED.value, {"song": self.meta} ) raise ApiException("获取 upload_id 错误") data = json.loads(resp.text) if data["OK"] != 1: self.dispatch( AudioUploaderEvents.PREUPLOAD_FAILED.value, {"song": self.meta} ) raise ApiException("获取 upload_id 错误：" + json.dumps(data)) preupload["upload_id"] = data["upload_id"] self.__song_id = preupload["biz_id"] return preupload async def _upload_cover(self, cover: str) -> str: return await upload_cover(cover, self.credential) async def _main(self): preupload = await self._preupload() await UposFileUploader(file=self.__upos_file, preupload=preupload).upload() if self.meta.lrc: lrc_url = await upload_lrc( song_id=self.__song_id, lrc=self.meta.lrc, credential=self.credential ) else: lrc_url = "" self.dispatch(AudioUploaderEvents.PRE_COVER) if self.meta.cover: try: cover_url = await self._upload_cover(self.meta.cover) except Exception as e: self.dispatch(AudioUploaderEvents.COVER_FAILED, {"err": e}) raise e self.dispatch(AudioUploaderEvents.AFTER_COVER.value, cover_url) self.dispatch(AudioUploaderEvents.PRE_SUBMIT.value) try: result = await self._submit(lrc_url=lrc_url, cover_url=cover_url) except Exception as e: self.dispatch(AudioUploaderEvents.SUBMIT_FAILED.value, {"err": e}) raise e self.dispatch(AudioUploaderEvents.AFTER_SUBMIT.value, result) return result async def _submit(self, cover_url: str, lrc_url: str = "") -> int: uploader = await user.get_self_info(self.credential) data = { "lyric_url": lrc_url, "cover_url": cover_url, "song_id": self.__song_id, "mid": uploader["mid"], "cr_type": self.meta.content_type.value, "creation_type_id": self.meta.creation_type.value, "music_type_id": self.meta.song_type.value, "style_type_id": self.meta.style.value if self.meta.style else 0, "theme_type_id": self.meta.theme.value if self.meta.theme else 0, "language_type_id": self.meta.language.value if self.meta.language else 0, "origin_title": self.meta.origin_title if self.meta.origin_title else "", "origin_url": self.meta.origin_url if self.meta.origin_url else "", "avid": self.meta.aid if self.meta.aid else "", "tid": self.meta.tid if self.meta.tid else "", "cid": self.meta.cid if self.meta.cid else "", "compilation_id": self.meta.compilation_id if self.meta.compilation_id else "", "title": self.meta.title, "intro": self.meta.desc, "member_with_type": [ { "m_type": 1, # 歌手 "members": [ {"name": singer.name, "mid": singer.uid} for singer in self.meta.singer ], }, { "m_type": 2, # 作词 "members": [ {"name": lyricist.name, "mid": lyricist.uid} for lyricist in self.meta.lyricist ], }, { "m_type": 3, "members": [ {"name": composer.name, "mid": composer.uid} for composer in self.meta.composer ], }, # 作曲 { "m_type": 4, "members": [ {"name": arranger.name, "mid": arranger.uid} for arranger in self.meta.arranger ], }, # 编曲 { "m_type": 5, "members": [ {"name": mixer.name, "mid": mixer.uid} for mixer in self.meta.mixer ], }, # 混音只能填一个人，你问我为什么我不知道 { "m_type": 6, "members": [ {"name": cover_maker.name, "mid": cover_maker.uid} for cover_maker in self.meta.cover_maker ], }, # 本家作者 { "m_type": 7, "members": [ {"name": cover_maker.name, "mid": cover_maker.uid} for cover_maker in self.meta.cover_maker ], }, # 封面 { "m_type": 8, "members": [ {"name": sound_source.name, "mid": sound_source.uid} for sound_source in self.meta.sound_source ], }, # 音源 { "m_type": 9, "members": [ {"name": tuning.name, "mid": tuning.uid} for tuning in self.meta.tuning ], }, # 调音 { "m_type": 10, "members": [ {"name": player.name, "mid": player.uid} for player in self.meta.player ], }, # 演奏 { "m_type": 11, "members": [ {"name": instrument} for instrument in self.meta.instrument ], }, # 乐器 { "m_type": 127, "members": [{"name": uploader["name"], "mid": uploader["mid"]}], }, # 上传者 ], "song_tags": [{"tagName": tag_name} for tag_name in self.meta.tags], "create_time": "%.3f" % time.time(), "activity_id": 0, "is_bgm": 1 if self.meta.is_bgm else 0, "source": 0, "album_id": 0, } api = _API["submit_single_song"] return ( await Api(**api, credential=self.credential, json_body=True, no_csrf=True) .update_data(**data) .result ) async def start(self) -> dict: """ 开始上传 """ task = asyncio.create_task(self._main()) self.__task = task try: result = await task self.__task = None return result except asyncio.CancelledError: # 忽略 task 取消异常 pass except Exception as e: self.dispatch(AudioUploaderEvents.FAILED.value, {"err": e}) raise e async def abort(self): """ 中断更改 """ if self.__task: self.__task.cancel("用户手动取消") self.dispatch(AudioUploaderEvents.ABORTED.value, None) async def upload_lrc(lrc: str, song_id: int, credential: Credential) -> str: """ 上传 LRC 歌词 Args: lrc (str): 歌词 credential (Credential): 凭据 """ api = _API["lrc"] data = {"song_id": song_id, "lrc": lrc} return await Api(**api, credential=credential).update_data(**data).result async def get_upinfo(param: Union[int, str], credential: Credential) -> List[dict]: """ 获取 UP 信息 Args: param (Union[int, str]): UP 主 ID 或者用户名 credential (Credential): 凭据 """ api = _API["upinfo"] data = {"param": param} return await Api(**api, credential=credential).update_data(**data).result async def upload_cover(cover: Picture, credential: Credential) -> str: api = _API["image"] # 小于 3MB raise_for_statement(os.path.getsize(cover) < 1024 * 1024 * 3, "3MB size limit") # 宽高比 1:1 raise_for_statement(cover.width == cover.height, "width == height, 600 * 600 recommanded") files = {"file": cover.content} return await Api(**api, credential=credential).update_files(**files).result