Quantumult-X/Scripts/imoutai/imoutaiHelper.py

import base64
import datetime
import json
import math
import os
import random
import time

import requests
from Crypto.Cipher import AES

from dailycheckin import CheckIn


class Encrypt:
    def __init__(self, key, iv):
        self.key = key.encode("utf-8")
        self.iv = iv.encode("utf-8")

    def pkcs7padding(self, text):
        """明文使用PKCS7填充"""
        bs = 16
        length = len(text)
        bytes_length = len(text.encode("utf-8"))
        padding_size = length if (bytes_length == length) else bytes_length
        padding = bs - padding_size % bs
        padding_text = chr(padding) * padding
        self.coding = chr(padding)
        return text + padding_text

    def aes_encrypt(self, content):
        """AES加密"""
        cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
        content_padding = self.pkcs7padding(content)
        encrypt_bytes = cipher.encrypt(content_padding.encode("utf-8"))
        result = str(base64.b64encode(encrypt_bytes), encoding="utf-8")
        return result

    def aes_decrypt(self, content):
        """AES解密"""
        cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
        content = base64.b64decode(content)
        text = cipher.decrypt(content).decode("utf-8")
        return text.rstrip(self.coding)


class IMAOTAI(CheckIn):
    name = "i茅台"

    def __init__(self, check_item):
        self.check_item = check_item
        self.RESERVE_RULE = 0
        self.mt_r = "clips_OlU6TmFRag5rCXwbNAQ/Tz1SKlN8THcecBp/"
        self.ITEM_MAP = {
            "10941": "53%vol 500ml贵州茅台酒（甲辰龙年）",
            "10942": "53%vol 375ml×2贵州茅台酒（甲辰龙年）",
            "10056": "53%vol 500ml茅台1935",
            "2478": "53%vol 500ml贵州茅台酒（珍品）",
        }
        self.ITEM_CODES = ["10941", "10942"]
        AES_KEY = "qbhajinldepmucsonaaaccgypwuvcjaa"
        AES_IV = "2018534749963515"
        self.encrypt = Encrypt(key=AES_KEY, iv=AES_IV)

        self.mt_version = json.loads(
            requests.get("https://itunes.apple.com/cn/lookup?id=1600482450").text
        )["results"][0]["version"]
        self.headers = {}
        self.header_context = """
MT-Lat: 28.499562
MT-K: 1675213490331
MT-Lng: 102.182324
Host: app.moutai519.com.cn
MT-User-Tag: 0
Accept: */*
MT-Network-Type: WIFI
MT-Token: 1
MT-Info: 028e7f96f6369cafe1d105579c5b9377
MT-Device-ID: 2F2075D0-B66C-4287-A903-DBFF6358342A
MT-Bundle-ID: com.moutai.mall
Accept-Language: en-CN;q=1, zh-Hans-CN;q=0.9
MT-Request-ID: 167560018873318465
MT-APP-Version: 1.3.7
User-Agent: iOS;16.3;Apple;?unrecognized?
MT-R: clips_OlU6TmFRag5rCXwbNAQ/Tz1SKlN8THcecBp/HGhHdw==
Content-Length: 93
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
Content-Type: application/json
userId: 2
"""

    def init_headers(
        self,
        user_id: str = "1",
        token: str = "2",
        lat: str = "29.83826",
        lng: str = "119.74375",
    ):
        for k in self.header_context.strip().split("\n"):
            temp_l = k.split(": ")
            dict.update(self.headers, {temp_l[0]: temp_l[1]})
        dict.update(self.headers, {"userId": user_id})
        dict.update(self.headers, {"MT-Token": token})
        dict.update(self.headers, {"MT-Lat": lat})
        dict.update(self.headers, {"MT-Lng": lng})
        dict.update(self.headers, {"MT-APP-Version": self.mt_version})

    def get_current_session_id(self):
        day_time = int(time.mktime(datetime.date.today().timetuple())) * 1000
        my_url = f"https://static.moutai519.com.cn/mt-backend/xhr/front/mall/index/session/get/{day_time}"
        responses = requests.get(my_url)
        if responses.status_code != 200:
            print(
                f"get_current_session_id : params : {day_time}, response code : {responses.status_code}, response body : {responses.text}"
            )
        current_session_id = responses.json()["data"]["sessionId"]
        dict.update(self.headers, {"current_session_id": str(current_session_id)})

    def get_map(self, lat: str = "28.499562", lng: str = "102.182324"):
        p_c_map = {}
        url = "https://static.moutai519.com.cn/mt-backend/xhr/front/mall/resource/get"
        headers = {
            "X-Requested-With": "XMLHttpRequest",
            "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 15_0_1 like Mac OS X)",
            "Referer": "https://h5.moutai519.com.cn/gux/game/main?appConfig=2_1_2",
            "Client-User-Agent": "iOS;16.0.1;Apple;iPhone 14 ProMax",
            "MT-R": "clips_OlU6TmFRag5rCXwbNAQ/Tz1SKlN8THcecBp/HGhHdw==",
            "Origin": "https://h5.moutai519.com.cn",
            "MT-APP-Version": self.mt_version,
            "MT-Request-ID": f"{int(time.time() * 1000)}{random.randint(1111111, 999999999)}{int(time.time() * 1000)}",
            "Accept-Language": "zh-CN,zh-Hans;q=1",
            "MT-Device-ID": f"{int(time.time() * 1000)}{random.randint(1111111, 999999999)}{int(time.time() * 1000)}",
            "Accept": "application/json, text/javascript, */*; q=0.01",
            "mt-lng": f"{lng}",
            "mt-lat": f"{lat}",
        }
        res = requests.get(url, headers=headers)
        mtshops = res.json().get("data", {}).get("mtshops_pc", {})
        urls = mtshops.get("url")
        r = requests.get(urls)
        for k, v in dict(r.json()).items():
            provinceName = v.get("provinceName")
            cityName = v.get("cityName")
            if not p_c_map.get(provinceName):
                p_c_map[provinceName] = {}
            if not p_c_map[provinceName].get(cityName, None):
                p_c_map[provinceName][cityName] = [k]
            else:
                p_c_map[provinceName][cityName].append(k)

        return p_c_map, dict(r.json())

    def max_shop(self, city, item_code, p_c_map, province, shops):
        max_count = 0
        max_shop_id = "0"
        shop_ids = p_c_map[province][city]
        for shop in shops:
            shopId = shop["shopId"]
            items = shop["items"]

            if shopId not in shop_ids:
                continue
            for item in items:
                if item["itemId"] != str(item_code):
                    continue
                if item["inventory"] > max_count:
                    max_count = item["inventory"]
                    max_shop_id = shopId
        print(
            f"item code {item_code}, max shop id : {max_shop_id}, max count : {max_count}"
        )
        return max_shop_id

    def distance_shop(
        self,
        item_code,
        shops,
        source_data,
        lat: str = "28.499562",
        lng: str = "102.182324",
    ):
        temp_list = []
        for shop in shops:
            shopId = shop["shopId"]
            items = shop["items"]
            item_ids = [i["itemId"] for i in items]
            if str(item_code) not in item_ids:
                continue
            shop_info = source_data.get(shopId)
            d = math.sqrt(
                (float(lat) - shop_info["lat"]) ** 2
                + (float(lng) - shop_info["lng"]) ** 2
            )
            temp_list.append((d, shopId))

        temp_list = sorted(temp_list, key=lambda x: x[0])
        if len(temp_list) > 0:
            return temp_list[0][1]
        else:
            return "0"

    def get_location_count(
        self,
        province: str,
        city: str,
        item_code: str,
        p_c_map: dict,
        source_data: dict,
        lat: str = "29.83826",
        lng: str = "102.182324",
        reserve_rule: int = 0,
    ):
        day_time = int(time.mktime(datetime.date.today().timetuple())) * 1000
        session_id = self.headers["current_session_id"]
        responses = requests.get(
            f"https://static.moutai519.com.cn/mt-backend/xhr/front/mall/shop/list/slim/v3/{session_id}/{province}/{item_code}/{day_time}"
        )
        if responses.status_code != 200:
            print(
                f"get_location_count : params : {day_time}, response code : {responses.status_code}, response body : {responses.text}"
            )
        shops = responses.json()["data"]["shops"]

        if reserve_rule == 0:
            return self.distance_shop(item_code, shops, source_data, lat, lng)
        if reserve_rule == 1:
            return self.max_shop(city, item_code, p_c_map, province, shops)

    def act_params(self, shop_id: str, item_id: str):
        session_id = self.headers["current_session_id"]
        userId = self.headers["userId"]
        params = {
            "itemInfoList": [{"count": 1, "itemId": item_id}],
            "sessionId": int(session_id),
            "userId": userId,
            "shopId": shop_id,
        }
        s = json.dumps(params)
        act = self.encrypt.aes_encrypt(s)
        params.update({"actParam": act})
        return params

    def reservation(self, params: dict):
        params.pop("userId")
        responses = requests.post(
            "https://app.moutai519.com.cn/xhr/front/mall/reservation/add",
            json=params,
            headers=self.headers,
        ).json()
        if responses.get("code") == 401:
            msg = {
                "name": "申购结果",
                "value": "token失效, 请重新抓包获取",
            }
        elif responses.get("code") != 2000:
            msg = {
                "name": "申购结果",
                "value": responses.get("message"),
            }
        else:
            msg = {
                "name": "申购结果",
                "value": responses.get("data", {}).get("successDesc"),
            }
        return msg

    def getUserEnergyAward(self):
        """
        耐力值
        """
        cookies = {
            "MT-Device-ID-Wap": self.headers["MT-Device-ID"],
            "MT-Token-Wap": self.headers["MT-Token"],
            "YX_SUPPORT_WEBP": "1",
        }
        response = requests.post(
            url="https://h5.moutai519.com.cn/game/isolationPage/getUserEnergyAward",
            cookies=cookies,
            headers=self.headers,
            json={},
        ).json()
        if response.get("code") == 200:
            msg = {
                "name": "耐力",
                "value": "✅领取耐力成功",
            }
        else:
            msg = {
                "name": "耐力",
                "value": response.get("message"),
            }
        return msg

    def main(self):
        msg = []
        mobile = self.check_item.get("mobile")
        province = self.check_item.get("province")
        city = self.check_item.get("city")
        token = self.check_item.get("token")
        userId = self.check_item.get("userid")
        lat = self.check_item.get("lat")
        lng = self.check_item.get("lng")
        item_codes = self.check_item.get("item_codes", self.ITEM_CODES)
        reserve_rule = self.check_item.get("reserve_rule", 0)
        msg = [
            {
                "name": "手机号",
                "value": f"{mobile}",
            },
            {
                "name": "省份城市",
                "value": f"{province}{city}",
            },
        ]
        p_c_map, source_data = self.get_map(lat=lat, lng=lng)
        self.get_current_session_id()
        self.init_headers(user_id=userId, token=token, lng=lng, lat=lat)
        try:
            for item in item_codes:
                max_shop_id = self.get_location_count(
                    province=province,
                    city=city,
                    item_code=item,
                    p_c_map=p_c_map,
                    source_data=source_data,
                    lat=lat,
                    lng=lng,
                    reserve_rule=reserve_rule,
                )
                if max_shop_id == "0":
                    continue
                reservation_params = self.act_params(max_shop_id, item)
                reservation_msg = self.reservation(reservation_params)
                time.sleep(20)
                award_msg = self.getUserEnergyAward()
                msg.append(reservation_msg)
                msg.append(award_msg)
        except BaseException as e:
            msg.append(
                {
                    "name": "申购结果",
                    "value": e,
                }
            )
        msg = "\n".join([f"{one.get('name')}: {one.get('value')}" for one in msg])
        return msg


if __name__ == "__main__":
    with open(
        os.path.join(os.path.dirname(os.path.dirname(__file__)), "config.json"),
        encoding="utf-8",
    ) as f:
        datas = json.loads(f.read())
    _check_item = datas.get("IMAOTAI", [])[0]
    print(IMAOTAI(check_item=_check_item).main())