bilibili_api/creative_center.py

"""
bilibili_api.creative_center

创作中心相关。

务必携带 Credential 信息，否则无法获取到数据。
"""

from enum import Enum
from typing import List, Union, Optional
from datetime import datetime

from .video_zone import VideoZoneTypes
from .utils.utils import get_api
from .utils.credential import Credential
from .utils.network import Api

API = get_api("creative_center")


class GraphPeriod(Enum):
    """
    统计图表的时间段。

    + YESTERDAY: 昨天
    + WEEK: 近一周
    + MONTH: 近一月
    + THREE_MONTH: 近三月
    + TOTAL: 历史累计
    """

    YESTERDAY = -1
    WEEK = 0
    MONTH = 1
    THREE_MONTH = 2
    TOTAL = 3


class FanGraphPeriod(Enum):
    """
    粉丝统计图表的时间段。

    + YESTERDAY: 昨天
    + WEEK: 近一周
    + MONTH: 近一月
    + THREE_MONTH: 近三月
    """

    YESTERDAY = -1
    WEEK = 0
    MONTH = 1
    THREE_MONTH = 2


class GraphType(Enum):
    """
    统计图表的类型。

    + PLAY: 播放量
    + VISITOR: 访问量
    + FAN: 粉丝数
    + LIKE: 点赞数
    + FAV: 收藏数
    + SHARE: 分享数
    + COMMENT: 评论数
    + DAMKU: 弹幕数
    + COIN: 投币数
    + ELEC: 充电数
    """

    PLAY = "play"
    VISITOR = "visitor"
    FAN = "fan"
    LIKE = "like"
    FAV = "fav"
    SHARE = "share"
    COMMENT = "comment"
    DAMKU = "dm"
    COIN = "coin"
    ELEC = "elec"


class FanGraphType(Enum):
    """
    粉丝统计图表的类型。

    + ALL_FANS: 粉丝总量
    + FAN: 新增粉丝
    + FOLLOW: 新增关注
    + UNFOLLOW: 取消关注
    """

    ALL_FANS = "all_fans"
    NEW_FANS = "fan"
    FOLLOW = "follow"
    UNFOLLOW = "unfollow"


class ArticleInfoType(Enum):
    """
    文章统计信息的类型。

    + READ: 阅读
    + COMMENT: 评论
    + SHARE: 分享
    + COIN: 投币
    + FAV: 收藏
    + LIKE: 点赞
    """

    READ = 1
    COMMENT = 2
    SHARE = 3
    COIN = 4
    FAV = 5
    LIKE = 6


class Copyright(Enum):
    """
    稿件播放完成率对比的版权类型。

    + ALL: 全部
    + ORIGINAL: 原创
    + REPRINT: 转载
    """

    ALL = 0
    ORIGINAL = 1
    REPRINT = 2


class UploadManagerOrder(Enum):
    """
    内容管理排序字段。

    + CLICK: 点击
    + STOW: 收藏
    + SENDDATE: 上传日期
    + DM_COUNT: 弹幕数量
    + COMMENT_COUNT: 评论数量
    """

    CLICK = "click"
    STOW = "stow"
    SENDDATE = "senddate"
    DM_COUNT = "dm_count"
    COMMENT_COUNT = "scores"  # wtf the scores means


class UploadManagerStatus(Enum):
    """
    内容管理稿件状态字段。

    + ALL: 全部稿件
    + PUBED: 已通过
    + IS_PUBING: 进行中
    + NOT_PUBED: 未通过
    """

    ALL = "is_pubing,pubed,not_pubed"
    PUBED = "pubed"
    IS_PUBING = "is_pubing"
    NOT_PUBED = "not_pubed"


class UploadManagerSort(Enum):
    """
    内容管理文章排序字段。

    + CREATED_TIME: 创建日期
    + LIKE: 点赞
    + COMMENT: 评论
    + FAV: 收藏
    + COIN: 投币
    """

    CREATED_TIME = 1
    LIKE = 2
    COMMENT = 3
    FAV = 5
    COIN = 6


class UploadManagerArticleStatus(Enum):
    """
    内容管理文章状态字段。

    + ALL: 全部稿件
    + PUBED: 已通过
    + IS_PUBING: 进行中
    + NOT_PUBED: 未通过
    """

    ALL = 0
    PUBED = 2
    IS_PUBING = 1
    NOT_PUBED = 3


