meituan/common.py

# 获取请求头
import base64
import os
import random
import re
import time
import zlib
import pandas
import requests
from requests.adapters import HTTPAdapter
from urllib3 import Retry




# 获取请求会话
def getSession():
    session = requests.session()
    retry = Retry(connect=3, backoff_factor=0.5)
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session


href = 'href'
zh = 'zh'
en = 'en'

# 城市参数类
class CityParam:

    def __init__(self, city) -> None:
        super().__init__()
        self.city = city
        self.uuid = self.get_uuid()
        self.data = self.getData()
        self.param = self.getParam()

    # 获取请求头
    def getHeaders(self):
        return {
            "Accept": "application/json",
            "Referer": "https://{}.meituan.com/".format(self.city[en]),
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.108 Safari/537.36",
        }

    # 获取请求参数uuid
    def get_uuid(self):
        """获取uuid"""
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.131 Safari/537.36"
        }
        res = requests.get(self.city[href], headers=headers).text
        return re.findall(r'"uuid":"(.*?)"', res, re.S)[0]

    # 初始化参数字典
    def getData(self):
        return {
            "cityName": self.city[zh],
            "cateId": '0',
            "areaId": "0",
            "sort": "",
            "dinnerCountAttrId": "",
            "page": "1",
            "userId": "",
            "uuid": self.uuid,
            "platform": "1",
            "partner": "126",
            "originUrl": href,
            "riskLevel": "1",
            "optimusCode": "1"
        }

    # 初始化请求参数
    def getParam(self):
        return {
            "cityName": self.data["cityName"],
            "cateId": self.data["cateId"],
            "areaId": self.data["areaId"],
            "sort": self.data["sort"],
            "dinnerCountAttrId": self.data["dinnerCountAttrId"],
            "page": self.data["page"],
            "userId": self.data["userId"],
            "uuid": self.data["uuid"],
            "platform": self.data["platform"],
            "partner": self.data["partner"],
            "originUrl": self.data["originUrl"],
            "riskLevel": self.data["riskLevel"],
            "optimusCode": self.data["optimusCode"],
        }

    # 生成请求sign参数
    def sign(self):
        """生成sign参数"""
        # 默认编码
        # coding = sys.getdefaultencoding()
        # 二进制压缩
        SIGN_PARAM = "areaId={}&cateId={}&cityName={}&dinnerCountAttrId={}&optimusCode={}&originUrl={}&page={}&partner={}&platform={}&riskLevel={}&sort={}&userId={}&uuid={}".format(
            self.data["areaId"],
            self.data["cateId"],
            self.data["cityName"],
            self.data["dinnerCountAttrId"],
            self.data["optimusCode"],
            self.data["originUrl"],
            self.data["page"],
            self.data["partner"],
            self.data["platform"],
            self.data["riskLevel"],
            self.data["sort"],
            self.data["userId"],
            self.data["uuid"]
        )
        binary_data = zlib.compress(SIGN_PARAM.encode())
        # base64编码
        base64_data = base64.b64encode(binary_data)
        # 返回utf8编码的字符串
        return base64_data.decode()

    # 生成请求token参数
    def encrypt_token(self):
        """生成_token参数"""
        ts = int(time.time() * 1000)  # time.time()返回1970年至今的时间(以秒为单位)
        # 伪装机型
        json_path = os.path.dirname(os.path.realpath(__file__)) + '\\config\\br.json'
        df = pandas.read_json(json_path)
        brVD, brR_one, brR_two = df.iloc[random.randint(0, len(df) - 1)]
        token_data = {
            "rId": 100900,
            "ver": "1.0.6",
            "ts": ts,
            "cts": ts + random.randint(100, 120),  # 经测,cts - ts 的差值大致在 90-130 之间
            # "cts": ts + 100,
            "brVD": eval(brVD),
            "brR": [eval(brR_one), eval(brR_two), 24, 24],
            "bI": [self.city[href], ""],
            "mT": [],
            "kT": [],
            "aT": [],
            "tT": [],
            "aM": "",
            "sign": self.sign()
        }
        # 二进制压缩
        binary_data = zlib.compress(str(token_data).encode())
        # base64编码
        base64_data = base64.b64encode(binary_data)
        return base64_data.decode()


# 等待服务器发送数据秒数
TIMEOUT = 5
data='data'
# 总记录数键
totalCounts = 'totalCounts'
# 商户键
poiInfos = 'poiInfos'