diff --git a/main.py b/main.py index c5579f1..a63c413 100644 --- a/main.py +++ b/main.py @@ -14,6 +14,17 @@ from fpioa_manager import fm import utime from machine import UART import os +try: + import ubinascii +except: + ubinascii = None +try: + import ucryptolib as crypto_lib +except: + try: + import cryptolib as crypto_lib + except: + crypto_lib = None # --- 增加堆内存大小 --- # K210有8MB SRAM，默认堆较小，增加到512KB用于图像压缩 @@ -41,6 +52,125 @@ ESP_BAUD_RATES = (115200, 9600, 74880, 57600, 38400) ESP_UART_BAUD = 115200 ESP_BOOT_WAIT_MS = 6000 # 等待ESP8266上电启动和自动连网 LOCAL_FACE_PHOTO_DIR = "/sd/face_photos" # 本地人脸照片缓存目录 +AES_KEY = b"\x2b\x7e\x15\x16\x28\xae\xd2\xa6\xab\xf7\x15\x88\x09\xcf\x4f\x3c" +AES_IV = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f" + +def crypto_ready(): + return crypto_lib is not None and ubinascii is not None + +def pkcs7_pad(data): + pad_len = 16 - (len(data) % 16) + return data + bytes([pad_len]) * pad_len + +def pkcs7_unpad(data): + if not data: + return b"" + pad_len = data[-1] + if pad_len < 1 or pad_len > 16 or pad_len > len(data): + return data + for b in data[-pad_len:]: + if b != pad_len: + return data + return data[:-pad_len] + +def aes_cbc_crypt(data, encrypt=True): + if not crypto_ready(): + print("[CRYPTO] ucryptolib/ubinascii unavailable, refuse plaintext HTTP") + return None + try: + cipher = crypto_lib.aes(AES_KEY, 2, AES_IV) + return cipher.encrypt(data) if encrypt else cipher.decrypt(data) + except Exception as e: + print("[CRYPTO] AES failed:", e) + return None + +def encrypt_payload_base64(data): + if isinstance(data, str): + data = data.encode("utf-8") + encrypted = aes_cbc_crypt(pkcs7_pad(data), True) + if encrypted is None: + return None + encoded = ubinascii.b2a_base64(encrypted).strip() + try: + return encoded.decode("utf-8") + except: + return str(encoded, "utf-8") + +def decrypt_payload_base64(data): + if data is None: + return None + if isinstance(data, str): + data = data.strip().encode("utf-8") + try: + encrypted = ubinascii.a2b_base64(data) + except Exception as e: + print("[CRYPTO] Base64 decode failed:", e) + return None + decrypted = aes_cbc_crypt(encrypted, False) + if decrypted is None: + return None + try: + return pkcs7_unpad(decrypted).decode("utf-8") + except Exception as e: + print("[CRYPTO] UTF-8 decode failed:", e) + return None + +def extract_http_body(resp): + if not resp: + return "" + marker = "\r\n\r\n" + start = resp.find(marker) + if start == -1: + return "" + body = resp[start + len(marker):] + content_length = -1 + cl_pos = resp.find("Content-Length:") + if cl_pos != -1: + cl_pos += len("Content-Length:") + while cl_pos < len(resp) and resp[cl_pos] == " ": + cl_pos += 1 + cl_end = cl_pos + while cl_end < len(resp) and resp[cl_end].isdigit(): + cl_end += 1 + try: + content_length = int(resp[cl_pos:cl_end]) + except: + content_length = -1 + if content_length >= 0 and len(body) >= content_length: + return body[:content_length] + for marker in ("\r\nCLOSED", "CLOSED", "\r\nOK"): + pos = body.find(marker) + if pos != -1: + body = body[:pos] + return body.strip() + +def http_body_complete(resp): + if not resp or "\r\n\r\n" not in resp: + return False + body = resp.split("\r\n\r\n", 1)[1] + cl_pos = resp.find("Content-Length:") + if cl_pos == -1: + return "CLOSED" in resp + cl_pos += len("Content-Length:") + while cl_pos < len(resp) and resp[cl_pos] == " ": + cl_pos += 1 + cl_end = cl_pos + while cl_end < len(resp) and resp[cl_end].isdigit(): + cl_end += 1 + try: + return len(body) >= int(resp[cl_pos:cl_end]) + except: + return "CLOSED" in resp + +def decode_http_json_body(resp): + body = extract_http_body(resp) + if not body: + return None + if "X-Encrypted: AES128-CBC" in resp: + return decrypt_payload_base64(body) + if body.startswith("{") or body.startswith("["): + return body + return decrypt_payload_base64(body) # 摄像头安装方向：当前实物画面上下左右都反，水平镜像+垂直翻转等价于旋转180度。 CAMERA_HMIRROR = 1 @@ -519,6 +649,10 @@ def update_face_record(face_id, new_student_id): face_id = str(face_id) new_student_id = str(new_student_id) + if not local_face_id_exists(face_id): + print("[UPDATE] Ignore mapping for missing local face_id {} -> {}".format(face_id, new_student_id)) + return False + face_id_mapping[face_id] = new_student_id save_local_mapping() @@ -560,7 +694,7 @@ def esp_http_get(url_path, timeout=8000): utime.sleep_ms(200) # 构建HTTP GET请求 - http_request = "GET {} HTTP/1.1\r\nHost: {}:{}\r\nX-Api-Token: {}\r\nConnection: close\r\n\r\n".format( + http_request = "GET {} HTTP/1.1\r\nHost: {}:{}\r\nX-Api-Token: {}\r\nX-Encrypted-Response: true\r\nConnection: close\r\n\r\n".format( url_path, SERVER_IP, SERVER_PORT, API_TOKEN) # 使用AT+CIPSEND发送 @@ -623,8 +757,8 @@ def esp_http_get(url_path, timeout=8000): print("[HTTP_GET] 累积数据: {} bytes".format(len(resp))) except: pass - # 检查是否收到完整响应（有JSON且连接关闭） - if "}" in resp and "{" in resp: + # 加密响应是Base64文本，按Content-Length或连接关闭判断完整性。 + if http_body_complete(resp): if "CLOSED" in resp: print("[HTTP_GET] 连接已关闭，数据完整") break @@ -644,103 +778,27 @@ def esp_http_get(url_path, timeout=8000): if len(resp) > 0: print("[HTTP_GET] 响应片段: {}".format(resp[:300].replace('\r\n', '\\n'))) - # 提取JSON - json_start = resp.find('{"') - if json_start != -1: - json_end = resp.rfind('}') - if json_end != -1 and json_end > json_start: - result = resp[json_start:json_end+1] - print("[HTTP_GET] 提取JSON成功: {}".format(result[:100])) - return result + result = decode_http_json_body(resp) + if result: + print("[HTTP_GET] 解密JSON成功: {}".format(result[:100])) + return result - print("[HTTP_GET] 未找到JSON") + print("[HTTP_GET] 未找到可用JSON") return None -def sync_face_id_mapping(): - """从服务器同步人脸ID映射配置并更新本地记录""" - global face_id_mapping - - if not wifi_connected: - print("[SYNC] WiFi not connected, skip sync") - return False - - try: - print("[SYNC] ========== 开始同步映射配置 ==========") - print("[SYNC] 请求: GET /api/v1/faces/mappings") - response = esp_http_get("/api/v1/faces/mappings", 2000) - - if response is None: - print("[SYNC] 服务器无响应") - return False - - print("[SYNC] 收到响应: {}".format(response[:150] if len(response) > 150 else response)) - - # 简单解析JSON - mappings_start = response.find('"mappings":{') - if mappings_start == -1: - print("[SYNC] 响应中无mappings字段") - return False - - mappings_start += len('"mappings":{') - mappings_end = response.find('}', mappings_start) - if mappings_end == -1: - return False - - mappings_str = response[mappings_start:mappings_end] - print("[SYNC] 解析映射字符串: {}".format(mappings_str)) - - # 解析键值对 - server_mappings = {} - parts = mappings_str.split(',') - for part in parts: - part = part.strip() - if ':' in part: - kv = part.split(':') - if len(kv) == 2: - key = kv[0].strip().strip('"') - value = kv[1].strip().strip('"') - if key and value: - server_mappings[key] = value - - print("[SYNC] 服务器映射数量: {}".format(len(server_mappings))) - for k, v in server_mappings.items(): - print("[SYNC] 人脸ID: {} -> 学号: {}".format(k, v)) - - if len(server_mappings) == 0: - print("[SYNC] 服务器无映射配置") - face_id_mapping = {} - save_local_mapping() - return True - - # 对比本地映射 - print("[SYNC] 本地映射数量: {}".format(len(face_id_mapping))) - for k, v in face_id_mapping.items(): - print("[SYNC] 本地: {} -> {}".format(k, v)) - - # 只更新 face_id -> 学号映射，保持 recordftr.txt 里的原始 face_id 稳定。 - updated_count = 0 - for face_id, student_id in server_mappings.items(): - print("[SYNC] 检查是否需要更新: ID:{} -> {}".format(face_id, student_id)) - if update_face_record(face_id, student_id): - updated_count += 1 - - # 更新内存中的映射表 - face_id_mapping = server_mappings - save_local_mapping() - - print("[SYNC] ========== 同步完成，更新了 {} 条记录 ==========".format(updated_count)) - return True - - except Exception as e: - print("[SYNC] 同步出错:", e) - return False - def extract_face_id(face_name): face_name = str(face_name) if face_name.startswith("ID:"): return face_name.split(":", 1)[1] return face_name +def local_face_id_exists(face_id): + face_id = str(face_id) + for p in peoples: + if extract_face_id(p.name) == face_id: + return True + return False + def get_mapped_student_id(face_name): face_id = extract_face_id(face_name) return face_id_mapping.get(face_id, "") @@ -783,9 +841,7 @@ def delete_face_by_id(face_id): deleted_name = peoples[deleted_index].name peoples.pop(deleted_index) try: - with open(record_path, "w") as f: - for p in peoples: - f.write(p.name + '#' + str(p.feature) + "\n") + write_face_record() print("[DELETE] 删除人脸成功: {}".format(deleted_name)) # 同时删除本地照片，避免下次启动补传已删除的人脸照片 @@ -820,6 +876,9 @@ def esp_http_post_json(url_path, json_data, timeout=2000): """发送HTTP POST请求（优化版）""" if not wifi_connected: return None + encrypted_data = encrypt_payload_base64(json_data) + if encrypted_data is None: + return None esp_send_cmd("AT+CIPCLOSE", 300) utime.sleep_ms(50) @@ -830,8 +889,8 @@ def esp_http_post_json(url_path, json_data, timeout=2000): if "OK" not in resp and "CONNECT" not in resp: return None - http_request = "POST {} HTTP/1.1\r\nHost: {}:{}\r\nContent-Type: application/json\r\nX-Api-Token: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}".format( - url_path, SERVER_IP, SERVER_PORT, API_TOKEN, len(json_data), json_data) + http_request = "POST {} HTTP/1.1\r\nHost: {}:{}\r\nContent-Type: application/octet-stream\r\nX-Api-Token: {}\r\nX-Encrypted: AES128-CBC\r\nX-Encrypted-Response: true\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}".format( + url_path, SERVER_IP, SERVER_PORT, API_TOKEN, len(encrypted_data), encrypted_data) if not esp_send_data(http_request, 1500): esp_send_cmd("AT+CIPCLOSE", 300) @@ -844,164 +903,12 @@ def esp_http_post_json(url_path, json_data, timeout=2000): data = uart2.read() if data: resp += data.decode('utf-8', 'ignore') - if "success" in resp or "}" in resp: + if http_body_complete(resp): break utime.sleep_ms(10) esp_send_cmd("AT+CIPCLOSE", 300) - return resp - -def sync_pending_delete(): - """从服务器同步待删除的人脸列表""" - if not wifi_connected: - print("[DELETE_SYNC] WiFi未连接，跳过") - return False - - try: - print("[DELETE_SYNC] ========== 开始同步删除列表 ==========") - print("[DELETE_SYNC] 请求: GET /api/v1/faces/pending-delete") - response = esp_http_get("/api/v1/faces/pending-delete", 8000) - - if response is None: - print("[DELETE_SYNC] 服务器无响应") - return False - - print("[DELETE_SYNC] 收到响应: {}".format(response)) - - # 解析JSON - start = response.find('"pending_delete":[') - if start == -1: - print("[DELETE_SYNC] 响应中无pending_delete字段") - return True - - start += len('"pending_delete":[') - end = response.find(']', start) - if end == -1: - return False - - arr_str = response[start:end] - print("[DELETE_SYNC] 待删除列表字符串: '{}'".format(arr_str)) - - if not arr_str.strip(): - print("[DELETE_SYNC] 无待删除的人脸") - return True - - # 解析数组 - ids_to_delete = [] - parts = arr_str.split(',') - for part in parts: - part = part.strip().strip('"') - if part: - ids_to_delete.append(part) - - print("[DELETE_SYNC] 待删除人脸ID列表: {}".format(ids_to_delete)) - - if not ids_to_delete: - print("[DELETE_SYNC] 解析后无待删除项") - return True - - # 显示当前本地人脸列表 - print("[DELETE_SYNC] 当前本地人脸数量: {}".format(len(peoples))) - for i, p in enumerate(peoples): - print("[DELETE_SYNC] [{}] {}".format(i, p.name)) - - # 执行删除（每次最多处理一个，避免长时间阻塞） - face_id = ids_to_delete[0] - print("[DELETE_SYNC] 正在删除: {}".format(face_id)) - - delete_result = delete_face_by_id(face_id) - print("[DELETE_SYNC] 删除结果: {}".format("成功" if delete_result else "失败")) - - if delete_result: - # 只有本地特征真正删除成功后才确认，避免服务端过早清掉待删除任务 - print("[DELETE_SYNC] 发送确认: POST /api/v1/faces/delete-ack") - confirm_result = esp_http_post_confirm(face_id) - print("[DELETE_SYNC] 确认结果: {}".format("成功" if confirm_result else "失败")) - else: - print("[DELETE_SYNC] 本地未删除成功，暂不确认，等待下次同步重试") - - # 显示删除后的人脸列表 - print("[DELETE_SYNC] 删除后本地人脸数量: {}".format(len(peoples))) - for i, p in enumerate(peoples): - print("[DELETE_SYNC] [{}] {}".format(i, p.name)) - - print("[DELETE_SYNC] ========== 删除同步完成 ==========") - return True - - except Exception as e: - print("[DELETE_SYNC] 同步出错:", e) - return False - -def esp_http_post_confirm(face_id): - """发送删除确认请求""" - if not wifi_connected: - return False - - try: - esp_send_cmd("AT+CIPCLOSE", 500) - utime.sleep_ms(200) - esp_send_cmd("AT+CIPMUX=0", 500) - - # 建立TCP连接 - cmd = 'AT+CIPSTART="TCP","{}",{}'.format(SERVER_IP, SERVER_PORT) - resp = esp_send_cmd(cmd, 3000) - if "OK" not in resp and "CONNECT" not in resp: - print("[CONFIRM] TCP连接失败") - return False - - # 构建POST请求 - json_data = '{{"face_id":"{}"}}'.format(face_id) - http_request = "POST /api/v1/faces/delete-ack HTTP/1.1\r\n" - http_request += "Host: {}:{}\r\n".format(SERVER_IP, SERVER_PORT) - http_request += "Content-Type: application/json\r\n" - http_request += "X-Api-Token: {}\r\n".format(API_TOKEN) - http_request += "Content-Length: {}\r\n".format(len(json_data)) - http_request += "Connection: close\r\n\r\n" - http_request += json_data - - # 发送请求 - cmd = "AT+CIPSEND={}".format(len(http_request)) - uart2.write(cmd + "\r\n") - utime.sleep_ms(100) - - # 等待>提示符 - start = utime.ticks_ms() - prompt = "" - while utime.ticks_diff(utime.ticks_ms(), start) < 2000: - if uart2.any(): - prompt += uart2.read().decode('utf-8', 'ignore') - if ">" in prompt: - break - utime.sleep_ms(10) - - if ">" not in prompt: - esp_send_cmd("AT+CIPCLOSE", 300) - return False - - # 发送数据 - uart2.write(http_request) - - # 等待响应 - resp = "" - start = utime.ticks_ms() - while utime.ticks_diff(utime.ticks_ms(), start) < 5000: - if uart2.any(): - data = uart2.read() - if data: - resp += data.decode('utf-8', 'ignore') - if "success" in resp: - print("[CONFIRM] 服务器确认成功") - esp_send_cmd("AT+CIPCLOSE", 300) - return True - utime.sleep_ms(20) - - esp_send_cmd("AT+CIPCLOSE", 300) - print("[CONFIRM] 响应: {}".format(resp[:100] if resp else "(空)")) - return "success" in resp - - except Exception as e: - print("[CONFIRM] 错误:", e) - return False + return decode_http_json_body(resp) def json_get_int(text, key, default=0): marker = '"{}":'.format(key) @@ -1118,15 +1025,33 @@ def parse_mapping_object(obj_str): mappings[key] = value return mappings -def apply_face_mappings(server_mappings): +def apply_face_profiles(server_mappings): global face_id_mapping print("[SYNC_STATE] 服务器映射数量: {}".format(len(server_mappings))) updated_count = 0 + valid_mappings = {} + for face_id, student_id in face_id_mapping.items(): + if local_face_id_exists(face_id) and student_id: + valid_mappings[face_id] = student_id + for face_id, student_id in server_mappings.items(): + if not local_face_id_exists(face_id): + print("[SYNC_STATE] 跳过本地不存在的face_id映射: ID:{} -> {}".format(face_id, student_id)) + continue + if student_id == "__PHOTO_ONLY__": + print("[SYNC_STATE] 服务端仅有照片占位，保留本地映射: ID:{}".format(face_id)) + continue + if not student_id: + if face_id in valid_mappings: + del valid_mappings[face_id] + updated_count += 1 + print("[SYNC_STATE] 解除本地face_id绑定: ID:{}".format(face_id)) + continue print("[SYNC_STATE] 映射: ID:{} -> {}".format(face_id, student_id)) if update_face_record(face_id, student_id): updated_count += 1 - face_id_mapping = server_mappings + valid_mappings[face_id] = student_id + face_id_mapping = valid_mappings save_local_mapping() print("[SYNC_STATE] 映射应用完成，更新 {} 条".format(updated_count)) @@ -1152,6 +1077,28 @@ def sync_unified_state(): save_sync_version(server_version) return True + profiles_arr = extract_json_block(response, "face_profiles", '[', ']') + server_mappings = {} + server_face_ids = {} + pos = 0 + while profiles_arr and pos < len(profiles_arr): + obj_start = profiles_arr.find('{', pos) + if obj_start == -1: + break + obj_end = profiles_arr.find('}', obj_start) + if obj_end == -1: + break + obj = profiles_arr[obj_start:obj_end + 1] + face_id = json_get_string(obj, "face_id", "") + student_id = json_get_string(obj, "student_id", "") + status = json_get_string(obj, "status", "") + if face_id: + server_face_ids[face_id] = 1 + if status == "photo_only" and not student_id: + student_id = "__PHOTO_ONLY__" + server_mappings[face_id] = student_id + pos = obj_end + 1 + tasks_arr = extract_json_block(response, "delete_tasks", '[', ']') delete_tasks = parse_delete_tasks(tasks_arr) pending_arr = extract_json_block(response, "pending_delete", '[', ']') @@ -1165,6 +1112,18 @@ def sync_unified_state(): for task in delete_tasks: face_id = task["face_id"] + if face_id in server_face_ids: + print("[SYNC_STATE] 忽略旧删除任务，服务端仍有人脸记录: {}".format(face_id)) + deleted_ids.append(face_id) + if task["task_id"]: + deleted_task_ids.append(task["task_id"]) + continue + if not local_face_id_exists(face_id): + print("[SYNC_STATE] 删除任务face_id本地不存在，跳过但ACK: {}".format(face_id)) + deleted_ids.append(face_id) + if task["task_id"]: + deleted_task_ids.append(task["task_id"]) + continue if delete_face_by_id(face_id): deleted_ids.append(face_id) if task["task_id"]: @@ -1184,9 +1143,7 @@ def sync_unified_state(): delete_failed = True print("[SYNC_STATE] 删除ACK失败，暂不推进本地版本，等待下次重试") - mappings_obj = extract_json_block(response, "face_mappings", '{', '}') - server_mappings = parse_mapping_object(mappings_obj) - apply_face_mappings(server_mappings) + apply_face_profiles(server_mappings) if not delete_failed: save_sync_version(server_version) @@ -1301,45 +1258,205 @@ class People: peoples = [] record_path = "/sd/recordftr.txt" -def parse_feature_vector(text): - """只解析浮点数组文本，避免执行SD卡中的任意代码。""" - allowed = "0123456789+-eE.,[] \t\r\n" - for ch in text: - if ch not in allowed: - raise ValueError("invalid char in feature") +def hex_value(ch): + if '0' <= ch <= '9': + return ord(ch) - ord('0') + if 'a' <= ch <= 'f': + return ord(ch) - ord('a') + 10 + if 'A' <= ch <= 'F': + return ord(ch) - ord('A') + 10 + return -1 - text = text.strip() - if not text.startswith("[") or not text.endswith("]"): - raise ValueError("feature must be list") +def parse_bytes_literal(text): + """安全解析 bytearray(b"...") / b"..." 文本，不使用 eval。""" + pos1 = text.find("b'") + pos2 = text.find('b"') + if pos1 < 0 and pos2 < 0: + raise ValueError("bytes literal not found") + if pos1 >= 0 and (pos2 < 0 or pos1 < pos2): + pos = pos1 + quote = "'" + else: + pos = pos2 + quote = '"' - inner = text[1:-1].strip() + i = pos + 2 + values = [] + while i < len(text): + ch = text[i] + if ch == quote: + if len(values) < 64: + raise ValueError("feature too short") + return bytearray(values) + + if ch != "\\": + values.append(ord(ch)) + i += 1 + else: + i += 1 + if i >= len(text): + raise ValueError("bad escape") + esc = text[i] + if esc == "x": + if i + 2 >= len(text): + raise ValueError("bad hex escape") + hi = hex_value(text[i + 1]) + lo = hex_value(text[i + 2]) + if hi < 0 or lo < 0: + raise ValueError("bad hex escape") + values.append(hi * 16 + lo) + i += 3 + elif esc == "n": + values.append(10) + i += 1 + elif esc == "r": + values.append(13) + i += 1 + elif esc == "t": + values.append(9) + i += 1 + elif esc == "b": + values.append(8) + i += 1 + elif esc == "f": + values.append(12) + i += 1 + elif esc == "v": + values.append(11) + i += 1 + elif esc == "a": + values.append(7) + i += 1 + elif esc == "\\": + values.append(92) + i += 1 + elif esc == quote: + values.append(ord(quote)) + i += 1 + elif '0' <= esc <= '7': + val = ord(esc) - ord('0') + i += 1 + count = 1 + while count < 3 and i < len(text) and '0' <= text[i] <= '7': + val = val * 8 + ord(text[i]) - ord('0') + i += 1 + count += 1 + values.append(val & 0xff) + else: + values.append(ord(esc)) + i += 1 + + if len(values) > 4096: + raise ValueError("feature too long") + + raise ValueError("unterminated bytes literal") + +def parse_number_list(text): + start = text.find("[") + end = text.rfind("]") + if start < 0 or end <= start: + raise ValueError("feature must be bytearray/list") + + inner = text[start + 1:end].strip() if not inner: raise ValueError("empty feature") values = [] + has_float = False + all_byte = True for token in inner.split(","): token = token.strip() if not token: raise ValueError("empty number") - values.append(float(token)) - if len(values) > 512: + if "." in token or "e" in token or "E" in token: + value = float(token) + has_float = True + all_byte = False + else: + value = int(token) + if value < 0 or value > 255: + all_byte = False + values.append(value) + if len(values) > 4096: raise ValueError("feature too long") if len(values) < 64: raise ValueError("feature too short") + if all_byte and not has_float: + return bytearray(values) return values +def parse_feature_vector(text): + text = text.strip() + if text.startswith("bytearray(") or text.startswith("b'") or text.startswith('b"'): + return parse_bytes_literal(text) + return parse_number_list(text) + +HEX_CHARS = "0123456789abcdef" + +def byte_to_hex(v): + return HEX_CHARS[(v >> 4) & 15] + HEX_CHARS[v & 15] + +def escape_byte_values(values): + out = "" + for v in values: + v = int(v) & 0xff + if v == 92: + out += "\\\\" + elif v == 34: + out += '\\"' + elif v == 10: + out += "\\n" + elif v == 13: + out += "\\r" + elif v == 9: + out += "\\t" + elif 32 <= v <= 126: + out += chr(v) + else: + out += "\\x" + byte_to_hex(v) + return out + +def feature_to_record_text(feature): + type_text = str(type(feature)) + if type_text.find("bytearray") >= 0 or type_text.find("bytes") >= 0: + values = [] + for v in feature: + values.append(int(v) & 0xff) + return 'bytearray(b"' + escape_byte_values(values) + '")' + return str(feature) + +def write_face_record(): + with open(record_path, "w") as f: + for p in peoples: + f.write(str(p.name) + '#' + feature_to_record_text(p.feature) + "\n") + def load_faces(): global peoples; peoples = [] + total_count = 0 + bad_count = 0 + print("[FACE_DB] Loading {}".format(record_path)) try: with open(record_path, "r") as f: for line in f: + line = line.strip() + if not line: + continue + total_count += 1 # 使用 maxsplit=1 参数，只按第一个 '#' 分割，防止特征数据中的'#'导致错误 - parts = line.strip().split('#', 1) + parts = line.split('#', 1) if len(parts) == 2 and parts[1]: - try: peoples.append(People(parts[0], parse_feature_vector(parts[1]))) - except Exception as e_eval: print("Feature parse error on line for {}: {}".format(parts[0], e_eval)) - print("成功加载 %d 条人脸数据" % len(peoples)) + try: + peoples.append(People(parts[0], parse_feature_vector(parts[1]))) + except Exception as e_eval: + bad_count += 1 + print("Feature parse error on line {} for {}: {}".format(total_count, parts[0], e_eval)) + else: + bad_count += 1 + print("[FACE_DB] Bad line {}: missing feature".format(total_count)) + if total_count == 0: + print("[FACE_DB] recordftr.txt is empty") + print("成功加载 %d 条人脸数据 (总行:%d, 失败:%d)" % (len(peoples), total_count, bad_count)) except Exception as e: print("读取人脸数据库失败:", e) def collect_used_face_ids(): @@ -1388,7 +1505,8 @@ def add_face(feature, face_img=None): peoples.pop() return False - with open(record_path, "a") as f: f.write(new_name + '#' + str(feature) + "\n") + with open(record_path, "a") as f: + f.write(new_name + '#' + feature_to_record_text(feature) + "\n") print("添加成功:", new_name) return True @@ -1402,8 +1520,7 @@ def delete_face_by_index(idx): deleted_name = peoples[idx].name peoples.pop(idx) try: - with open(record_path, "w") as f: - for p in peoples: f.write(p.name + '#' + str(p.feature) + "\n") + write_face_record() if deleted_name.startswith("ID:"): delete_local_face_photo(deleted_name.split(":", 1)[1]) else: @@ -1612,21 +1729,12 @@ while True: face_id = extract_face_id(best_face['state']) mapped_student_id = get_mapped_student_id(best_face['state']) if not mapped_student_id: - print("[UART SEND] Face {} has no student mapping, send UNKNOWN".format(face_id)) - msg_to_send = "UNKNOWN\r\n" - last_saved_id = "" - if msg_to_send != last_sent_data: - uart.write(msg_to_send) - print("[UART SEND] -> {}".format(msg_to_send.strip())) - last_sent_data = msg_to_send - last_send_time = time_now - lcd.display(img) - continue - student_id = mapped_student_id - student_name = mapped_student_id + print("[UART SEND] Face {} has no student mapping, fallback to face_id".format(face_id)) + student_id = mapped_student_id if mapped_student_id else face_id + student_name = mapped_student_id if mapped_student_id else "FACE_{}".format(face_id) confidence = best_face['score'] / 100.0 - # 构造STM32期望的格式：ID:<学号>,NAME:<学号>,CONF:0.95 + # 构造STM32期望的格式；未绑定学号时默认使用K210原始face_id。 msg_to_send = "ID:{},NAME:{},CONF:{:.2f}\r\n".format( student_id, student_name,