class ArchiveType(Enum):
    """
    评论管理中的稿件类型。

    + VIDEO: 视频
    + ARTICLE: 文章
    + AUDIO: 音频
    """

    VIDEO = 1  # 视频
    ARTICLE = 12  # 文章
    AUDIO = 14  # 音频


class CommentManagerOrder(Enum):
    """
    评论管理中的排序字段。

    + RECENTLY: 最近
    + LIKE: 点赞
    + REPLY: 回复
    """

    RECENTLY = 1  # 最近
    LIKE = 2  # 点赞
    REPLY = 3  # 回复


class DanmakuMode(Enum):
    """
    弹幕模式。

    + ROLL: 滚动
    + BOTTOM: 底端
    + TOP: 顶端
    + REVERSE: 逆向
    + ADVANCED: 高级
    + CODE: 代码
    + BAS: BAS 补充注释
    """

    ROLL = 1  # 滚动
    BOTTOM = 4  # 底端
    TOP = 5  # 顶端
    REVERSE = 6  # 逆向
    ADVANCED = 7  # 高级
    CODE = 8  # 代码
    BAS = 9  # BAS


class DanmakuPool(Enum):
    """
    子弹幕池类型。

    + NORMAL: 普通
    + SUBTITLE: 字幕
    + SPECIAL: 特殊
    """

    NORMAL = 0  # 普通
    SUBTITLE = 1  # 字幕
    SPECIAL = 2  # 特殊

class DanmakuOrder(Enum):
    """
    弹幕排序依据

    + CTIME: 发送时间
    + LIKE_COUNT: 点赞数
    """

    CTIME = "ctime"
    LIKE_COUNT = "like_count"


class DanmakuSort(Enum):
    """
    弹幕排序顺序

    + DESC: 降序
    + ASC: 升序
    """

    DESC = "desc"
    ASC = "asc"


class DanmakuType(Enum):
    """
    弹幕筛选类型

    + ALL: 全部
    + PROTECT: 保护弹幕
    """

    ALL = 0
    PROTECT = 2


async def get_compare(credential: Credential) -> dict:
    """
    获取对比数据。

    Args:
        credentials (Credential): Credential 凭据。

    Returns:
        dict: 视频对比数据。
    """
    api = API["overview"]["compare"]
    return await Api(**api, credential=credential).result


async def get_graph(
    credential: Credential,
    period: GraphPeriod = GraphPeriod.WEEK,
    graph_type: GraphType = GraphType.PLAY,
) -> dict:
    """
    获取统计图表数据。

    Args:
        credentials (Credential): Credential 凭据。

        period      (GraphPeriod): 时间段。

        graph_type  (GraphType):   图表类型。

    Returns:
        dict: 视频统计图表数据。
    """
    api = API["overview"]["graph"]
    params = {
        "period": period.value,
        "s_locale": "zh_CN",
        "type": graph_type.value,
    }
    return await Api(**api, credential=credential).update_params(**params).result


