import json import requests from hytest.common import * from cfg import cfg from bs4 import BeautifulSoup from requests.packages import urllib3 #存放公用方法 # 存储 全局共享 数据 GSTORE = {} class APIMgr(): #打印https请求与消息 def __init__(self): self.ui = None self.token = None def printRequest(self,req): if req.body==None: msgBody='' else: msgBody=req.body self.ui.outputWindow.append( '{}\n{}\n{}\n\n{}'.format( '\n\n-------发送请求--------', req.method+''+req.url, '\n'.join('{}:{}'.format(k,v) for k,v in req.headers.items()), msgBody, )) # 打印http相应消息的函数 def printResponse(self, response): print('\n\n----- https response begin -----') print(response.status_code) # print(response.headers) for k, v in response.headers.items(): print(f'{k}:{v}') print(response.content.decode('utf8')) print('----- https response end-----\n\n') headers = { 'Accept-Charset':'utf-8', 'Accept':'*/*', 'Accept-Encoding':'gzip, deflate, br', 'User-Agent':'Apifox/1.0.0 (https://apifox.com)', 'Content-Type':'application/json', 'MiniprogramCode':'wy_zbxm' } #headers设置为全局变量 GSTORE['headers'] = headers # session对象设置为全局变量 s = requests.Session() GSTORE['s'] = s # 小程序登录接口 def mgr_login_wxapi(self, useproxies=False): headers=GSTORE['headers'] s = GSTORE['s'] if useproxies: self.s.proxies.update({'http':'127.0.0.1:8888'}) response = self.s.post(f"{cfg.target_host_wxapi}/debrisproduct/free/autologin",headers=headers,json= { 'phone':'15037870765', 'token':'8817684142977367381' }) self.printResponse(response) # 把response对象返回出去 return response """退出登录小程序""" def mgr_logout_wxapi(self): url = f"{cfg.target_host_wxapi}/debrisproduct/logout" headers = GSTORE['headers'] s = GSTORE['s'] s.post(url=url, headers=headers) """我的用户信息""" def myuser_information(self): hearders = GSTORE['headers'] s = GSTORE['s'] response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/myinfo", headers=hearders) return response """招标项目搜索-搜索入口""" def bidding_project_search(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "area": "河南", "propertyForm": "医院", "OwnerType": "医疗单位", "Page": 1, "PageSize": 1, "KeyWords": "物业" } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/search/bidList", headers=hearders, json=json) return response """采购意向搜索-搜索入口""" def purchase_intention_search(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "area": "安徽", "propertyForm": "学校", "OwnerType": "教育单位", "Page": 1, "PageSize": 1, "KeyWords": "数据" } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/search/buyPurposeList", headers=hearders, json=json) return response """招标项目搜索-订阅入口""" def tender_subscription(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "area": "河南", "propertyForm": "医院", "OwnerType": "医疗单位", "Page": 1, "PageSize": 1, "KeyWords": "物业" } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/search/bidSubscribeList", headers=hearders, json=json) return response """采购意向搜索-订阅入口""" def purchase_intention_subscription(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "area": "安徽", "propertyForm": "学校", "OwnerType": "教育单位", "Page": 1, "PageSize": 1, "KeyWords": "数据" } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/search/buyPurposeSubscribeList", headers=hearders, json=json) return response """情报热词""" def information_buzzword(self): hearders = GSTORE['headers'] s = GSTORE['s'] response = s.get(f"{cfg.target_host_wxapi}/debrisproduct/info/getHotInfo", headers=hearders) return response """情报列表/搜索-搜索""" def information_search(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "page_size": 1, "page_num": 1, "keyWords": "科技" } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/info/getInfoList", headers=hearders, json=json) return response """情报列表/搜索-订阅""" def information_subscription(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "page_size": 1, "page_num": 1, "keyWords": "科技" } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/info/subscribeList", headers=hearders, json=json) return response """监控列表""" def monitor_list(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "Size": 1, "Page": 1 } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/monitorList", headers=hearders, json=json) return response """监控操作""" def monitor_operate(self): hearders = GSTORE['headers'] s = GSTORE['s'] json={ "CompanyName":"东莞市中医院", "OperateType":1 } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/monitorOperate", headers=hearders, json=json) return response """详情页信息""" def detailed_information(self): hearders = GSTORE['headers'] s = GSTORE['s'] json = { "_id": "ABCY1xZcDIvJyssAlV1cHU8CicoEjRjYWR1Pw4jIy4wc3xzfTNUCec%3D" } response = s.post(f"{cfg.target_host_wxapi}/debrisproduct/getBiddingDetail", headers=hearders, json=json) return response #获取订单详情 def debrisproduct_orderDetail(self, orderCode): headers = GSTORE['headers'] s = GSTORE['s'] # 从全局变量中获取 cookie INFO(f"使用的 cookies: {s}") # 打印 cookie，确保它正确 response = s.post( f"{cfg.target_host_wxapi}/debrisproduct/orderDetail", headers=headers, data={"orderCode": orderCode} ) return response #获取订单列表 def debrisproduct_orderList(self, orderCode): headers = GSTORE['headers'] s = GSTORE['s'] # 从全局变量中获取 cookie INFO(f"使用的 cookies: {s}") # 打印 cookie，确保它正确 response = s.post( f"{cfg.target_host_wxapi}/debrisproduct/orderList", headers=headers, data={"orderCode": orderCode} ) apimgr = APIMgr()