ck_site.Site.haidan() - Code Metrics - Inspection of ":racehorse: Improve code quality" - OreosLab/checkinpanel - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (8b9c3e)

by Leon

created 2022-10-14 07:44 UTC

ck_site.Site.haidan() A

↳ Parent: ck_site

Complexity

Conditions

Size

Total Lines	14
Code Lines	13

Duplication

Lines	14
Ratio	100 %

Importance

Changes

Metric	Value
cc	4
eloc	13
nop	2
dl	14
loc	14
rs	9.75
c	0
b	0
f	0

# -*- coding: utf-8 -*-
"""
:author @gnodgl
cron: 11 6 * * *
new Env('Site');
"""

import json
import re
from json.decoder import JSONDecodeError

import requests

from notify_mtr import send
from utils import get_data

desp = ""


def log(info):
    global desp
    desp = desp + info + "\n\n"


class Site:
    def __init__(self, check_items):
        self.check_items = check_items
        self.error_tip = "cookie 已过期或网站类型不对"
        self.url = ""

    @staticmethod
    def generate_headers(url):
        return {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.82 Safari/537.36",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Referer": url,
        }

    @staticmethod
    def cookie_parse(cookie_str):
        cookie_dict = {}
        cookies = cookie_str.split(";")
        for cookie in cookies:
            cookie = cookie.split("=")
            cookie_dict[cookie[0]] = cookie[1]
        return cookie_dict

    def signin(self, session, url):
        # hdarea 签到
        if url == "https://www.hdarea.co":
            self.hdarea(session, url)
        elif url == "https://pterclub.com":
            self.pterclub(session, url)
        elif url == "https://www.haidan.video":
            self.haidan(session, url)
        elif url == "https://pt.btschool.club":
            self.btschool(session, url)
        elif url == "https://lemonhd.org":
            self.lemonhd(session, url)
        elif url in ["https://hdtime.org", "https://www.pttime.org"]:
            self.hdtime(session, url)
        else:
            self.common(session, url)

    def common(self, session, url):
        attendance_url = f"{url}/attendance.php"
        with session.get(attendance_url) as response:
            r = re.compile(r"请勿重复刷新")
            r1 = re.compile(r"签到已得\s*\d+")
            location = r1.search(response.text).span()
            if r.search(response.text):
                tip = "重复签到"
            elif location:
                tip = response.text[location[0], location[1]]
            else:
                tip = self.url
                print(f"{url} {tip}")
            log(f"{url} {tip}")

    @staticmethod

    def hdtime(session, url):
        attendance_url = f"{url}/attendance.php"
        with session.get(attendance_url) as response:
            r = re.compile(r"签到成功")
            r1 = re.compile(r"请勿重复刷新")
            if r.search(response.text):
                tip = "签到成功"
            elif r1.search(response.text):
                tip = "重复签到"
            else:
                tip = "cookie已过期"
                print(f"{url} {tip}")
            log(f"{url} {tip}")

    def lemonhd(self, session, url):

        attendance_url = f"{url}/attendance.php"
        with session.get(attendance_url) as response:
            r = re.compile(r"已签到")
            r1 = re.compile(r"请勿重复刷新")
            if r.search(response.text):
                tip = "签到成功"
            elif r1.search(response.text):
                tip = "重复签到"
            else:
                tip = self.url
                print(f"{url} {tip}")
            log(f"{url} {tip}")

    @staticmethod
    def btschool(session, url):
        attendance_url = f"{url}/index.php?action=addbonus"
        with session.get(attendance_url) as response:
            r = re.compile(r"今天签到您获得\d+点魔力值")
            r1 = re.compile(r"退出")
            location = r.search(response.text)
            if location:
                tip = location.group()
            elif r1.search(response.text):
                tip = "重复签到"
            else:
                tip = "cookie已过期"
                print(f"{url} {tip}")
            log(f"{url} {tip}")

    @staticmethod

    def haidan(session, url):
        attendance_url = f"{url}/signin.php"
        with session.get(attendance_url) as response:
            r = re.compile(r"已经打卡")
            r1 = re.compile(r"退出")
            if r.search(response.text):
                tip = "签到成功"
            elif r1.search(response.text):
                tip = "重复签到"
            else:
                tip = "cookie 已过期或网站类型不对!"
                print(f"{url} {tip}")
            log(f"{url} {tip}")

    def pterclub(self, session, url):
        attendance_url = f"{url}/attendance-ajax.php"
        with session.get(attendance_url) as response:
            try:
                msg = json.loads(
                    response.text.encode("utf-8").decode("unicode-escape")
                ).get("message")
            except JSONDecodeError:
                msg = response.text
            if "连续签到" in msg:
                pattern = re.compile(r"</?.>")
                tip = f"签到成功, {re.sub(pattern, '', msg)}"
            elif "重复刷新" in msg:
                tip = "重复签到"
            else:
                tip = self.url
                print(f"{url} {tip}")
            log(f"{url} {tip}")

    def hdarea(self, session, url):

        attendance_url = f"{url}/sign_in.php"
        data = {"action": "sign_in"}
        with session.post(attendance_url, data) as response:
            r = re.compile(r"获得了\d+魔力值")
            r1 = re.compile(r"重复")
            log(response.text)
            if r.search(response.text):
                tip = "签到成功"
            elif r1.search(response.text):
                tip = "重复签到"
            else:
                tip = self.url
                print(f"{url} {tip}")
            log(f"{url} {tip}")

    @staticmethod
    def signin_discuz_dsu(session, url):
        attendance_url = (
            f"{url}/plugin.php?"
            f"id=dsu_paulsign:sign&operation=qiandao&infloat=1&sign_as=1&inajax=1"
        )
        hash_url = f"{url}/plugin.php?id=dsu_paulsign:sign"
        with session.get(hash_url) as hashurl:
            h = re.compile(r'name="formhash" value="(.*?)"')
            formhash = h.search(hashurl.text)[1]
        data = {
            "qdmode": 3,
            "qdxq": "kx",
            "fastreply": 0,
            "formhash": formhash,
            "todaysay": "",
        }
        with session.post(attendance_url, data) as response:
            r = re.compile(r"签到成功")
            r1 = re.compile(r"已经签到")
            if r.search(response.text):
                log(f"{url} 签到成功")
            elif r1.search(response.text):
                log(f"{url} 重复签到")
            else:
                log(f"{url} {response.text}")
                print(f"{url} {response.text}")

    @staticmethod
    def signin_hifi(session, url):
        attendance_url = f"{url}/sg_sign.htm"
        with session.post(attendance_url) as response:
            r = re.compile(r"成功")
            r1 = re.compile(r"今天已经")
            if r.search(response.text):
                log(f"{url} 签到成功")
            elif r1.search(response.text):
                log(f"{url} 重复签到")
            else:
                log(f"{url} {response.text}")
                print(f"{url} {response.text}")

    def main(self):
        for check_item in self.check_items:
            s = requests.session()
            url = check_item.get("url")
            site_type = check_item.get("type")
            cookie = self.cookie_parse(check_item.get("cookie"))
            header = self.generate_headers(url)
            s.headers.update(header)
            s.cookies.update(cookie)
            if site_type == "pt":
                self.signin(s, url)
            elif site_type == "discuz":
                self.signin_discuz_dsu(s, url)
            elif site_type == "hifi":
                self.signin_hifi(s, url)
            else:
                log("请在配置文件中配置网站类型，如 type: 'pt'")
                print("请在配置文件中配置网站类型，如 type: 'pt'")
        return desp


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


