某了么数据获取脚本

在这里插入图片描述

这段代码定义了一个名为 ElemeH5 的类，继承自 Base 类，用于处理与饿了么平台的API交互。该类包括了多种方法来进行网络请求、数据处理和API接口的动态生成。以下是对主要组成部分的详细解析：

类属性定义：

func_template：字符串，用作动态生成函数的模板。
appKey 和 secret：API使用的密钥和安全码。
platform 和 sandbox：用于定义API请求的平台和是否是沙箱环境。
domain：API请求的基础域名。

构造函数 (init):

初始化基类并设置函数模板。

方法定义：

onekeylogin()：

发起POST请求到指定的URL，尝试一键登录功能。

h5_token和 umidToken (使用 @property 装饰器)：

这两个属性方法分别用来获取会话中的token和生成umidToken，umidToken是根据当前时间和随机数生成的。

sign()：

生成请求的签名，采用MD5加密方法，确保请求的安全性。

_execute()：

执行实际的HTTP请求。根据传入的请求参数，进行GET或POST请求，并处理返回结果。

render_template()：

用于根据传入的参数动态生成函数的字符串。

urlCreateFunc()：

动态解析API链接并生成相应的函数，增强了代码的灵活性和可维护性。该方法还将生成的函数代码追加到当前文件。

createTypes()：

动态创建类型定义，用于强类型语言的接口定义，便于后续维护和类型检查。

searchPoiNearby() 和 reverseGeoCoding()：

特定的业务方法，用于调用地理位置相关的API，如搜索附近的地点和进行地理位置逆编码。

