ck_nga.NGA.view_video() - Code Metrics - Inspection of ":art: Format with black (88)" - OreosLab/checkinpanel - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (67948d)

by Leon

created 2022-10-14 06:55 UTC

ck_nga.NGA.view_video() B

↳ Parent: ck_nga

Complexity

Conditions

Size

Total Lines	32
Code Lines	31

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	31
nop	3
dl	0
loc	32
rs	7.736
c	0
b	0
f	0

# -*- coding: utf-8 -*-
"""
cron: 26 13 * * *
new Env('NGA');
"""

import re
import time

import requests

from notify_mtr import send
from utils import get_data


class NGA:
    def __init__(self, check_items):
        self.check_items = check_items
        self.url = "https://ngabbs.com/nuke.php"
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Linux; Android 10; NOH-AN00 Build/HUAWEINOH-AN00; wv) "
            "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
            "Chrome/83.0.4103.106 Mobile Safari/537.36 Nga_Official/90021",
            "X-Requested-With": "gov.pianzong.androidnga",
            "X-USER-AGENT": "Nga_Official/90021(HUAWEI NOH-AN00;Android 10)",
        }

    def signin(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "app_id": "1010",
            "__act": "check_in",
            "__lib": "check_in",
            "__output": "12",
        }
        return requests.post(self.url, data, headers=self.headers).json()

    def get_silver_coin(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "app_id": "1010",
            "mid": "2",
            "__act": "check_mission",
            "__lib": "mission",
            "__output": "11",
        }
        try:
            res = requests.post(self.url, data, headers=self.headers).json()
            stat = res["data"][0]
            silver_coin_get_stat = stat[4] if "已经" in stat[4] else stat[2]["2"]
        except Exception as e:
            silver_coin_get_stat = str(e)
        return silver_coin_get_stat

    def get_n_coin(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "app_id": "1010",
            "mid": "30",
            "__act": "check_mission",
            "__lib": "mission",
            "__output": "11",
        }
        try:
            res = requests.post(self.url, data, headers=self.headers).json()
            stat = res["data"][0]
            n_coin_get_stat = stat[4] if "已经" in stat[4] else stat[2]["30"]
        except Exception as e:
            n_coin_get_stat = str(e)
        return n_coin_get_stat

    def view_video(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "app_id": "1010",
            "__act": "video_view_task_counter_add_v2",
            "__lib": "mission",
            "__output": "11",
        }
        success_sum = 0
        failure_sum = 0
        failure_msg = ""
        failure_msg_all = ""
        for _ in range(4):
            try:
                res = requests.post(self.url, data, headers=self.headers).json()
                time.sleep(30)
                raw_stat = re.search(r"\'raw_stat\':\s*{([^}]+)", str(res))[1]
                task_code = re.search(r"\'6\':\s(\d)", raw_stat)[1]
                time_code = re.search(r"\'5\':\s(\d)", raw_stat)[1]
                if task_code == "1" or (task_code == "0" and time_code == "1"):
                    success_sum += 1
            except Exception as e:
                failure_msg = str(e)
                failure_sum += 1
            failure_msg_all += failure_msg + "\n"
        video_coin = success_sum // 2 * 1
        return (
            f"观看视频成功次数：{success_sum}，共获得N币：{video_coin}"
            if failure_sum == 0
            else f"观看视频成功次数：{success_sum}，共获得N币：{video_coin}；\n"
            f"观看视频失败次数：{failure_sum}；\n"
            f"错误信息：{failure_msg_all}"
        )

    def view_video_for_adfree_24h(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "app_id": "1010",
            "__act": "video_view_task_counter_add_v2_for_adfree_sp1",
            "__lib": "mission",
            "__output": "11",
        }
        try:
            res = requests.post(self.url, data, headers=self.headers).json()
            time.sleep(30)
            if str(res["data"][1][0]) == "{}":
                code = res["data"][1][1]["141"]["raw_stat"]["6"]
            else:
                code = res["data"][1][0]["141"]["raw_stat"]["6"]
            adfree_24h_stat = "已获得免广告状态：24h" if code == 1 else "观看视频失败！"
        except Exception as e:
            adfree_24h_stat = str(e)
        return adfree_24h_stat

    def view_video_for_adfree(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "app_id": "1010",
            "__act": "video_view_task_counter_add_v2_for_adfree",
            "__lib": "mission",
            "__output": "11",
        }
        ids = ("142", "143", "144", "145")
        success_sum = 0
        failure_sum = 0
        failure_msg = ""
        failure_msg_all = ""
        code = {}
        for i, item in enumerate(ids):
            try:
                res = requests.post(self.url, data, headers=self.headers).json()
                time.sleep(30)
                code[i] = res["data"][1][0][item]["raw_stat"]["6"]
                if code[i] == 1:
                    success_sum += 1
                else:
                    failure_sum += 1
            except Exception as e:
                failure_msg = str(e)
                failure_sum += 1
            failure_msg_all += failure_msg + "\n"
        adfree_time = success_sum * 6
        return (
            f"观看视频成功次数：{success_sum}，共获得免广告时长：{adfree_time}h"
            if failure_sum == 0
            else f"观看视频成功次数：{success_sum}，共获得免广告时长：{adfree_time} h；\n"
            f"观看视频失败次数：{failure_sum}；\n"
            f"错误信息：{failure_msg_all}"
        )

    def get_signin_stat(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "sign": "",
            "app_id": "1010",
            "__act": "get_stat",
            "__lib": "check_in",
            "__output": "14",
        }
        res = requests.post(self.url, data, headers=self.headers).json()["result"][0]
        continued = res["continued"]
        total = res["sum"]
        return continued, total

    def get_user(self, token, uid):
        data = {
            "access_token": token,
            "t": round(time.time()),
            "access_uid": uid,
            "sign": "",
            "app_id": "1010",
            "__act": "iflogin",
            "__lib": "login",
            "__output": "12",
        }
        res = requests.post(self.url, data, headers=self.headers).json()
        return res["result"]["username"]

    def main(self):
        msg_all = ""
        for check_item in self.check_items:
            token = check_item.get("token")
            uid = check_item.get("uid")
            signin_res = self.signin(token, uid)
            try:
                continued, total = self.get_signin_stat(token, uid)
                username = self.get_user(token, uid)
                if signin_res["code"] == 0:
                    signin_stat = (
                        f"用户：{username}\n统计信息：签到成功，连续签到{continued}天，累计签到{total}天"
                    )
                elif signin_res["code"] == 1:
                    signin_stat = (
                        f"用户：{username}\n统计信息：今日已签，连续签到{continued}天，累计签到{total}天"
                    )
                else:
                    signin_stat = f'用户：{username}\n统计信息：{signin_res["msg"]}'
                time.sleep(1)
                silver_coin_get_stat = self.get_silver_coin(token, uid)
                n_coin_get_stat = self.get_n_coin(token, uid)
                video_view_stat = self.view_video(token, uid)
                adfree_24h_stat = self.view_video_for_adfree_24h(token, uid)
                msg = (
                    f"{signin_stat}\n"
                    f"------【每日签到得银币】------\n"
                    f"{silver_coin_get_stat}\n"
                    f"------【每日签到得N币】------\n"
                    f"{n_coin_get_stat}\n"
                    f"------【每天看两次视频】------\n"
                    f"{video_view_stat}\n"
                    f"------【看视频免广告(限时任务)】------\n"
                    f"{adfree_24h_stat}\n"
                )
            except Exception as e:
                msg = str(e)
            msg_all += msg + "\n\n"
        return msg_all


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


