2022年 11月 8日

python—接口调用

文章目录

      • 一、接口请求
      • 二、二进制请求与响应
          • GET 请求
          • POST 请求
      • 三、关于图片的请求
      • 四、Python调用WebService接口
      • 五、Python HmacSHA1 加密与签名

一、接口请求

http 请求包含:请求行、请求头、请求体。【关于详情】

http协议报文

1.请求报文 (请求行/请求头/请求数据/空行) 请求行:
  请求方法字段、URL字段和HTTP协议版本
  例如:GET/index.htmlHTTP/1.1
    get方法将数据拼接在url后面,传递参数受限
  请求方法:
    GET、POST、HEAD、PUT、DELETE、OPTIONS、TRACE、CONNECT
请求头(key-value形式):
  User-Agent:产生请求的浏览器类型。
  Accept:客户端可识别的内容类型列表。
  Host:主机地址
请求数据:
  post方法中,会把数据以keyvalue形式发送请求
空行:
  发送回车符和换行符,通知服务器以下不再有请求头
2.响应报文 (状态行、消息报头、响应正文) 状态行
消息报头
响应正文

request

import requests

r = requests.get('https://api.github.com/events')                      # 获取网页信息
r = requests.post('http://httpbin.org/post', data = {'key':'value'})   # 发送信息请求

r = requests.put('http://httpbin.org/put', data = {'key':'value'})
r = requests.delete('http://httpbin.org/delete')
r = requests.head('http://httpbin.org/get')
r = requests.options('http://httpbin.org/get')

payload = {'key1': 'value1', 'key2': 'value2'}
r = requests.get("http://httpbin.org/get", params=payload)

print(r.url)       # 有返回,则URL已被正确编码
print(r.text)      # 推测的文本编码
print(r.content)   # 找到 (HTTP 和 XML自身可以指定编码格式)文本编码
print(r.encoding)  # 常用于中文乱码解决。用于修改编码方式(使用 codecs 模块进行注册)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

Requests 也可以使用定制的编码:
创建了自己的编码,如r.encoding = ‘ISO-8859-1’ 并使用 codecs 模块进行注册,你就可以轻松地使用这个解码器名称作为 r.encoding 的值, 然后由 Requests 来为你处理编码。
关于【响应中文乱码问题请点击】。

二、二进制请求与响应

GET 请求

以请求返回的二进制数据保存图片,保存文件。

import requests
from PIL import Image
from io import BytesIO


url = 'https://api.github.com/some/endpoint'
headers = {'user-agent': 'my-app/0.0.1'}
r = requests.get(url, headers=headers)

i = Image.open(BytesIO(r.content))

with open(filename, 'wb') as fd:
    for chunk in r.iter_content(chunk_size):
        fd.write(chunk)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
POST 请求
import requests


payload = {'key1': 'value1', 'key2': 'value2'}

r = requests.post("http://httpbin.org/post", data=payload)  # 简单地传递一个字典给 data 参数
r = requests.post(url, data=json.dumps(payload))            # 传递str 而不是dict,数据会被直接发布出去。
print(r.text)
print(r.content)
print(r.url)
print(r.status_code)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

多部分编码(Multipart-Encoded)的文件

url = 'http://httpbin.org/post'
files = {'file': open('report.xls', 'rb')}

# 显式地设置文件名,文件类型和请求头
files = {'file': ('report.xls', open('report.xls', 'rb'), 'application/vnd.ms-excel', {'Expires': '0'})}

# 发送作为文件来接收的字符串
files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\n')}

>>> r = requests.post(url, files=files)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

建议你用二进制模式(binary mode)打开文件。这是因为 Requests 可能会试图为你提供 Content-Length header,在它这样做的时候,这个值会被设为文件的字节数(bytes)。如果用文本模式(text mode)打开文件,就可能会发生错误。
最好先用二进制的形式打开文件,将其编码成base64来进行传输:
a = open("pdf_reference.pdf", "rb").read().encode("base64")

带参数,带请求头
小示例:

import requests
import json


