jianyu_auto_api/lib/wxapi.py

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()