1			# -- coding: utf-8 --
2			"""
3			cron: 26 13 * * *
4			new Env('NGA');
5			"""
6
7			import re
8			import time
9
10			import requests
11
12			from notify_mtr import send
13			from utils import get_data
14
15
16			class NGA:
17			def __init__(self, check_items):
18			self.check_items = check_items
19			self.url = "https://ngabbs.com/nuke.php"
20			self.headers = {
21			"User-Agent": "Mozilla/5.0 (Linux; Android 10; NOH-AN00 Build/HUAWEINOH-AN00; wv) "
22			"AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
23			"Chrome/83.0.4103.106 Mobile Safari/537.36 Nga_Official/90021",
24			"X-Requested-With": "gov.pianzong.androidnga",
25			"X-USER-AGENT": "Nga_Official/90021(HUAWEI NOH-AN00;Android 10)",
26			}
27
28			def signin(self, token, uid):
29			data = {
30			"access_token": token,
31			"t": round(time.time()),
32			"access_uid": uid,
33			"app_id": "1010",
34			"__act": "check_in",
35			"__lib": "check_in",
36			"__output": "12",
37			}
38			return requests.post(self.url, data, headers=self.headers).json()
39
40			def get_silver_coin(self, token, uid):
41			data = {
42			"access_token": token,
43			"t": round(time.time()),
44			"access_uid": uid,
45			"app_id": "1010",
46			"mid": "2",
47			"__act": "check_mission",
48			"__lib": "mission",
49			"__output": "11",
50			}
51			try:
52			res = requests.post(self.url, data, headers=self.headers).json()
53			stat = res["data"][0]
54			silver_coin_get_stat = stat[4] if "已经" in stat[4] else stat[2]["2"]
55			except Exception as e:
56			silver_coin_get_stat = str(e)
57			return silver_coin_get_stat
58
59			def get_n_coin(self, token, uid):
60			data = {
61			"access_token": token,
62			"t": round(time.time()),
63			"access_uid": uid,
64			"app_id": "1010",
65			"mid": "30",
66			"__act": "check_mission",
67			"__lib": "mission",
68			"__output": "11",
69			}
70			try:
71			res = requests.post(self.url, data, headers=self.headers).json()
72			stat = res["data"][0]
73			n_coin_get_stat = stat[4] if "已经" in stat[4] else stat[2]["30"]
74			except Exception as e:
75			n_coin_get_stat = str(e)
76			return n_coin_get_stat
77
78			def view_video(self, token, uid):
79			data = {
80			"access_token": token,
81			"t": round(time.time()),
82			"access_uid": uid,
83			"app_id": "1010",
84			"__act": "video_view_task_counter_add_v2",
85			"__lib": "mission",
86			"__output": "11",
87			}
88			success_sum = 0
89			failure_sum = 0
90			failure_msg = ""
91			failure_msg_all = ""
92			for _ in range(4):
93			try:
94			res = requests.post(self.url, data, headers=self.headers).json()
95			time.sleep(30)
96			raw_stat = re.search(r"\'raw_stat\':\s*{([^}]+)", str(res))[1]
97			task_code = re.search(r"\'6\':\s(\d)", raw_stat)[1]
98			time_code = re.search(r"\'5\':\s(\d)", raw_stat)[1]
99			if task_code == "1" or (task_code == "0" and time_code == "1"):
100			success_sum += 1
101			except Exception as e:
102			failure_msg = str(e)
103			failure_sum += 1
104			failure_msg_all += failure_msg + "\n"
105			video_coin = success_sum // 2 * 1
106			return (
107			f"观看视频成功次数：{success_sum}，共获得N币：{video_coin}"
108			if failure_sum == 0
109			else f"观看视频成功次数：{success_sum}，共获得N币：{video_coin}；\n"
110			f"观看视频失败次数：{failure_sum}；\n"
111			f"错误信息：{failure_msg_all}"
112			)
113
114			def view_video_for_adfree_24h(self, token, uid):
115			data = {
116			"access_token": token,
117			"t": round(time.time()),
118			"access_uid": uid,
119			"app_id": "1010",
120			"__act": "video_view_task_counter_add_v2_for_adfree_sp1",
121			"__lib": "mission",
122			"__output": "11",
123			}
124			try:
125			res = requests.post(self.url, data, headers=self.headers).json()
126			time.sleep(30)
127			if str(res["data"][1][0]) == "{}":
128			code = res["data"][1][1]["141"]["raw_stat"]["6"]
129			else:
130			code = res["data"][1][0]["141"]["raw_stat"]["6"]
131			adfree_24h_stat = "已获得免广告状态：24h" if code == 1 else "观看视频失败！"
132			except Exception as e:
133			adfree_24h_stat = str(e)
134			return adfree_24h_stat
135
136			def view_video_for_adfree(self, token, uid):
137			data = {
138			"access_token": token,
139			"t": round(time.time()),
140			"access_uid": uid,
141			"app_id": "1010",
142			"__act": "video_view_task_counter_add_v2_for_adfree",
143			"__lib": "mission",
144			"__output": "11",
145			}
146			ids = ("142", "143", "144", "145")
147			success_sum = 0
148			failure_sum = 0
149			failure_msg = ""
150			failure_msg_all = ""
151			code = {}
152			for i, item in enumerate(ids):
153			try:
154			res = requests.post(self.url, data, headers=self.headers).json()
155			time.sleep(30)
156			code[i] = res["data"][1][0][item]["raw_stat"]["6"]
157			if code[i] == 1:
158			success_sum += 1
159			else:
160			failure_sum += 1
161			except Exception as e:
162			failure_msg = str(e)
163			failure_sum += 1
164			failure_msg_all += failure_msg + "\n"
165			adfree_time = success_sum * 6
166			return (
167			f"观看视频成功次数：{success_sum}，共获得免广告时长：{adfree_time}h"
168			if failure_sum == 0
169			else f"观看视频成功次数：{success_sum}，共获得免广告时长：{adfree_time} h；\n"
170			f"观看视频失败次数：{failure_sum}；\n"
171			f"错误信息：{failure_msg_all}"
172			)
173
174			def get_signin_stat(self, token, uid):
175			data = {
176			"access_token": token,
177			"t": round(time.time()),
178			"access_uid": uid,
179			"sign": "",
180			"app_id": "1010",
181			"__act": "get_stat",
182			"__lib": "check_in",
183			"__output": "14",
184			}
185			res = requests.post(self.url, data, headers=self.headers).json()["result"][0]
186			continued = res["continued"]
187			total = res["sum"]
188			return continued, total
189
190			def get_user(self, token, uid):
191			data = {
192			"access_token": token,
193			"t": round(time.time()),
194			"access_uid": uid,
195			"sign": "",
196			"app_id": "1010",
197			"__act": "iflogin",
198			"__lib": "login",
199			"__output": "12",
200			}
201			res = requests.post(self.url, data, headers=self.headers).json()
202			return res["result"]["username"]
203
204			def main(self):
205			msg_all = ""
206			for check_item in self.check_items:
207			token = check_item.get("token")
208			uid = check_item.get("uid")
209			signin_res = self.signin(token, uid)
210			try:
211			continued, total = self.get_signin_stat(token, uid)
212			username = self.get_user(token, uid)
213			if signin_res["code"] == 0:
214			signin_stat = (
215			f"用户：{username}\n统计信息：签到成功，连续签到{continued}天，累计签到{total}天"
216			)
217			elif signin_res["code"] == 1:
218			signin_stat = (
219			f"用户：{username}\n统计信息：今日已签，连续签到{continued}天，累计签到{total}天"
220			)
221			else:
222			signin_stat = f'用户：{username}\n统计信息：{signin_res["msg"]}'
223			time.sleep(1)
224			silver_coin_get_stat = self.get_silver_coin(token, uid)
225			n_coin_get_stat = self.get_n_coin(token, uid)
226			video_view_stat = self.view_video(token, uid)
227			adfree_24h_stat = self.view_video_for_adfree_24h(token, uid)
228			msg = (
229			f"{signin_stat}\n"
230			f"------【每日签到得银币】------\n"
231			f"{silver_coin_get_stat}\n"
232			f"------【每日签到得N币】------\n"
233			f"{n_coin_get_stat}\n"
234			f"------【每天看两次视频】------\n"
235			f"{video_view_stat}\n"
236			f"------【看视频免广告(限时任务)】------\n"
237			f"{adfree_24h_stat}\n"
238			)
239			except Exception as e:
240			msg = str(e)
241			msg_all += msg + "\n\n"
242			return msg_all
243
244
245			if __name__ == "__main__":
246			_data = get_data()
247			_check_items = _data.get("NGA", [])
248			result = NGA(check_items=_check_items).main()
249			send("NGA", result)
250

OreosLab / checkinpanel

Branch — master (67948d)

ck_nga.NGA.view_video() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like