服务运行成功！以下是可用的API接口：
+ +
+# 获取状态
+curl http://localhost:8080/api/status
+
+# 用户登录
+curl -X POST -H "Content-Type: application/json" \\
+ -d '{"username":"admin","password":"123456"}' \\
+ http://localhost:8080/api/login
+
+# 获取用户信息
+curl http://localhost:8080/api/user?id=1
+
+
+
+ """
+
+ def do_OPTIONS(self):
+ """处理预检请求（CORS）"""
+ self.send_response(200)
+ self.send_header("Access-Control-Allow-Origin", "*")
+ self.send_header(
+ "Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS"
+ )
+ self.send_header("Access-Control-Allow-Headers", "Content-Type")
+ self.end_headers()
+
+ def log_message(self, format, *args):
+ """自定义日志格式"""
+ logger.info(f"{self.address_string()} - {format % args}")
+
+
+def run_server(host="localhost", port=8080):
+ """启动HTTP服务器"""
+ server_address = (host, port)
+ httpd = HTTPServer(server_address, CustomHTTPRequestHandler)
+
+ print(f"🌟 HTTP服务器启动成功!")
+ print(f"📍 地址: http://{host}:{port}")
+ print(f"🔗 在浏览器中访问: http://{host}:{port}")
+ print(f"⏹️ 按 Ctrl+C 停止服务器\n")
+
+ try:
+ httpd.serve_forever()
+ except KeyboardInterrupt:
+ print("\n🛑 服务器已停止")
+ httpd.server_close()
+
+
+if __name__ == "__main__":
+ # 启动服务器
+ run_server(host="10.0.0.202", port=9909)
diff --git a/ruoyi-fastapi-backend/config/env.py b/ruoyi-fastapi-backend/config/env.py
index e66b0d9..13df5e2 100644
--- a/ruoyi-fastapi-backend/config/env.py
+++ b/ruoyi-fastapi-backend/config/env.py
@@ -154,8 +154,9 @@ class HaiKangSettings:
"""
海康平台配置
"""
- HAIKANG_URL = ''
- HAIKANG_PORT = 443
+ HAIKANG_URL = 'http://10.0.0.202'
+ # HAIKANG_PORT = 443
+ HAIKANG_PORT = 9909
HAIKANG_AK = ''
HAIKANG_SK = ''
HAIKANG_ACCESS_TOKEN_URL = '/api/v1/oauth/token'
diff --git a/ruoyi-fastapi-backend/utils/haikang_util.py b/ruoyi-fastapi-backend/utils/haikang_util.py
index 1ec0702..fca8f12 100644
--- a/ruoyi-fastapi-backend/utils/haikang_util.py
+++ b/ruoyi-fastapi-backend/utils/haikang_util.py
@@ -1,16 +1,16 @@
import hashlib
import hmac
-import json
+import json
import base64
import requests
-import httpx
+import httpx
from fastapi import HTTPException
from datetime import datetime, timezone
from email.utils import format_datetime
from urllib.parse import urlparse
from dateutil.relativedelta import relativedelta
-from config.env import HaiKangConfig
+from config.env import HaiKangConfig
from utils.log_util import logger
@@ -23,7 +23,7 @@ class HaikangUtil:
# self.HAIKANG_URL = HaiKangConfig.HAIKANG_URL
# self.HAIKANG_PORT = HaiKangConfig.HAIKANG_PORT
# self.HAIKANG_AK = HaiKangConfig.HAIKANG_AK
- # self.HAIKANG_SK = HaiKangConfig.HAIKANG_SK
+ # self.HAIKANG_SK = HaiKangConfig.HAIKANG_SK
# 通用请求头
@classmethod
def build_signed_headers(cls, method, url, body, app_key, app_secret):
@@ -40,28 +40,23 @@ class HaikangUtil:
# 1. 基本 headers
accept = "*/*"
content_type = "application/json"
-
+
# 2. 计算 MD5（可选，如果 body 存在）
content_md5 = ""
if body:
-
+
# .digest()返回原始二进制md5
- md5_digest = hashlib.md5(body.encode('utf-8')).digest()
- content_md5 = base64.b64encode(md5_digest).decode('utf-8')
+ md5_digest = hashlib.md5(body.encode("utf-8")).digest()
+ content_md5 = base64.b64encode(md5_digest).decode("utf-8")
# 3. 生成 Date header（HTTP 规范格式）
now = datetime.now(timezone.utc)
date = format_datetime(now, usegmt=True)
# 4. 构造 httpHeaders 部分
- http_headers_str = "\n".join([
- method.upper(),
- accept,
- content_md5,
- content_type,
- date,
- ""
- ])
+ http_headers_str = "\n".join(
+ [method.upper(), accept, content_md5, content_type, date, ""]
+ )
# 5. 自定义 headers 部分
custom_headers_str = f"x-ca-key:{app_key}\n"
@@ -76,10 +71,10 @@ class HaikangUtil:
string_to_sign = http_headers_str + custom_headers_str + path_and_query
# 8. 使用 HmacSHA256 + Base64 签名
- h = hmac.new(app_secret.encode('utf-8'),
- string_to_sign.encode('utf-8'),
- hashlib.sha256)
- signature = base64.b64encode(h.digest()).decode('utf-8')
+ h = hmac.new(
+ app_secret.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha256
+ )
+ signature = base64.b64encode(h.digest()).decode("utf-8")
# 9. 返回完整 headers
headers = {
@@ -89,20 +84,23 @@ class HaikangUtil:
"Date": date,
"X-Ca-Key": app_key,
"X-Ca-Signature": signature,
- "X-Ca-Signature-Headers": "x-ca-key"
+ "X-Ca-Signature-Headers": "x-ca-key",
}
return headers
-
+
# 发送请求
@classmethod
- async def send_request(method: str, url: str, headers: dict, body: str | None):
+ async def send_request(cls, method: str, url: str, headers: dict, body: str | None):
async with httpx.AsyncClient() as client:
- response = await client.request(method, url, headers=headers, content=body, timeout=60)
+ response = await client.request(
+ method, url, headers=headers, content=body, timeout=60
+ )
response.raise_for_status()
# if response.status_code >= 400:
# logger.error(f"发送请求失败: {url} , {response.status_code} {response.text}")
# raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
+
# 获取access_token
@classmethod
async def get_access_token(cls):
@@ -111,11 +109,13 @@ class HaikangUtil:
Returns:
_type_: json
"""
-
+
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_ACCESS_TOKEN_URL}"
- headers = cls.build_signed_headers('POST', url, None, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
- back = await cls.send_request('POST', url, headers, None)
-
+ headers = cls.build_signed_headers(
+ "POST", url, None, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+ back = await cls.send_request("POST", url, headers, None)
+
if back["code"] == "0":
logger.info("获取access_token成功")
return [True, back]
@@ -123,8 +123,7 @@ class HaikangUtil:
else:
logger.error("获取access_token失败", back["code"], back["msg"])
return [False, back]
-
-
+
# 查询门禁点列表v2
@classmethod
async def get_door_list_v2(cls, pageNo: int = 1, pageSize: int = 10):
@@ -137,51 +136,59 @@ class HaikangUtil:
Returns:
_type_: _description_
"""
- url = f'{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_SEARCH}'
- body_dict ={
+ url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_SEARCH}"
+ body_dict = {
"pageNo": max(pageNo, 1),
"pageSize": min(pageSize, 999),
}
body_json = json.dumps(body_dict, separators=(",", ":"))
-
- headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, body_json)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, body_json)
+
if back["code"] == "0":
logger.info("查询门禁点列表成功")
return [True, back["data"]]
else:
logger.error("查询门禁点列表失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
+
# 查询门禁状态
@classmethod
- async def get_door_status(cls, door_index_codes:list):
+ async def get_door_status(cls, door_index_codes: list):
"""查询门禁状态
Args:
door_index_codes (list): 门禁点唯一标识
"""
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_STATES_URL}"
-
+
body_dict = {"doorIndexCodes": door_index_codes}
body_json = json.dumps(body_dict, separators=(",", ":"))
-
- headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, body_json)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, body_json)
+
if back["code"] == "0":
logger.info("查询门禁状态成功")
- return [True, back["data"]["authDoorList"], back["data"]["noAuthDoorIndexCodeList"]]
+ return [
+ True,
+ back["data"]["authDoorList"],
+ back["data"]["noAuthDoorIndexCodeList"],
+ ]
else:
logger.error("查询门禁状态失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
-
-
+
# 门禁点反控
@classmethod
- async def door_do_control(cls, door_index_code: list, control_type: int):
+ async def door_do_control(cls, door_index_code: list, control_type: int):
"""门禁点反控
Args:
@@ -191,82 +198,101 @@ class HaikangUtil:
if control_type not in [0, 1, 2]:
logger.error("control_type参数错误 ", control_type)
return [False, 400, "control_type参数错误"]
-
+
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_DOCONTROL_URL}"
- body_dict = {
- "doorIndexCodes": door_index_code,
- "controlType": control_type
- }
+ body_dict = {"doorIndexCodes": door_index_code, "controlType": control_type}
body_json = json.dumps(body_dict, separators=(",", ":"))
-
- headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, body_json)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, body_json)
+
+ """
+ {
+ "code": "0",
+ "msg": "success",
+ "data": [
+ {
+ "doorIndexCode": "2c95c028a809448f962a969e3ab34f",
+ "controlResultCode": 0, # 0表示反控成功, 其他表示失败, 门禁控制是否成功主要看这个值
+ "controlResultDesc": "success",
+ }
+ ],
+ },
+ """
+
if back["code"] == "0":
- logger.info(f'执行门禁控制成功 door_index_code:{door_index_code} control_type:{control_type} ')
+ logger.info(
+ f"执行门禁控制接口成功 door_index_code:{door_index_code} control_type:{control_type} "
+ )
return [True, back["data"]]
else:
- logger.error(f'执行门禁控制失败 ', back["code"], back["msg"])
+ logger.error(f"执行门禁控制失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
-
+
@classmethod
# 查询门禁点事件v2
async def query_door_events_v2(cls, door_index_code: str, **args):
- """ 查询门禁点事件v2
+ """查询门禁点事件v2
Args:
door_index_code (str): 门禁唯一标识
"""
-
+
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_ENVENTS_URL}"
-
+
# current_time = datetime.now(timezone.utc)
# three_months_ago = current_time - relativedelta(months=3)
-
+
# # ISO8601 时间格式
# current_time_iso = current_time.isoformat()
# three_months_ago_iso = three_months_ago.isoformat()
-
- body_dict ={
+
+ body_dict = {
"pageNo": max(args.get("pageNo", 1), 1),
"pageSize": min(args.get("pageSize", 10), 999),
"doorIndexCode": door_index_code,
-
# 排序字段
"sort": "eventTime",
# 倒序返回
"order": "desc",
}
-
+
if args.get("startTime"):
body_dict["startTime"] = args.get("startTime")
if args.get("endTime"):
body_dict["endTime"] = args.get("endTime")
if args.get("eventType"):
body_dict["eventType"] = args.get("eventType")
-
+
body_json = json.dumps(body_dict, separators=(",", ":"))
-
- headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, body_json)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, body_json)
+
if back["code"] == "0":
- logger.info(f'获取门禁事件成功')
+ logger.info(f"获取门禁事件成功")
return [True, back["data"]]
else:
- logger.error(f'获取门禁事件失败 ', back["code"], back["msg"] )
+ logger.error(f"获取门禁事件失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
-
+
# 人脸分组1vN检索
@classmethod
- async def face_group_1vN_search(cls,
- facePicBinaryData: str, # base64编码后的字符串
- pageNo: int=1, pageSize:int=20, searchNum: int =99,
- minSimilarity: int = 50,
- faceGroupIndexCodes: list[str] = None,
- ):
+ async def face_group_1vN_search(
+ cls,
+ facePicBinaryData: str, # base64编码后的字符串
+ pageNo: int = 1,
+ pageSize: int = 20,
+ searchNum: int = 99,
+ minSimilarity: int = 50,
+ faceGroupIndexCodes: list[str] = None,
+ ):
"""人脸分组1vN检索
Args:
@@ -280,8 +306,8 @@ class HaikangUtil:
_type_: _description_
"""
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_APPLICATION_ONETOMANY_URL}"
-
- body_dict ={
+
+ body_dict = {
"facePicBinaryData": facePicBinaryData,
"pageNo": pageNo,
"pageSize": pageSize,
@@ -289,24 +315,28 @@ class HaikangUtil:
"minSimilarity": minSimilarity,
"faceGroupIndexCodes": faceGroupIndexCodes,
}
-
+
body_json = json.dumps(body_dict, separators=(",", ":"))
-
- headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, body_json)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, body_json)
+
if back["code"] == "0":
- logger.info(f'获取人脸分组检索成功')
+ logger.info(f"获取人脸分组检索成功")
return [True, back["data"]]
else:
- logger.error(f'获取人脸分组检索失败 ', back["code"], back["msg"] )
+ logger.error(f"获取人脸分组检索失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
-
+
# 人脸评分
- async def face_picture_check(cls,
- facePicBinaryData: str , #
- ):
+ @classmethod
+ async def face_picture_check(
+ cls,
+ facePicBinaryData: str, #
+ ):
"""人脸评分
Args:
@@ -316,30 +346,27 @@ class HaikangUtil:
_type_: _description_
"""
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_PICTURE_CHECK_URL}"
-
- body_dict ={
- "facePicBinaryData": facePicBinaryData
- }
-
+
+ body_dict = {"facePicBinaryData": facePicBinaryData}
+
body_json = json.dumps(body_dict, separators=(",", ":"))
-
- headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, body_json)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, body_json)
+
if back["code"] == "0":
- logger.info(f'获取人脸评分成功')
+ logger.info(f"获取人脸评分成功")
return [True, back["data"]]
else:
- logger.error(f'获取人脸评分失败 ', back["code"], back["msg"] )
+ logger.error(f"获取人脸评分失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
-
+
# 按条件查询人员识别事件
@classmethod
- async def get_face_capture_list_event(cls,
- startTime,
- endTime
- ):
+ async def get_face_capture_list_event(cls, startTime, endTime):
"""按条件查询人员识别事件, 暂无实现.
Args:
@@ -347,26 +374,30 @@ class HaikangUtil:
endTime (_type_): _description_
"""
pass
-
- # 按条件查询人脸分组
+
+ # 按条件查询人脸分组
@classmethod
async def get_face_group(cls):
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_FACE_GROUP_URL}"
-
- headers = cls.build_signed_headers('POST', url, None, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, None)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, None, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, None)
+
if back["code"] == "0":
- logger.info(f'按条件查询人脸成功')
+ logger.info(f"按条件查询人脸成功")
return [True, back["data"]]
else:
- logger.error(f'按条件查询人脸失败 ', back["code"], back["msg"] )
+ logger.error(f"按条件查询人脸失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
-
+
# 查询访客预约记录, visitorStatus: 0 待审核, 1 正常, 2 迟到, 3 失效, 4 审核退回, 9 审核失效, 10 邀约中, 11 邀约失效
@classmethod
- async def query_visitor_record(cls, pageNo: int = 1, pageSize: int = 10, visitorStatus:int = 1):
+ async def query_visitor_record(
+ cls, pageNo: int = 1, pageSize: int = 10, visitorStatus: int = 1
+ ):
"""查询访客预约记录
Args:
@@ -378,21 +409,23 @@ class HaikangUtil:
_type_: _description_
"""
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_VISITOR_RECORD_SEARCH}"
- body_dict ={
+ body_dict = {
"pageNo": pageNo,
"pageSize": pageSize,
- "visitorStatus": visitorStatus
+ "visitorStatus": visitorStatus,
}
-
+
body_json = json.dumps(body_dict, separators=(",", ":"))
-
- headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
-
- back = await cls.send_request('POST', url, headers, body_json)
-
+
+ headers = cls.build_signed_headers(
+ "POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
+ )
+
+ back = await cls.send_request("POST", url, headers, body_json)
+
if back["code"] == "0":
- logger.info(f'查询访客预约记录成功')
+ logger.info(f"查询访客预约记录成功")
return [True, back["data"]]
else:
- logger.error(f'查询访客预约记录失败 ', back["code"], back["msg"] )
- return [False, back["code"], back["msg"]]
\ No newline at end of file
+ logger.error(f"查询访客预约记录失败 ", back["code"], back["msg"])
+ return [False, back["code"], back["msg"]]