url = 'http://official-account/app/messages/group'
body = {"type": "text", "content": "text", "tag_id": "20717"}
headers = {'content-type': "application/json", 'Authorization': 'APP appid = 4abf1a,token = 9480295ab2e2eddb8'}

response = requests.post(url, data = json.dumps(body), headers = headers)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

三、关于图片的请求

图片和二进制格式互转

# 以 二进制方式 进行图片读取
with open("img.jpg","rb") as f:
	img_bin = f.read() # 以二进制方式读取的图片内容

# 将 图片的二进制内容 转成 真实图片
with open("img.jpg","wb") as f:
    f.write(img_bin) # img_bin里面保存着二进制流的图片内容
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

图片转成字节数组 image2bytearray
我打印了这两者输出,有些微差别,可以都试一下

import io
from PIL import Image


#================ 方法一 ===================
img = Image.open(r"C:\Users\xxx\Desktop\111.png", mode='r')

imgByteArr = io.BytesIO()
img.save(imgByteArr, format='png')

imgByteArr = imgByteArr.getvalue()
print('imgByteArr:',imgByteArr)



#================ 方法二 ==================
data = open(r"C:\Users\xxx\Desktop\111.png", 'rb').read()
print('bytearray:',bytearray(data))
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

安装依赖:pip3 install Pillow

from io import BytesIO
from PIL import Image
import requests

res = requests.get('http://p1.pstatp.com/list/300x196/pgc-image/152923179745640a81b1fdc.webp', stream=True)  # 获取字节流最好加stream这个参数
byte_stream = BytesIO(res.content)   # 把请求到的数据转换为Bytes字节流(这样解释不知道对不对,可以参照[廖雪峰](https://www.liaoxuefeng.com/wiki/0014316089557264a6b348958f449949df42a6d3a2e542c000/001431918785710e86a1a120ce04925bae155012c7fc71e000)的教程看一下)

roiImg = Image.open(byte_stream)      # Image打开Byte字节流数据

imgByteArr = io.BytesIO()             # 创建一个空的Bytes对象
roiImg.save(imgByteArr, format='PNG') # PNG就是图片格式,我试过换成JPG/jpg都不行
imgByteArr = imgByteArr.getvalue()    # 这个就是保存的图片字节流

# 写入本地
with open("./abc.png", "wb") as f:
    f.write(imgByteArr)


#================ 方法三 ==================
img = Image.open(BytesIO(byte_stream))
img.save(save_name + img.format, quality=95)


img = Image.open(BytesIO(byte_stream))
img.save(save_name + img.format, quality=95)


img = Image.frombytes("RGB", (18,18), img) # 图片格式,图片大小,图片数据流
img.save("111.png")
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

四、Python调用WebService接口

关于WebService查看【wiki网址】。
Python的WebService开发库是soaplib。 【关于开发示例】
Suds: 是一个轻量级的SOAP客户端,suds-jurko是suds的一个分支版本。

suds:https://pypi.org/project/suds/ 最近维护时间: Sep 15, 2010
suds-jurko 0.4.1.jurko.2:https://pypi.org/project/suds-jurko/0.4.1.jurko.2/最近维护时间:Dec 25, 2011
安装 pip install suds
安装:pip install suds-jurko==0.4.1.jurko.2

免费的webserver网址:http://www.webxml.com.cn/zh_cn/web_services.aspx
测试工具:
接口使用cxf编写:SoapUI 5.2.1:下载地址,测试查看1。测试查看2。
也可以用:Webservice测试工具 Storm测试。

#encoding:utf-8
import json
from suds.client import Client

# import logging
# logging.basicConfig(level=logging.INFO)
# logging.getLogger('suds.client').setLevel(logging.DEBUG)

# username = 'api_user'
# password = 'api_pwd'

url="http://192.168.18.74/api/Calculator/soap11/description"
headers = {'Content-Type': 'application/soap+xml; charset="UTF-8"'}

client = Client(url,headers=headers,faults=False,timeout=15)

