ck_bilibili.BiliBili.get_followings() - Code Metrics - OreosLab/checkinpanel - Measure and Improve Code Quality continuously with Scrutinizer

ck_bilibili.BiliBili.get_followings() A
last analyzed 2023-01-04 04:52 UTC

↳ Parent: ck_bilibili

Complexity

Conditions

Size

Total Lines	30
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	16
nop	6
dl	0
loc	30
rs	9.6
c	0
b	0
f	0

# -*- coding: utf-8 -*-
"""
cron: 30 8 * * *
new Env('Bilibili');
"""

import time

import requests

from notify_mtr import send
from utils import get_data


class BiliBili:
    def __init__(self, check_items: list):
        self.check_items = check_items

    @staticmethod
    def get_nav(session: requests.Session) -> tuple:
        """GET 登录基本信息-导航栏用户信息

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/login/login_info.md

        :param requests.Session session:
        :return (str, int, int, bool, int, int):
        (用户昵称, 用户 mid, 登录状态, 硬币数, 会员类型, 当前经验)
        """
        url = "https://api.bilibili.com/x/web-interface/nav"
        data = session.get(url=url).json().get("data", {})
        uname = data.get("uname")
        uid = data.get("mid")
        is_login = data.get("isLogin")
        coin = data.get("money")
        vip_type = data.get("vipType")
        current_exp = data.get("level_info", {}).get("current_exp")
        return uname, uid, is_login, coin, vip_type, current_exp

    @staticmethod
    def get_today_exp(session: requests.Session) -> list:
        """GET 获取今日经验信息

        :param requests.Session session:
        :return list: 今日经验信息列表
        """
        url = "https://api.bilibili.com/x/member/web/exp/log?jsonp=jsonp"
        today = time.strftime("%Y-%m-%d", time.localtime())
        return list(
            filter(
                lambda x: x["time"].split()[0] == today,
                session.get(url=url).json().get("data").get("list"),
            )
        )

    @staticmethod
    def sign_live(session: requests.Session) -> str:
        """GET 直播间用户实用 API-直播签到

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/live/user.md

        :param requests.Session session:
        :return str: 直播签到结果，成功/已签/失败/异常
        """
        try:
            url = "https://api.live.bilibili.com/xlive/web-ucenter/v1/sign/DoSign"
            res = session.get(url=url).json()
            if res["code"] == 0:
                msg = (
                    f'签到成功，{res["data"]["text"]}，'
                    f'特别信息：{res["data"]["specialText"]}，'
                    f'本月已签到 {res["data"]["hadSignDays"]} 天'
                )
            elif res["code"] == 1011040:
                msg = "今日已签到过，无法重复签到"
            else:
                msg = f'签到失败，信息为：{res["message"]}'
        except Exception as e:
            msg = f"签到异常，原因为 {str(e)}"
            print(msg)
        return msg

    @staticmethod
    def clockin_manga(session: requests.Session, platform: str = "android") -> str:
        """POST 模拟漫画客户端签到

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/manga/ClockIn.md

        :param requests.Session session:
        :param str platform: 签到平台, defaults to "android"
        :return str: 漫画客户端签到结果，成功/已签/失败/异常
        """
        try:
            url = "https://manga.bilibili.com/twirp/activity.v1.Activity/ClockIn"
            post_data = {"platform": platform}
            res = session.post(url=url, data=post_data).json()
            if res["code"] == 0:
                msg = "签到成功"
            elif res["msg"] == "clockin clockin is duplicate":
                msg = "今天已经签到过了"
            else:
                msg = f'签到失败，信息为 {res["msg"]}'
                print(msg)
        except Exception as e:
            msg = f"签到异常，原因为：{str(e)}"
            print(msg)
        return msg

    @staticmethod
    def receive_vip_privilege(
        session: requests.Session, bili_jct: str, receive_type: int = 1
    ) -> dict:
        """POST 大会员兑换福利-兑换卡券

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/user/vip.md

        :param requests.Session session:
        :param str bili_jct: Cookie bili_jct 字段
        :param int receive_type: 1~5: B币券 会员购优惠券 漫画福利券 会员购包邮券 漫画商城优惠券, defaults to 1
        :return dict: {"code": 0, "message": "0", "ttl": 1}
        """
        url = "https://api.bilibili.com/x/vip/privilege/receive"
        post_data = {"type": receive_type, "csrf": bili_jct}
        return session.post(url=url, data=post_data).json()

    @staticmethod
    def get_manga_vip_reward(session: requests.Session) -> dict:
        """POST 获取漫画大会员福利

        :param requests.Session session:
        :return dict: json 回复
        """
        url = "https://manga.bilibili.com/twirp/user.v1.User/GetVipReward"
        return session.post(url=url, json={"reason_id": 1}).json()

    @staticmethod
    def report_video_history(
        session: requests.Session,
        bili_jct: str,
        aid: int,
        cid: int,
        progress: int = 300,
    ) -> dict:
        """POST 视频观看数据上报-上报观看进度（双端）

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/video/report.md

        :param requests.Session session:
        :param str bili_jct: Cookie bili_jct 字段
        :param int aid: 稿件 avid
        :param int cid: 视频 cid
        :param int progress: 观看进度(s), defaults to 300
        :return dict: {"code": 0, "message": "0", "ttl": 1}
        """
        url = "http://api.bilibili.com/x/v2/history/report"
        post_data = {"aid": aid, "cid": cid, "progress": progress, "csrf": bili_jct}
        return session.post(url=url, data=post_data).json()

    @staticmethod
    def share(session: requests.Session, bili_jct: str, aid: int) -> dict:
        """POST 分享指定 av 号视频（web 端）

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/video/like_coin_fav.md

        :param requests.Session session:
        :param str bili_jct: Cookie bili_jct 字段
        :param int aid: 稿件 avid
        :return dict: {"code": 0, "message": "0", "ttl": 1, "data":19}
        """
        url = "https://api.bilibili.com/x/web-interface/share/add"
        post_data = {"aid": aid, "csrf": bili_jct}
        return session.post(url=url, data=post_data).json()

    @staticmethod
    def get_followings(
        session: requests.Session,
        uid: int,
        pn: int = 1,
        ps: int = 50,
        order: str = "desc",
        order_type: str = "attention",
    ) -> dict:
        """GET 用户关系相关-查询用户关注明细

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/user/relation.md

        :param requests.Session session:
        :param int uid: 目标用户 mid
        :param int pn: 页码, defaults to 1
        :param int ps: 每页项数, defaults to 50
        :param str order: defaults to "desc"
        :param str order_type: 排序方式, 按照关注顺序排列: 留空 按照最常访问排列: attention, defaults to "attention"
        :return dict: {"code": 0, "message": "0", "ttl": 1, "data": {"list": [], re_version: 3228575555, "total": 699}}
        """
        params: dict = {
            "vmid": uid,
            "pn": pn,
            "ps": ps,
            "order": order,
            "order_type": order_type,
        }
        url = "https://api.bilibili.com/x/relation/followings"
        return session.get(url=url, params=params).json()

    @staticmethod
    def search_space_arc(
        session: requests.Session,
        uid: int,
        pn: int = 1,
        ps: int = 100,
        tid: int = 0,
        order: str = "pubdate",
        keyword: str = "",
    ) -> list:
        """GET 获取指定 up 主空间视频投稿信息

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/user/space.md

        :param requests.Session session:
        :param int uid: 目标用户 mid
        :param int pn: 页码, defaults to 1
        :param int ps: 每页项数, defaults to 100
        :param int tid: 筛选目标分区, 0-不进行分区筛选, defaults to 0
        :param str order: 排序方式, 最新发布-pubdate, 最多播放-click, 最多收藏-stow, defaults to "pubdate"
        :param str keyword: 关键词筛选, 用于使用关键词搜索该 UP 主视频稿件, defaults to ""
        :return list: [{"aid": 585275804, "cid": 0, "title": "*", "owner": "apple"}, ...]
        """
        params: dict = {
            "mid": uid,
            "pn": pn,
            "ps": ps,
            "tid": tid,
            "order": order,
            "keyword": keyword,
        }
        url = "https://api.bilibili.com/x/space/arc/search"
        res = session.get(url=url, params=params).json()
        return [
            {
                "aid": one.get("aid"),
                "cid": 0,
                "title": one.get("title"),
                "owner": one.get("author"),
            }
            for one in res.get("data", {}).get("list", {}).get("vlist", [])
        ]

    @staticmethod
    def pay_elec_new(
        session: requests.Session, bili_jct: str, uid: int, num: int = 50
    ) -> dict:
        """POST B 币方式充电-新版本 B 币充电

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/electric/Bcoin.md

        :param requests.Session session:
        :param str bili_jct: Cookie bili_jct 字段
        :param int uid: 充电对象用户 mid
        :param int num: 贝壳数量, 必须在 2-9999 之间, defaults to 50
        :return dict: {"code": 0,"message": "0", "ttl": 1, "data":{...}}
        """
        url = "https://api.bilibili.com/x/ugcpay/trade/elec/pay/quick"
        post_data = {
            "bp_num": num,
            "is_bp_remains_prior": True,  # 是否优先扣除 B 币余额, B 币充电 true
            "up_mid": uid,
            "otype": "up",  # 充电来源, up-空间充电 archive-视频充电
            "oid": uid,
            "csrf": bili_jct,
        }
        return session.post(url=url, data=post_data).json()

    @staticmethod
    def add_coin(
        session: requests.Session,
        bili_jct: str,
        aid: int,
        num: int = 1,
        select_like: int = 1,
    ) -> dict:
        """POST 给指定 av 号视频投币

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/article/like_coin_fav.md

        :param requests.Session session:
        :param str bili_jct: Cookie bili_jct 字段
        :param int aid: 稿件 avid
        :param int num: 投币数量, 上限为 2, defaults to 1
        :param int select_like: 是否附加点赞, 0-不点赞 1-同时点赞, defaults to 1
        :return dict: {"code": 0, "message": "0", "ttl": 1, "data": {"like": true}}
        """
        url = "https://api.bilibili.com/x/web-interface/coin/add"
        post_data = {
            "aid": aid,
            "multiply": num,
            "select_like": select_like,
            "cross_domain": "true",
            "csrf": bili_jct,
        }
        return session.post(url=url, data=post_data).json()

    @staticmethod
    def get_live_status(session: requests.Session) -> str:
        """GET 直播获取金银瓜子硬币数量

        :param requests.Session session:
        :return str: "银瓜子数量 金瓜子数量 硬币数量"
        """
        url = "https://api.live.bilibili.com/pay/v1/Exchange/getStatus"
        res = session.get(url=url).json()
        data = res.get("data")
        silver = data.get("silver", 0)
        gold = data.get("gold", 0)
        coin = data.get("coin", 0)
        return f"银瓜子数量: {silver}\n金瓜子数量: {gold}\n硬币数量: {coin}"

    @staticmethod
    def silver2coin(session: requests.Session, bili_jct: str) -> dict:
        """POST 银瓜子兑换硬币

        :param requests.Session session:
        :param str bili_jct: Cookie bili_jct 字段
        :return dict: json 返回
        """
        url = "https://api.live.bilibili.com/pay/v1/Exchange/silver2coin"
        post_data = {"csrf_token": bili_jct, "csrf": bili_jct}
        return session.post(url=url, data=post_data).json()

    @staticmethod
    def get_dynamic_videos(
        session: requests.Session, rid: int = 1, ps: int = 6
    ) -> list:
        """GET 分区最新视频-获取分区最新视频列表

        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/ranking&dynamic/dynamic.md

        :param requests.Session session:
        :param int rid: 目标分区 tid, defaults to 1
        :param int ps: 每页项数, defaults to 6
        :return list: [{"aid": 56998612, "cid": 99548502, "title": "abc", "owner": "abc"}, ...]
        """
        url = (
            f"https://api.bilibili.com/x/web-interface/dynamic/region?ps={ps}&rid={rid}"
        )
        res = session.get(url=url).json()
        return [
            {
                "aid": one.get("aid"),
                "cid": one.get("cid"),
                "title": one.get("title"),
                "owner": one.get("owner", {}).get("name"),
            }
            for one in res.get("data", {}).get("archives", [])
        ]

    def main(self) -> str:
        msg_all = ""
        for check_item in self.check_items:
            cookie = {
                item.split("=")[0]: item.split("=")[1]
                for item in check_item.get("cookie").split("; ")
            }

            # csrf_token
            bili_jct = cookie.get("bili_jct", "")
            if not bili_jct:
                msg_all += "未获取到 bili_jct, 请检查 cookie 是否有效"
                continue
            # Config
            expected_coin = check_item.get("coin_num", 0)
            coin_type = check_item.get("coin_type", 1)
            silver2coin = check_item.get("silver2coin", True)

            # generate session
            session = requests.session()
            session.cookies.update(cookie)
            session.headers.update(
                {
                    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                    "AppleWebKit/537.36 (KHTML, like Gecko)"
                    "Chrome/91.0.4472.124 Safari/537.36 Edg/91.0.864.64",
                    "Referer": "https://www.bilibili.com/",
                    "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
                    "Connection": "keep-alive",
                }
            )

            # GET 导航栏用户信息 1/2
            uname, uid, is_login, coin, _, _ = self.get_nav(session=session)
            if not is_login:
                msg_all += "非登录状态，请检查\n\n"
                continue

            # POST 模拟漫画客户端签到
            manga_msg = self.clockin_manga(session=session)
            # GET 直播签到
            live_msg = self.sign_live(session=session)

            # GET 获取分区最新视频列表
            aid_list = self.get_dynamic_videos(session=session)

            # 今日已投币数量信息
            coins_av_count = len(
                list(
                    filter(
                        lambda x: x["reason"] == "视频投币奖励",
                        self.get_today_exp(session=session),
                    )
                )
            )
            coin_num = min(expected_coin - coins_av_count, coin)
            if coin_type == 1:
                following_list = self.get_followings(session=session, uid=uid)
                for following in following_list.get("data", {}).get("list", []):
                    mid = following.get("mid")
                    if mid:
                        aid_list += self.search_space_arc(session=session, uid=mid)

            if coin_num > 0:
                success_count = 0
                for aid in aid_list[::-1]:
                    res = self.add_coin(
                        session=session, aid=aid.get("aid"), bili_jct=bili_jct
                    )
                    if res["code"] == 0:
                        coin_num -= 1
                        print(f'成功给 {aid.get("title")} 投一个币')
                        success_count += 1
                    # -104 硬币不够了 -111 csrf 失败 34005 投币达到上限
                    elif res["code"] == 34005:
                        print(f'投币 {aid.get("title")} 失败，原因为 {res["message"]}')
                        continue
                    else:
                        print(f'投币 {aid.get("title")} 失败，原因为 {res["message"]}，跳过投币')
                        break
                    if coin_num <= 0:
                        break
                coin_msg = f"今日成功投币 {success_count + coins_av_count}/{expected_coin} 个"
            else:
                coin_msg = f"今日成功投币 {coins_av_count}/{expected_coin} 个"

            # POST 上报观看进度（双端）
            aid = aid_list[0].get("aid")
            cid = aid_list[0].get("cid")
            title = aid_list[0].get("title")
            report_res = self.report_video_history(
                session=session, bili_jct=bili_jct, aid=aid, cid=cid
            )
            if report_res.get("code") == 0:
                report_msg = f"观看《{title}》 300 秒"
            else:
                report_msg = "任务失败"
                print(report_msg)

            # POST 分享指定 av 号视频（web 端）
            share_res = self.share(session=session, bili_jct=bili_jct, aid=aid)
            if share_res.get("code") == 0:
                share_msg = f"分享《{title}》成功"
            else:
                share_msg = "分享失败"
                print(share_msg)

            # POST 银瓜子兑换硬币
            if silver2coin:
                silver2coin_res = self.silver2coin(session=session, bili_jct=bili_jct)
                if silver2coin_res["code"] == 0:
                    silver2coin_msg = "成功将银瓜子兑换为 1 个硬币"
                else:
                    silver2coin_msg = silver2coin_res["message"]
            else:
                silver2coin_msg = "未开启银瓜子兑换硬币功能"

            # GET 直播获取金银瓜子硬币数量
            live_stats = self.get_live_status(session=session)

            # GET 导航栏用户信息 2/2
            uname, uid, is_login, _, _, new_current_exp = self.get_nav(session=session)

            # 今日获得经验
            today_exp = sum(
                map(lambda x: x["delta"], self.get_today_exp(session=session))
            )
            # 预测升级天数
            update_data = (28800 - new_current_exp) // (today_exp or 1)

            msg = (
                f"帐号信息: {uname}\n"
                f"漫画签到: {manga_msg}\n"
                f"直播签到: {live_msg}\n"
                f"观看视频: {report_msg}\n"
                f"分享任务: {share_msg}\n"
                f"投币任务: {coin_msg}\n"
                f"银瓜子兑换硬币: {silver2coin_msg}\n"
                f"今日获得经验: {today_exp}\n"
                f"当前经验: {new_current_exp}\n"
                f"按当前速度升级还需: {update_data} 天\n"
                f"{live_stats}"
            )
            msg_all += msg + "\n\n"
        return msg_all