1		# -- coding: utf-8 --
2		"""
3		:author @gnodgl
4		cron: 11 6 * * *
5		new Env('Site');
6		"""
7
8		import json
9		import re
10		from json.decoder import JSONDecodeError
11
12		import requests
13
14		from notify_mtr import send
15		from utils import get_data
16
17		desp = ""
18
19
20		def log(info):
21		global desp
22		desp = desp + info + "\n\n"
23
24
25		class Site:
26		def __init__(self, check_items):
27		self.check_items = check_items
28		self.error_tip = "cookie 已过期或网站类型不对"
29		self.url = ""
30
31		@staticmethod
32		def generate_headers(url):
33		return {
34		"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
35		"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.82 Safari/537.36",
36		"Accept-Language": "zh-CN,zh;q=0.9",
37		"Referer": url,
38		}
39
40		@staticmethod
41		def cookie_parse(cookie_str):
42		cookie_dict = {}
43		cookies = cookie_str.split(";")
44		for cookie in cookies:
45		cookie = cookie.split("=")
46		cookie_dict[cookie[0]] = cookie[1]
47		return cookie_dict
48
49		def signin(self, session, url):
50		# hdarea 签到
51		if url == "https://www.hdarea.co":
52		self.hdarea(session, url)
53		elif url == "https://pterclub.com":
54		self.pterclub(session, url)
55		elif url == "https://www.haidan.video":
56		self.haidan(session, url)
57		elif url == "https://pt.btschool.club":
58		self.btschool(session, url)
59		elif url == "https://lemonhd.org":
60		self.lemonhd(session, url)
61		elif url in ["https://hdtime.org", "https://www.pttime.org"]:
62		self.hdtime(session, url)
63		else:
64		self.common(session, url)
65
66		def common(self, session, url):
67		attendance_url = f"{url}/attendance.php"
68		with session.get(attendance_url) as response:
69		r = re.compile(r"请勿重复刷新")
70		r1 = re.compile(r"签到已得\s*\d+")
71		location = r1.search(response.text).span()
72		if r.search(response.text):
73		tip = "重复签到"
74		elif location:
75		tip = response.text[location[0], location[1]]
76		else:
77		tip = self.url
78		print(f"{url} {tip}")
79		log(f"{url} {tip}")
80
81	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2022-10-14 07:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
82		def hdtime(session, url):
83		attendance_url = f"{url}/attendance.php"
84		with session.get(attendance_url) as response:
85		r = re.compile(r"签到成功")
86		r1 = re.compile(r"请勿重复刷新")
87		if r.search(response.text):
88		tip = "签到成功"
89		elif r1.search(response.text):
90		tip = "重复签到"
91		else:
92		tip = "cookie已过期"
93		print(f"{url} {tip}")
94		log(f"{url} {tip}")
95
96	View Code Duplication	def lemonhd(self, session, url):
		0 ignored issues – show Duplication introduced 2022-10-14 07:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
