python_api00000

from flask import Blueprint from flask import request import random import requests import re from comm.common_fun import success, error import gzip import json import time from Crypto.Cipher import AES import uuid import base64, zlib, time import hashlib from lxml import etree import urllib from urllib import parse # 注册时候前缀，之后本文件的每一个路由都有 dianping = Blueprint('dianping', __name__, url_prefix='/dianping') # 中文营业状态 @dianping.route('/get_business_status', methods=['POST']) def get_business_status(): try: data = json.loads(request.get_data(as_text=True)) shopId = data.get("shopuuid") ip = data.get("ip") proxie_ip = { "http": "http://" + ip, "https": "http://" + ip } shopbaseinfo = requests.get( f"https://mapi.dianping.com/gateway/nibmp/mva/poiext/shopbaseinfo?shopId={shopId}&clientType=1", proxies=proxie_ip, timeout=10).json() if shopbaseinfo["code"] == 200: shop_item = shopbaseinfo["data"] shop_item["shopTagsRichText"] = [x["text"] for x in shop_item["shopTagsRichText"]] del shop_item["jumpUrl"] return success(shop_item) else: return error("获取失败") except Exception as ex: return error(ex) # 评价标签 @dianping.route('/shopreviewlist', methods=['POST']) def shopreviewlist(): try: data = json.loads(request.get_data(as_text=True)) shopId = data["shopuuid"] ip = data.get("ip") proxie_ip = { "http": "http://" + ip, "https": "http://" + ip, } url = "http://www.dianping.com/ajax/json/shopDynamic/allReview" # token算法 br_w = random.choice([1366, 1920, 1280, 1680, 480, 1440, 1600, 1280, 320, 1280]) br_h = random.choice([768, 1080, 800, 1050, 800, 900, 1024, 480, 768]) info = { "rId": 100041, "ver": "1.0.6", "ts": int(time.time() * 1000), "cts": int(time.time() * 1000) + 1000000, "brVD": [ 649, 969 ], "brR": [ [ br_w, br_h ], [ br_w, br_h ], 24, 24 ], "bI": [ f"https://www.dianping.com/shop/{shopId}", "" ], "mT": [], "kT": [], "aT": [], "tT": [], "aM": "", "sign": "YnKwEAA1cjLw==" } token = base64.b64encode(zlib.compress(str(info).encode())).decode() params = { "shopId": shopId, "cityId": "2", "shopType": "20", "_token": token, # "uuid":uuid_str, "platform": "1", "partner": "150", "optimusCode": "10", "originUrl": f"https://www.dianping.com/shop/{shopId}?source=pc_jump" } ua_varsion = random.randint(500, 600) header = { "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-CN,zh;q=0.9", "Connection": "keep-alive", "Host": "www.dianping.com", "Referer": f"https://www.dianping.com/shop/{shopId}", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "User-Agent": f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/{ua_varsion}.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/{ua_varsion}.36", "X-Requested-With": "XMLHttpRequest" } json_data = requests.get(url, headers=header, timeout=3, proxies=proxie_ip, params=params).json() return success(json_data) except Exception as ex: return error(ex) # 响应字段里面的：nextUrl中的poi_id即为美团外卖的店铺id @dianping.route('/iswmid', methods=['POST']) def iswmid(): try: data = json.loads(request.get_data(as_text=True)) shopuuid = data["shopuuid"] ip = data.get("ip") proxie_ip = { "http": "http://" + ip, "https": "http://" + ip, } url = "https://apimeishi.meituan.com/meishi/dppoi/v1/poi/extraBusiness/0" \ f"?shopuuid={shopuuid}" \ "&phoneNo=" \ "&version_name=10.52.15" \ "&platform=20020500" \ "&deviceVersion=22" \ "&appVersion=10.52.15" \ f"&deviceId={random.randint(1000000, 99999999)}" \ "&dpid=&uuid=undefined" \ "&utm_medium=android" header = { "M-SHARK-TRACEID": "1160abcc84371847f0a701e6f1ec6ed43ea16371900864706932726809a1637311097357cb212b", "User-Agent": f"Dalvik/2.1.0 (Linux; U; Android 6.1.1; SM-N{random.randint(1, 9999)} Build/LMY49I)", "Host": "apimeishi.meituan.com" } resp = requests.get(url, headers=header, proxies=proxie_ip, timeout=3).json() return success(resp) except Exception as ex: return error(ex) # 店铺收录时间获取 @dianping.route('/shop_included_date', methods=['POST']) def shop_included_date(): try: data = json.loads(request.get_data(as_text=True)) shopId = data["shopid"] cookie_text = data["cookie"] shopuuid = data["shopuuid"] ip = data.get("ip") proxie_ip = { "http": "http://" + ip, "https": "http://" + ip, } url = "https://pdc.dianping.com/shopinfo" params = { "shopId": shopId, "latitude": "39.915", "longitude": "116.404", "notitlebar": "1", "shopuuid": shopuuid, "usermode": "2", "pushEnabled": "1", "product": "dpapp" } header = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-CN,zh;q=0.9", "Cache-Control": "max-age=0", "Connection": "keep-alive", "Cookie": cookie_text, "Host": "pdc.dianping.com", "Upgrade-Insecure-Requests": "1", "User-Agent": f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.69 Safari/5{random.randint(30, 50)}.36" } resp = requests.get(url, params=params, headers=header, proxies=proxie_ip, timeout=3) p = 'window.__INITIAL_STATE__=([\s\S+]*?);\(function' json_text = re.findall(p, resp.text) if json_text: p_data = json.loads(json_text[0])["shopinfo"] return success(p_data) else: return error(resp) except Exception as ex: return error(ex) # 推荐菜 @dianping.route('/recommenddishshoppagedish', methods=['POST']) def recommenddishshoppagedish(): try: data = json.loads(request.get_data(as_text=True)) shopid = data["shopid"] shopuuid = data["shopuuid"] cityid = data["cityid"] lng = data["lng"] lat = data["lat"] ip = data.get("ip") proxie_ip = { "http": "http://" + ip, "https": "http://" + ip, } time_int = int(time.time()) ua = f"MApi 1.4 (com.dianping.v1 10.29.3 om_{random.randint(1, 999)} GM1900; Android 5.1.1)" dpid =