-MtopVenusShopresourceserviceGetshopresource() 和 `MtopAlscWamaiStoreDetailBusinessTabPhone()：

这两个方法分别封装了对特定API的调用，包括参数的准备、请求的发起和响应的处理。

总结：

这个类是一个高度封装的API客户端，利用面向对象的方法管理饿了么平台的多个API接口。通过动态函数生成和类型定义，极大地提高了代码的可维护性和可扩展性。此外，它还包括了错误处理和日志记录功能，这有助于问题追踪和调试。整体上，这是一个典型的用于商业API交互的工具类设计。


from ..types import taobao, eleme
from ..base import Base
from pathlib import Path
import re
from urllib.parse import urlparse, parse_qsl, quote
from typing import Any, Dict, List
import json
from collections import OrderedDict
import time
import random
import hashlib
import datetime

class ElemeH5(Base):

    func_template: str
    appKey: str = '12574478'
    secret: str = ''
    platform: str = ''
    sandbox: bool = False
    domain: str = 'https://waimai-guide.ele.me'

    def __init__(self) -> None:
        super().__init__()

        self.func_template = ''
    def onekeylogin(self):
        res = self.post('https://www.cmpassport.com/h5/onekeylogin/getNewTelecomPhonescrip?_bx-v=2.5.3',json={
            'businessType': '8',
            'encrypted': '',
            'reqdata': '',
        })
    @property
    def h5_token(self):
        token = self.cookies.get('_m_h5_tk', domain='', default='')
        if not token is None and len(token) > 32:
            return token[0:32]

        self.logger.error('从cookie获取的token为空')
        return ''

    @property
    def umidToken(self):
        '''获取umidToken'''
        return 'C' + str(int(time.time() * 1000)) + \
            ''.join(str(random.choice(range(10))) for _ in range(11)) + \
            str(int(time.time() * 1000)) + \
            ''.join(str(random.choice(range(10))) for _ in range(3))

    def sign(self, token: str, t: str, appkey: str, data: str, binary: bool = False):
        '''sign签名加密方式采用淘宝H5网页的加密流程
        data传递使用的是字符串，一是为了少加密一次，二是为了直接说明这个要转成json字符串，还需要去掉空格'''

        sign_func = 'digest' if binary else 'hexdigest'
        sign_str = f'{token}&{t}&{appkey}&{data}'
        self.logger.debug('sign签名字符串:{signStr}',signStr=sign_str)
        return getattr(hashlib.md5(sign_str.encode('utf-8')),sign_func)()

    def _execute(self, request_options: Any):
        '''解析请求参数并将参数进行加密'''
        method = request_options.get('method')
        if method.upper() == 'GET':
            payload = request_options.get('params')
        else:
            payload = request_options.get('data')

        if payload is None:
            raise Exception('提交的数据不能为None')

        timestamp = str(int(datetime.datetime.now().timestamp() * 1000))
        appKey = payload.get('appKey', self.appKey)
        payload.update({
            't': timestamp,
            'type': 'json',
            'dataType': 'json',
        })
        payloadData = payload.get('data', {})
        dataStr = json.dumps(payloadData, separators=(',', ':'))
        payload.update({
            'sign': self.sign(self.h5_token, timestamp, appKey, dataStr),
            'data': dataStr
        })
        self.logger.debug('{method}请求:{payload}', payload=payload, method=method.upper())
        res = self.request(**request_options)
        resj: taobao.ApiRes = res.json()

        if res.status_code != 200:
            self.logger.error('API请求失败:{rtxt}', rtxt=res.text)
        else:
            ret = resj.get('ret')
            retMsg = ret[0]
            if 'SUCCESS' not in retMsg.upper():
                self.logger.error('API请求失败:{ret}', ret=ret)

        return (resj, res)

    def render_template(self, kwargs: Any):
        return self.func_template.format(**kwargs)

    def urlCreateFunc(self, api: str, method: str = 'get', func_name: str | None = None, desc: str = '', formdata: Any = {}):
        '''解析API链接为函数'''
        urlObj = urlparse(api)

        queryParams = dict(parse_qsl(urlObj.query))
        [platform, service_name, version, SV] = urlObj.path.strip('/').split('/')

        func_name = func_name if func_name else ''.join([i.capitalize() for i in service_name.split('.')])
        if hasattr(self, func_name):
            return

        for key in queryParams:
            val = queryParams[key]
            if len(re.findall(r'\{.*?\}|\[.*?\]', val)) > 0:
                queryParams[key] = json.loads(val)

        with open(__file__, 'a+', encoding='utf-8') as f:
            funcStr = self.render_template({
                'payload': queryParams,
                'hostname': urlObj.hostname,
                'scheme': urlObj.scheme,
                'path': urlObj.path,
                'params': urlObj.params,
                'func_name': func_name,
                'service_name': service_name,
                'platform': platform,
                'version': version,
                'desc': desc if desc else f'{service_name}函数',
                'method': method,
                'formdata': formdata,
            })
            f.write(funcStr)

    def createTypes(self, typeName: str, data: Dict[str, Any]):
        '''创建类型'''
        if not data:
            return

        formatter = '''

class {typeName}(TypedDict):
{fieldStr}
'''
        fieldStr = ''
        for key in data:
            val = data[key]
            if type(val) is str:
                fieldStr += f'    {key}: str\n'
            elif type(val) is list:
                fieldStr += f'    {key}: List'
            elif type(val) is dict:
                fieldStr += f'    {key}: Dict'
            elif type(val) is int:
                fieldStr += f'    {key}: int'
            elif type(val) is float:
                fieldStr += f'    {key}: float'
            elif type(val) is bool:
                fieldStr += f'    {key}: bool'
            else:
                fieldStr += f'    {key}: Any'

        nowFile = Path(__file__)
        typeFile = nowFile.parent.parent.joinpath('types/taobao.py')
        with open(typeFile, 'a+', encoding='utf-8') as f:
            typeStr = formatter.format(**{
                'typeName': typeName,
                'fieldStr': fieldStr,
            })
            f.write(typeStr)

    def searchPoiNearby(self, keyword: str, latitude: float, longitue: float, offset: int = 0, limit: int = 20):
        '''搜索附近地址'''
        url = 'https://h5.ele.me/restapi/bgs/poi/search_poi_nearby'
        res = self.get(url, params={
            'keyword': quote(keyword),
            'offset': offset,
            'limit': limit,
            'latitude': latitude,
            'longitude': longitue,
        })
        resj: List[eleme.SearchPoiNearbyRes] = res.json()
        return resj

    def reverseGeoCoding(self, longitude: float, latitude: float):
        '''重新定位'''
        url = 'https://h5.ele.me/restapi/bgs/poi/reverse_geo_coding'
        res = self.get(url,params={
            'latitude': latitude,
            'longitude': longitude
        })
        resj: eleme.reverseGeoCodingRes = res.json()
        return resj


    def MtopVenusShopresourceserviceGetshopresource(self, data: Any = {}):
        """mtop.venus.shopresourceservice.getshopresource函数"""

        method = 'get'
        params = {'jsv': '2.7.1', 'appKey': '12574478', 't': '1701086935133', 'sign': '21f5d98752b48dcb5e1a2d59eacb6529', 'api': 'mtop.venus.ShopResourceService.getShopResource', 'v': '1.4', 'type': 'originaljson', 'dataType': 'json', 'ecode': '1', 'SV': '5.0', 'data': {'lat': 30.452488, 'lng': 114.319347, 'from': 'native', 'bizChannel': 'mobile.default.default', 'deviceId': '500B511F28CE4C29996AC6EC4EDCA549|1701086828893', 'livingShowChannel': 'other', 'channel': 'PC', 'subChannel': 'ELE_APP', 'store_id': '1102570163', 'ele_id': 'E14249670395629855804', 'itemId': '', 'scentExtend': '{"businessComeFrom":""}'}, 'bx_et': 'dxvJc1wTGq0k4LyJbzhmY30xbsomsLKrMU-_KwbuOELvzGCBZemyMpLGu6AuE62dAeYRKa6hFwZplH1WK07hpvLVueCE4wbCJexBZmDiI3-PL9_KSADgxFBFd8rH-Jxy49WQmyGMG3yp7A4VU9PjobZ_QdJhVfI_sbXxGKsRwNehkJeKxgCR53_YurWNfdvOWeaGWMVSDmFUT_sqdoAHc'}
        url = 'https://waimai-guide.ele.me/h5/mtop.venus.shopresourceservice.getshopresource/1.4/5.0/'
        if data:
            params['data'].update(data)

        request_options = OrderedDict()
        request_options.setdefault('method', method)
        request_options.setdefault('url', url)
        if method.upper() == 'GET':
            request_options.setdefault('params', params)
        else:
            request_options.setdefault('data', params)

        return self._execute(request_options)

    def MtopAlscWamaiStoreDetailBusinessTabPhone(self, data: Any = {}):
        """mtop.alsc.wamai.store.detail.business.tab.phone函数"""

        method = 'POST'
        params = {''}
        url = 'https://waimai-guide.ele.me/h5/mtop.alsc.wamai.store.detail.business.tab.phone/1.0/5.0/'
        payload = params.get('data', {'eleStoreId': 'E6480404258831972301'})
        if data:
            payload.update(data)

        request_options = OrderedDict()
        request_options.setdefault('method', method)
        request_options.setdefault('url', url)
        request_options.setdefault('params', params)
        if method.upper() == 'POST':
            request_options.setdefault('data', payload)

        return self._execute(request_options)