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