-
Notifications
You must be signed in to change notification settings - Fork 0
/
face_compare_python3_demo.py
136 lines (120 loc) · 4.4 KB
/
face_compare_python3_demo.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
#
# 人脸比对 WebAPI 接口调用示例
# 运行前:请先填写Appid、APIKey、APISecret以及图片路径
# 运行方法:直接运行 main 即可
# 结果: 控制台输出结果信息
#
# 接口文档(必看):https://www.xfyun.cn/doc/face/xffaceComparisonRecg/API.html
#
from datetime import datetime
from wsgiref.handlers import format_date_time
from time import mktime
import hashlib
import base64
import hmac
from urllib.parse import urlencode
import os
import traceback
import json
import requests
class AssembleHeaderException(Exception):
def __init__(self, msg):
self.message = msg
class Url:
def __init__(this, host, path, schema):
this.host = host
this.path = path
this.schema = schema
pass
# 进行sha256加密和base64编码
def sha256base64(data):
sha256 = hashlib.sha256()
sha256.update(data)
digest = base64.b64encode(sha256.digest()).decode(encoding='utf-8')
return digest
def parse_url(requset_url):
stidx = requset_url.index("://")
host = requset_url[stidx + 3:]
schema = requset_url[:stidx + 3]
edidx = host.index("/")
if edidx <= 0:
raise AssembleHeaderException("invalid request url:" + requset_url)
path = host[edidx:]
host = host[:edidx]
u = Url(host, path, schema)
return u
def assemble_ws_auth_url(requset_url, method="GET", api_key="", api_secret=""):
u = parse_url(requset_url)
host = u.host
path = u.path
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
print(date)
# date = "Thu, 12 Dec 2019 01:57:27 GMT"
signature_origin = "host: {}\ndate: {}\n{} {} HTTP/1.1".format(host, date, method, path)
print(signature_origin)
signature_sha = hmac.new(api_secret.encode('utf-8'), signature_origin.encode('utf-8'),
digestmod=hashlib.sha256).digest()
signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
api_key, "hmac-sha256", "host date request-line", signature_sha)
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
print(authorization_origin)
values = {
"host": host,
"date": date,
"authorization": authorization
}
return requset_url + "?" + urlencode(values)
def gen_body(appid, img1_path, img2_path, server_id):
with open(img1_path, 'rb') as f:
img1_data = f.read()
with open(img2_path, 'rb') as f:
img2_data = f.read()
body = {
"header": {
"app_id": appid,
"status": 3
},
"parameter": {
server_id: {
"service_kind": "face_compare",
"face_compare_result": {
"encoding": "utf8",
"compress": "raw",
"format": "json"
}
}
},
"payload": {
"input1": {
"encoding": "jpg",
"status": 3,
"image": str(base64.b64encode(img1_data), 'utf-8')
},
"input2": {
"encoding": "jpg",
"status": 3,
"image": str(base64.b64encode(img2_data), 'utf-8')
}
}
}
return json.dumps(body)
def run(appid, apikey, apisecret, img1_path, img2_path, server_id='s67c9c78c'):
url = 'http://api.xf-yun.com/v1/private/{}'.format(server_id)
request_url = assemble_ws_auth_url(url, "POST", apikey, apisecret)
headers = {'content-type': "application/json", 'host': 'api.xf-yun.com', 'app_id': appid}
print(request_url)
response = requests.post(request_url, data=gen_body(appid, img1_path, img2_path, server_id), headers=headers)
resp_data = json.loads(response.content.decode('utf-8'))
print(resp_data)
print(base64.b64decode(resp_data['payload']['face_compare_result']['text']).decode())
#请填写控制台获取的APPID、APISecret、APIKey以及要比对的图片路径
if __name__ == '__main__':
run(
appid='bbbd9a32',
apisecret='YmZhZDJmOGZlYzdlZDQ5ZTdmYzUyNjU3',
apikey='0012cfa7e8b4bb50639efd7dbfa5de4f',
img1_path=r'IMG_0797.png',
img2_path=r'IMG_1495.png',
)