def call_api_test(num1,num2):
    try:
        #--WSDL请求Header----------------------------------------------
        # auth = client.factory.create('AuthenticationInfo')
        # auth.userName = username
        # auth.password = password
        # client.set_options(soapheaders=auth)

        #--WSDL请求Body----------------------------------------------
        result = client.service.add(a=num1,b=num2)
        
        if result[0] == 200:
            return (True, result[1])
        else:
            return (False, result[1])
    except Exception as e:
        return (False, e)

if __name__ == '__main__': #get_software
    a= call_api_test(2,3)
    print (json.loads(a[1]))
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

python3测试一个小案例

from suds.client import Client


url = "http://www.webxml.com.cn/webservices/qqOnlineWebService.asmx?wsdl"

client = Client(url)    #创建Webservice Client对象
print(client)           #打印出Client对象所有的方法

result = client.service.qqCheckOnline("xxxxxxxx")  #使用client.service.qqCheckOnline方法
print("结果为:"+result)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

suds-community 0.8.4:https://pypi.org/project/suds-community/ 最近维护时间: Dec 22, 2019
安装:pip install suds-community

from suds.client import Client

# 连接到webservice服务,获取查询天气服务方法
client = Client('http://ws.webxml.com.cn/WebServices/WeatherWS.asmx?wsdl')
print(client)

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

报错:suds.TypeNotFound: Type not found:

# 解决错误
from suds.client import Client
from suds.xsd.doctor import ImportDoctor, Import

# 导入正确的命名空间。
imp = Import('http://www.w3.org/2001/XMLSchema', location='http://www.w3.org/2001/XMLSchema.xsd')
imp.filter.add('http://WebXml.com.cn/')
doctor = ImportDoctor(imp)
client = Client('http://ws.webxml.com.cn/WebServices/WeatherWS.asmx?wsdl', doctor=doctor)
print(client)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

五、Python HmacSHA1 加密与签名

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import math
import os
import io
import requests
import time
import datetime
import hashlib
import codecs
import uuid
import traceback

def testapi():
    appkey = "5daexxxx"                             #这里输入提供的app_key
    appsecret = "dcf85ca2a9a2c1dfe3e45905xxxxxxxx"  #这里输入提供的app_secret

    api_url = "https://xxxx/v1/item/xxxx"
    image_url = "http://xxxxt/img/sample.jpg"
    result = {}
    try:
        timestamp = int(time.time())               # generate timestamp

        m = hashlib.md5()
        token = appkey + "+" + str(timestamp) + "+" + appsecret
        m.update(token.encode('utf-8'))
        token = m.hexdigest()                      # generate token

        # post request
        data = {'image_url': image_url, 'app_key': appkey, 'timestamp': str(timestamp), 'token': token}
        r = requests.post(api_url, data=data)
        if r.status_code != 200:
            print("failed to get info from : ", image_url)
        else:
            result = r.json()
            print(result)
    except:
        traceback.print_exc()
    return result


if __name__ == '__main__':
    testapi()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
import requests, json
import base64
import time
import hmac
import hashlib
import os


def get_img(img_paths):
    img_list = []
    for paths,dirs,filenames in os.walk(img_paths):
        for filename in filenames:
            if filename.endswith('.jpg'):
                img_list.append(paths+'/'+filename)
    return img_list


def POST(img_path):
    api_url = "http://192.168.x.x:8000/xx/xx/xxxx"
    encryptKey = 'xx+zkudjZIGM/KbppeVSTemJpMErh31+xx/xx=='
    with open(img_path, 'rb') as f:
        base64_data = base64.b64encode(f.read())
        imgdata = base64_data.decode()

    params = {"appId":'xxxxxxxxxxx',
              'data': {"imgBase64":imgdata},
              "format": "json",
              "noncestr": "sdkfhjsdjkfsdgf",
              "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
              "version": "V1.0"
              }
    encryptText = 'appId='+params["appId"]+'&data={"imgBase64":"'+params["data"]['imgBase64']+\
                  '"}&format=json&noncestr=xxxxxx&timestamp='+params['timestamp']+'version=V1.0'
    print(encryptText)

    sign = base64.b64encode(hmac.new(bytes(encryptKey,'utf-8'), bytes(encryptText,'utf-8'), digestmod='sha1').digest())
    r = requests.post(api_url, sign)
    print(r)
    print(r.status_code)
    print(r.json)


