jianyu_auto_api/lib/webapi.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 = {

        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.67"
    }
    #打开链接
    def open_url(self,url):
        response = requests.get(url)
        response.encoding = 'utf-8'  # 设置编码为gbk
        soup = BeautifulSoup(response.text, 'html.parser')
        title = soup.title.string
        return title

    #headers设置为全局变量
    GSTORE['headers'] = headers
    # session对象设置为全局变量
    s = requests.Session()
    GSTORE['s'] = s
    #pc登录接口
    def mgr_login(self, phone='18211989987', password='123456',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}/phone/login",headers=headers,data=
                                    {
                                     'reqType': 'phoneLogin',
                                     'isAutoLogin': 'false',
                                     'phone':phone,
                                     'password':password
                                    })
        self.printResponse(response)
        # 把response对象返回出去
        return response

    # app登录接口
    def mgr_login_app(self, phone='18211989987', password='123456',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_app}/jyapp/free/login",headers=headers,params=
                                    {
                                     'reqType': 'phoneLogin',
                                     'phone':phone,
                                     'password':password,
                                     'rid':'',
                                     'oid': '',
                                     'phoneType': '',
                                     'channel': '',
                                     'deviceId': ''
                                    })
        self.printResponse(response)
        # 把response对象返回出去
        return response
    """退出登录pc"""
    def mgr_logout(self):
        url = f"{cfg.target_host}/front/signOut"
        headers = GSTORE['headers']
        s = GSTORE['s']
        s.post(url=url, headers=headers)
        # self.printResponse(res)
        # return res

    """退出登录app"""
    def mgr_logout_app(self):
        url = f"{cfg.target_host_app}/jyapp/free/signOut"
        headers = GSTORE['headers']
        s = GSTORE['s']
        s.post(url=url, headers=headers)
    #招标搜索
    def bidsearch(self, keywords="建筑", publishtime="fiveyear", selectType="content"):
        #使用全局变量
        headers = GSTORE['headers']
        params={"keywords":keywords,"publishtime":publishtime,"timeslot":"","area":"","subtype":"","minprice":"","maxprice":"","industry":"","buyerclass":"","buyertel":"","winnertel":"","selectType":selectType,"notkey":"","fileExists":"0","city":"","searchGroup":"0","searchMode":"0","wordsMode":"0","additionalWords":""}
        #保存session
        session = self.s
        response = session.post(f"{cfg.target_host}/jylab/supsearch/index.html", headers=headers, params=params)
        self.printResponse(response)
        return response

     # 企业搜索
    def enterpriseSearch(self, match="北京剑鱼信息技术有限公司河南分公司", matchType="A", pageSize="10", pageNum="0"):
        # 使用全局变量
        headers = GSTORE['headers']
        # 保存session
        session = self.s
        response = session.post(f"{cfg.target_host}/publicapply/enterpriseSearch/doQuery", headers=headers, data={
            'match': match,
            'matchType': matchType,
            'pageSize': pageSize,
            'pageNum': pageNum
        })
        self.printResponse(response)
        return response

        # 供应搜索
    def supplySearch(self, keywords="PH计", searchType="title", province="", city="", time="", status="0",
                         pageSize=50, pageIndex=1):
        headers = GSTORE['headers']
        headers['Content-Type'] = 'application/json'  # 添加Content-Type头部
        url = f"{cfg.target_host}/jyinfo/supplySearch"
        data = {
                "keywords": keywords,
                "searchType": searchType,
                "province": province,
                "city": city,
                "time": time,
                "status": status,
                "pageSize": pageSize,
                "pageIndex": pageIndex
            }
        session=self.s
        response = session.post(url=url, headers=headers, data=json.dumps(data))
        self.printResponse(response)
        return response

    #采购单位搜索
    def buyersousuo(self, buyerName, province=None, city=None, buyerClass=None, isCheckFollow=True, isCheckReceive=True,
                    isContact=0, pageSize=10, pageNum=1):
        if buyerClass is None:
            buyerClass = []
        if city is None:
            city = []
        if province is None:
            province = []
        url = f"{cfg.target_host}/jyapi/jybx/buyer/eType/buyerList"
        data = {
            "buyerName": buyerName,
            "province": province,
            "city": city,
            "buyerClass": buyerClass,
            "isCheckFollow": isCheckFollow,
            "isCheckReceive": isCheckReceive,
            "isContact": isContact,
            "pageSize": pageSize,
            "pageNum":pageNum
        }
        response = requests.post(url=url, json=data, headers=self.headers)
        self.printResponse(response)
        return response
    #接口数据传值常用三种方式：urlencoded---params，键值对---data，json格式---json
    #获取推送记录接口
    def push_list(self):
        headers=GSTORE['headers']
        s = GSTORE['s']
        response = self.s.post(f"{cfg.target_host}/jyapi/jybx/subscribe/fType/list",headers=headers,json=
                         {"pageNum": 1, "pageSize": 50, "format": "table", "area": "", "selectTime": "all", "city": "", "buyerClass": "",
                            "subtype": "", "industry": "", "keyWords": "", "fileExists": "", "price": "", "source": "", "exportNum": "",
                            "vt": ""})
        return  response

    # 不登录招标搜索
    def notloggedin_search(self, keyword='科技', publishtime='thisyear', selectType='content,title'):
        headers = GSTORE['headers']
        params={"keywords": keyword , "publishtime": publishtime, "timeslot": "", "area": "", "subtype": "",
         "minprice": "", "maxprice": "", "industry": "", "buyerclass": "", "buyertel": "", "winnertel": "",
         "selectType": selectType, "notkey": "", "fileExists": "0", "city": "", "searchGroup": "0",
         "searchMode": "0", "wordsMode": "0", "additionalWords": ""}
        response = requests.post(f"{cfg.target_host}/jylab/supsearch/index.html", headers=headers, params=params)
        response.raise_for_status()  # 如果请求失败，会抛出异常
        return response

    #不登录采购单位搜索
    def notloggedin_buysearch(self,keyword):
        headers = GSTORE['headers']
        s = GSTORE['s']
        data = {"buyerName":keyword,"province":[],"city":[],"buyerClass":[],"isCheckFollow":False,"isCheckReceive":False,"isContact":0,"pageSize":10,"pageNum":1}

        response =s.post(f"{cfg.target_host}/jyapi/jybx/buyer/eType/buyerList", headers=headers, json=data)
        # response.raise_for_status()  # 如果请求失败，会抛出异常
        return response
    #消息中心列表
    def get_messagelist(self):
        headers = GSTORE['headers']
        s = GSTORE['s']
        response=s.post(f"{cfg.target_host}/jyapi/messageCenter/MessageList",headers=headers,json=
                    {"msgType":-1,"isRead":-1,"offset":1,"size":20})
        return response

    """用户中台"""
    def get_userCenter(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        response = s.post(f"{cfg.target_host}/userCenter/workDesktop/menuInfo", headers=hearders)
        return response

    """我的订单"""
    def get_myOrder(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params = {
            "type": 0,
            "pageNum": 1,
            "fromPage": "pc",
            "page_size": 10
        }
        response = s.post(f"{cfg.target_host}/subscribepay/orderListDetails/myOrder", headers=hearders, params=params)
        return response

    """优惠卷"""
    def get_coupon(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "mold": 4,
            "currentPage": 1,
            "pageSize": 8,
            "platform": "P"
             }
        response= s.post(f"{cfg.target_host}/jyCoupon/getInfoByUser", headers=hearders,params=params)
        return response

    """数据自动导出"""
    def get_dataExport(self,publishtime=1672502400_1688951529,keyword='数据',selectType='title'):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "publishtime":publishtime,
            "area":"",
            "city":"",
            "region":"",
            "industry":"",
            "buyerclass":"",
            "keyword": [{"keyword":keyword,"appended":[],"exclude":[]}],
            "selectType":selectType,
            "minprice":"",
            "maxprice":"",
            "subtype":"",
            "buyer":"",
            "winner":"",
            "dataType": 2
        }
        response = s.post(f"{cfg.target_host}/front/dataExport/sieveData", headers=hearders, params=params)
        return response

    """数据导出记录"""
    def Export_recordList(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "pageNum": 0,
            "pageSize": 10
        }
        response = s.post(f"{cfg.target_host}/subscribepay/dataExportPack/recordList", headers=hearders, params=params)
        return response

    """剑鱼文库搜索"""
    def Library_search(self,keyWord='数据'):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "keyWord":keyWord,
            "tag":"",
            "sort": "tSort",
            "num": 1,
            "size": 10
        }
        response = s.post(f"{cfg.target_host}/jydocs/search", headers=hearders, params=params)
        return response

    """剑鱼文库收藏"""
    def Library_collection(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "sign": 1,
            "num": 1,
            "size": 10
        }
        response = s.post(f"{cfg.target_host}/jydocs/user/list", headers=hearders, params=params)
        return response

    """剑鱼文库我的文库"""
    def My_library(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "sign": 0,
            "num": 1,
            "size": 10
        }
        response = s.post(f"{cfg.target_host}/jydocs/user/list", headers=hearders, params=params)
        return response

    """项目进度监控"""
    def Project_monitoring(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "pageNum": 0,
            "pageSize": 500
        }
        response = s.post(f"{cfg.target_host}/bigmember/follow/project/list", headers=hearders, params=params)
        return response

    """企业情报监控"""
    def Enterprise_monitoring(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "pageNum": 0,
            "pageSize": 10,
            "match":"",
            "group":"",
        }
        response = s.post(f"{cfg.target_host}/bigmember/follow/ent/list", headers=hearders, params=params)
        return response
    """客户监控"""
    def Customer_monitoring(self,):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params = {
            "pagesize": 10,
            "pageno": 0,
            "keyword":""
        }
        response = s.post(f"{cfg.target_host}/publicapply/customer/list", headers=hearders, params=params)
        return response

    """标讯收藏"""
    def Message_Collection(self):
        hearders = GSTORE['headers']
        s = GSTORE['s']
        params={
            "buyerPhone":0,
            "buyerclass":"",
            "label":"",
            "pagenum":1,
            "pagesize":50,
            "selectTime":"",
            "winnerPhone":0
        }
        response = s.post(f"{cfg.target_host}/publicapply/bidcoll/list", headers=hearders, params=params)
        return response

    """标讯收藏"""
    #信息获取
    def Getuser(self):
        headers = GSTORE['headers']
        s = GSTORE['s']
        response = s.get(f"{cfg.target_host}/jypay/user/getAccountInfo", headers=headers)
        return response
    #密码校验
    def Check_password(self):
        headers = GSTORE['headers']
        s = GSTORE['s']
        params = {
            "password": "123456"
        }
        response = s.post(f"{cfg.target_host}/publicapply/password/check", headers=headers, params=params)
        return response
    #身份获取
    def Identity_list(self):
        headers = GSTORE['headers']
        s = GSTORE['s']
        params = {
        }
        response = s.post(f"{cfg.target_host}/publicapply/identity/list",headers=headers, params=params)
        # 解析响应内容为JSON
        response_json = response.json()
        # 从JSON响应中提取token
        self.token = response_json['data'][0]['token']
        return response
    def Identity_switch(self):
        headers = GSTORE["headers"]
        s = GSTORE["s"]
        params = {
            "token":self.token
        }
        response =s.post(f"{cfg.target_host}/publicapply/identity/switch", headers=self.headers,params=params)
        return response
    def User_info(self):
        headers = {
            'content-Type': 'application/json',
            'appId': '10000',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.67'
        }
        s = GSTORE["s"]
        params = {}
        response =s.post(f"{cfg.target_host}/userCenter/ent/userInfo",headers=headers,params=params)
        return response
    def Whether_buy(self):
        headers = GSTORE["headers"]
        s = GSTORE["s"]
        response =s.get(f"{cfg.target_host}/entnicheNew/buy/whetherbuy",headers=headers)
        return response

    def Commonly_List(self):
        headers = {
            'content-Type': 'application/json',
            'appId': '10000',
            'userid': '63a41aa5cd7ea10389b2a8f3'
        }
        s = GSTORE['s']
        response =s.post(f"{cfg.target_host}/userCenter/workDesktop/renew/commonlyList",headers=headers)
        return response
    def Authorised_info(self):
        headers = {
            "Content-Type":"application/json",
            "functionCode":"znsj_kf_use"
        }
        s = GSTORE['s']
        response =s.post(f"{cfg.target_host}/resourceCenter/waitEmpowerDetail",headers=headers)
        return response
apimgr = APIMgr()