97		attendance_url = f"{url}/attendance.php"
98		with session.get(attendance_url) as response:
99		r = re.compile(r"已签到")
100		r1 = re.compile(r"请勿重复刷新")
101		if r.search(response.text):
102		tip = "签到成功"
103		elif r1.search(response.text):
104		tip = "重复签到"
105		else:
106		tip = self.url
107		print(f"{url} {tip}")
108		log(f"{url} {tip}")
109
110		@staticmethod
111		def btschool(session, url):
112		attendance_url = f"{url}/index.php?action=addbonus"
113		with session.get(attendance_url) as response:
114		r = re.compile(r"今天签到您获得\d+点魔力值")
115		r1 = re.compile(r"退出")
116		location = r.search(response.text)
117		if location:
118		tip = location.group()
119		elif r1.search(response.text):
120		tip = "重复签到"
121		else:
122		tip = "cookie已过期"
123		print(f"{url} {tip}")
124		log(f"{url} {tip}")
125
126	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2022-10-14 07:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
127		def haidan(session, url):
128		attendance_url = f"{url}/signin.php"
129		with session.get(attendance_url) as response:
130		r = re.compile(r"已经打卡")
131		r1 = re.compile(r"退出")
132		if r.search(response.text):
133		tip = "签到成功"
134		elif r1.search(response.text):
135		tip = "重复签到"
136		else:
137		tip = "cookie 已过期或网站类型不对!"
138		print(f"{url} {tip}")
139		log(f"{url} {tip}")
140
141		def pterclub(self, session, url):
142		attendance_url = f"{url}/attendance-ajax.php"
143		with session.get(attendance_url) as response:
144		try:
145		msg = json.loads(
146		response.text.encode("utf-8").decode("unicode-escape")
147		).get("message")
148		except JSONDecodeError:
149		msg = response.text
150		if "连续签到" in msg:
151		pattern = re.compile(r"</?.>")
152		tip = f"签到成功, {re.sub(pattern, '', msg)}"
153		elif "重复刷新" in msg:
154		tip = "重复签到"
155		else:
156		tip = self.url
157		print(f"{url} {tip}")
158		log(f"{url} {tip}")
159
160	View Code Duplication	def hdarea(self, session, url):
		0 ignored issues – show Duplication introduced 2022-10-14 07:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
