rogerxavier's picture
Upload 258 files
0aee47a verified
raw
history blame
21.6 kB
"""
bilibili_api.note
笔记相关
"""
import re
import json
from enum import Enum
from html import unescape
from typing import List, Union, overload
import yaml
import httpx
from yarl import URL
from .utils.initial_state import get_initial_state
from .utils.utils import get_api, raise_for_statement
from .utils.picture import Picture
from .utils.credential import Credential
from .exceptions import ApiException, ArgsException
from .utils.network import Api, get_session
from .video import get_cid_info_sync
from . import article
API = get_api("note")
API_ARTICLE = get_api("article")
class NoteType(Enum):
PUBLIC = "public"
PRIVATE = "private"
class Note:
"""
笔记相关
"""
def __init__(
self,
cvid: Union[int, None] = None,
aid: Union[int, None] = None,
note_id: Union[int, None] = None,
note_type: NoteType = NoteType.PUBLIC,
credential: Union[Credential, None] = None,
):
"""
Args:
cvid (int) : 公开笔记 ID (对应专栏的 cvid) (公开笔记必要)
aid (int) : 稿件 ID(oid_type 为 0 时是 avid) (私有笔记必要)
note_id (int) : 私有笔记 ID (私有笔记必要)
note_type (str) : 笔记类型 (private, public)
credential (Credential, optional) : Credential. Defaults to None.
"""
self.__oid = -1
self.__note_id = -1
self.__cvid = -1
# ID 和 type 检查
if note_type == NoteType.PRIVATE:
if not aid or not note_id:
raise ArgsException("私有笔记需要 oid 和 note_id")
self.__oid = aid
self.__note_id = note_id
elif note_type == NoteType.PUBLIC:
if not cvid:
raise ArgsException("公开笔记需要 cvid")
self.__cvid = cvid
else:
raise ArgsException("type_ 只能是 public 或 private")
self.__type = note_type
# 未提供 credential 时初始化该类
# 私有笔记需要 credential
self.credential: Credential = Credential() if credential is None else credential
# 用于存储视频信息,避免接口依赖视频信息时重复调用
self.__info: Union[dict, None] = None
# 用于存储正文的节点
self.__children: List[Node] = []
# 用于存储是否解析
self.__has_parsed: bool = False
# 用于存储转换为 markdown 和 json 时使用的信息
self.__meta: dict = {}
def get_cvid(self) -> int:
return self.__cvid
def get_aid(self) -> int:
return self.__oid
def get_note_id(self) -> int:
return self.__note_id
def turn_to_article(self) -> "article.Article":
"""
将笔记类转为专栏类。需要保证笔记是公开笔记。
Returns:
Note: 专栏类
"""
raise_for_statement(self.__type == NoteType.PUBLIC)
return article.Article(cvid=self.get_cvid(), credential=self.credential)
async def get_info(self) -> dict:
"""
获取笔记信息
Returns:
dict: 笔记信息
"""
if self.__type == NoteType.PRIVATE:
return await self.get_private_note_info()
else:
return await self.get_public_note_info()
async def __get_info_cached(self) -> dict:
"""
获取视频信息,如果已获取过则使用之前获取的信息,没有则重新获取。
Returns:
dict: 调用 API 返回的结果。
"""
if self.__info is None:
return await self.get_info()
return self.__info
async def get_private_note_info(self) -> dict:
"""
获取私有笔记信息。
Returns:
dict: 调用 API 返回的结果。
"""
raise_for_statement(self.__type == NoteType.PRIVATE)
api = API["private"]["detail"]
# oid 为 0 时指 avid
params = {"oid": self.get_aid(), "note_id": self.get_note_id(), "oid_type": 0}
resp = (
await Api(**api, credential=self.credential).update_params(**params).result
)
# 存入 self.__info 中以备后续调用
self.__info = resp
return resp
async def get_public_note_info(self) -> dict:
"""
获取公有笔记信息。
Returns:
dict: 调用 API 返回的结果。
"""
raise_for_statement(self.__type == NoteType.PUBLIC)
api = API["public"]["detail"]
params = {"cvid": self.get_cvid()}
resp = (
await Api(**api, credential=self.credential).update_params(**params).result
)
# 存入 self.__info 中以备后续调用
self.__info = resp
return resp
async def get_images_raw_info(self) -> List["dict"]:
"""
获取笔记所有图片原始信息
Returns:
list: 图片信息
"""
result = []
content = (await self.__get_info_cached())["content"]
for line in content:
if type(line["insert"]) == dict:
if "imageUpload" in line["insert"]:
img_info = line["insert"]["imageUpload"]
result.append(img_info)
return result
async def get_images(self) -> List["Picture"]:
"""
获取笔记所有图片并转为 Picture 类
Returns:
list: 图片信息
"""
result = []
images_raw_info = await self.get_images_raw_info()
for image in images_raw_info:
result.append(Picture().from_url(url=f'https:{image["url"]}'))
return result
async def get_all(self) -> dict:
"""
(仅供公开笔记)
一次性获取专栏尽可能详细数据,包括原始内容、标签、发布时间、标题、相关专栏推荐等
Returns:
dict: 调用 API 返回的结果
"""
raise_for_statement(self.__type == NoteType.PUBLIC)
return await get_initial_state(f"https://www.bilibili.com/read/cv{self.__cvid}")
async def set_like(self, status: bool = True) -> dict:
"""
(仅供公开笔记)
设置专栏点赞状态
Args:
status (bool, optional): 点赞状态. Defaults to True
Returns:
dict: 调用 API 返回的结果
"""
raise_for_statement(self.__type == NoteType.PUBLIC)
self.credential.raise_for_no_sessdata()
api = API_ARTICLE["operate"]["like"]
data = {"id": self.__cvid, "type": 1 if status else 2}
return await Api(**api, credential=self.credential).update_data(**data).result
async def set_favorite(self, status: bool = True) -> dict:
"""
(仅供公开笔记)
设置专栏收藏状态
Args:
status (bool, optional): 收藏状态. Defaults to True
Returns:
dict: 调用 API 返回的结果
"""
raise_for_statement(self.__type == NoteType.PUBLIC)
self.credential.raise_for_no_sessdata()
api = (
API_ARTICLE["operate"]["add_favorite"]
if status
else API_ARTICLE["operate"]["del_favorite"]
)
data = {"id": self.__cvid}
return await Api(**api, credential=self.credential).update_data(**data).result
async def add_coins(self) -> dict:
"""
(仅供公开笔记)
给笔记投币,目前只能投一个。
Returns:
dict: 调用 API 返回的结果
"""
raise_for_statement(self.__type == NoteType.PUBLIC)
self.credential.raise_for_no_sessdata()
upid = (await self.get_info())["mid"]
api = API_ARTICLE["operate"]["coin"]
data = {"aid": self.__cvid, "multiply": 1, "upid": upid, "avtype": 2}
return await Api(**api, credential=self.credential).update_data(**data).result
async def fetch_content(self) -> None:
"""
获取并解析笔记内容
该返回不会返回任何值,调用该方法后请再调用 `self.markdown()` 或 `self.json()` 来获取你需要的值。
"""
def parse_note(data: List[dict]):
for field in data:
if not isinstance(field["insert"], str):
if "tag" in field["insert"].keys():
node = VideoCardNode()
node.aid = get_cid_info_sync(field["insert"]["tag"]["cid"])[
"cid"
]
self.__children.append(node)
elif "imageUpload" in field["insert"].keys():
node = ImageNode()
node.url = field["insert"]["imageUpload"]["url"]
self.__children.append(node)
elif "cut-off" in field["insert"].keys():
node = ImageNode()
node.url = field["insert"]["cut-off"]["url"]
self.__children.append(node)
else:
raise Exception()
else:
node = TextNode(field["insert"])
if "attributes" in field.keys():
if field["attributes"].get("bold") == True:
bold = BoldNode()
bold.children = [node]
node = bold
if field["attributes"].get("strike") == True:
delete = DelNode()
delete.children = [node]
node = delete
if field["attributes"].get("underline") == True:
underline = UnderlineNode()
underline.children = [node]
node = underline
if field["attributes"].get("background") == True:
# FIXME: 暂不支持背景颜色
pass
if field["attributes"].get("color") != None:
color = ColorNode()
color.color = field["attributes"]["color"].replace("#", "")
color.children = [node]
node = color
if field["attributes"].get("size") != None:
size = FontSizeNode()
size.size = field["attributes"]["size"]
size.children = [node]
node = size
else:
pass
self.__children.append(node)
info = await self.get_info()
content = info["content"]
content = unescape(content)
parse_note(json.loads(content))
self.__has_parsed = True
self.__meta = await self.__get_info_cached()
del self.__meta["content"]
def markdown(self) -> str:
"""
转换为 Markdown
请先调用 fetch_content()
Returns:
str: Markdown 内容
"""
if not self.__has_parsed:
raise ApiException("请先调用 fetch_content()")
content = ""
for node in self.__children:
try:
markdown_text = node.markdown()
except Exception as e:
pass
else:
content += markdown_text
meta_yaml = yaml.safe_dump(self.__meta, allow_unicode=True)
content = f"---\n{meta_yaml}\n---\n\n{content}"
return content
def json(self) -> dict:
"""
转换为 JSON 数据
请先调用 fetch_content()
Returns:
dict: JSON 数据
"""
if not self.__has_parsed:
raise ApiException("请先调用 fetch_content()")
return {
"type": "Note",
"meta": self.__meta,
"children": list(map(lambda x: x.json(), self.__children)),
}
# TODO: 笔记上传/编辑/删除
class Node:
def __init__(self):
pass
@overload
def markdown(self) -> str: # type: ignore
pass
@overload
def json(self) -> dict: # type: ignore
pass
class ParagraphNode(Node):
def __init__(self):
self.children = []
self.align = "left"
def markdown(self):
content = "".join([node.markdown() for node in self.children])
return content + "\n\n"
def json(self):
return {
"type": "ParagraphNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class HeadingNode(Node):
def __init__(self):
self.children = []
def markdown(self):
text = "".join([node.markdown() for node in self.children])
if len(text) == 0:
return ""
return f"## {text}\n\n"
def json(self):
return {
"type": "HeadingNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class BlockquoteNode(Node):
def __init__(self):
self.children = []
def markdown(self):
t = "".join([node.markdown() for node in self.children])
# 填补空白行的 > 并加上标识符
t = "\n".join(["> " + line for line in t.split("\n")]) + "\n\n"
return t
def json(self):
return {
"type": "BlockquoteNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class ItalicNode(Node):
def __init__(self):
self.children = []
def markdown(self):
text = "".join([node.markdown() for node in self.children])
if len(text) == 0:
return ""
return f" *{text}*"
def json(self):
return {
"type": "ItalicNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class BoldNode(Node):
def __init__(self):
self.children = []
def markdown(self):
t = "".join([node.markdown() for node in self.children])
if len(t) == 0:
return ""
return f" **{t.lstrip().rstrip()}** "
def json(self):
return {
"type": "BoldNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class DelNode(Node):
def __init__(self):
self.children = []
def markdown(self):
text = "".join([node.markdown() for node in self.children])
if len(text) == 0:
return ""
return f" ~~{text}~~"
def json(self):
return {
"type": "DelNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class UnderlineNode(Node):
def __init__(self):
self.children = []
def markdown(self):
text = "".join([node.markdown() for node in self.children])
if len(text) == 0:
return ""
return " $\\underline{" + text + "}$ "
class UlNode(Node):
def __init__(self):
self.children = []
def markdown(self):
return "\n".join(["- " + node.markdown() for node in self.children])
def json(self):
return {
"type": "UlNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class OlNode(Node):
def __init__(self):
self.children = []
def markdown(self):
t = []
for i, node in enumerate(self.children):
t.append(f"{i + 1}. {node.markdown()}")
return "\n".join(t)
def json(self):
return {
"type": "OlNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class LiNode(Node):
def __init__(self):
self.children = []
def markdown(self):
return "".join([node.markdown() for node in self.children])
def json(self):
return {
"type": "LiNode",
"children": list(map(lambda x: x.json(), self.children)),
}
class ColorNode(Node):
def __init__(self):
self.color = "000000"
self.children = []
def markdown(self):
return "".join([node.markdown() for node in self.children])
def json(self):
return {
"type": "ColorNode",
"color": self.color,
"children": list(map(lambda x: x.json(), self.children)),
}
class FontSizeNode(Node):
def __init__(self):
self.size = 16
self.children = []
def markdown(self):
return "".join([node.markdown() for node in self.children])
def json(self):
return {
"type": "FontSizeNode",
"size": self.size,
"children": list(map(lambda x: x.json(), self.children)),
}
# 特殊节点,即无子节点
class TextNode(Node):
def __init__(self, text: str):
self.text = text
def markdown(self):
return self.text
def json(self):
return {"type": "TextNode", "text": self.text}
class ImageNode(Node):
def __init__(self):
self.url = ""
self.alt = ""
def markdown(self):
if URL(self.url).scheme == "":
self.url = "https:" + self.url
alt = self.alt.replace("[", "\\[")
return f"![{alt}]({self.url})\n\n"
def json(self):
if URL(self.url).scheme == "":
self.url = "https:" + self.url
return {"type": "ImageNode", "url": self.url, "alt": self.alt}
class LatexNode(Node):
def __init__(self):
self.code = ""
def markdown(self):
if "\n" in self.code:
# 块级公式
return f"$$\n{self.code}\n$$"
else:
# 行内公式
return f"${self.code}$"
def json(self):
return {"type": "LatexNode", "code": self.code}
class CodeNode(Node):
def __init__(self):
self.code = ""
self.lang = ""
def markdown(self):
return f"```{self.lang if self.lang else ''}\n{self.code}\n```\n\n"
def json(self):
return {"type": "CodeNode", "code": self.code, "lang": self.lang}
# 卡片
class VideoCardNode(Node):
def __init__(self):
self.aid = 0
def markdown(self):
return f"[视频 av{self.aid}](https://www.bilibili.com/av{self.aid})\n\n"
def json(self):
return {"type": "VideoCardNode", "aid": self.aid}
class ArticleCardNode(Node):
def __init__(self):
self.cvid = 0
def markdown(self):
return f"[文章 cv{self.cvid}](https://www.bilibili.com/read/cv{self.cvid})\n\n"
def json(self):
return {"type": "ArticleCardNode", "cvid": self.cvid}
class BangumiCardNode(Node):
def __init__(self):
self.epid = 0
def markdown(self):
return f"[番剧 ep{self.epid}](https://www.bilibili.com/bangumi/play/ep{self.epid})\n\n"
def json(self):
return {"type": "BangumiCardNode", "epid": self.epid}
class MusicCardNode(Node):
def __init__(self):
self.auid = 0
def markdown(self):
return f"[音乐 au{self.auid}](https://www.bilibili.com/audio/au{self.auid})\n\n"
def json(self):
return {"type": "MusicCardNode", "auid": self.auid}
class ShopCardNode(Node):
def __init__(self):
self.pwid = 0
def markdown(self):
return f"[会员购 {self.pwid}](https://show.bilibili.com/platform/detail.html?id={self.pwid})\n\n"
def json(self):
return {"type": "ShopCardNode", "pwid": self.pwid}
class ComicCardNode(Node):
def __init__(self):
self.mcid = 0
def markdown(self):
return (
f"[漫画 mc{self.mcid}](https://manga.bilibili.com/m/detail/mc{self.mcid})\n\n"
)
def json(self):
return {"type": "ComicCardNode", "mcid": self.mcid}
class LiveCardNode(Node):
def __init__(self):
self.room_id = 0
def markdown(self):
return f"[直播 {self.room_id}](https://live.bilibili.com/{self.room_id})\n\n"
def json(self):
return {"type": "LiveCardNode", "room_id": self.room_id}
class AnchorNode(Node):
def __init__(self):
self.url = ""
self.text = ""
def markdown(self):
text = self.text.replace("[", "\\[")
return f"[{text}]({self.url})"
def json(self):
return {"type": "AnchorNode", "url": self.url, "text": self.text}
class SeparatorNode(Node):
def __init__(self):
pass
def markdown(self):
return "\n------\n"
def json(self):
return {"type": "SeparatorNode"}