async def get_overview(
    credential: Credential, period: GraphPeriod = GraphPeriod.WEEK
) -> dict:
    """
    获取概览数据。

    Args:
        credentials (Credential): Credential 凭据。

        period      (GraphPeriod): 时间段。

    Returns:
        dict: 视频概览数据。
    """
    api = API["overview"]["num"]
    # 不知道 tab 的作用是什么，但是不传会报错
    params = {"period": period.value, "s_locale": "zh_CN", "tab": 0}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_video_survey(credential: Credential) -> dict:
    """
    获取视频各分区中占比排行。

    Args:
        credentials (Credential): Credential 凭据。

    Returns:
        dict: 视频分区排行数据。
    """
    api = API["data-up"]["video"]["survey"]
    params = {"type": 1}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_video_playanalysis(
    credential: Credential, copyright: Copyright = Copyright.ALL
) -> dict:
    """
    获取稿件播放完成率对比。

    Args:
        credentials (Credential): Credential 凭据。

        copyright   (Copyright):   版权类型。

    Returns:
        dict: 稿件播放完成率对比数据。
    """
    api = API["data-up"]["video"]["playanalysis"]
    params = {"copyright": copyright.value}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_video_source(credential: Credential) -> dict:
    """
    获取稿件播放来源分布。

    Args:
        credentials (Credential): Credential 凭据。

    Returns:
        dict: 视频来源分布数据。
    """
    api = API["data-up"]["video"]["source"]
    params = {"s_locale": "zh_CN"}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_fan_overview(
    credential: Credential, period: FanGraphPeriod = FanGraphPeriod.WEEK
) -> dict:
    """
    获取粉丝概览数据。

    Args:
        credentials (Credential): Credential 凭据。

        period      (FanGraphPeriod): 时间段。

    Returns:
        dict: 粉丝概览数据。
    """
    api = API["data-up"]["fan"]["overview"]
    params = {"period": period.value}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_fan_graph(
    credential: Credential,
    period: FanGraphPeriod = FanGraphPeriod.WEEK,
    graph_type: FanGraphType = FanGraphType.ALL_FANS,
) -> dict:
    """
    获取粉丝图表数据。

    Args:
        credentials (Credential): Credential 凭据。

        period      (FanGraphPeriod): 时间段。

        graph_type  (FanGraphType):   图表类型。

    Returns:
        dict: 粉丝图表数据。
    """
    api = API["data-up"]["fan"]["graph"]
    params = {"period": period.value, "type": graph_type.value}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_article_overview(credential: Credential) -> dict:
    """
    获取文章概览数据。

    Args:
        credentials (Credential): Credential 凭据。

    Returns:
        dict: 文章概览数据。
    """
    api = API["data-up"]["article"]["overview"]
    return await Api(**api, credential=credential).result


async def get_article_graph(
    credential: Credential, graph_type: ArticleInfoType = ArticleInfoType.READ
) -> dict:
    """
    获取文章图表数据。

    Args:
        credentials (Credential): Credential 凭据。

        graph_type  (ArticleInfoType):   图表类型。

    Returns:
        dict: 文章图表数据。
    """

    api = API["data-up"]["article"]["graph"]
    params = {"type": graph_type.value}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_article_rank(
    credential: Credential, rank_type: ArticleInfoType = ArticleInfoType.READ
) -> dict:
    """
    获取文章排行数据。

    Args:
        credentials (Credential): Credential 凭据。

        rank_type  (ArticleInfoType):   排行依据。

    Returns:
        dict: 文章排行数据。
    """

    api = API["data-up"]["article"]["rank"]
    params = {"type": rank_type.value}
    return await Api(**api, credential=credential).update_params(**params).result


async def get_article_source(credential: Credential) -> dict:
    """
    获取文章阅读终端数据

    Args:
        credentials (Credential): Credential 凭据。

    Returns:
        dict: 文章阅读终端数据。
    """

    api = API["data-up"]["article"]["source"]
    return await Api(**api, credential=credential).result


"""
upload manager

https://member.bilibili.com/platform/upload-manager
"""


async def get_video_draft_upload_manager_info(credential: Credential) -> dict:
    """
    获取内容管理视频草稿信息

    Args:
        credentials (Credential): Credential 凭据。

    Returns:
        dict: 内容管理视频草稿信息。
    """

    api = API["upload-manager"]["video_draft"]
    return await Api(**api, credential=credential).result


async def get_video_upload_manager_info(
    credential: Credential,
    is_interative: bool = False,
    pn: int = 1,
    ps: int = 10,
    order: UploadManagerOrder = UploadManagerOrder.CLICK,
    tid: Union[VideoZoneTypes, None, int] = None,
    status: UploadManagerStatus = UploadManagerStatus.ALL,
) -> dict:
    """
    获取内容管理视频信息

    Args:
        credentials (Credential): Credential 凭据。

        is_interative (bool): 是否为互动视频

        pn (int): 页码

        ps (int): 每页项数

        tid: (VideoZoneTypes, None, int): 分区

        status (UploadManagerStatus): 稿件状态

        order (UploadManagerOrder): 稿件排序

    Returns:
        dict: 内容管理视频信息。
    """
    params = {
        "pn": pn,
        "ps": ps,
        "coop": 1,
        "status": status.value,
        "order": order.value,
        "interactive": 2 if is_interative else 1,
        "tid": tid.value if isinstance(tid, Enum) else tid,
    }
    api = API["upload-manager"]["video"]
    return await Api(**api, credential=credential).update_params(**params).result