161		attendance_url = f"{url}/sign_in.php"
162		data = {"action": "sign_in"}
163		with session.post(attendance_url, data) as response:
164		r = re.compile(r"获得了\d+魔力值")
165		r1 = re.compile(r"重复")
166		log(response.text)
167		if r.search(response.text):
168		tip = "签到成功"
169		elif r1.search(response.text):
170		tip = "重复签到"
171		else:
172		tip = self.url
173		print(f"{url} {tip}")
174		log(f"{url} {tip}")
175
176		@staticmethod
177		def signin_discuz_dsu(session, url):
178		attendance_url = (
179		f"{url}/plugin.php?"
180		f"id=dsu_paulsign:sign&operation=qiandao&infloat=1&sign_as=1&inajax=1"
181		)
182		hash_url = f"{url}/plugin.php?id=dsu_paulsign:sign"
183		with session.get(hash_url) as hashurl:
184		h = re.compile(r'name="formhash" value="(.*?)"')
185		formhash = h.search(hashurl.text)[1]
186		data = {
187		"qdmode": 3,
188		"qdxq": "kx",
189		"fastreply": 0,
190		"formhash": formhash,
191		"todaysay": "",
192		}
193		with session.post(attendance_url, data) as response:
194		r = re.compile(r"签到成功")
195		r1 = re.compile(r"已经签到")
196		if r.search(response.text):
197		log(f"{url} 签到成功")
198		elif r1.search(response.text):
199		log(f"{url} 重复签到")
200		else:
201		log(f"{url} {response.text}")
202		print(f"{url} {response.text}")
203
204		@staticmethod
205		def signin_hifi(session, url):
206		attendance_url = f"{url}/sg_sign.htm"
207		with session.post(attendance_url) as response:
208		r = re.compile(r"成功")
209		r1 = re.compile(r"今天已经")
210		if r.search(response.text):
211		log(f"{url} 签到成功")
212		elif r1.search(response.text):
213		log(f"{url} 重复签到")
214		else:
215		log(f"{url} {response.text}")
216		print(f"{url} {response.text}")
217
218		def main(self):
219		for check_item in self.check_items:
220		s = requests.session()
221		url = check_item.get("url")
222		site_type = check_item.get("type")
223		cookie = self.cookie_parse(check_item.get("cookie"))
224		header = self.generate_headers(url)
225		s.headers.update(header)
226		s.cookies.update(cookie)
227		if site_type == "pt":
228		self.signin(s, url)
229		elif site_type == "discuz":
230		self.signin_discuz_dsu(s, url)
231		elif site_type == "hifi":
232		self.signin_hifi(s, url)
233		else:
234		log("请在配置文件中配置网站类型，如 type: 'pt'")
235		print("请在配置文件中配置网站类型，如 type: 'pt'")
236		return desp
237
238
239		if __name__ == "__main__":
240		_data = get_data()
241		_check_items = _data.get("SITE", [])
242		result = Site(check_items=_check_items).main()
243		send("Site", result)
244

OreosLab / checkinpanel

Branch — master (8b9c3e)

ck_site.Site.haidan() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like