星星充电 充电桩数据爬虫

某星充电APP 充电桩信息

环境

• win10
• 某星充电 APP 7.9.0版本
• Android8.1

X-Ca-Signature参数分析

APP有防抓包检测，使用算法助手启动可正常抓包，分析数据包后发现关键的就是X-Ca-Timestamp和X-Ca-Signature， Timestamp很明显看出来是时间戳，重点分析一下Signature参数

脱壳后搜索关键词：X-Ca-Signature
只有一处跟进去分析一下具体的参数构造


可以很明显看到是参数加上固定的key拼接后做了md5处理

Signature参数分析

站点详情有一个Signature参数是H5加载js生成的，定位到具体赋值的位置，然后往上跟可以看到是经过一个随机uid+时间戳生成的

    def get_signtrue(self, dic: dict):
        """
        生成signature参数：7.9.0版本
        """
        if (not "userId" in dic):
            dic["userId"] = ""
        new_list = []
        for i in self.run_sort(dic.keys()):
            new_list.append(f"{i}={dic[i]}")
        return self.get_md5("&".join(new_list))

获取指定城市所有站点数据

    def get_SearchStation(self, k, v):
        """
        获取指定城市所有站点数据
        """
        # 提取城市经纬度
        center = v['districts'][0]['center']
        lng = center.split(',')[0]  # 经度
        lat = center.split(',')[1]  # 维度
        CityCode = v['districts'][0]['adcode']  # 城市代码
        _time = self.get_time()  # 时间戳
        page = 1
        while True:
            data = {
                "stubGroupTypes": "0",
                "currentLat": lat,  # 当前坐标经度
                "orderType": "1",
                "orCityCode": CityCode,  # 城市代码
                "lng": lng,  # 城市坐标维度
                "pagecount": "50",
                "currentLng": lng,  # 当前坐标维度
                "page": page,
                "radius": "10000",
                "lat": lat,  # 城市坐标经度
                "showPromoteLabel": "1"
            }
            X_Ca_Signature = self.get_X_Ca_Signature(data, _time)
            headers = {
                'appVersion': '7.9.1.1',
                'X-Ca-Timestamp': _time,
                'X-Ca-Signature': X_Ca_Signature,
                'Connection': 'Keep-Alive',
                'User-Agent': 'okhttp/3.12.1',
            }
            # res = requests.post(url=url, data=data, headers=headers, proxies=self.proxies, verify=True, allow_redirects=True, timeout=60)
            res = self._parse_url(url=url, data=data, headers=headers)
            # print(res.text)
            if not res.text or res.status_code != 200:
                logger.info(f'{k} 第{page}页数据获取失败')
                return
            if not res.json().get('data'):
                logger.info(f'{k} 数据获取完成')
                return

            new_path = '数据' + os.sep + k
            if not os.path.exists(new_path):
                os.makedirs(new_path)
            filename = new_path + os.sep + '站点列表.json'
            # 多页的情况直接把新数据添加至已有文件
            if os.path.exists(filename):
                with open(filename, 'r+', encoding='utf-8') as f:
                    row_data = json.load(f)
                    row_data['data'] += res.json().get('data')
                    with open(filename, 'w', encoding='utf-8') as f1:
                        json.dump(row_data, f1, ensure_ascii=False)
            else:
                with open(filename, 'w', encoding='utf-8') as f:
                    # ensure_ascii 不适用ascii编码 解决不能显示中文问题
                    json.dump(res.json(), f, ensure_ascii=False)

            logger.info(f'{k} 第{page}页数据获取成功')
            page += 1

获取站点详情信息

        def get_detail(self, stationId, name, lat, lng):
        """
        获取站点详情
        """
        _time = self.get_time()  # 时间戳
        nonce = self.get_nonce()  # nonce参数
        test_ = {
            "id": stationId,
            "gisType": "1",
            "lat": lat,
            "lng": lng,
            "stubType": "0",
            "versionFlag": "1",
            "nonce": nonce,
            "timestamp": _time
        }
        get_signtrue = self.get_signtrue(test_)
        headers = {
            "referrer": "web",
            "Accept": "application/json, text/plain, */*",
            "timestamp": _time,
            "signature": get_signtrue,
            "User-Agent": "Mozilla/5.0 (Linux; Android 8.1.0; Nexus 5X Build/OPM7.181205.001; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/61.0.3163.98 Mobile Safari/537.36",
            "appVersion": "7.9.0",
            "Accept-Language": "zh-CN,en-CA;q=0.8,en-US;q=0.6",
        }
        params = {
            "id": stationId,
            "gisType": "1",
            "lat": lat,
            "lng": lng,
            "stubType": "0",
            "versionFlag": "1",
            "nonce": nonce,
        }
        response = requests.get(url, headers=headers, params=params)
        print(response.text)
        if not response.text or response.status_code != 200:
            logger.info(f'{name} 数据获取失败')
            return

        if not response.json().get('data'):
            logger.info(f'{name} 数据获取完成')
            return
        return response.json()

效果

资源下载

https://download.csdn.net/download/qq_38154948/87484473