async def get_article_upload_manager_info(
    credential: Credential,
    status: UploadManagerArticleStatus = UploadManagerArticleStatus.ALL,
    sort: UploadManagerSort = UploadManagerSort.CREATED_TIME,
    pn: int = 1,
) -> dict:
    """
    获取内容管理文章信息

    Args:
        credentials (Credential): Credential 凭据。

        pn (int): 页码

        status (UploadManagerArticleStatus): 稿件状态

        sort (UploadManagerSort): 稿件排序

    Returns:
        dict: 内容管理文章信息。
    """

    params = {"pn": pn, "group": status.value, "sort": sort.value, "mobi_app": "pc"}
    api = API["upload-manager"]["article"]
    return await Api(**api, credential=credential).update_params(**params).result


async def get_article_list_upload_manager_info(credential: Credential) -> dict:
    """
    获取内容管理文章信息

    Args:
        credentials (Credential): Credential 凭据。

    Returns:
        dict: 内容管理文集信息。
    """

    api = API["upload-manager"]["article_list"]
    return await Api(**api, credential=credential).result


"""
comment manager

https://member.bilibili.com/platform/comment
"""


async def get_comments(
    credential: Credential,
    oid: Optional[int] = None,
    keyword: Optional[str] = None,
    archive_type: ArchiveType = ArchiveType.VIDEO,
    order: CommentManagerOrder = CommentManagerOrder.RECENTLY,
    filter: int = -1,
    pn: int = 1,
    ps: int = 10,
    charge_plus_filter: bool = False,
) -> dict:
    """
    获取评论

    Args:
        credentials (Credential): Credential 凭据。

        oid (Optional, int): 指定稿件

        keyword (Optional, str): 关键词

        archive_type (ArchiveType): 稿件类型

        order (CommentManagerOrder): 排序字段

        filter (int): 筛选器，作用未知

        pn (int): 页码

        ps (int): 每页项数

        charge_plus_filter (bool): charge_plus_filter

    Returns:
        dict: 评论管理评论信息。
    """

    params = {
        "order": order.value,
        "filter": filter,
        "type": archive_type.value,
        "pn": pn,
        "ps": ps,
        "charge_plus_filter": charge_plus_filter,
    }

    if keyword is not None:
        params["keyword"] = keyword
    if oid is not None:
        params["oid"] = oid

    api = API["comment-manager"]["fulllist"]
    return await Api(**api, credential=credential).update_params(**params).result


async def del_comments(
    credential: Credential,
    oid: Union[int, List[int]],
    rpid: Union[int, List[int]],
    archive_type: ArchiveType = ArchiveType.VIDEO,
):
    """
    删除评论

    每个评论对应一个 oid

    Args:
        credentials (Credential): Credential 凭据。

        oid (int, lsit): 指定稿件

        rpid (int, lsit): 指定评论

        archive_type (ArchiveType): 稿件类型
    """
    data = {
        "oid": ",".join(oid) if isinstance(oid, list) else oid,
        "type": archive_type.value,
        "rpid": ",".join(rpid) if isinstance(rpid, list) else rpid,
        "jsonp": "jsonp",
        "csrf": credential.bili_jct,
    }

    api = API["comment-manager"]["del"]
    return await Api(**api, credential=credential).update_data(**data).result


"""
danmaku manager

https://member.bilibili.com/platform/inter-active/danmu
"""


async def get_recently_danmakus(
    credential: Credential, pn: int = 1, ps: int = 50
) -> dict:
    """
    最近弹幕

    Args:
        credential (Credential): Credential 凭据。

        pn (int): 页码。

        ps (int): 每页项数。

    Returns:
        dict: 弹幕管理最近弹幕信息。
    """

    params = {"pn": pn, "ps": ps}

    api = API["danmaku-manager"]["recent"]
    return await Api(**api, credential=credential).update_params(**params).result





