修改

Scripts/imaotai/imaotaiHelper.py

@@ -0,0 +1,357 @@
+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())