if __name__ == "__main__":
    _data = get_data()
    _check_items = _data.get("BILIBILI", [])
    result = BiliBili(check_items=_check_items).main()
    send("Bilibili", result)


1			# -- coding: utf-8 --
2			"""
3			cron: 30 8 * * *
4			new Env('Bilibili');
5			"""
6
7			import time
8
9			import requests
10
11			from notify_mtr import send
12			from utils import get_data
13
14
15			class BiliBili:
16			def __init__(self, check_items: list):
17			self.check_items = check_items
18
19			@staticmethod
20			def get_nav(session: requests.Session) -> tuple:
21			"""GET 登录基本信息-导航栏用户信息
22
23			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/login/login_info.md
24
25			:param requests.Session session:
26			:return (str, int, int, bool, int, int):
27			(用户昵称, 用户 mid, 登录状态, 硬币数, 会员类型, 当前经验)
28			"""
29			url = "https://api.bilibili.com/x/web-interface/nav"
30			data = session.get(url=url).json().get("data", {})
31			uname = data.get("uname")
32			uid = data.get("mid")
33			is_login = data.get("isLogin")
34			coin = data.get("money")
35			vip_type = data.get("vipType")
36			current_exp = data.get("level_info", {}).get("current_exp")
37			return uname, uid, is_login, coin, vip_type, current_exp
38
39			@staticmethod
40			def get_today_exp(session: requests.Session) -> list:
41			"""GET 获取今日经验信息
42
43			:param requests.Session session:
44			:return list: 今日经验信息列表
45			"""
46			url = "https://api.bilibili.com/x/member/web/exp/log?jsonp=jsonp"
47			today = time.strftime("%Y-%m-%d", time.localtime())
48			return list(
49			filter(
50			lambda x: x["time"].split()[0] == today,
51			session.get(url=url).json().get("data").get("list"),
52			)
53			)
54
55			@staticmethod
56			def sign_live(session: requests.Session) -> str:
57			"""GET 直播间用户实用 API-直播签到
58
59			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/live/user.md
60
61			:param requests.Session session:
62			:return str: 直播签到结果，成功/已签/失败/异常
63			"""
64			try:
65			url = "https://api.live.bilibili.com/xlive/web-ucenter/v1/sign/DoSign"
66			res = session.get(url=url).json()
67			if res["code"] == 0:
68			msg = (
69			f'签到成功，{res["data"]["text"]}，'
70			f'特别信息：{res["data"]["specialText"]}，'
71			f'本月已签到 {res["data"]["hadSignDays"]} 天'
72			)
73			elif res["code"] == 1011040:
74			msg = "今日已签到过，无法重复签到"
75			else:
76			msg = f'签到失败，信息为：{res["message"]}'
77			except Exception as e:
78			msg = f"签到异常，原因为 {str(e)}"
79			print(msg)
80			return msg
81
82			@staticmethod
83			def clockin_manga(session: requests.Session, platform: str = "android") -> str:
84			"""POST 模拟漫画客户端签到
85
86			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/manga/ClockIn.md
87
88			:param requests.Session session:
89			:param str platform: 签到平台, defaults to "android"
90			:return str: 漫画客户端签到结果，成功/已签/失败/异常
91			"""
92			try:
93			url = "https://manga.bilibili.com/twirp/activity.v1.Activity/ClockIn"
94			post_data = {"platform": platform}
95			res = session.post(url=url, data=post_data).json()
96			if res["code"] == 0:
97			msg = "签到成功"
98			elif res["msg"] == "clockin clockin is duplicate":
99			msg = "今天已经签到过了"
100			else:
101			msg = f'签到失败，信息为 {res["msg"]}'
102			print(msg)
103			except Exception as e:
104			msg = f"签到异常，原因为：{str(e)}"
105			print(msg)
106			return msg
107
108			@staticmethod
109			def receive_vip_privilege(
110			session: requests.Session, bili_jct: str, receive_type: int = 1
111			) -> dict:
112			"""POST 大会员兑换福利-兑换卡券
113
114			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/user/vip.md
115
116			:param requests.Session session:
117			:param str bili_jct: Cookie bili_jct 字段
118			:param int receive_type: 1~5: B币券会员购优惠券漫画福利券会员购包邮券漫画商城优惠券, defaults to 1
119			:return dict: {"code": 0, "message": "0", "ttl": 1}
120			"""
121			url = "https://api.bilibili.com/x/vip/privilege/receive"
122			post_data = {"type": receive_type, "csrf": bili_jct}
123			return session.post(url=url, data=post_data).json()
124
125			@staticmethod
126			def get_manga_vip_reward(session: requests.Session) -> dict:
127			"""POST 获取漫画大会员福利
128
129			:param requests.Session session:
130			:return dict: json 回复
131			"""
132			url = "https://manga.bilibili.com/twirp/user.v1.User/GetVipReward"
133			return session.post(url=url, json={"reason_id": 1}).json()
134
135			@staticmethod
136			def report_video_history(
137			session: requests.Session,
138			bili_jct: str,
139			aid: int,
140			cid: int,
141			progress: int = 300,
142			) -> dict:
143			"""POST 视频观看数据上报-上报观看进度（双端）
144
145			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/video/report.md
146
147			:param requests.Session session:
148			:param str bili_jct: Cookie bili_jct 字段
149			:param int aid: 稿件 avid
150			:param int cid: 视频 cid
151			:param int progress: 观看进度(s), defaults to 300
152			:return dict: {"code": 0, "message": "0", "ttl": 1}
153			"""
154			url = "http://api.bilibili.com/x/v2/history/report"
155			post_data = {"aid": aid, "cid": cid, "progress": progress, "csrf": bili_jct}
156			return session.post(url=url, data=post_data).json()
157
158			@staticmethod
159			def share(session: requests.Session, bili_jct: str, aid: int) -> dict:
160			"""POST 分享指定 av 号视频（web 端）
161
162			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/video/like_coin_fav.md
163
164			:param requests.Session session:
165			:param str bili_jct: Cookie bili_jct 字段
166			:param int aid: 稿件 avid
167			:return dict: {"code": 0, "message": "0", "ttl": 1, "data":19}
168			"""
169			url = "https://api.bilibili.com/x/web-interface/share/add"
170			post_data = {"aid": aid, "csrf": bili_jct}
171			return session.post(url=url, data=post_data).json()
172
173			@staticmethod
174			def get_followings(
175			session: requests.Session,
176			uid: int,
177			pn: int = 1,
178			ps: int = 50,
179			order: str = "desc",
180			order_type: str = "attention",
181			) -> dict:
182			"""GET 用户关系相关-查询用户关注明细
183
184			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/user/relation.md
185
186			:param requests.Session session:
187			:param int uid: 目标用户 mid
188			:param int pn: 页码, defaults to 1
189			:param int ps: 每页项数, defaults to 50
190			:param str order: defaults to "desc"
191			:param str order_type: 排序方式, 按照关注顺序排列: 留空按照最常访问排列: attention, defaults to "attention"
192			:return dict: {"code": 0, "message": "0", "ttl": 1, "data": {"list": [], re_version: 3228575555, "total": 699}}
193			"""
194			params: dict = {
195			"vmid": uid,
196			"pn": pn,
197			"ps": ps,
198			"order": order,
199			"order_type": order_type,
200			}
201			url = "https://api.bilibili.com/x/relation/followings"
202			return session.get(url=url, params=params).json()
203
204			@staticmethod
205			def search_space_arc(
206			session: requests.Session,
207			uid: int,
208			pn: int = 1,
209			ps: int = 100,
210			tid: int = 0,
211			order: str = "pubdate",
212			keyword: str = "",
213			) -> list:
214			"""GET 获取指定 up 主空间视频投稿信息
215
216			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/user/space.md
217
218			:param requests.Session session:
219			:param int uid: 目标用户 mid
220			:param int pn: 页码, defaults to 1
221			:param int ps: 每页项数, defaults to 100
222			:param int tid: 筛选目标分区, 0-不进行分区筛选, defaults to 0
223			:param str order: 排序方式, 最新发布-pubdate, 最多播放-click, 最多收藏-stow, defaults to "pubdate"
224			:param str keyword: 关键词筛选, 用于使用关键词搜索该 UP 主视频稿件, defaults to ""
225			:return list: [{"aid": 585275804, "cid": 0, "title": "*", "owner": "apple"}, ...]
226			"""
227			params: dict = {
228			"mid": uid,
229			"pn": pn,
230			"ps": ps,
231			"tid": tid,
232			"order": order,
233			"keyword": keyword,
234			}
235			url = "https://api.bilibili.com/x/space/arc/search"
236			res = session.get(url=url, params=params).json()
237			return [
238			{
239			"aid": one.get("aid"),
240			"cid": 0,
241			"title": one.get("title"),
242			"owner": one.get("author"),
243			}
244			for one in res.get("data", {}).get("list", {}).get("vlist", [])
245			]
246
247			@staticmethod
248			def pay_elec_new(
249			session: requests.Session, bili_jct: str, uid: int, num: int = 50
250			) -> dict:
251			"""POST B 币方式充电-新版本 B 币充电
252
253			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/electric/Bcoin.md
254
255			:param requests.Session session:
256			:param str bili_jct: Cookie bili_jct 字段
257			:param int uid: 充电对象用户 mid
258			:param int num: 贝壳数量, 必须在 2-9999 之间, defaults to 50
259			:return dict: {"code": 0,"message": "0", "ttl": 1, "data":{...}}
260			"""
261			url = "https://api.bilibili.com/x/ugcpay/trade/elec/pay/quick"
262			post_data = {
263			"bp_num": num,
264			"is_bp_remains_prior": True, # 是否优先扣除 B 币余额, B 币充电 true
265			"up_mid": uid,
266			"otype": "up", # 充电来源, up-空间充电 archive-视频充电
267			"oid": uid,
268			"csrf": bili_jct,
269			}
270			return session.post(url=url, data=post_data).json()
271
272			@staticmethod
273			def add_coin(
274			session: requests.Session,
275			bili_jct: str,
276			aid: int,
277			num: int = 1,
278			select_like: int = 1,
279			) -> dict:
280			"""POST 给指定 av 号视频投币
281
282			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/article/like_coin_fav.md
283
284			:param requests.Session session:
285			:param str bili_jct: Cookie bili_jct 字段
286			:param int aid: 稿件 avid
287			:param int num: 投币数量, 上限为 2, defaults to 1
288			:param int select_like: 是否附加点赞, 0-不点赞 1-同时点赞, defaults to 1
289			:return dict: {"code": 0, "message": "0", "ttl": 1, "data": {"like": true}}
290			"""
291			url = "https://api.bilibili.com/x/web-interface/coin/add"
292			post_data = {
293			"aid": aid,
294			"multiply": num,
295			"select_like": select_like,
296			"cross_domain": "true",
297			"csrf": bili_jct,
298			}
299			return session.post(url=url, data=post_data).json()
300
301			@staticmethod
302			def get_live_status(session: requests.Session) -> str:
303			"""GET 直播获取金银瓜子硬币数量
304
305			:param requests.Session session:
306			:return str: "银瓜子数量金瓜子数量硬币数量"
307			"""
308			url = "https://api.live.bilibili.com/pay/v1/Exchange/getStatus"
309			res = session.get(url=url).json()
310			data = res.get("data")
311			silver = data.get("silver", 0)
312			gold = data.get("gold", 0)
313			coin = data.get("coin", 0)
314			return f"银瓜子数量: {silver}\n金瓜子数量: {gold}\n硬币数量: {coin}"
315
316			@staticmethod
317			def silver2coin(session: requests.Session, bili_jct: str) -> dict:
318			"""POST 银瓜子兑换硬币
319
320			:param requests.Session session:
321			:param str bili_jct: Cookie bili_jct 字段
322			:return dict: json 返回
323			"""
324			url = "https://api.live.bilibili.com/pay/v1/Exchange/silver2coin"
325			post_data = {"csrf_token": bili_jct, "csrf": bili_jct}
326			return session.post(url=url, data=post_data).json()
327
328			@staticmethod
329			def get_dynamic_videos(
330			session: requests.Session, rid: int = 1, ps: int = 6
331			) -> list:
332			"""GET 分区最新视频-获取分区最新视频列表
333
334			https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/ranking&dynamic/dynamic.md
335
336			:param requests.Session session:
337			:param int rid: 目标分区 tid, defaults to 1
338			:param int ps: 每页项数, defaults to 6
339			:return list: [{"aid": 56998612, "cid": 99548502, "title": "abc", "owner": "abc"}, ...]
340			"""
341			url = (
342			f"https://api.bilibili.com/x/web-interface/dynamic/region?ps={ps}&rid={rid}"
343			)
344			res = session.get(url=url).json()
345			return [
346			{
347			"aid": one.get("aid"),
348			"cid": one.get("cid"),
349			"title": one.get("title"),
350			"owner": one.get("owner", {}).get("name"),
351			}
352			for one in res.get("data", {}).get("archives", [])
353			]
354
355			def main(self) -> str:
356			msg_all = ""
357			for check_item in self.check_items:
358			cookie = {
359			item.split("=")[0]: item.split("=")[1]
360			for item in check_item.get("cookie").split("; ")
361			}
362
363			# csrf_token
364			bili_jct = cookie.get("bili_jct", "")
365			if not bili_jct:
366			msg_all += "未获取到 bili_jct, 请检查 cookie 是否有效"
367			continue
368			# Config
369			expected_coin = check_item.get("coin_num", 0)
370			coin_type = check_item.get("coin_type", 1)
371			silver2coin = check_item.get("silver2coin", True)
372
373			# generate session
374			session = requests.session()
375			session.cookies.update(cookie)
376			session.headers.update(
377			{
378			"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
379			"AppleWebKit/537.36 (KHTML, like Gecko)"
380			"Chrome/91.0.4472.124 Safari/537.36 Edg/91.0.864.64",
381			"Referer": "https://www.bilibili.com/",
382			"accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
383			"Connection": "keep-alive",
384			}
385			)
386
387			# GET 导航栏用户信息 1/2
388			uname, uid, is_login, coin, _, _ = self.get_nav(session=session)
389			if not is_login:
390			msg_all += "非登录状态，请检查\n\n"
391			continue
392
393			# POST 模拟漫画客户端签到
394			manga_msg = self.clockin_manga(session=session)
395			# GET 直播签到
396			live_msg = self.sign_live(session=session)
397
398			# GET 获取分区最新视频列表
399			aid_list = self.get_dynamic_videos(session=session)
400
401			# 今日已投币数量信息
402			coins_av_count = len(
403			list(
404			filter(
405			lambda x: x["reason"] == "视频投币奖励",
406			self.get_today_exp(session=session),
407			)
408			)
409			)
410			coin_num = min(expected_coin - coins_av_count, coin)
411			if coin_type == 1:
412			following_list = self.get_followings(session=session, uid=uid)
413			for following in following_list.get("data", {}).get("list", []):
414			mid = following.get("mid")
415			if mid:
416			aid_list += self.search_space_arc(session=session, uid=mid)
417
418			if coin_num > 0:
419			success_count = 0
420			for aid in aid_list[::-1]:
421			res = self.add_coin(
422			session=session, aid=aid.get("aid"), bili_jct=bili_jct
423			)
424			if res["code"] == 0:
425			coin_num -= 1
426			print(f'成功给 {aid.get("title")} 投一个币')
427			success_count += 1
428			# -104 硬币不够了 -111 csrf 失败 34005 投币达到上限
429			elif res["code"] == 34005:
430			print(f'投币 {aid.get("title")} 失败，原因为 {res["message"]}')
431			continue
432			else:
433			print(f'投币 {aid.get("title")} 失败，原因为 {res["message"]}，跳过投币')
434			break
435			if coin_num <= 0:
436			break
437			coin_msg = f"今日成功投币 {success_count + coins_av_count}/{expected_coin} 个"
438			else:
439			coin_msg = f"今日成功投币 {coins_av_count}/{expected_coin} 个"
440
441			# POST 上报观看进度（双端）
442			aid = aid_list[0].get("aid")
443			cid = aid_list[0].get("cid")
444			title = aid_list[0].get("title")
445			report_res = self.report_video_history(
446			session=session, bili_jct=bili_jct, aid=aid, cid=cid
447			)
448			if report_res.get("code") == 0:
449			report_msg = f"观看《{title}》 300 秒"
450			else:
451			report_msg = "任务失败"
452			print(report_msg)
453
454			# POST 分享指定 av 号视频（web 端）
455			share_res = self.share(session=session, bili_jct=bili_jct, aid=aid)
456			if share_res.get("code") == 0:
457			share_msg = f"分享《{title}》成功"
458			else:
459			share_msg = "分享失败"
460			print(share_msg)
461
462			# POST 银瓜子兑换硬币
463			if silver2coin:
464			silver2coin_res = self.silver2coin(session=session, bili_jct=bili_jct)
465			if silver2coin_res["code"] == 0:
466			silver2coin_msg = "成功将银瓜子兑换为 1 个硬币"
467			else:
468			silver2coin_msg = silver2coin_res["message"]
469			else:
470			silver2coin_msg = "未开启银瓜子兑换硬币功能"
471
472			# GET 直播获取金银瓜子硬币数量
473			live_stats = self.get_live_status(session=session)
474
475			# GET 导航栏用户信息 2/2
476			uname, uid, is_login, _, _, new_current_exp = self.get_nav(session=session)
477
478			# 今日获得经验
479			today_exp = sum(
480			map(lambda x: x["delta"], self.get_today_exp(session=session))
481			)
482			# 预测升级天数
483			update_data = (28800 - new_current_exp) // (today_exp or 1)
484
485			msg = (
486			f"帐号信息: {uname}\n"
487			f"漫画签到: {manga_msg}\n"
488			f"直播签到: {live_msg}\n"
489			f"观看视频: {report_msg}\n"
490			f"分享任务: {share_msg}\n"
491			f"投币任务: {coin_msg}\n"
492			f"银瓜子兑换硬币: {silver2coin_msg}\n"
493			f"今日获得经验: {today_exp}\n"
494			f"当前经验: {new_current_exp}\n"
495			f"按当前速度升级还需: {update_data} 天\n"
496			f"{live_stats}"
497			)
498			msg_all += msg + "\n\n"
499			return msg_all
500
501
502			if __name__ == "__main__":
503			_data = get_data()
504			_check_items = _data.get("BILIBILI", [])
505			result = BiliBili(check_items=_check_items).main()
506			send("Bilibili", result)
507

OreosLab / checkinpanel

ck_bilibili.BiliBili.get_followings() A last analyzed 2023-01-04 04:52 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ck_bilibili.BiliBili.get_followings() A
last analyzed 2023-01-04 04:52 UTC