import face_recognition import cv2 import os import time import threading import queue import base64 import zipfile import json import csv from datetime import datetime from flask import Flask, request, render_template_string, Response, jsonify, send_from_directory, abort from flask_socketio import SocketIO, emit app = Flask(__name__) socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading') # === КОНФИГУРАЦИЯ === CONFIG_FILE = "config.json" DEFAULT_CONFIG = { "camera_ip": "192.168.1.100", "camera_stream": 1, # 1 - низкое качество, 2 - среднее, 0 - высокое "retention_days": 7, "process_interval": 0.2 } # === ЖЕСТКИЕ НАСТРОЙКИ КАМЕРЫ (СКРЫТЫ ОТ ОПЕРАТОРА) === CAMERA_USER = "admin" CAMERA_PASS = "4NUDZhJ7" # Формат URL для Dahua/Hikvision. {stream} заменится на номер потока URL_TEMPLATE = "rtsp://{user}:{password}@{ip}/cam/realmonitor?channel=1&subtype={stream}" def load_config(): if not os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, 'w') as f: json.dump(DEFAULT_CONFIG, f, indent=4) return DEFAULT_CONFIG with open(CONFIG_FILE, 'r') as f: return json.load(f) config = load_config() # Глобальные переменные known_face_encodings = [] known_face_names = [] frame_queue = queue.Queue(maxsize=2) last_processed_time = 0 PROCESS_INTERVAL = float(config.get("process_interval", 0.2)) SAVE_DIR = "Лица" DATASET_DIR = "dataset" LOG_FILE = "access_log.csv" ARCHIVE_DIR = "logs_archive" RETENTION_DAYS = int(config.get("retention_days", 7)) last_sent_identity = None is_person_visible = False camera_thread_instance = None camera_stop_event = threading.Event() # === ИНИЦИАЛИЗАЦИЯ ПАПОК === for d in [SAVE_DIR, DATASET_DIR, ARCHIVE_DIR, os.path.join(SAVE_DIR, "Unknown")]: os.makedirs(d, exist_ok=True) if not os.path.exists(LOG_FILE): with open(LOG_FILE, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f, delimiter=';') writer.writerow(["Дата", "Время", "Имя", "Скриншот"]) def load_faces(): global known_face_encodings, known_face_names print("Loading faces...") known_face_encodings = [] known_face_names = [] for filename in os.listdir(DATASET_DIR): if filename.endswith((".jpg", ".jpeg", ".png")): try: image = face_recognition.load_image_file(os.path.join(DATASET_DIR, filename)) encodings = face_recognition.face_encodings(image) if len(encodings) > 0: known_face_encodings.append(encodings[0]) known_face_names.append(os.path.splitext(filename)[0]) except Exception as e: print(f"Error loading {filename}: {e}") print(f"✅ Loaded {len(known_face_names)} faces.") load_faces() def get_base64_image(path): try: with open(path, "rb") as img_file: return base64.b64encode(img_file.read()).decode('utf-8') except: return "" def cleanup_old_files(): now = time.time() cutoff = now - (RETENTION_DAYS * 86400) for root, dirs, files in os.walk(SAVE_DIR): for file in files: if file.endswith(".jpg"): filepath = os.path.join(root, file) if os.path.getmtime(filepath) < cutoff: try: os.remove(filepath) except: pass if os.path.exists(LOG_FILE) and os.path.getsize(LOG_FILE) > 0: if os.path.getmtime(LOG_FILE) < cutoff: zip_name = os.path.join(ARCHIVE_DIR, f"log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip") try: with zipfile.ZipFile(zip_name, 'w', zipfile.ZIP_DEFLATED) as zipf: zipf.write(LOG_FILE) open(LOG_FILE, 'w').close() except: pass def write_log(name, screenshot_path): timestamp = datetime.now() with open(LOG_FILE, "a", newline='', encoding="utf-8") as f: writer = csv.writer(f, delimiter=';') writer.writerow([ timestamp.strftime("%Y-%m-%d"), timestamp.strftime("%H:%M:%S"), name, screenshot_path ]) def process_frame(frame): global last_processed_time, last_sent_identity, is_person_visible small_frame = cv2.resize(frame, (0, 0), fx=0.25, fy=0.25) rgb_small_frame = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB) face_locations = face_recognition.face_locations(rgb_small_frame) if not face_locations: if is_person_visible: is_person_visible = False last_sent_identity = None return frame top, right, bottom, left = face_locations[0] face_encoding = face_recognition.face_encodings(rgb_small_frame, [face_locations[0]])[0] matches = face_recognition.compare_faces(known_face_encodings, face_encoding) name = "Unknown" if True in matches: name = known_face_names[matches.index(True)] full_top, full_right = top * 4, right * 4 full_bottom, full_left = bottom * 4, left * 4 should_notify = (name != last_sent_identity) or (not is_person_visible) if should_notify: screenshot_path = "" b64_img = "" pad_y, pad_x = 60, 40 h, w = frame.shape[:2] crop = frame[max(0,full_top-pad_y):min(h,full_bottom+pad_y), max(0,full_left-pad_x):min(w,full_right+pad_x)] if crop.size > 0: user_dir = os.path.join(SAVE_DIR, name) os.makedirs(user_dir, exist_ok=True) filename = f"{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.jpg" filepath = os.path.join(user_dir, filename) cv2.imwrite(filepath, crop, [int(cv2.IMWRITE_JPEG_QUALITY), 100]) screenshot_path = filepath b64_img = get_base64_image(filepath) write_log(name, filepath) threading.Thread(target=cleanup_old_files).start() if name != "Unknown": if not b64_img: emp_photo_path = os.path.join(DATASET_DIR, f"{name}.jpg") b64_img = get_base64_image(emp_photo_path) socketio.emit('new_event', {'type': 'known', 'name': name, 'photo': b64_img, 'time': datetime.now().strftime("%H:%M:%S")}) else: if b64_img: socketio.emit('new_event', {'type': 'unknown', 'name': 'НЕИЗВЕСТНЫЙ', 'photo': b64_img, 'path': screenshot_path, 'time': datetime.now().strftime("%H:%M:%S")}) last_sent_identity = name is_person_visible = True color = (0, 255, 0) if name != "Unknown" else (0, 0, 255) cv2.rectangle(frame, (full_left, full_top), (full_right, full_bottom), color, 2) cv2.putText(frame, name, (full_left + 6, full_bottom - 6), cv2.FONT_HERSHEY_DUPLEX, 1.0, (255, 255, 255), 1) return frame def camera_thread(camera_url): print(f"🔌 Connecting to: {camera_url}") video_capture = cv2.VideoCapture(camera_url) video_capture.set(cv2.CAP_PROP_BUFFERSIZE, 1) while not camera_stop_event.is_set(): success, frame = video_capture.read() if not success: time.sleep(1) video_capture.release() video_capture = cv2.VideoCapture(camera_url) continue if frame_queue.full(): try: frame_queue.get_nowait() except: pass frame_queue.put(frame) time.sleep(0.01) video_capture.release() def start_camera(): global camera_thread_instance, camera_stop_event if camera_thread_instance and camera_thread_instance.is_alive(): camera_stop_event.set() camera_thread_instance.join(timeout=5) camera_stop_event.clear() ip = config.get("camera_ip", "") stream = config.get("camera_stream", 1) if ip: # ИСПРАВЛЕНО: pass заменено на password url = URL_TEMPLATE.format(user=CAMERA_USER, password=CAMERA_PASS, ip=ip, stream=stream) camera_thread_instance = threading.Thread(target=camera_thread, args=(url,), daemon=True) camera_thread_instance.start() print(f"📷 Camera started for IP: {ip} (Stream: {stream})") else: print("⚠️ Camera IP not configured") start_camera() # === API ENDPOINTS === @app.route('/api/config', methods=['GET', 'POST']) def api_config(): global config, PROCESS_INTERVAL, RETENTION_DAYS if request.method == 'POST': data = request.json if 'camera_ip' in data: config['camera_ip'] = data['camera_ip'] if 'camera_stream' in data: config['camera_stream'] = int(data['camera_stream']) config['retention_days'] = int(data.get('retention_days', 7)) config['process_interval'] = float(data.get('process_interval', 0.2)) with open(CONFIG_FILE, 'w') as f: json.dump(config, f, indent=4) PROCESS_INTERVAL = config['process_interval'] RETENTION_DAYS = config['retention_days'] # Перезапускаем камеру при изменении IP или потока if any(k in data for k in ['camera_ip', 'camera_stream']): start_camera() return jsonify({"status": "ok"}) return jsonify(config) @app.route('/api/log') def api_log(): entries = [] if os.path.exists(LOG_FILE): with open(LOG_FILE, 'r', encoding='utf-8') as f: reader = csv.reader(f, delimiter=';') next(reader, None) for row in reader: if len(row) >= 4: safe_path = row[3].replace("\\", "/") entries.append({ 'date': row[0], 'time': row[1], 'name': row[2], 'path': safe_path }) entries.reverse() return jsonify(entries) @app.route('/api/upload', methods=['POST']) def api_upload(): name = request.form.get('name', '').strip() photo = request.files.get('photo') if not name or not photo: return jsonify({"error": "Name and photo required"}), 400 filename = f"{name.replace(' ', '_')}.jpg" path = os.path.join(DATASET_DIR, filename) photo.save(path) load_faces() return jsonify({"status": "ok", "name": name}) @app.route('/api/image/') def serve_image(filename): safe_dir = os.path.abspath(SAVE_DIR) requested_path = os.path.abspath(os.path.join(SAVE_DIR, filename)) if requested_path.startswith(safe_dir) and os.path.exists(requested_path): return send_from_directory(SAVE_DIR, filename) return abort(404) @app.route('/video_feed') def video_feed(): def generate(): global last_processed_time while True: try: frame = frame_queue.get(timeout=5) if time.time() - last_processed_time >= PROCESS_INTERVAL: frame = process_frame(frame) last_processed_time = time.time() ret, buffer = cv2.imencode('.jpg', frame) yield (b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n') except: continue return Response(generate(), mimetype='multipart/x-mixed-replace; boundary=frame') @app.route('/') def index(): return render_template_string(''' Face Recognition System
📹 Монитор
📋 Журнал
⚙️ Настройки
✅ СОТРУДНИКИ
LIVE
⚠️ НЕИЗВЕСТНЫЕ

Журнал посещений

ДатаВремяИмяСтатусФото

Настройки системы

➕ Добавить сотрудника

📷 Подключение камеры

⚙️ Система

''') if __name__ == '__main__': socketio.run(app, host='0.0.0.0', port=5000, debug=False)