async def get_danmakus(
    credential: Credential,
    oid: int,
    select_type: DanmakuType = DanmakuType.ALL,
    archive_type: ArchiveType = ArchiveType.VIDEO,
    mids: Optional[Union[int, List[int]]] = None,
    keyword: Optional[str] = None,
    progress_from: Optional[int] = None,
    progress_to: Optional[int] = None,
    ctime_from: Optional[datetime] = None,
    ctime_to: Optional[datetime] = None,
    modes: Optional[Union[DanmakuMode, List[DanmakuMode]]] = None,
    pools: Optional[Union[DanmakuPool, List[DanmakuPool]]] = None,
    attrs=None,  # 未知参数，我在高级筛选里面找不到
    order: DanmakuOrder = DanmakuOrder.CTIME,
    sort: DanmakuSort = DanmakuSort.DESC,
    pn: int = 1,
    ps: int = 50,
    cp_filter: bool = False,
) -> dict:
    """
    弹幕搜索

    Args:
        credential (Credential): Credential 凭据

        oid (int): 稿件oid，用逗号分隔

        select_type (DanmakuType): 弹幕类型

        archive_type (ArchiveType): 稿件类型

        mids (list[int], int): 用户mids，用逗号分隔或者直接 int

        keyword (str): 关键词

        progress_from (int): 进度开始

        progress_to (int): 进度结束

        ctime_from (datetime.datetime): 创建时间起始

        ctime_to (datetime.datetime): 创建时间结束

        modes (DanmakuMode): 弹幕模式。

        pool (DanmakuPool): 弹幕池

        attrs (Unknown): 弹幕属性，未知参数

        order (DanmakuOrder): 排序字段

        sort (DanmakuSort): 排序方式

        pn (int): 页码。

        ps (int): 每页项数。

        cp_filter (bool): 是否过滤CP弹幕。未知参数，默认为 False

    Returns:
        dict: 弹幕搜索结果
    """
    params = {
        "oid": oid,
        "type": archive_type.value,
        "mids": ",".join(mids) if isinstance(mids, list) else mids,
        "select_type": select_type.value,
        "keyword": keyword,
        "progress_from": progress_from,
        "progress_to": progress_to,
        "ctime_from": ctime_from.strftime("%d-%m-%Y %H:%M:%S")
        if ctime_from is not None
        else None,
        "ctime_to": ctime_to.strftime("%d-%m-%Y %H:%M:%S")
        if ctime_to is not None
        else None,
        "modes": (
            ",".join([mode.value for mode in modes])
            if isinstance(modes, list)
            else modes.value
        )
        if modes is not None
        else None,
        "pool": (
            ",".join([pool.value for pool in pools])
            if isinstance(pools, list)
            else pools.value
        )
        if pools is not None
        else None,
        "attrs": attrs,
        "order": order.value,
        "sort": sort.value,
        "pn": pn,
        "ps": ps,
        "cp_filter": cp_filter,
    }

    api = API["danmaku-manager"]["search"]
    return await Api(**api, credential=credential).update_params(**params).result


async def del_danmaku(
    credential: Credential, oid: int, dmids: Union[int, List[int]]
) -> dict:
    """
    删除弹幕

    Args:
        oid (int): 稿件 oid

        dmids (list[int], int): 弹幕 id，可以传入列表和 int
    """

    return await edit_danmaku_state(credential=credential, oid=oid, dmids=dmids, state=1)


async def edit_danmaku_state(
    credential: Credential,
    oid: int,
    dmids: Union[int, List[int]],
    state: Optional[int] = None,
) -> dict:
    """
    操作弹幕状态

    Args:
        oid (int): 稿件 oid

        dmids (list[int], int): 弹幕 id，可以传入列表和 int

        state (int, Optional): 弹幕状态 1 删除 2 保护 3 取消保护

    Returns:
        dict: API 返回信息
    """
    data = {
        "type": 1,
        "oid": oid,
        "dmids": ",".join(dmids) if isinstance(dmids, list) else dmids,
        "state": state,
    }

    api = API["danmaku-manager"]["state"]
    return await Api(**api, credential=credential).update_data(**data).result


async def edit_danmaku_pool(
    credential: Credential,
    oid: int,
    dmids: Union[int, List[int]],
    is_subtitle: bool = True,
) -> dict:
    """
    操作弹幕池

    Args:
        oid (int): 稿件 oid

        dmids (list[int], int): 弹幕 id，可以传入列表和 int

        is_subtitle (bool): 是否为字幕

    Returns:
        dict: API 返回信息
    """
    data = {
        "type": 1,
        "oid": oid,
        "dmids": ",".join(dmids) if isinstance(dmids, list) else dmids,
        "pool": 1 if is_subtitle else 0,
    }

    api = API["danmaku-manager"]["pool"]
    return await Api(**api, credential=credential).update_data(**data).result