if __name__ == '__main__':
    img_path = './201920100013253001_30302_02.jpg'
    POST(img_path)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45

python2.7post请求示例

# -*- coding: utf-8 -*-
import urllib
import urllib2
import mimetools, mimetypes
import os
import json
import sys
import shutil
import base64


def get_files(input_path):
    file_list = []
    for root,dirs,filenames in os.walk(input_path):
        for filename in filenames:
            if filename.endswith((".jpg",".tif",".jpeg",".png",".JPG",".TIF",".JPEG",".PNG")):
                file_list.append(root+"/"+filename)
    return file_list


class Callable:
    def __init__(self, anycallable):
        self.__call__ = anycallable

class MultipartPostHandler(urllib2.BaseHandler):
    handler_order = urllib2.HTTPHandler.handler_order - 10  # needs to run first

    def http_request(self, request):
        data = request.get_data()
        if data is not None and type(data) != str:
            v_files = []
            v_vars = []
            try:
                for key, value in data.items():
                    if type(value) == file:
                        v_files.append((key, value))
                    else:
                        v_vars.append((key, value))
            except TypeError:
                systype, value, traceback = sys.exc_info()
                raise TypeError, "not a valid non-string sequence or mapping object", traceback

            if len(v_files) == 0:
                data = urllib.urlencode(v_vars, 1)
            else:
                boundary, data = self.multipart_encode(v_vars, v_files)
                contenttype = 'multipart/form-data; boundary=%s' % boundary
                if request.has_header('Content-Type') \
                        and request.get_header('Content-Type').find('multipart/form-data') != 0:
                    print "Replacing %s with %s" % (request.get_header('content-type'), 'multipart/form-data')
                request.add_unredirected_header('Content-Type', contenttype)
            request.add_data(data)
        return request

    def multipart_encode(vars, files, boundary=None, buffer=None):
        if boundary is None:
            boundary = mimetools.choose_boundary()
        if buffer is None:
            buffer = ''
        for key, value in vars:
            buffer += '--%s\r\n' % boundary
            buffer += 'Content-Disposition: form-data; name="%s"' % key
            buffer += '\r\n\r\n' + value + '\r\n'
        for key, fd in files:
            filename = os.path.basename(fd.name)
            contenttype = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
            buffer += '--%s\r\n' % boundary
            buffer += 'Content-Disposition: form-data; name="%s"; filename="%s"\r\n' % (key, filename)
            buffer += 'Content-Type: %s\r\n' % contenttype
            fd.seek(0)
            buffer += '\r\n' + fd.read() + '\r\n'
        buffer += '--%s--\r\n\r\n' % boundary
        return boundary, buffer

    multipart_encode = Callable(multipart_encode)
    https_request = http_request


class Verify_Code_Test:
    def __init__(self):
        self.data = {"channel_num": 'TEST', "file_id": "", "file_content": ""}

    def image_2_string(self, img_path):
        try:
            self.data["file_id"] = img_path.split('/')[-1]
            with open(img_path, 'rb') as rf:
                data0 = rf.read()
            image_base64 = base64.b64encode(data0)
            self.data["file_content"] = image_base64.decode()
            return True
        except Exception as e:
            print(str(e))
        return False

    def verify_code(self, img_path):
        if self.image_2_string(img_path):
            opener = urllib2.build_opener(MultipartPostHandler)
            data = json.dumps(self.data)
            resp = opener.open('http://127.0.0.1:5000/verify_code', data)
            print resp.read()

if __name__ == '__main__':
    verify_code_test = Verify_Code_Test()
    verify_code_test.verify_code("./images/icbc.png")
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104

参考与鸣谢:
字节数组 https://www.cnblogs.com/fieldtianye/p/8276552.html
http://2.python-requests.org/zh_CN/latest/user/quickstart.html
http://2.python-requests.org/zh_CN/latest/user/advanced.html#advanced
https://blog.csdn.net/qq_22034353/article/details/94398536