# update: 2022-7-18
"""
中建八局一公司业务接口
"""
from urllib import parse
import requests
import time
import traceback
import sys
import importlib
sys.path.append('/home/server/projects/taiwuict/cscec-8bur-vms/supplement-python')
# sys.path.append(r'D:\share\gitee\casper.supplement-python') # for pc
methods = importlib.import_module(f"libraries.base_original")
class Api(object):
# def __init__(self, username='System001', password='Swdx@143',
def __init__(self, username='500A7062', password='198797#cjhxbin',
project_id='d0f254cc-8196-499a-8f0d-52bf8ba91486',
project_name='山东省公共卫生临床中心（济南市传染病医院二期）项目工程总承包（EPC）'):
"""
第1现场：
济南市中心医院（东院区）项目工程总承包（EPC）
7eeb53a1-2bbd-4efa-8c8e-66936476226c
第2现场：
山东省大数据产业基地建设项目施工总承包
项目编号：500D10301958
ORGID：63729db2-ea9c-4096-b5d9-3ad0369903c8
第3现场：
山东省公共卫生临床中心（济南市传染病医院二期）项目工程总承包（EPC）
ORGID：d0f254cc-8196-499a-8f0d-52bf8ba91486
第4现场：
济南奥体东 13 及 17-2 号地块开发项目工程总 承包（EPC）
a8defc53-76b3-4f4e-931d-8dd35ba310c7
"""
# --- define ---
self.service_url = 'http://app.cscec81.com/api' # 正式环境
# self.service_url = f"http://pmdev.cscec81.com/api" # 正式环境
# self.service_url = 'http://221.214.64.195:31229/api' # 公网测试地址
# self.service_url = 'http://10.198.6.181:31229/api' # vpn测试环境 https://221.214.64.195 gelubo01 Gelubo@2022
# --- release ---
self.project_id = project_id
self.project_name = project_name
# --- release ---
self.token = None
if username and password:
session_id, user_code, user_id = self.get_user_info(username, password)
self.token = f"Bearer {self.get_token(session_id, user_code, user_id)}"
# --- test ---
# self.token_test = 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjAxMTgyODIsInVzZXJfbmFtZSI6Iui1tei_nuWxsTM0NTBhMjI0NmZiLTkwMWYtNGIwNi1hMjU2LTY4YTI4Y2VkZTU2Y2UwZDk5OTk3LTkzNzEtNGMwZi04NDNjLTUyNTMzYjE4ZTYxNSIsImp0aSI6Ijg3MzcxZTY0LTgwNzEtNGJlYy04YWMzLTUxMzViMTU5MzhjMyIsImNsaWVudF9pZCI6ImJyb3dzZXIiLCJzY29wZSI6WyJjbGllbnQiXX0.NF0IXabm_7RRqcdra6A7fsBvyBWa_62rkppQAFxalc4'
@staticmethod
def get_user_info(username, password):
"""
获取用户信息
url: http://pmdev.cscec81.com/Login/Login.ashx?t=' + new Date().getTime() + '&method=CheckLogin
"""
# url = f"{self.service_url}/Login/Login.ashx"
url = f"http://app.cscec81.com/Login/Login.ashx" # 正式环境
# url = f"http://pm.cscec81.com/Login/Login.ashx" # 测试环境
# url = f"http://221.214.64.195:31229/Login/Login.ashx" # 测试环境
url += f"?t={int(time.time() * 1000)}&method=CheckLogin"
data = {
'Username': username,
'password': password,
'usertype': '0',
'sys': 'sitemanage',
'clientType': '5',
}
payload = parse.urlencode(data, encoding='utf-8')
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Cookie': 'ASP.NET_SessionId=luy3wg10b2kymc2nbiytn4xt'
}
print('Api.get_user_info.url:', url)
# print('Api.get_user_info.headers:', headers)
# print('Api.get_user_info.payload:', payload)
response = requests.request('POST', url, headers=headers, data=payload)
if response.status_code > 300:
print('Api.get_user_info.result:', response.status_code)
print('Api.get_user_info.detail:', response.text)
return '', '', ''
print('Api.get_user_info.result:', response.status_code)
json_data = response.json()
# print('Api.get_user_info.json_data:', json_data)
if not json_data or not json_data.get('data'):
return '', '', ''
session_id = json_data.get('data').get('ticketID')
user_code = json_data.get('data').get('userCode')
user_id = json_data.get('data').get('userID')
# print('Api.get_user_info.session_id:', session_id)
# print('Api.get_user_info.user_code:', user_code)
# print('Api.get_user_info.user_id:', user_id)
return session_id, user_code, user_id
@staticmethod
def get_token(session_id, user_code, user_id):
"""
获取令牌
url: http://pmdev.cscec81.com/api/oauth/token?grant_type=session&sessionId="
+ ticketID + "&userCode=" + userCode + "&userId=" + userID + "&client_type=web&system=1
"""
# url = f"{self.service_url}/oauth/token"
url = f"http://app.cscec81.com/api/oauth/token" # 正式环境
# url = f"http://pm.cscec81.com/api/oauth/token" # 测试环境
url += f"?grant_type=session&sessionId={session_id}&userCode={user_code}&userId={user_id}"
url += f"&client_type=web&system=1"
headers = {
'Authorization': 'Basic YnJvd3Nlcjo='
}
print('Api.get_token.url:', url)
# print('Api.get_token.headers:', headers)
response = requests.post(url=url, headers=headers)
if response.status_code > 300:
print('Api.get_token.result:', response.status_code)
print('Api.get_token.detail:', response.text)
return ''
else:
print('Api.get_token.result:', response.status_code)
# return response.headers.get('authorization')
return response.json().get('access_token')
def get_cscec8_user_list(self):
"""
获取八局管理人员
/api/economic/indoor/getProjectUserList?projectId=05037f9b-5cb6-4a43-9410-c0e6a83f280e
return [
{
cardId: 身份证号
imgUrl: 人脸图片
dutyId: 职务id
dutyName: 职务名称
}
]
"""
url = f"http://app.cscec81.com/api/economic/indoor/getProjectUserFaceList?projectId={self.project_id}"
print('Api.get_cscec8_user_list.url:', url)
response = requests.get(url, headers={'content-type': 'application/json'})
if response.status_code > 300:
print('Api.get_cscec8_user_list.result:', response.status_code)
print('Api.get_cscec8_user_list.detail:', response.text)
return []
json_data = response.json()
# print(json_data)
return json_data
def get_worker_list(self):
"""
获取工人列表（云筑网工人信息）
url: http://app.cscec81.com/api/safeeducation/workerInfo/workers?projectName=济南市中心医院（东院区）项目工程总承包（EPC）
url: http://221.214.64.195:31229/api/safeeducation/workerInfo/workers?projectName=丽泽SOHO项目
return [
{
headImagePath: 人脸图像地址
subContractorName: 劳务公司
workTypeName: 工种
}
]
"""
url = f"http://app.cscec81.com/api/safeeducation/workerInfo/workers?projectName={self.project_name}"
url += f"&page=0&size=5000"
print('Api.get_worker_list.url:', url)
response = requests.get(url, headers={'content-type': 'application/json'})
print('Api.get_worker_list.result:', response.status_code)
if response.status_code > 300:
print('Api.get_worker_list.result:', response.status_code)
print('Api.get_worker_list.detail:', response.text)
return []
json_data = response.json()
# {'projectName': '济南市中心医院（东院区）项目工程总承包（EPC）', 'workerName': '高玉锋',
# 'idCardNumber': '372527198311022813', 'workTypeName': '电气设备安装工', 'headImagePath': None}
if not json_data or not json_data.get('success') or not json_data.get('data'):
return []
if not json_data.get('data').get('content'):
return []
# print(json_data.get('data').keys())
# print(json_data.get('data').get('totalElements'))
# print(len(json_data.get('data').get('content')))
# print(json_data.get('data').get('content')[0])
return json_data.get('data').get('content')
@staticmethod
def get_face_image_by_path(image_path):
"""
获取工人照片
url: https://lwres.yzw.cn/worker-avatar/Original/2018/0612/f3230a72-57ab-4aaf-a741-61d8c078fe89.jpg
"""
while True:
try:
url = f"https://lwres.yzw.cn/{image_path}"
# print('Api.get_face_image_by_path.url:', url)
response = requests.get(url, headers={'content-type': 'application/json'})
# print('Api.get_face_image_by_path.result:', response.status_code)
if response.status_code > 300:
print('Api.get_face_image_by_path.result:', response.status_code)
print('Api.get_face_image_by_path.detail:', response.text)
return b''
else:
return response.content
except Exception as exception:
if exception.__class__.__name__ == 'ConnectionError':
print(f"Api.get_face_image_by_path: sleep 60s")
time.sleep(60)
else:
print(f"Api.get_face_image_by_path.exception:{exception.__class__.__name__}")
print(f"Api.get_face_image_by_path.traceback:{traceback.format_exc()}")
return b''
@staticmethod
def get_face_image_by_path_v2(image_path):
"""
获取工人高清头像
url: http://app.cscec81.com/media/202110/25/20211025-161726-381-4862.jpg
"""
while True:
try:
url = f"http://app.cscec81.com/media/{image_path}"
# print('Api.get_face_image_by_path_v2.url:', url)
response = requests.get(url, headers={'content-type': 'application/json'})
# print('Api.get_face_image_by_path_v2.result:', response.status_code)
if response.status_code > 300:
print('Api.get_face_image_by_path_v2.result:', response.status_code)
print('Api.get_face_image_by_path_v2.detail:', response.text)
return b''
else:
return response.content
except Exception as exception:
if exception.__class__.__name__ == 'ConnectionError':
print(f"Api.get_face_image_by_path_v2: sleep 60s")
time.sleep(60)
else:
print(f"Api.get_face_image_by_path_v2.exception:{exception.__class__.__name__}")
print(f"Api.get_face_image_by_path_v2.traceback:{traceback.format_exc()}")
return b''
@staticmethod
def get_face_image_by_url(image_url):
"""
获取八局管理人员人脸图像
url: https://hcm.cscec81.com/img?type=photo&size=120&index=ae2f8424-c63f-11ea-b0a9-ea6f5cf6388d&rnd=1653618903725
"""
while True:
try:
# print('Api.get_face_image_by_url.url:', url)
response = requests.get(image_url, headers={'content-type': 'application/json'})
# print('Api.get_face_image_by_url.result:', response.status_code)
if response.status_code > 300:
print('Api.get_face_image_by_url.result:', response.status_code)
print('Api.get_face_image_by_url.detail:', response.text)
return b''
else:
return response.content
except Exception as exception:
if exception.__class__.__name__ == 'ConnectionError':
print(f"Api.get_face_image_by_url: sleep 60s")
time.sleep(60)
else:
print(f"Api.get_face_image_by_url.exception:{exception.__class__.__name__}")
print(f"Api.get_face_image_by_url.traceback:{traceback.format_exc()}")
return b''
def upload_file(self, face_uuid, file_path):
"""
上传文件图片
"""
# service_url = 'http://221.214.64.195:31229/api'
# url = f"{service_url}/storages/files/anonymousUpload"
url = f"{self.service_url}/storages/files/anonymousUpload"
files = {
'id': (None, face_uuid),
'file': ('1.jpg', open(file_path, 'rb'), 'image/jpg'),
}
headers = {
'Authorization': self.token,
}
print('Api.upload_file.url:', url)
response = requests.post(url=url, files=files, headers=headers)
if response.status_code > 300:
print('Api.upload_file.result:', response.status_code)
# print('Api.upload_file.detail:', response.text)
return ''
else:
print('Api.upload_file.result:', response.status_code)
"""
原地址
http://10.198.6.181:31900/cscec81-cloud/202110/12/20211012-161022-713-7035.jpg
http://10.198.6.173:30900/cscec81-cloud/202206/13/20220613-022118-96-3795.jpg
替换后地址
http://221.214.64.195:31229/media/202110/12/20211012-161022-713-7035.jpg
http://app.cscec81.com/media/202110/12/20211012-161022-713-7035.jpg
"""
link = response.json().get('link')
# link = link.replace('http://10.198.6.181:31900/cscec81-cloud', 'http://221.214.64.195:31229/media') # 测试
# link = link.replace('http://10.198.6.173:30900/cscec81-cloud', 'http://app.cscec81.com/media') # 正式
# link = f"http://221.214.64.195:31229/media{link.split('cscec81-cloud')[1]}" # 测试环境
link = f"http://app.cscec81.com/media{link.split('cscec81-cloud')[1]}" # 正式环境
return link
def get_user_list(self):
"""
获取在职人员列表
"""
url = f"{self.service_url}/system/users/getUserList"
data = {
'companyId': self.project_id, # 单位id
'userName': '', # 人员名称
'isMainPosition': '', # 1 主职 0 主职+兼职
'postId': '', # 职务
'postName': '',
}
headers = {
'Authorization': self.token
}
# print('Api.get_user_list.url:', url)
response = requests.get(url, params=data, headers=headers)
# print('Api.get_user_list.result:', response.status_code)
if response.status_code > 300:
print('Api.get_user_list.result:', response.status_code)
print('Api.get_user_list.detail:', response.text)
return []
json_data = response.json()
if not json_data.get('content'):
return []
print(json_data.get('content')[0])
return json_data.get('content')
def push_face(self, send_at, face_file_link, face_name, cscec8_duty_name, cscec8_id, age, group, source):
"""
上传新增人脸（上传八局管理人员接口）
"""
# test_service_url = 'http://10.198.6.181:31229/api'
# url = f"{test_service_url}/worker/userInfo"
url = f"{self.service_url}/worker/userInfo"
data = [{
'groupId': group, # 分组ID 001 陌生人 002 劳务工人 003 临时用工 004 项目管理人员 005 VIP 006 黑名单 007 其他
'equipmentId': self.project_id, # 设备id
'userId': cscec8_id, # 用户唯一标识 人脸id
'userName': face_name, # 用户名 人脸名称
'age': age, # 年龄
'postInfoName': cscec8_duty_name, # 职务
'lastEnterTime': send_at, # 最后进入时间 '2021-08-10 08:00:00'
'faceInfo': face_file_link, # 文件链接地址
'sourceType': source, # 是否是八局人脸库信息 0 不是 1 是
}]
# print('Api.push_face:data:', data)
# methods.debug_log('Api.push_face', f"m-418 | data: {data}")
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
# print('Api.push_face.url:', url)
response = requests.post(url, json=data, headers=headers)
if response.status_code > 300:
print('Api.push_face.result:', response.status_code)
print('Api.push_face.detail:', response.text)
methods.debug_log('apis.u3_for_cscec', f"m-395: url -> {url}")
methods.debug_log('apis.u3_for_cscec', f"m-395: data -> {data}")
methods.debug_log('apis.u3_for_cscec', f"m-395: result -> {response.status_code}")
methods.debug_log('apis.u3_for_cscec', f"m-395: detail -> {response.text}")
return False
else:
# print('Api.push_face.result:', response.status_code)
# print('Api.push_face:json:', response.json())
return True
def push_work_info(self, send_at, face_file_link, face_name, prc_id, worker_contractor, worker_type_name, age,
find_at):
"""
添加劳务人员接口（上传农民工接口）
"""
# test_service_url = 'http://425ax87563.zicp.vip'
# url = f"{test_service_url}/worker/userInfo/workerInfo"
url = f"{self.service_url}/worker/userInfo/workerInfo"
data = [{
'equipmentId': self.project_id, # 设备id
'userId': prc_id, # 用户唯一标识 农民工身份证id
'userName': face_name, # 用户名 工人名称
'userCerd': prc_id, # 用户身份证号
'age': age, # 年龄
'postInfoName': '', # 职务
'workType': worker_type_name, # 工种（农民工）
'labourCompany': worker_contractor, # 劳务公司（农民工）
'enterState': '', # 进场状态
'enterTime': '', # 进场时间
'synTime': find_at, # 同步时间（同步云筑网时间）
'lastEnterTime': send_at, # 最后进入时间 '2021-08-10 08:00:00'
'faceInfo': face_file_link, # 文件链接地址 人脸信息（图片传八局服务器系统，入参只传服务器文件地址）
'labourSource': str(0), # 是否八局底库人员 0 不是 1 是
}]
# data = [
# {'equipmentId': '63729db2-ea9c-4096-b5d9-3ad0369903c8', 'userId': '511025199009011013', 'userName': '赵润',
# 'userCerd': '511025199009011013', 'age': '', 'postInfoName': '', 'workType': None, 'labourCompany': None,
# 'enterState': '', 'enterTime': '', 'synTime': '', 'lastEnterTime': '2022-05-27 10:10:56',
# 'faceInfo': 'http://221.214.64.195:31229/media/202205/27/20220527-150651-60-5508.jpg',
# 'labourSource': '0'}]
# print('Api.push_work_info:data:', data)
# methods.debug_log('Api.push_work_info', f"m-464 | data: {data}")
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
# print('Api.push_work_info.url:', url)
response = requests.post(url, json=data, headers=headers)
if response.status_code > 300:
# print('Api.push_work_info.result:', response.status_code)
# print('Api.push_work_info.detail:', response.text)
methods.debug_log('apis.u3_for_cscec', f"m-457: url -> {url}")
methods.debug_log('apis.u3_for_cscec', f"m-457: data -> {data}")
methods.debug_log('apis.u3_for_cscec', f"m-457: result -> {response.status_code}")
methods.debug_log('apis.u3_for_cscec', f"m-457: detail -> {response.text}")
return False
else:
# print('Api.push_work_info:json:', response.json())
return True
def pull_update_info(self, start_at):
"""
获取线上更新数据 todo 定时请求更新本地数据相关的数据
"""
# test_service_url = 'http://425ax87563.zicp.vip'
# url = f"{test_service_url}/worker/userInfo/findUpdateUserByTime"
url = f"{self.service_url}/worker/userInfo/findUpdateUserByTime"
# url = f"http://425ax87563.zicp.vip/worker/userInfo/findUpdateUserByTime?queryStartTime=2022-05-27 11:09:00"
# url += '?queryStartTime=2022-05-27 11:09:00'
url += f"?queryStartTime={start_at}"
data = {
# 'companyId': self.project_id, # 单位id
# 'queryStartTime': '2021-10-21 11:09:00',
# 'queryStartTime': '2022-05-27 11:09:00',
}
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
print('Api.pull_update_info.url:', url)
response = requests.get(url, params=data, headers=headers)
print('Api.pull_update_info.result:', response.status_code)
print('Api.pull_update_info:text:', response.json())
return []
# if response.status_code > 300:
# print('Api.pull_update_info:text:', response.text)
# return []
# json_data = response.json()
# if not json_data.get('content'):
# return []
# print(json_data.get('content')[0])
# return json_data.get('content')
def pull_alarm_list(self, create_at):
"""
获取告警列表
"""
# test_service_url = 'http://425ax87563.zicp.vip'
# url = f"{test_service_url}/worker/alarm/alarmPageList"
# url += f"?createTime={create_at}"
url = f"{self.service_url}/worker/alarm/alarmPageList"
url += f"?createTime={create_at}"
params = {
# 'createTime': '2021-08-10 08:00:00'
}
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
print('Api.pull_alarm_list.url:', url)
response = requests.get(url, params=params, headers=headers)
if response.status_code > 300:
print('Api.pull_alarm_list.result:', response.status_code)
print('Api.pull_alarm_list.detail:', response.text)
return {}
else:
print('Api.pull_alarm_list.result:', response.status_code)
return response.json()
def pull_alarm_group_list(self, create_at):
"""
获取报警分组列表
return [
{
ALARM_GROUP_STRATEGY_IDS: 1 现场屏显报警 2 现场语音报警 3 微信报警
}
]
"""
# test_service_url = 'http://425ax87563.zicp.vip'
# url = f"{test_service_url}/worker/alarm/alarmGroupList"
url = f"{self.service_url}/worker/alarm/alarmGroupList"
params = {
'projectId': self.project_id,
'createTime': create_at,
}
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
print('Api.pull_alarm_group_list.url:', url)
response = requests.get(url, params=params, headers=headers)
if response.status_code > 300:
print('Api.pull_alarm_group_list.result:', response.status_code)
print('Api.pull_alarm_group_list.detail:', response.text)
return []
else:
print('Api.pull_alarm_group_list.result:', response.status_code)
return response.json()
def push_message(self):
"""
推送微信消息
{'ALARM_TYPE': 1, 'ALARM_USER_ID': 'f6bae517-0a28-429c-b65b-b16548ab2ca4', 'ALARM_GROUP_NAME': '屏显报警',
'id': '17ee224382634361bc5b9ef41ce09eb1', 'ALARM_STRATEGY_GROUP_ID': '5e58ed8828c84c779b3cddb0257d6f5f',
'ALARM_PROJECT_ID': '63729db2-ea9c-4096-b5d9-3ad0369903c8', 'ALARM_NAME': '刘美琪'}
{'ALARM_TYPE': 1, 'ALARM_USER_ID': '7a574580-6af1-40af-a3c8-1c0d5aee567b', 'ALARM_GROUP_NAME': '微信报警',
'id': '04bb6b4628954b2282b15a3cb2e74ab8', 'ALARM_STRATEGY_GROUP_ID': '0f695744b43447b2b4cbad8dc78389f4',
'ALARM_PROJECT_ID': '63729db2-ea9c-4096-b5d9-3ad0369903c8', 'ALARM_NAME': '吴非'}
微信推送结果查询接口 worker/alarm/getWxPushState
状态 00 未发送 01 发送成功 02 发送失败 03 发送异常
"""
# test_service_url = 'http://425ax87563.zicp.vip'
# url = f"{test_service_url}/worker/alarm/getWxAlarmUser"
url = f"{self.service_url}/worker/alarm/getWxAlarmUser"
data = {
'projectId': self.project_id, # 项目id
# 'groupId': '5e58ed8828c84c779b3cddb0257d6f5f', # 告警组id ALARM_STRATEGY_GROUP_ID
'groupId': '0f695744b43447b2b4cbad8dc78389f4', # 告警组id ALARM_STRATEGY_GROUP_ID
# 'userId': 'f6bae517-0a28-429c-b65b-b16548ab2ca4', # 告警用户id ALARM_USER_ID
'userId': '7a574580-6af1-40af-a3c8-1c0d5aee567b', # 告警用户id ALARM_USER_ID
}
print(f"Api.push_message:data: {data}")
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
print('Api.push_message.url:', url)
response = requests.post(url, json=data, headers=headers)
if response.status_code > 300:
print('Api.push_message.result:', response.status_code)
print('Api.push_message.detail:', response.text)
return {}
else:
print('Api.push_message.result:', response.status_code)
return response.json()
def check_message_state(self):
"""
微信推送结果查询接口 worker/alarm/getWxPushState
状态 00 未发送 01 发送成功 02 发送失败 03 发送异常
"""
# test_service_url = 'http://425ax87563.zicp.vip'
# url = f"{test_service_url}/worker/alarm/getWxPushState"
url = f"{self.service_url}/worker/alarm/getWxPushState"
# url += f"?wxPushId=24290bde-09d7-444b-b2ea-039213f8b628"
url += f"?wxPushId=77c36ad2-ad35-464b-bf52-6d16fd9c74e0"
data = {
# 'wxPushId': mid,
# 'wxPushId': '24290bde-09d7-444b-b2ea-039213f8b628',
}
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
print('Api.check_message_state.url:', url)
# response = requests.get(url, json=data, headers=headers)
response = requests.get(url, headers=headers)
if response.status_code > 300:
print('Api.check_message_state.result:', response.status_code)
print('Api.check_message_state.detail:', response.text)
return {}
else:
print('Api.check_message_state.result:', response.status_code)
# return response.json()
return response.text
def push_state(self):
"""
每分钟发送心跳请求接口
"""
# test_service_url = 'http://425ax87563.zicp.vip'
# url = f"{test_service_url}/worker/equipmentMaintain/updateEquStateTime"
url = f"{self.service_url}/worker/equipmentMaintain/updateEquStateTime"
data = {
'projectId': self.project_id,
}
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
# print('Api.push_state.url:', url)
response = requests.post(url, json=data, headers=headers)
if response.status_code > 300:
print('Api.push_state.result:', response.status_code)
print('Api.push_state.detail:', response.text)
return {}
else:
# print('Api.push_state.result:', response.status_code)
return response.json()
def push_face_log(self, user_id, find_at, face_link=''):
"""
上传识别记录
methods.ts_to_string(item.get('create_at'), '%Y-%m-%d %H:%M:%S')
"""
# test_service_url = 'http://10.198.6.181:31229/api'
# url = f"{test_service_url}/worker/ledger/addLedger"
url = f"{self.service_url}/worker/ledger/addLedger"
data = [{
'projectId': self.project_id, # 项目id
'equipmentInfo': self.project_id, # 设备id
'equipmentRecognitionTime': find_at, # 识别时间 '2021-10-21 11:09:00'
'userId': user_id, # 人脸标识 八局人员使用uuid 劳务人员使用身份证号
'faceInfo': face_link, # 文件链接地址 人脸信息（图片传八局服务器系统，入参只传服务器文件地址）
}]
headers = {
'Authorization': self.token,
# 'Authorization': self.token_test,
}
# methods.debug_log('apis.u3_for_cscec', f"m-651: project_id -> {self.project_id}")
print('Api.push_face_log.url:', url)
# print('Api.push_face_log:data:', data)
response = requests.post(url, json=data, headers=headers)
if response.status_code > 300:
# print('Api.push_face_log.result:', response.status_code)
# print('Api.push_face_log.detail:', response.text)
methods.debug_log('apis.u3_for_cscec', f"m-678: url -> {url}")
methods.debug_log('apis.u3_for_cscec', f"m-678: data -> {data}")
methods.debug_log('apis.u3_for_cscec', f"m-678: result -> {response.status_code}")
methods.debug_log('apis.u3_for_cscec', f"m-678: detail -> {response.text}")
return False
else:
print('Api.push_face_log.result:', response.status_code)
print('Api.push_face_log.detail:', response.text)
# methods.debug_log('apis.u3_for_cscec', f"m-668: result -> {response.status_code}")
# methods.debug_log('apis.u3_for_cscec', f"m-668: detail -> {response.text}")
# methods.debug_log('apis.u3_for_cscec', f"m-668: detail -> {response.json()}")
return True
if __name__ == '__main__':
# --- init ---
# api = Api(username='500A4475', password='cscec81#')
# api = Api(username='500A7062', password='198797#cjhxbin') # 侯经理 正式环境
api = Api(project_id='63729db2-ea9c-4096-b5d9-3ad0369903c8',
project_name='山东省大数据产业基地建设项目施工总承包') # 正式环境账号
# api = Api(project_id='a8defc53-76b3-4f4e-931d-8dd35ba310c7',
# project_name='济南奥体东 13 及 17-2 号地块开发项目工程总 承包（EPC）') # 正式环境账号
# api = Api() # 正式环境账号
# --- test ---
# out = api.upload_file('aabbccdd-aabbccss-ssggaaff-sseeqqhh', r"D:\3.jpg")
# print(out)
# --- test ---
# api.push_state()
# api.push_face(
# send_at='2022-07-18 08:09:00',
# face_file_link='http://app.cscec81.com/media/202207/18/20220718-105339-394-2140.jpg',
# face_name='张三001',
# cscec8_duty_name='司机',
# cscec8_id='aabbccdd-aabbccss-ssggaaff-sseeqqhh',
# age=60,
# group='004',
# source=1)
# api.push_face_log('aabbccdd-aabbccss-ssggaaff-sseeqqhh', '2022-06-02 08:09:01',
# 'http://app.cscec81.com/media/202207/18/20220718-111202-730-7523.jpg')
# --- test ---
# _dict = {}
# items = api.get_worker_list()
# print(items[0])
# print(len(items))
# for item in items:
# _dict[item.get('idCardNumber')] = 1
# print(len(_dict.keys()))
# --- check 八局人员 ---
# items = api.get_cscec8_user_list()
# for item in items:
# if item.get('userName') in ['孙会沅']:
# print(item)
# --- check avatarAddress ---
# out = api.get_worker_list()
# for one in out:
# if one.get('workerName') in ['鲁成', '韩继涛', '丁保树', '金义莲']:
# print(one)
# # if one.get('subContractorName'):
# # print(one.get('avatarAddress'))
# # print(one.get('flag'), type(one.get('flag')))
# # image = api.get_face_image_by_path_v2(one.get('avatarAddress'))
# # print(type(image), len(image))
# --- test ---
# uuid = '62a166f324bfcf82f7066b65'
# face_image_path = '/home/server/resources/vms-files/2022-0609-112017-618668-raw.jpg'
# face_image_path = '/home/server/resources/vms-files/2022-0601-172536-193936.jpg'
# out = api.upload_file(uuid, face_image_path)
# print(out)
# --- test ---
# o = api.pull_alarm_list('2021-06-01 08:00:00')
# print(o)
# o = api.pull_alarm_group_list('2021-06-01 08:00:00')
# print(o)
# o = api.push_message()
# print(o)
# o = api.check_message_state()
# print(o)
# --- test ---
# o = api.pull_update_info('2022-05-27 11:09:00')
# o = api.pull_alarm_list('2022-05-27 11:09:00')
# o = api.pull_alarm_group_list()
# print(o)
# --- test ---
# o = api.push_work_info(1, 2, 3, 4, 1, 1)
# print(o)
# --- test ---
# u = 'https://hcm.cscec81.com/img?type=photo&size=120&index=ae2f8424-c63f-11ea-b0a9-ea6f5cf6388d&rnd=1653618903725'
# u = 'https://hcm.cscec81.com/img?type=photo&size=120&index=6605f16a-095a-11ec-99ad-da7e2f6980d1&rnd=1653620550071'
# o = api.get_face_image_by_url(u)
# print(o)
# --- test ---
# out = api.get_cscec8_user_list()
# print(len(out))
# for i in out:
# if i.get('userName') == '杜兴昌':
# print(i)
# --- test ---
# out = api.push_face('6166f0ab95a68293064d97c4', '2021-04-14 14:43:55',
# 'http://221.214.64.195:31229/media/202110/21/20211021-120733-570-3862.jpg', 1, '张001')
# out = api.push_work_info('6166f0ab95a68293064d97c4', '2021-10-13 14:43:55',
# 'http://221.214.64.195:31229/media/202110/21/20211021-120733-570-3862.jpg')
# print(out)
# --- test ---
# out = api.pull_alarm_group_list()
# print(out)
# --- test ---
# alarm_group_id = '1dfcd494affd4ee9aeb533c2d12558e7'
# out = api.push_message(alarm_group_id)
# --- test ---
# out = api.pull_alarm_group_list()
# out = api.get_user_list()
# print(out)
# --- test ---
# out = api.get_user_list()
# print(out)
# --- test ---
# def write_bytes(path, data=b''):
# with open(path, 'wb') as f:
# f.write(data)
# outputs = api.get_worker_list_by_project_name('济南市中心医院（东院区）项目工程总承包（EPC）')
# count = 0
# for i in outputs:
# if not i.get('headImagePath'):
# continue
# url = f"https://lwres.yzw.cn/{i.get('headImagePath')}"
# r = requests.get(url, headers={'content-type': 'application/json'})
# count += 1
# name = str(count).zfill(6)
# write_bytes(f"D:\\test\\111\\{name}.jpg", r.content)
# --- test ---
# def write_bytes(path, data=b''):
# with open(path, 'wb') as f:
# f.write(data)
# _url = f"https://lwres.yzw.cn/worker-avatar/Original/2021/0303/0f735c37-118f-4cd8-a0a5-753e0e2fc7d7.jpg"
# r = requests.get(_url, headers={'content-type': 'application/json'})
# write_bytes(f"D:\\test\\111\\{1}.jpg", r.content)
# --- test ---
# d1 = {}
# outputs = api.get_worker_list_by_project_name('济南市中心医院（东院区）项目工程总承包（EPC）')
# for i in outputs:
# if i.get('workTypeName') not in d1:
# d1[i.get('workTypeName')] = 0
# d1[i.get('workTypeName')] += 1
# print(d1)
# --- test ---
# api.get_image('worker-avatar/Original/2018/0612/f3230a72-57ab-4aaf-a741-61d8c078fe89.jpg')
# out = api.get_worker_list_by_project_name('济南市中心医院（东院区）项目工程总承包（EPC）')
# print(len(out))