ck_heytap.Heytap.daily_viewgoods() - Code Metrics - Inspection of ":zap: 本地化日志输出：EUserv" - OreosLab/checkinpanel - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( c3d8b2...2da959 )

by Leon

created 2021-12-18 07:46 UTC

ck_heytap.Heytap.daily_viewgoods() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	57
Code Lines	47

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
eloc	47
nop	1
dl	0
loc	57
rs	5.4
c	0
b	0
f	0

How to fix Long Method Complexity

# -*- coding: utf-8 -*-
"""
:author @yyzhou94
cron: 19 5,21 * * *
new Env('欢太商城');
"""

import json
import re
import time

import requests

import utils_tmp
from notify_mtr import send
from utils import get_data


class Heytap:
    def __init__(self, check_items):
        self.check_items = check_items
        self.client = None
        self.session = requests.session()
        self.log = ""
        self.cookies = "cookie"
        self.user_agent = "ua"
        self.s_channel = "oppostore"
        self.source_type = "505"  # 初始化设置为505，会从cookie获取实际数据
        self.sa_distinct_id = ""
        self.sa_device_id = ""
        self.s_version = ""
        self.brand = "iPhone"  # 初始化设置为iPhone，会从cookie获取实际机型
        self.act_task = utils_tmp.act_list  # 修改为从文件获取，避免更新代码后丢失原有活动配置
        self.if_draw = False  # 初始化设置为False，会从配置文件获取实际设置

        self.accept = "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9"
        self.accept_encoding1 = "gzip, deflate, br"
        self.accept_encoding2 = "gzip, deflate"
        self.client_package = "com.oppo.store"
        self.content_type1 = "application/x-www-form-urlencoded"
        self.content_type2 = "application/x-www-form-urlencoded;charset=UTF-8"
        self.host1 = "store.oppo.com"
        self.host2 = "msec.opposhop.cn"
        self.user_agent2 = "okhttp/3.12.12.200sp1"

        self.point_got = "任务完成！积分领取+"
        self.point_err = "领取积分奖励出错！\n"
        self.done_msg = "任务已完成\n"
        self.err_msg = "错误，原因为: "
        self.amount = "amount="

    # 获取cookie里的一些参数，部分请求需要使用到  hss修改
    def get_cookie_data(self):
        try:
            app_param = re.findall("app_param=(.*?)}", self.cookies)[0] + "}"
            app_param = json.loads(app_param)
            self.sa_device_id = app_param["sa_device_id"]
            self.brand = app_param["brand"]
            self.sa_distinct_id = re.findall("sa_distinct_id=(.*?);", self.cookies)[0]
            self.source_type = re.findall("source_type=(.*?);", self.cookies)[0]
            self.s_version = re.findall("s_version=(.*?);", self.cookies)[0]
            self.s_channel = re.findall("s_channel=(.*?);", self.cookies)[0]
        except Exception as e:
            print(
                "获取Cookie部分数据失败，将采用默认设置，请检查Cookie是否包含s_channel，s_version，source_type，sa_distinct_id\n",
                e,
            )
            self.s_channel = "ios_oppostore"
            self.source_type = "505"

    # 获取个人信息，判断登录状态
    def get_user_info(self):
        flag = False
        headers = {
            "Host": "store.oppo.com",
            "Accept": self.accept,
            "Content-Type": self.content_type1,
            "Connection": "keep-alive",
            "User-Agent": self.user_agent,
            "Accept-Language": "zh-cn",
            "Accept-Encoding": self.accept_encoding1,
            "cookie": self.cookies,
        }
        response = self.session.get(
            "https://store.oppo.com/cn/oapi/users/web/member/info", headers=headers
        )
        response.encoding = "utf-8"
        try:
            result = response.json()
            if result["code"] == 200:
                self.log += f'======== {result["data"]["userName"]} ========\n'
                self.log += (
                    "【登录成功】："
                    + result["data"]["userName"]
                    + f"\n【抽奖开关】：{self.if_draw}\n"
                )
                flag = True
            else:
                self.log += "【登录失败】：" + result["errorMessage"] + "\n"
        except Exception as e:
            self.log += "【登录】：错误，原因为: " + str(e) + "\n"

        if flag:
            self.get_cookie_data()
            return self.session
        else:
            return False

    # 任务中心列表，获取任务及任务状态
    def taskCenter(self):
        headers = {
            "Host": self.host1,
            "Accept": self.accept,
            "Content-Type": self.content_type1,
            "Connection": "keep-alive",
            "User-Agent": self.user_agent,
            "Accept-Language": "zh-cn",
            "Accept-Encoding": self.accept_encoding1,
            "cookie": self.cookies,
            "referer": "https://store.oppo.com/cn/app/taskCenter/index",
        }
        res1 = self.client.get(
            "https://store.oppo.com/cn/oapi/credits/web/credits/show", headers=headers
        )
        res1 = res1.json()
        return res1

    # 每日签到
    # 位置: APP → 我的 → 签到
    def daily_bonus(self):
        try:
            dated = time.strftime("%Y-%m-%d")
            headers = {
                "Host": self.host1,
                "Accept": self.accept,
                "Content-Type": self.content_type1,
                "Connection": "keep-alive",
                "User-Agent": self.user_agent,
                "Accept-Language": "zh-cn",
                "Accept-Encoding": self.accept_encoding1,
                "cookie": self.cookies,
                "referer": "https://store.oppo.com/cn/app/taskCenter/index",
            }
            res = self.taskCenter()
            status = res["data"]["userReportInfoForm"]["status"]
            if status == 0:
                res = res["data"]["userReportInfoForm"]["gifts"]
                for data in res:
                    if data["date"] == dated:
                        qd = data
                        if not qd["today"]:
                            data = self.amount + str(qd["credits"])
                            res1 = self.client.post(
                                "https://store.oppo.com/cn/oapi/credits/web/report/immediately",
                                headers=headers,
                                data=data,
                            )
                            res1 = res1.json()
                            if res1["code"] == 200:
                                self.log += "【每日签到成功】：" + res1["data"]["message"] + "\n"
                            else:
                                self.log += "【每日签到失败】：" + res1 + "\n"
                        else:
                            # print(str(qd["credits"]), str(qd["type"]), str(qd["gift"]))
                            if not qd["type"]:
                                data = self.amount + str(qd["credits"])
                            else:
                                data = (
                                    self.amount
                                    + str(qd["credits"])
                                    + "&type="
                                    + str(qd["type"])
                                    + "&gift="
                                    + str(qd["gift"])
                                )
                            res1 = self.client.post(
                                "https://store.oppo.com/cn/oapi/credits/web/report/immediately",
                                headers=headers,
                                data=data,
                            )
                            res1 = res1.json()
                            if res1["code"] == 200:
                                self.log += "【每日签到成功】：" + res1["data"]["message"] + "\n"
                            else:
                                self.log += "【每日签到失败】：" + str(res1) + "\n"
            else:
                self.log += "【每日签到】：已经签到过了！\n"
            time.sleep(1)
        except Exception as e:
            self.log += "【每日签到】：错误，原因为: " + str(e) + "\n"

    # 浏览商品 10个sku +20 分
    # 位置: APP → 我的 → 签到 → 每日任务 → 浏览商品
    def daily_viewgoods(self):
        self.log += "【每日浏览商品】\n"
        try:
            headers = {
                "clientPackage": self.client_package,
                "Host": self.host2,
                "Accept": self.accept,
                "Content-Type": self.content_type1,
                "Connection": "keep-alive",
                "User-Agent": self.user_agent2,
                "Accept-Encoding": "gzip",
                "cookie": self.cookies,
            }
            res = self.taskCenter()
            res = res["data"]["everydayList"]
            for data in res:
                if data["name"] == "浏览商品":
                    if data["completeStatus"] == 0:
                        # 原链接貌似获取不到商品id，更换一个 原链接https://msec.opposhop.cn/goods/v1/SeckillRound/goods/3016?pageSize=12&currentPage=1
                        shoplist = self.client.get(
                            "https://msec.opposhop.cn/goods/v1/products/010239"
                        )
                        res = shoplist.json()
                        if res["meta"]["code"] == 200:
                            i = 0
                            for skuinfo in res["details"][0]["infos"]:
                                skuid = skuinfo["skuId"]
                                self.client.get(
                                    "https://msec.opposhop.cn/goods/v1/info/sku?skuId="
                                    + str(skuid),
                                    headers=headers,
                                )
                                i += 1
                                if i > 10:
                                    break
                                time.sleep(7)
                            res2 = self.cashingCredits(
                                data["marking"], data["type"], data["credits"]
                            )
                            if res2:
                                self.log += self.point_got + str(data["credits"]) + "\n"
                            else:
                                self.log += self.point_err
                        else:
                            self.log += "错误，获取商品列表失败\n"
                    elif data["completeStatus"] == 1:
                        res2 = self.cashingCredits(
                            data["marking"], data["type"], data["credits"]
                        )
                        if res2:
                            self.log += self.point_got + str(data["credits"]) + "\n"
                        else:
                            self.log += self.point_err
                    else:
                        self.log += self.done_msg
        except Exception as e:
            self.log += self.err_msg + str(e) + "\n"

    def daily_sharegoods(self):
        self.log += "【每日分享商品】\n"
        try:
            headers = {
                "clientPackage": self.client_package,
                "Host": self.host2,
                "Accept": self.accept,
                "Content-Type": self.content_type1,
                "Connection": "keep-alive",
                "User-Agent": self.user_agent2,
                "Accept-Encoding": "gzip",
                "cookie": self.cookies,
            }
            daysignlist = self.taskCenter()
            res = daysignlist
            res = res["data"]["everydayList"]
            for data in res:
                if data["name"] == "分享商品到微信":
                    qd = data
            if qd["completeStatus"] == 0:
                count = qd["readCount"]
                endcount = qd["times"]
                while count <= endcount:
                    self.client.get(
                        "https://msec.opposhop.cn/users/vi/creditsTask/pushTask?marking=daily_sharegoods",
                        headers=headers,
                    )
                    count += 1
                res2 = self.cashingCredits(qd["marking"], qd["type"], qd["credits"])
                if res2:
                    self.log += self.point_got + str(qd["credits"]) + "\n"
                else:
                    self.log += self.point_err
            elif qd["completeStatus"] == 1:
                res2 = self.cashingCredits(qd["marking"], qd["type"], qd["credits"])
                if res2:
                    self.log += self.point_got + str(qd["credits"]) + "\n"
                else:
                    self.log += self.point_err
            else:
                self.log += self.done_msg
        except Exception as e:
            self.log += self.err_msg + str(e) + "\n"

    # 执行完成任务领取奖励
    def cashingCredits(self, info_marking, info_type, info_credits):
        headers = {
            "Host": self.host1,
            "clientPackage": self.client_package,
            "Accept": "application/json, text/plain, */*",
            "Content-Type": self.content_type1,
            "Connection": "keep-alive",
            "User-Agent": self.user_agent,
            "Accept-Language": "zh-cn",
            "Accept-Encoding": self.accept_encoding1,
            "cookie": self.cookies,
            "Origin": "https://store.oppo.com",
            "X-Requested-With": self.client_package,
            "referer": "https://store.oppo.com/cn/app/taskCenter/index?us=gerenzhongxin&um=hudongleyuan&uc=renwuzhongxin",
        }

        data = (
            "marking="
            + str(info_marking)
            + "&type="
            + str(info_type)
            + "&amount="
            + str(info_credits)
        )

        res = self.client.post(
            "https://store.oppo.com/cn/oapi/credits/web/credits/cashingCredits",
            data=data,
            headers=headers,
        )

        res = res.json()

        if res["code"] == 200:
            return True
        else:
            return False

    # 活动平台抽奖通用接口
    def lottery(self, datas, referer="", extra_draw_cookie=""):

        headers = {
            "Host": "hd.oppo.com",
            "User-Agent": self.user_agent,
            "Cookie": extra_draw_cookie + self.cookies,
            "Referer": referer,
            "Accept": "application/json, text/javascript, */*; q=0.01",
            "Accept-Language": "zh-cn",
            "Accept-Encoding": "br, gzip, deflate",
            "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        }
        res = self.client.get("https://hd.oppo.com/user/login", headers=headers).json()
        if res["no"] == "200":
            res = self.client.post(
                "https://hd.oppo.com/platform/lottery", data=datas, headers=headers
            )
            res = res.json()
            return res
        else:
            return res

    # 活动平台完成任务接口
    def task_finish(self, aid, t_index):
        headers = {
            "Accept": "application/json, text/plain, */*;q=0.01",
            "Content-Type": self.content_type2,
            "Connection": "keep-alive",
            "User-Agent": self.user_agent,
            "Accept-Encoding": self.accept_encoding2,
            "cookie": self.cookies,
            "Origin": "https://hd.oppo.com",
            "X-Requested-With": "XMLHttpRequest",
        }
        datas = "aid=" + str(aid) + "&t_index=" + str(t_index)
        res = self.client.post(
            "https://hd.oppo.com/task/finish", data=datas, headers=headers
        )
        res = res.json()
        return res

    # 活动平台领取任务奖励接口
    def task_award(self, aid, t_index):
        headers = {
            "Accept": "application/json, text/plain, */*;q=0.01",
            "Content-Type": self.content_type2,
            "Connection": "keep-alive",
            "User-Agent": self.user_agent,
            "Accept-Encoding": self.accept_encoding2,
            "cookie": self.cookies,
            "Origin": "https://hd.oppo.com",
            "X-Requested-With": "XMLHttpRequest",
        }
        datas = "aid=" + str(aid) + "&t_index=" + str(t_index)
        res = self.client.post(
            "https://hd.oppo.com/task/award", data=datas, headers=headers
        )
        res = res.json()
        return res

    # 做活动任务和抽奖通用接口  hss修改
    def do_task_and_draw(self):
        try:

            for act_list in self.act_task:
                act_name = act_list["act_name"]
                aid = act_list["aid"]
                referer = act_list["referer"]
                if_draw = act_list["if_draw"]
                if_task = act_list["if_task"]
                end_time = act_list["end_time"]
                headers = {
                    "Accept": "application/json, text/javascript, */*; q=0.01",
                    "Connection": "keep-alive",
                    "User-Agent": self.user_agent,
                    "Accept-Encoding": self.accept_encoding2,
                    "cookie": self.cookies,
                    "X-Requested-With": "XMLHttpRequest",
                    "Referer": referer,
                }
                dated = int(time.time())
                end_time = time.mktime(
                    time.strptime(end_time, "%Y-%m-%d %H:%M:%S")
                )  # 设置活动结束日期

                if dated < end_time:
                    if if_task:
                        res = self.client.get(
                            f"https://hd.oppo.com/task/list?aid={aid}", headers=headers
                        )
                        tasklist = res.json()
                        self.log += f"【{act_name}-任务】\n"
                        for _, jobs in enumerate(tasklist["data"]):
                            title = jobs["title"]
                            t_index = jobs["t_index"]
                            aid = t_index[: t_index.index("i")]
                            if jobs["t_status"] == 0:
                                finishmsg = self.task_finish(aid, t_index)
                                if finishmsg["no"] == "200":
                                    time.sleep(1)
                                    awardmsg = self.task_award(aid, t_index)
                                    msg = awardmsg["msg"]
                                    self.log += f"{title}：{msg}\n"
                                    time.sleep(3)
                            elif jobs["t_status"] == 1:
                                awardmsg = self.task_award(aid, t_index)
                                msg = awardmsg["msg"]
                                self.log += f"{title}：{msg}\n"
                                time.sleep(3)
                            else:
                                self.log += f"{title}：任务已完成\n"
                    if self.if_draw and if_draw:  # 判断当前用户是否抽奖 和 判断当前活动是否抽奖
                        lid = act_list["lid"]
                        extra_draw_cookie = act_list["extra_draw_cookie"]
                        draw_times = act_list["draw_times"]
                        self.log += f"【{act_name}-抽奖】："
                        x = 0
                        while x < draw_times:
                            data = f"aid={aid}&lid={lid}&mobile=&authcode=&captcha=&isCheck=0&source_type={self.source_type}&s_channel={self.s_channel}&sku=&spu="
                            res = self.lottery(data, referer, extra_draw_cookie)
                            msg = res["msg"]
                            if "次数已用完" in msg:
                                self.log += "  第" + str(x + 1) + "抽奖：抽奖次数已用完\n"
                                break
                            if "活动已结束" in msg:
                                self.log += "  第" + str(x + 1) + "抽奖：活动已结束，终止抽奖\n"
                                break
                            goods_name = res["data"]["goods_name"]
                            if goods_name:
                                self.log += (
                                    "  第" + str(x + 1) + "次抽奖：" + str(goods_name) + "\n"
                                )
                            elif "提交成功" in msg:
                                self.log += "  第" + str(x + 1) + "次抽奖：未中奖\n"
                            x += 1
                            time.sleep(5)
                else:
                    self.log += f"【{act_name}】：活动已结束，不再执行\n"
        except Exception as e:
            self.log += "【执行任务和抽奖】：错误，原因为: " + str(e) + "\n"

    # 早睡打卡
    def zaoshui_task(self):
        try:
            headers = {
                "Host": self.host1,
                "Connection": "keep-alive",
                "s_channel": self.s_channel,
                "utm_term": "direct",
                "utm_campaign": "direct",
                "utm_source": "direct",
                "ut": "direct",
                "uc": "zaoshuidaka",
                "sa_device_id": self.sa_device_id,
                "guid": self.sa_device_id,
                "sa_distinct_id": self.sa_distinct_id,
                "clientPackage": self.client_package,
                "Cache-Control": "no-cache",
                "um": "hudongleyuan",
                "User-Agent": self.user_agent,
                "ouid": "",
                "Accept": "application/json, text/plain, */*",
                "source_type": self.source_type,
                "utm_medium": "direct",
                "brand": "iPhone",
                "appId": "",
                "s_version": self.s_version,
                "us": "gerenzhongxin",
                "appKey": "",
                "X-Requested-With": self.client_package,
                "Referer": "https://store.oppo.com/cn/app/cardingActivities?utm_source=opposhop&utm_medium=task",
                "Accept-Encoding": self.accept_encoding2,
                "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
                "cookie": self.cookies,
            }
            res = self.client.get(
                "https://store.oppo.com/cn/oapi/credits/web/clockin/applyOrClockIn",
                headers=headers,
            ).json()
            if "余额不足" in str(res):
                self.log += "【早睡打卡】：申请失败，积分余额不足\n"
            else:
                applyStatus = res["data"]["applyStatus"]
                # 2 19:30打卡成功 0 8：00打卡成功
                if applyStatus == 1:
                    self.log += "【早睡打卡】：报名成功，请当天19:30-22:00留意打卡状态\n"
                if applyStatus == 0:
                    self.log += "【早睡打卡】：申请失败，积分不足或报名时间已过\n"
                if applyStatus == 2:
                    self.log += "【早睡打卡】：已报名成功，请当天19:30-22:00留意打卡状态\n"
                if res["data"]["clockInStatus"] == 1:
                    self.log += "【早睡打卡】：打卡成功，积分将于24:00前到账\n"
                if res["data"]["clockInStatus"] == 2:
                    self.log += "【早睡打卡】：打卡成功，积分将于24:00前到账\n"
            # 打卡记录
            res = self.client.get(
                "https://store.oppo.com/cn/oapi/credits/web/clockin/getMyRecord",
                headers=headers,
            ).json()
            if res["code"] == 200:
                record = res["data"]["everydayRecordForms"]
                self.log += "【早睡打卡记录】\n"
                i = 0
                for data in record:
                    self.log += (
                        data["everydayDate"]
                        + "-"
                        + data["applyClockInStatus"]
                        + "："
                        + data["credits"]
                        + "\n"
                    )
                    i += 1
                    if i == 4:  # 最多显示最近4条记录
                        break

        except Exception as e:
            self.log += "【早睡打卡】：错误，原因为: " + str(e) + "\n"

    # 主程序
    def main(self):
        i = 1
        for check_item in self.check_items:
            self.cookies = check_item.get("cookie")
            self.user_agent = check_item.get("useragent")
            self.if_draw = check_item.get("draw")
            self.client = self.get_user_info()
            if self.client:
                try:
                    self.daily_bonus()  # 执行每日签到
                    self.daily_viewgoods()  # 执行每日商品浏览任务
                    self.daily_sharegoods()  # 执行每日商品分享任务
                    self.do_task_and_draw()  # 自己修改的接口，针对活动任务及抽奖，新增及删除活动请修改act_list.py
                    # self.zaoshui_task()  # 早睡报名 由于自己不能及时更新cookie，就关闭了打卡
                except Exception as e:
                    self.log += f"账号{i}执行出错：{e}\n"
            else:
                self.log += f"账号{i}已失效，请及时更新cookies\n"
            i += 1
            self.log += "\n\n"
        return self.log


if __name__ == "__main__":
    data = get_data()
    _check_items = data.get("HEYTAP", [])
    res = Heytap(check_items=_check_items).main()
    send("欢太商城", res)


1			# -- coding: utf-8 --
2			"""
3			:author @yyzhou94
4			cron: 19 5,21 * * *
5			new Env('欢太商城');
6			"""
7
8			import json
9			import re
10			import time
11
12			import requests
13
14			import utils_tmp
15			from notify_mtr import send
16			from utils import get_data
17
18
19			class Heytap:
20			def __init__(self, check_items):
21			self.check_items = check_items
22			self.client = None
23			self.session = requests.session()
24			self.log = ""
25			self.cookies = "cookie"
26			self.user_agent = "ua"
27			self.s_channel = "oppostore"
28			self.source_type = "505" # 初始化设置为505，会从cookie获取实际数据
29			self.sa_distinct_id = ""
30			self.sa_device_id = ""
31			self.s_version = ""
32			self.brand = "iPhone" # 初始化设置为iPhone，会从cookie获取实际机型
33			self.act_task = utils_tmp.act_list # 修改为从文件获取，避免更新代码后丢失原有活动配置
34			self.if_draw = False # 初始化设置为False，会从配置文件获取实际设置
35
36			self.accept = "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,/;q=0.8,application/signed-exchange;v=b3;q=0.9"
37			self.accept_encoding1 = "gzip, deflate, br"
38			self.accept_encoding2 = "gzip, deflate"
39			self.client_package = "com.oppo.store"
40			self.content_type1 = "application/x-www-form-urlencoded"
41			self.content_type2 = "application/x-www-form-urlencoded;charset=UTF-8"
42			self.host1 = "store.oppo.com"
43			self.host2 = "msec.opposhop.cn"
44			self.user_agent2 = "okhttp/3.12.12.200sp1"
45
46			self.point_got = "任务完成！积分领取+"
47			self.point_err = "领取积分奖励出错！\n"
48			self.done_msg = "任务已完成\n"
49			self.err_msg = "错误，原因为: "
50			self.amount = "amount="
51
52			# 获取cookie里的一些参数，部分请求需要使用到 hss修改
53			def get_cookie_data(self):
54			try:
55			app_param = re.findall("app_param=(.*?)}", self.cookies)[0] + "}"
56			app_param = json.loads(app_param)
57			self.sa_device_id = app_param["sa_device_id"]
58			self.brand = app_param["brand"]
59			self.sa_distinct_id = re.findall("sa_distinct_id=(.*?);", self.cookies)[0]
60			self.source_type = re.findall("source_type=(.*?);", self.cookies)[0]
61			self.s_version = re.findall("s_version=(.*?);", self.cookies)[0]
62			self.s_channel = re.findall("s_channel=(.*?);", self.cookies)[0]
63			except Exception as e:
64			print(
65			"获取Cookie部分数据失败，将采用默认设置，请检查Cookie是否包含s_channel，s_version，source_type，sa_distinct_id\n",
66			e,
67			)
68			self.s_channel = "ios_oppostore"
69			self.source_type = "505"
70
71			# 获取个人信息，判断登录状态
72			def get_user_info(self):
73			flag = False
74			headers = {
75			"Host": "store.oppo.com",
76			"Accept": self.accept,
77			"Content-Type": self.content_type1,
78			"Connection": "keep-alive",
79			"User-Agent": self.user_agent,
80			"Accept-Language": "zh-cn",
81			"Accept-Encoding": self.accept_encoding1,
82			"cookie": self.cookies,
83			}
84			response = self.session.get(
85			"https://store.oppo.com/cn/oapi/users/web/member/info", headers=headers
86			)
87			response.encoding = "utf-8"
88			try:
89			result = response.json()
90			if result["code"] == 200:
91			self.log += f'======== {result["data"]["userName"]} ========\n'
92			self.log += (
93			"【登录成功】："
94			+ result["data"]["userName"]
95			+ f"\n【抽奖开关】：{self.if_draw}\n"
96			)
97			flag = True
98			else:
99			self.log += "【登录失败】：" + result["errorMessage"] + "\n"
100			except Exception as e:
101			self.log += "【登录】：错误，原因为: " + str(e) + "\n"
102
103			if flag:
104			self.get_cookie_data()
105			return self.session
106			else:
107			return False
108
109			# 任务中心列表，获取任务及任务状态
110			def taskCenter(self):
111			headers = {
112			"Host": self.host1,
113			"Accept": self.accept,
114			"Content-Type": self.content_type1,
115			"Connection": "keep-alive",
116			"User-Agent": self.user_agent,
117			"Accept-Language": "zh-cn",
118			"Accept-Encoding": self.accept_encoding1,
119			"cookie": self.cookies,
120			"referer": "https://store.oppo.com/cn/app/taskCenter/index",
121			}
122			res1 = self.client.get(
123			"https://store.oppo.com/cn/oapi/credits/web/credits/show", headers=headers
124			)
125			res1 = res1.json()
126			return res1
127
128			# 每日签到
129			# 位置: APP → 我的 → 签到
130			def daily_bonus(self):
131			try:
132			dated = time.strftime("%Y-%m-%d")
133			headers = {
134			"Host": self.host1,
135			"Accept": self.accept,
136			"Content-Type": self.content_type1,
137			"Connection": "keep-alive",
138			"User-Agent": self.user_agent,
139			"Accept-Language": "zh-cn",
140			"Accept-Encoding": self.accept_encoding1,
141			"cookie": self.cookies,
142			"referer": "https://store.oppo.com/cn/app/taskCenter/index",
143			}
144			res = self.taskCenter()
145			status = res["data"]["userReportInfoForm"]["status"]
146			if status == 0:
147			res = res["data"]["userReportInfoForm"]["gifts"]
148			for data in res:
149			if data["date"] == dated:
150			qd = data
151			if not qd["today"]:
152			data = self.amount + str(qd["credits"])
153			res1 = self.client.post(
154			"https://store.oppo.com/cn/oapi/credits/web/report/immediately",
155			headers=headers,
156			data=data,
157			)
158			res1 = res1.json()
159			if res1["code"] == 200:
160			self.log += "【每日签到成功】：" + res1["data"]["message"] + "\n"
161			else:
162			self.log += "【每日签到失败】：" + res1 + "\n"
163			else:
164			# print(str(qd["credits"]), str(qd["type"]), str(qd["gift"]))
165			if not qd["type"]:
166			data = self.amount + str(qd["credits"])
167			else:
168			data = (
169			self.amount
170			+ str(qd["credits"])
171			+ "&type="
172			+ str(qd["type"])
173			+ "&gift="
174			+ str(qd["gift"])
175			)
176			res1 = self.client.post(
177			"https://store.oppo.com/cn/oapi/credits/web/report/immediately",
178			headers=headers,
179			data=data,
180			)
181			res1 = res1.json()
182			if res1["code"] == 200:
183			self.log += "【每日签到成功】：" + res1["data"]["message"] + "\n"
184			else:
185			self.log += "【每日签到失败】：" + str(res1) + "\n"
186			else:
187			self.log += "【每日签到】：已经签到过了！\n"
188			time.sleep(1)
189			except Exception as e:
190			self.log += "【每日签到】：错误，原因为: " + str(e) + "\n"
191
192			# 浏览商品 10个sku +20 分
193			# 位置: APP → 我的 → 签到 → 每日任务 → 浏览商品
194			def daily_viewgoods(self):
195			self.log += "【每日浏览商品】\n"
196			try:
197			headers = {
198			"clientPackage": self.client_package,
199			"Host": self.host2,
200			"Accept": self.accept,
201			"Content-Type": self.content_type1,
202			"Connection": "keep-alive",
203			"User-Agent": self.user_agent2,
204			"Accept-Encoding": "gzip",
205			"cookie": self.cookies,
206			}
207			res = self.taskCenter()
208			res = res["data"]["everydayList"]
209			for data in res:
210			if data["name"] == "浏览商品":
211			if data["completeStatus"] == 0:
212			# 原链接貌似获取不到商品id，更换一个原链接https://msec.opposhop.cn/goods/v1/SeckillRound/goods/3016?pageSize=12&currentPage=1
213			shoplist = self.client.get(
214			"https://msec.opposhop.cn/goods/v1/products/010239"
215			)
216			res = shoplist.json()
217			if res["meta"]["code"] == 200:
218			i = 0
219			for skuinfo in res["details"][0]["infos"]:
220			skuid = skuinfo["skuId"]
221			self.client.get(
222			"https://msec.opposhop.cn/goods/v1/info/sku?skuId="
223			+ str(skuid),
224			headers=headers,
225			)
226			i += 1
227			if i > 10:
228			break
229			time.sleep(7)
230			res2 = self.cashingCredits(
231			data["marking"], data["type"], data["credits"]
232			)
233			if res2:
234			self.log += self.point_got + str(data["credits"]) + "\n"
235			else:
236			self.log += self.point_err
237			else:
238			self.log += "错误，获取商品列表失败\n"
239			elif data["completeStatus"] == 1:
240			res2 = self.cashingCredits(
241			data["marking"], data["type"], data["credits"]
242			)
243			if res2:
244			self.log += self.point_got + str(data["credits"]) + "\n"
245			else:
246			self.log += self.point_err
247			else:
248			self.log += self.done_msg
249			except Exception as e:
250			self.log += self.err_msg + str(e) + "\n"
251
252			def daily_sharegoods(self):
253			self.log += "【每日分享商品】\n"
254			try:
255			headers = {
256			"clientPackage": self.client_package,
257			"Host": self.host2,
258			"Accept": self.accept,
259			"Content-Type": self.content_type1,
260			"Connection": "keep-alive",
261			"User-Agent": self.user_agent2,
262			"Accept-Encoding": "gzip",
263			"cookie": self.cookies,
264			}
265			daysignlist = self.taskCenter()
266			res = daysignlist
267			res = res["data"]["everydayList"]
268			for data in res:
269			if data["name"] == "分享商品到微信":
270			qd = data
271			if qd["completeStatus"] == 0:
272			count = qd["readCount"]
273			endcount = qd["times"]
274			while count <= endcount:
275			self.client.get(
276			"https://msec.opposhop.cn/users/vi/creditsTask/pushTask?marking=daily_sharegoods",
277			headers=headers,
278			)
279			count += 1
280			res2 = self.cashingCredits(qd["marking"], qd["type"], qd["credits"])
281			if res2:
282			self.log += self.point_got + str(qd["credits"]) + "\n"
283			else:
284			self.log += self.point_err
285			elif qd["completeStatus"] == 1:
286			res2 = self.cashingCredits(qd["marking"], qd["type"], qd["credits"])
287			if res2:
288			self.log += self.point_got + str(qd["credits"]) + "\n"
289			else:
290			self.log += self.point_err
291			else:
292			self.log += self.done_msg
293			except Exception as e:
294			self.log += self.err_msg + str(e) + "\n"
295
296			# 执行完成任务领取奖励
297			def cashingCredits(self, info_marking, info_type, info_credits):
298			headers = {
299			"Host": self.host1,
300			"clientPackage": self.client_package,
301			"Accept": "application/json, text/plain, /",
302			"Content-Type": self.content_type1,
303			"Connection": "keep-alive",
304			"User-Agent": self.user_agent,
305			"Accept-Language": "zh-cn",
306			"Accept-Encoding": self.accept_encoding1,
307			"cookie": self.cookies,
308			"Origin": "https://store.oppo.com",
309			"X-Requested-With": self.client_package,
310			"referer": "https://store.oppo.com/cn/app/taskCenter/index?us=gerenzhongxin&um=hudongleyuan&uc=renwuzhongxin",
311			}
312
313			data = (
314			"marking="
315			+ str(info_marking)
316			+ "&type="
317			+ str(info_type)
318			+ "&amount="
319			+ str(info_credits)
320			)
321
322			res = self.client.post(
323			"https://store.oppo.com/cn/oapi/credits/web/credits/cashingCredits",
324			data=data,
325			headers=headers,
326			)
327
328			res = res.json()
329
330			if res["code"] == 200:
331			return True
332			else:
333			return False
334
335			# 活动平台抽奖通用接口
336			def lottery(self, datas, referer="", extra_draw_cookie=""):
337
338			headers = {
339			"Host": "hd.oppo.com",
340			"User-Agent": self.user_agent,
341			"Cookie": extra_draw_cookie + self.cookies,
342			"Referer": referer,
343			"Accept": "application/json, text/javascript, /; q=0.01",
344			"Accept-Language": "zh-cn",
345			"Accept-Encoding": "br, gzip, deflate",
346			"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
347			}
348			res = self.client.get("https://hd.oppo.com/user/login", headers=headers).json()
349			if res["no"] == "200":
350			res = self.client.post(
351			"https://hd.oppo.com/platform/lottery", data=datas, headers=headers
352			)
353			res = res.json()
354			return res
355			else:
356			return res
357
358			# 活动平台完成任务接口
359			def task_finish(self, aid, t_index):
360			headers = {
361			"Accept": "application/json, text/plain, /;q=0.01",
362			"Content-Type": self.content_type2,
363			"Connection": "keep-alive",
364			"User-Agent": self.user_agent,
365			"Accept-Encoding": self.accept_encoding2,
366			"cookie": self.cookies,
367			"Origin": "https://hd.oppo.com",
368			"X-Requested-With": "XMLHttpRequest",
369			}
370			datas = "aid=" + str(aid) + "&t_index=" + str(t_index)
371			res = self.client.post(
372			"https://hd.oppo.com/task/finish", data=datas, headers=headers
373			)
374			res = res.json()
375			return res
376
377			# 活动平台领取任务奖励接口
378			def task_award(self, aid, t_index):
379			headers = {
380			"Accept": "application/json, text/plain, /;q=0.01",
381			"Content-Type": self.content_type2,
382			"Connection": "keep-alive",
383			"User-Agent": self.user_agent,
384			"Accept-Encoding": self.accept_encoding2,
385			"cookie": self.cookies,
386			"Origin": "https://hd.oppo.com",
387			"X-Requested-With": "XMLHttpRequest",
388			}
389			datas = "aid=" + str(aid) + "&t_index=" + str(t_index)
390			res = self.client.post(
391			"https://hd.oppo.com/task/award", data=datas, headers=headers
392			)
393			res = res.json()
394			return res
395
396			# 做活动任务和抽奖通用接口 hss修改
397			def do_task_and_draw(self):
398			try:
399
400			for act_list in self.act_task:
401			act_name = act_list["act_name"]
402			aid = act_list["aid"]
403			referer = act_list["referer"]
404			if_draw = act_list["if_draw"]
405			if_task = act_list["if_task"]
406			end_time = act_list["end_time"]
407			headers = {
408			"Accept": "application/json, text/javascript, /; q=0.01",
409			"Connection": "keep-alive",
410			"User-Agent": self.user_agent,
411			"Accept-Encoding": self.accept_encoding2,
412			"cookie": self.cookies,
413			"X-Requested-With": "XMLHttpRequest",
414			"Referer": referer,
415			}
416			dated = int(time.time())
417			end_time = time.mktime(
418			time.strptime(end_time, "%Y-%m-%d %H:%M:%S")
419			) # 设置活动结束日期
420
421			if dated < end_time:
422			if if_task:
423			res = self.client.get(
424			f"https://hd.oppo.com/task/list?aid={aid}", headers=headers
425			)
426			tasklist = res.json()
427			self.log += f"【{act_name}-任务】\n"
428			for _, jobs in enumerate(tasklist["data"]):
429			title = jobs["title"]
430			t_index = jobs["t_index"]
431			aid = t_index[: t_index.index("i")]
432			if jobs["t_status"] == 0:
433			finishmsg = self.task_finish(aid, t_index)
434			if finishmsg["no"] == "200":
435			time.sleep(1)
436			awardmsg = self.task_award(aid, t_index)
437			msg = awardmsg["msg"]
438			self.log += f"{title}：{msg}\n"
439			time.sleep(3)
440			elif jobs["t_status"] == 1:
441			awardmsg = self.task_award(aid, t_index)
442			msg = awardmsg["msg"]
443			self.log += f"{title}：{msg}\n"
444			time.sleep(3)
445			else:
446			self.log += f"{title}：任务已完成\n"
447			if self.if_draw and if_draw: # 判断当前用户是否抽奖和判断当前活动是否抽奖
448			lid = act_list["lid"]
449			extra_draw_cookie = act_list["extra_draw_cookie"]
450			draw_times = act_list["draw_times"]
451			self.log += f"【{act_name}-抽奖】："
452			x = 0
453			while x < draw_times:
454			data = f"aid={aid}&lid={lid}&mobile=&authcode=&captcha=&isCheck=0&source_type={self.source_type}&s_channel={self.s_channel}&sku=&spu="
455			res = self.lottery(data, referer, extra_draw_cookie)
456			msg = res["msg"]
457			if "次数已用完" in msg:
458			self.log += " 第" + str(x + 1) + "抽奖：抽奖次数已用完\n"
459			break
460			if "活动已结束" in msg:
461			self.log += " 第" + str(x + 1) + "抽奖：活动已结束，终止抽奖\n"
462			break
463			goods_name = res["data"]["goods_name"]
464			if goods_name:
465			self.log += (
466			" 第" + str(x + 1) + "次抽奖：" + str(goods_name) + "\n"
467			)
468			elif "提交成功" in msg:
469			self.log += " 第" + str(x + 1) + "次抽奖：未中奖\n"
470			x += 1
471			time.sleep(5)
472			else:
473			self.log += f"【{act_name}】：活动已结束，不再执行\n"
474			except Exception as e:
475			self.log += "【执行任务和抽奖】：错误，原因为: " + str(e) + "\n"
476
477			# 早睡打卡
478			def zaoshui_task(self):
479			try:
480			headers = {
481			"Host": self.host1,
482			"Connection": "keep-alive",
483			"s_channel": self.s_channel,
484			"utm_term": "direct",
485			"utm_campaign": "direct",
486			"utm_source": "direct",
487			"ut": "direct",
488			"uc": "zaoshuidaka",
489			"sa_device_id": self.sa_device_id,
490			"guid": self.sa_device_id,
491			"sa_distinct_id": self.sa_distinct_id,
492			"clientPackage": self.client_package,
493			"Cache-Control": "no-cache",
494			"um": "hudongleyuan",
495			"User-Agent": self.user_agent,
496			"ouid": "",
497			"Accept": "application/json, text/plain, /",
498			"source_type": self.source_type,
499			"utm_medium": "direct",
500			"brand": "iPhone",
501			"appId": "",
502			"s_version": self.s_version,
503			"us": "gerenzhongxin",
504			"appKey": "",
505			"X-Requested-With": self.client_package,
506			"Referer": "https://store.oppo.com/cn/app/cardingActivities?utm_source=opposhop&utm_medium=task",
507			"Accept-Encoding": self.accept_encoding2,
508			"Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
509			"cookie": self.cookies,
510			}
511			res = self.client.get(
512			"https://store.oppo.com/cn/oapi/credits/web/clockin/applyOrClockIn",
513			headers=headers,
514			).json()
515			if "余额不足" in str(res):
516			self.log += "【早睡打卡】：申请失败，积分余额不足\n"
517			else:
518			applyStatus = res["data"]["applyStatus"]
519			# 2 19:30打卡成功 0 8：00打卡成功
520			if applyStatus == 1:
521			self.log += "【早睡打卡】：报名成功，请当天19:30-22:00留意打卡状态\n"
522			if applyStatus == 0:
523			self.log += "【早睡打卡】：申请失败，积分不足或报名时间已过\n"
524			if applyStatus == 2:
525			self.log += "【早睡打卡】：已报名成功，请当天19:30-22:00留意打卡状态\n"
526			if res["data"]["clockInStatus"] == 1:
527			self.log += "【早睡打卡】：打卡成功，积分将于24:00前到账\n"
528			if res["data"]["clockInStatus"] == 2:
529			self.log += "【早睡打卡】：打卡成功，积分将于24:00前到账\n"
530			# 打卡记录
531			res = self.client.get(
532			"https://store.oppo.com/cn/oapi/credits/web/clockin/getMyRecord",
533			headers=headers,
534			).json()
535			if res["code"] == 200:
536			record = res["data"]["everydayRecordForms"]
537			self.log += "【早睡打卡记录】\n"
538			i = 0
539			for data in record:
540			self.log += (
541			data["everydayDate"]
542			+ "-"
543			+ data["applyClockInStatus"]
544			+ "："
545			+ data["credits"]
546			+ "\n"
547			)
548			i += 1
549			if i == 4: # 最多显示最近4条记录
550			break
551
552			except Exception as e:
553			self.log += "【早睡打卡】：错误，原因为: " + str(e) + "\n"
554
555			# 主程序
556			def main(self):
557			i = 1
558			for check_item in self.check_items:
559			self.cookies = check_item.get("cookie")
560			self.user_agent = check_item.get("useragent")
561			self.if_draw = check_item.get("draw")
562			self.client = self.get_user_info()
563			if self.client:
564			try:
565			self.daily_bonus() # 执行每日签到
566			self.daily_viewgoods() # 执行每日商品浏览任务
567			self.daily_sharegoods() # 执行每日商品分享任务
568			self.do_task_and_draw() # 自己修改的接口，针对活动任务及抽奖，新增及删除活动请修改act_list.py
569			# self.zaoshui_task() # 早睡报名由于自己不能及时更新cookie，就关闭了打卡
570			except Exception as e:
571			self.log += f"账号{i}执行出错：{e}\n"
572			else:
573			self.log += f"账号{i}已失效，请及时更新cookies\n"
574			i += 1
575			self.log += "\n\n"
576			return self.log
577
578
579			if __name__ == "__main__":
580			data = get_data()
581			_check_items = data.get("HEYTAP", [])
582			res = Heytap(check_items=_check_items).main()
583			send("欢太商城", res)
584

OreosLab / checkinpanel

Push — master ( c3d8b2...2da959 )

ck_heytap.Heytap.daily_viewgoods() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like