app.py # Line 1 should have no leading space from flask import Flask, render_template, Response, request, jsonify, redirect, url_for, flash, session, send_from_directory, send_file from flask_login import LoginManager, login_user, logout_user, login_required, current_user from flask_socketio import SocketIO, emit, join_room, leave_room from werkzeug.security import generate_password_hash, check_password_hash from werkzeug.utils import secure_filename from flask_wtf.csrf import CSRFProtect from functools import wraps import secrets from werkzeug.utils import secure_filename import database # Uses SQLite now import video_processing import time import os import psutil from models import User import threading import datetime # For date parsing import zipfile # For backup/restore import shutil # For backup/restore file operations from fpdf import FPDF # For PDF reports from fpdf.fonts import FontFace from fpdf.enums import Align import requests # For testing telegram bot # Define project root early as it's used in helpers/classes project_root = os.path.dirname(os.path.abspath(__file__)) # --- Constants --- UPLOAD_FOLDER = 'uploads' ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'} ALLOWED_BACKUP_EXTENSIONS = {'zip'} # For restore # --- App Setup --- app = Flask(__name__) app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'a_very_secret_key_that_should_be_changed') app.config['UPLOAD_FOLDER'] = 'uploads' # Main folder for uploads app.config['ALLOWED_EXTENSIONS'] = {'png', 'jpg', 'jpeg', 'gif'} # For images primarily # Set max content length to None (unlimited) - Use with caution! app.config['MAX_CONTENT_LENGTH'] = None # app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024 * 1024 # Example: 1 GB limit # Initialize SocketIO # Use 'threading' mode for development with Flask dev server. # For production, consider eventlet or gevent. socketio = SocketIO(app, async_mode='threading') if not os.path.exists(UPLOAD_FOLDER): try: os.makedirs(UPLOAD_FOLDER) except OSError as e: print(f"Warning: Could not create upload folder {UPLOAD_FOLDER}: {e}") # --- Login Manager Setup --- login_manager = LoginManager() login_manager.init_app(app) login_manager.login_view = 'login' login_manager.login_message = "الرجاء تسجيل الدخول للوصول إلى هذه الصفحة." login_manager.login_message_category = 'info' @login_manager.user_loader def load_user(user_id): return User.get(user_id) # --- Helper Functions --- def allowed_file(filename, allowed_set=ALLOWED_EXTENSIONS): return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowed_set # --- PDF Report Generation (Keep as before) --- class PDFReport(FPDF): def __init__(self, title, *args, **kwargs): super().__init__(*args, **kwargs) self.report_title = title font_path = os.path.join(project_root, "static", "fonts", "Amiri-Regular.ttf") if os.path.exists(font_path): self.add_font("Amiri", "", font_path) self.set_font("Amiri", size=12) else: print("Warning: Arabic font file not found at", font_path, ". Using default font.") self.set_font("helvetica", size=12) self.alias_nb_pages() self.add_page() def header(self): self.set_font(self.font_family, "B", 15); self.cell(0, 10, self.report_title, align='C', new_x="LMARGIN", new_y="NEXT"); self.ln(10) def footer(self): self.set_y(-15); self.set_font(self.font_family, "I", 8); page_num_text = f"صفحة {self.page_no()}/{{nb}}"; self.cell(0, 10, page_num_text, align='C') def chapter_title(self, title): self.set_font(self.font_family, "B", 12); self.cell(0, 10, title, new_x="LMARGIN", new_y="NEXT", border='B'); self.ln(5) def event_table(self, header, data): if not data: self.set_font(self.font_family, "", 11); self.cell(0, 10, "لا توجد بيانات للعرض لهذه الفترة.", align='C'); self.ln(); return self.set_font(self.font_family, "B", 10); page_w = self.epw; col_widths = [page_w * 0.25, page_w * 0.2, page_w * 0.3, page_w * 0.25] self.set_fill_color(230, 230, 230) for i, col_name in enumerate(header): self.cell(col_widths[i], 7, col_name, border=1, fill=True, align='C') self.ln() self.set_font(self.font_family, "", 9); self.set_fill_color(255, 255, 255); fill = False for row in data: ts = datetime.datetime.fromtimestamp(row.get('timestamp', 0)).strftime('%Y-%m-%d %H:%M:%S') event_type = str(row.get('type', '')); name = str(row.get('name', '')); video_name = str(row.get('video_name', '')) max_h = 7 if len(name) > 25 or len(video_name) > 20: max_h = 14 self.cell(col_widths[0], max_h, ts, border='LR', align='C', fill=fill) self.cell(col_widths[1], max_h, event_type, border='LR', align='C', fill=fill) current_x = self.get_x(); current_y = self.get_y() self.multi_cell(col_widths[2], 7, name, border='LR', align='C', fill=fill) self.set_xy(current_x + col_widths[2], current_y) self.multi_cell(col_widths[3], 7, video_name, border='LR', align='C', fill=fill) self.ln(max_h); fill = not fill self.cell(sum(col_widths), 0, '', border='T') # --- Video Stream Management (Keep as before) --- def start_all_streams(): print("Starting all video streams..."); videos = database.get_videos() for video in videos: video_processing.start_stream(video) print(f"Attempted to start {len(videos)} streams.") def stop_all_streams(): print("Stopping all video streams...") # Get a copy of the keys to avoid modification issues during iteration video_ids = list(video_processing.video_streams.keys()) threads_to_join = [] # First, signal all streams to stop and collect threads print(f"Signaling {len(video_ids)} streams to stop...") for video_id in video_ids: thread = video_processing.signal_stop_stream(video_id) if thread: threads_to_join.append(thread) # Now, wait for all threads to finish print(f"Waiting for {len(threads_to_join)} stream threads to join...") start_join_time = time.time() for thread in threads_to_join: try: # Calculate remaining timeout elapsed = time.time() - start_join_time remaining_timeout = max(0, 5.0 - elapsed) # Use a total timeout of 5s thread.join(timeout=remaining_timeout) if thread.is_alive(): print(f"Warning: Thread {thread.name} did not stop within the allocated time.") except KeyboardInterrupt: print(f"KeyboardInterrupt received while joining thread {thread.name}. Continuing shutdown...") # Optionally break the loop if immediate shutdown is desired on Ctrl+C # break except Exception as e: print(f"Error joining thread {thread.name}: {e}") # Finally, clear the dictionary print("Clearing video streams dictionary...") video_processing.video_streams.clear() print("All video streams stopped and cleared.") # --- Routes --- # Authentication (Keep as before) @app.route('/login', methods=['GET', 'POST']) def login(): if current_user.is_authenticated: return redirect(url_for('videos_page')) if request.method == 'POST': user = User.get_by_username(request.form['username']) if not user or not user.check_password(request.form['password']): flash('الرجاء التحقق من بيانات الدخول والمحاولة مرة أخرى.', 'danger'); return redirect(url_for('login')) login_user(user, remember=True if request.form.get('remember') else False); flash(f'مرحباً بعودتك، {user.username}!', 'success') next_page = request.args.get('next'); return redirect(next_page or url_for('videos_page')) return render_template('login.html') @app.route('/logout') @login_required def logout(): logout_user(); flash('تم تسجيل خروجك بنجاح.', 'info'); return redirect(url_for('login')) # Main Pages (Keep as before) @app.route('/') @login_required def index(): return redirect(url_for('videos_page')) @app.route('/videos') @login_required def videos_page(): videos = database.get_videos(); settings = database.get_settings(); grid_cols = settings.get('live_view_grid_cols', 4) for video in videos: if video.get('id') not in video_processing.video_streams: video_processing.start_stream(video) return render_template('videos.html', title='العرض المباشر', videos=videos, grid_cols=grid_cols) # Settings Page (Keep as before) @app.route('/settings', methods=['GET', 'POST']) @login_required def settings_page(): if current_user.role != 'admin': flash('ليس لديك الصلاحية للوصول إلى الإعدادات.', 'warning'); return redirect(url_for('videos_page')) if request.method == 'POST': settings_data = request.form.to_dict() if database.save_settings(settings_data): new_frame_skip = database.get_settings().get('frame_skip', 5) for video_id in video_processing.video_streams: video_processing.update_stream_settings(video_id, {'frame_skip': new_frame_skip}) flash('تم تحديث الإعدادات بنجاح.', 'success') else: flash('فشل تحديث الإعدادات.', 'danger') return redirect(url_for('settings_page')) settings = database.get_settings(); return render_template('settings.html', title='الإعدادات', settings=settings) # Management Pages (Keep as before) @app.route('/manage/users') @login_required def users_manage_page(): if current_user.role != 'admin': flash('ليس لديك الصلاحية لإدارة المستخدمين.', 'warning'); return redirect(url_for('videos_page')) users = database.get_users(); return render_template('manage_users.html', title='إدارة المستخدمين', users=users) @app.route('/manage/cameras') @login_required def cameras_manage_page(): cameras = database.get_cameras(); return render_template('manage_cameras.html', title='إدارة الكاميرات', cameras=cameras) @app.route('/manage/videos') @login_required def videos_manage_page(): videos = database.get_videos(); return render_template('manage_videos.html', title='إدارة الفيديوهات', videos=videos) @app.route('/manage/people') @login_required def people_manage_page(): people = database.get_people(); return render_template('manage_people.html', title='إدارة الأشخاص', people=people) @app.route('/manage/trucks') @login_required def trucks_manage_page(): # Removed passing trucks data, will be fetched via API by Alpine.js return render_template('manage_trucks.html', title='إدارة الشاحنات') @app.route('/manage/telegram') @login_required def telegram_manage_page(): bots = database.get_telegram_bots(); return render_template('manage_telegram.html', title='إدارة بوتات التليجرام', bots=bots) # Reports Page (Keep as before) @app.route('/reports') @login_required def reports_page(): return render_template('reports.html', title='التقارير') # Recordings Management Page @app.route('/manage/recordings') @login_required def recordings_manage_page(): return render_template('manage_recordings.html', title='إدارة التسجيلات') # API to get detection events with filtering @app.route('/api/detection_events') @login_required def api_get_detection_events(): start = request.args.get('start', type=float) end = request.args.get('end', type=float) event_type = request.args.get('type', default=None, type=str) name = request.args.get('name', default=None, type=str) events = database.get_detection_events(start_time=start, end_time=end, event_type=event_type, specific_name=name) return jsonify({"success": True, "events": events}) # --- Chat Page --- @app.route('/chat') @login_required def chat_page(): bots = database.get_telegram_bots() return render_template('chat.html', title='محادثة تليجرام', bots=bots) # --- Video Feed (Keep as before) --- def generate_frames(video_id): print(f"Frame generator started for video ID: {video_id}") last_frame_time = time.time() frame_count = 0 while True: if video_id not in video_processing.video_streams: print(f"Stream {video_id} stopped. Stopping generator.") break frame_bytes = video_processing.get_stream_frame(video_id) if not frame_bytes: time.sleep(0.05) continue # Calculate FPS and adjust sleep time current_time = time.time() frame_count += 1 elapsed_time = current_time - last_frame_time fps = frame_count / elapsed_time # Adjust sleep time based on FPS target_fps = 20 # Target FPS sleep_time = max(0, 1/target_fps - (current_time - last_frame_time)) yield (b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n') time.sleep(sleep_time) @app.route('/video_feed/') @login_required def video_feed(video_id): try: video = database.get_video_by_id(video_id) if not video: return jsonify({"error": "Video not found"}), 404 if video_id not in video_processing.video_streams: try: video_processing.start_stream(video) time.sleep(1) except Exception as e: print(f"Error starting video stream: {str(e)}") return jsonify({"error": "Failed to start video stream"}), 500 response = Response( generate_frames(video_id), mimetype='multipart/x-mixed-replace; boundary=frame' ) # Add security headers response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate' response.headers['Pragma'] = 'no-cache' response.headers['Expires'] = '0' response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-Frame-Options'] = 'SAMEORIGIN' return response except Exception as e: print(f"Error in video_feed: {str(e)}") return jsonify({"error": "Internal server error"}), 500 # --- API Routes --- # System Monitoring API (Keep as before) @app.route('/api/system_stats') @login_required def system_stats(): try: cpu = psutil.cpu_percent(interval=0.1); mem = psutil.virtual_memory().percent; return jsonify({"success": True, "cpu_percent": cpu, "memory_percent": mem}) except Exception as e: print(f"Stats Error: {e}"); return jsonify({"success": False, "message": str(e)}), 500 # Video Settings API (Refined to handle recording_enabled and correct key/response) @app.route('/api/video//toggle_setting', methods=['POST']) @login_required def toggle_video_setting(video_id): video = database.get_video_by_id(video_id); if not video: return jsonify({"success": False, "message": "لم يتم العثور على الفيديو"}), 404 # Read setting_key from request.json (matches frontend call) data = request.get_json() print(f"--- Debug toggle_video_setting: Received data: {data}") setting_key = data.get('setting_key') print(f"--- Debug toggle_video_setting: Extracted setting_key: {setting_key}") # Add 'recording_enabled' to the list of valid settings valid_settings = [ 'face_recognition_enabled', 'telegram_enabled', 'motion_detection_enabled', 'recording_enabled' # Added recording toggle ] if not setting_key or setting_key not in valid_settings: print(f"Invalid setting key received: {setting_key} for video {video_id}") return jsonify({"success": False, "message": "اسم الإعداد غير صالح"}), 400 try: current_value = video.get(setting_key, False) new_state = not current_value update_data = {setting_key: new_state} # Update the database first if not database.update_video(video_id, update_data): return jsonify({"success": False, "message": "فشل تحديث إعداد الفيديو في قاعدة البيانات"}), 500 # Update the running stream settings (like frame skip, etc., if applicable) # Note: video_processing.update_stream_settings might not need recording_enabled # Only pass relevant settings if necessary. # For now, we pass it, but it might be ignored by the function. video_processing.update_stream_settings(video_id, update_data) # --- Handle Recording Start/Stop --- if setting_key == 'recording_enabled': video_name = video.get('name', f'video_{video_id}') # Get video name for filename if new_state: print(f"---> Received request to START recording for: {video_name} ({video_id})") try: # TODO: Implement or verify start_recording_process # Ensure this function exists and handles starting the recording logic. # It should ideally use video_name to construct the filename. video_processing.start_recording_process(video_id, video_name) print(f"--- Recording process initiated for {video_name}") except AttributeError: print("Error: video_processing.start_recording_process function not found!") # Optionally revert the DB change if starting fails critically? # For now, just log the error. except Exception as e: print(f"Error starting recording process for {video_name}: {e}") else: print(f"---> Received request to STOP recording for: {video_name} ({video_id})") try: # TODO: Implement or verify stop_recording_process # Ensure this function exists and handles stopping the recording. video_processing.stop_recording_process(video_id) print(f"--- Recording process stopped for {video_name}") except AttributeError: print("Error: video_processing.stop_recording_process function not found!") except Exception as e: print(f"Error stopping recording process for {video_name}: {e}") # ------------------------------------- # Return the new state in the expected format for the frontend print(f"Successfully toggled {setting_key} for {video_id} to {new_state}") return jsonify({"success": True, "setting_key": setting_key, "new_state": new_state}) except Exception as e: db.session.rollback() # Rollback DB if any exception occurs after potential update print(f"Unexpected error in toggle_video_setting for {video_id} ({setting_key}): {e}") return jsonify({"success": False, "message": f"خطأ داخلي في الخادم: {e}"}), 500 # --- CRUD APIs (Keep as before, ensure GET endpoints exist) --- # Generic GET by ID def get_item_api(func, item_id): item = func(item_id) if item: return jsonify({"success": True, func.__name__.replace('get_',''): item}) else: return jsonify({"success": False, "message": "العنصر غير موجود"}), 404 @app.route('/api/user/', methods=['GET']) @login_required def api_get_user(user_id): return get_item_api(database.get_user_by_id, user_id) @app.route('/api/camera/', methods=['GET']) @login_required def api_get_camera(cam_id): return get_item_api(database.get_camera_by_id, cam_id) @app.route('/api/video/', methods=['GET']) @login_required def api_get_video(video_id): return get_item_api(database.get_video_by_id, video_id) @app.route('/api/person/', methods=['GET']) @login_required def api_get_person(person_id): return get_item_api(database.get_person_by_id, person_id) @app.route('/api/truck/', methods=['GET']) @login_required def api_get_truck(truck_id): return get_item_api(database.get_truck_by_id, truck_id) @app.route('/api/telegram_bot/', methods=['GET']) @login_required def api_get_telegram_bot(bot_id): return get_item_api(database.get_telegram_bot_by_id, bot_id) # Videos @app.route('/api/videos', methods=['POST']) @login_required def api_add_video(): # Read form data and file uploads stype = request.form.get('source_type') name = request.form.get('name') vdata = {"name": name} # Start with name source_identifier = None # To store path/url for logging/stream start if stype == 'file': if 'video_file' not in request.files: return jsonify({"success": False, "message": "لم يتم رفع ملف الفيديو."}), 400 file = request.files['video_file'] if file.filename == '': return jsonify({"success": False, "message": "لم يتم اختيار ملف فيديو."}), 400 # Define allowed video extensions (adjust as needed) allowed_extensions = {'.mp4', '.avi', '.mov', '.mkv'} file_ext = os.path.splitext(file.filename)[1].lower() if file_ext not in allowed_extensions: return jsonify({"success": False, "message": f"نوع ملف الفيديو غير مدعوم ({file_ext}). الأنواع المدعومة: {', '.join(allowed_extensions)}"}), 400 # Ensure upload folder exists upload_folder = os.path.join(app.config.get('UPLOAD_FOLDER', 'uploads'), 'video_files') os.makedirs(upload_folder, exist_ok=True) # Create a secure filename (original filename + timestamp/uuid) base_name = secure_filename(os.path.splitext(file.filename)[0]) timestamp = int(time.time()) safe_filename = f"{base_name}_{timestamp}{file_ext}" file_path = os.path.join(upload_folder, safe_filename) relative_path = os.path.join('video_files', safe_filename) # Path to store in DB try: file.save(file_path) vdata['path'] = relative_path # Store relative path in DB vdata['name'] = name or base_name # Use original base name if no display name given source_identifier = relative_path except Exception as e: print(f"Error saving uploaded video file: {e}") return jsonify({"success": False, "message": "حدث خطأ أثناء حفظ ملف الفيديو المرفوع."}), 500 elif stype == 'url': url = request.form.get('source_url') if not url: return jsonify({"success": False, "message": "رابط البث مطلوب."}), 400 # Basic URL validation (optional, can be more strict) if not (url.startswith('http://') or url.startswith('https://') or url.startswith('rtsp://')): return jsonify({"success": False, "message": "رابط البث غير صالح (يجب أن يبدأ بـ http/https/rtsp)."}), 400 vdata['url'] = url vdata['name'] = name or "فيديو رابط" # Default name for URLs source_identifier = url else: return jsonify({"success": False, "message": "نوع المصدر غير صالح."}), 400 # Add video data to database added = database.add_video(vdata) if not added: # If DB add failed and we saved a file, attempt to remove it if stype == 'file' and file_path and os.path.exists(file_path): try: os.remove(file_path) print(f"Removed orphaned video file after DB error: {file_path}") except OSError as e: print(f"Error removing orphaned video file {file_path}: {e}") return jsonify({"success": False, "message": "فشل حفظ بيانات الفيديو في قاعدة البيانات."}), 500 # Start video stream processing stream = video_processing.start_stream(added) warn = f"تمت إضافة الفيديو، لكن فشل بدء البث. تحقق من المصدر/السجلات: {source_identifier}" if not stream else None return jsonify({"success": True, "video": added, "warning": warn}), 201 @app.route('/api/video/', methods=['PUT']) @login_required def api_update_video(video_id): data = request.json update_data = {} if 'name' in data: update_data['name'] = data['name'] if not update_data: return jsonify({"success": False, "message": "لا توجد حقول قابلة للتعديل"}), 400 try: updated = database.update_video(video_id, update_data) if updated: # Update name in stream config if stream is running if 'name' in update_data and video_id in video_processing.video_streams: try: video_processing.video_streams[video_id].video_config['name'] = update_data['name'] except Exception as stream_update_err: # Log the error but don't fail the whole request app.logger.warning(f"Failed to update stream config name for {video_id}: {stream_update_err}") return jsonify({"success": True, "video": updated}) else: # If database.update_video returns None/False, it means video not found or update failed return jsonify({"success": False, "message": "الفيديو غير موجود أو فشل التحديث"}), 404 except Exception as e: app.logger.error(f"Error updating video {video_id}: {e}", exc_info=True) return jsonify({"success": False, "message": f"خطأ داخلي في الخادم أثناء تحديث الفيديو: {e}"}), 500 @app.route('/api/video/', methods=['DELETE']) @login_required def api_delete_video(video_id): video_processing.signal_stop_stream(video_id) if database.delete_video(video_id): return jsonify({"success": True, "message": "تم حذف الفيديو"}) else: return jsonify({"success": False, "message": "فشل حذف الفيديو"}), 404 # Users (Keep POST/PUT/DELETE as before) @app.route('/api/users', methods=['POST']) @login_required def api_add_user(): if current_user.role != 'admin': return jsonify({"success": False, "message": "غير مصرح لك"}), 403 data = request.json if not data.get('username') or not data.get('password') or not data.get('full_name'): return jsonify({"success": False, "message": "الحقول المطلوبة مفقودة"}), 400 result = database.add_user(data) if result == "username_exists": return jsonify({"success": False, "message": "اسم المستخدم موجود بالفعل"}), 409 elif result: result.pop('password_hash', None); return jsonify({"success": True, "user": result}), 201 else: return jsonify({"success": False, "message": "فشل إضافة المستخدم"}), 500 @app.route('/api/user/', methods=['PUT']) @login_required def api_update_user(user_id): if current_user.role != 'admin': return jsonify({"success": False, "message": "غير مصرح لك"}), 403 data = request.json if user_id == current_user.id and 'role' in data and data['role'] != current_user.role: return jsonify({"success": False, "message": "لا يمكنك تغيير صلاحيتك بنفسك"}), 400 # Prepare update data, filter out disallowed fields if needed allowed_updates = ['full_name', 'password', 'role'] update_data = {k: v for k, v in data.items() if k in allowed_updates} # Handle password separately (hash it) if 'password' in update_data and update_data['password']: # Check if password is provided and not empty update_data['password_hash'] = generate_password_hash(update_data.pop('password')) elif 'password' in update_data: # Remove empty password field if present del update_data['password'] if not update_data: return jsonify({"success": False, "message": "لا توجد بيانات قابلة للتحديث"}), 400 try: updated = database.update_user(user_id, update_data) if updated: # Remove password hash before sending back updated.pop('password_hash', None) return jsonify({"success": True, "user": updated}) else: # If database.update_user returns None/False return jsonify({"success": False, "message": "المستخدم غير موجود أو فشل التحديث"}), 404 except Exception as e: app.logger.error(f"Error updating user {user_id}: {e}", exc_info=True) return jsonify({"success": False, "message": f"خطأ داخلي في الخادم أثناء تحديث المستخدم: {e}"}), 500 @app.route('/api/user/', methods=['DELETE']) @login_required def api_delete_user(user_id): if current_user.role != 'admin': return jsonify({"success": False, "message": "غير مصرح لك"}), 403 if user_id == current_user.id: return jsonify({"success": False, "message": "لا يمكنك حذف نفسك"}), 400 users = database.get_users(); admins = [u for u in users if u.get('role') == 'admin']; target = database.get_user_by_id(user_id) if len(admins) <= 1 and target and target.get('role') == 'admin': return jsonify({"success": False, "message": "لا يمكن حذف المسؤول الأخير"}), 400 if database.delete_user(user_id): return jsonify({"success": True, "message": "تم حذف المستخدم"}) else: return jsonify({"success": False, "message": "فشل حذف المستخدم"}), 404 # People # Add GET method to fetch all people @app.route('/api/people', methods=['GET', 'POST']) @login_required def api_people(): if request.method == 'GET': people = database.get_people() return jsonify({"success": True, "people": people}) elif request.method == 'POST': # Existing POST logic for adding a person - Indent this block if 'image' not in request.files: return jsonify({"success": False, "message": "ملف الصورة مفقود"}), 400 file = request.files['image']; name = request.form.get('name') if not name: return jsonify({"success": False, "message": "الاسم مطلوب"}), 400 if file.filename == '': return jsonify({"success": False, "message": "لم يتم اختيار ملف"}), 400 if file and allowed_file(file.filename): fname = secure_filename(f"{name}_{int(time.time())}.{file.filename.rsplit('.', 1)[1].lower()}") folder = os.path.join(app.config['UPLOAD_FOLDER'], 'people_images') if not os.path.exists(folder): try: os.makedirs(folder) except OSError: return jsonify({"success": False, "message": "خطأ في الخادم عند إنشاء مجلد الصور"}), 500 fpath = os.path.join(folder, fname); rel_path = os.path.join('people_images', fname) try: file.save(fpath); print(f"--- Debug: Saved person image to: {fpath}") # Log full save path pdata = {"name": name, "image_path": rel_path} print(f"--- Debug: Person data before DB add: {pdata}") # Log data being sent to DB added = database.add_person(pdata) if added: print(f"--- Debug: Person added to DB with ID: {added.get('id')}, Image Path: {added.get('image_path')}") # Log path from DB result threading.Thread(target=video_processing.load_known_data, daemon=True).start() return jsonify({"success": True, "person": added}), 201 else: # DB add failed if os.path.exists(fpath): os.remove(fpath) return jsonify({"success": False, "message": "فشل حفظ بيانات الشخص"}), 500 except Exception as e: print(f"Save person error: {e}") if os.path.exists(fpath): os.remove(fpath) return jsonify({"success": False, "message": f"خطأ في حفظ الصورة: {e}"}), 500 else: return jsonify({"success": False, "message": "نوع الملف غير مسموح به"}), 400 # Keep PUT/DELETE for People as before @app.route('/api/person/', methods=['GET','PUT','DELETE']) @login_required def api_person_detail(person_id): if request.method == 'GET': person = database.get_person_by_id(person_id) if person: return jsonify({"success": True, "person": person}) else: return jsonify({"success": False, "message": "الشخص غير موجود"}), 404 elif request.method == 'PUT': # Existing PUT logic for updating a person - Indent this block data = request.form.to_dict(); new_image_path = None old_person_data = database.get_person_by_id(person_id) if not old_person_data: return jsonify({"success": False, "message": "الشخص غير موجود"}), 404 if 'image' in request.files: file = request.files['image'] if file and file.filename != '' and allowed_file(file.filename): name = data.get('name', old_person_data.get('name')) fname = secure_filename(f"{name}_{int(time.time())}.{file.filename.rsplit('.', 1)[1].lower()}") folder = os.path.join(app.config['UPLOAD_FOLDER'], 'people_images') if not os.path.exists(folder): os.makedirs(folder) fpath = os.path.join(folder, fname); rel_path = os.path.join('people_images', fname) try: file.save(fpath); new_image_path = rel_path; data['image_path'] = new_image_path old_rel_path = old_person_data.get('image_path') if old_rel_path and old_rel_path != new_image_path: old_fpath = os.path.join(app.config['UPLOAD_FOLDER'], old_rel_path) if os.path.exists(old_fpath): os.remove(old_fpath) except Exception as e: return jsonify({"success": False, "message": "خطأ في حفظ الصورة الجديدة"}), 500 elif file.filename != '': return jsonify({"success": False, "message": "نوع الملف الجديد غير مسموح به"}), 400 updated_person_data = {k:v for k,v in data.items() if k != 'person_id'} # Exclude person_id from update data updated = database.update_person(person_id, updated_person_data) if updated: threading.Thread(target=video_processing.load_known_data, daemon=True).start() return jsonify({"success": True, "person": updated}) else: # This 'else' block also needs to be indented correctly if new_image_path: new_fpath = os.path.join(app.config['UPLOAD_FOLDER'], new_image_path) if os.path.exists(new_fpath): os.remove(new_fpath) return jsonify({"success": False, "message": "فشل تحديث الشخص"}), 404 elif request.method == 'DELETE': # Existing DELETE logic for deleting a person - Indent this block if database.delete_person(person_id): threading.Thread(target=video_processing.load_known_data, daemon=True).start() return jsonify({"success": True, "message": "تم حذف الشخص"}) else: return jsonify({"success": False, "message": "فشل حذف الشخص"}), 404 # Cameras (Keep POST/PUT/DELETE as before) @app.route('/api/cameras', methods=['POST']) @login_required def api_add_camera(): data = request.json if not data.get('name') or not data.get('url'): return jsonify({"success": False, "message": "اسم الكاميرا والرابط مطلوبان"}), 400 added = database.add_camera(data) if added: return jsonify({"success": True, "camera": added}), 201 else: return jsonify({"success": False, "message": "فشل إضافة الكاميرا"}), 500 @app.route('/api/camera/', methods=['PUT']) @login_required def api_update_camera(cam_id): data = request.json updated = database.update_camera(cam_id, data) if updated: return jsonify({"success": True, "camera": updated}) else: return jsonify({"success": False, "message": "الكاميرا غير موجودة أو فشل التحديث"}), 404 @app.route('/api/camera/', methods=['DELETE']) @login_required def api_delete_camera(cam_id): if database.delete_camera(cam_id): return jsonify({"success": True, "message": "تم حذف الكاميرا"}) else: return jsonify({"success": False, "message": "فشل حذف الكاميرا"}), 404 # Trucks (Keep POST/PUT/DELETE as before) @app.route('/api/trucks', methods=['GET', 'POST']) @login_required def api_trucks(): if request.method == 'GET': trucks = database.get_trucks() return jsonify({"success": True, "trucks": trucks}) elif request.method == 'POST': form_data = request.form.to_dict() image_file = request.files.get('image') image_path = None full_path = None # Initialize full_path if not form_data.get('number'): return jsonify({"success": False, "message": "رقم الشاحنة مطلوب."}), 400 # Handle image upload if image_file and allowed_file(image_file.filename): filename = secure_filename(f"truck_{form_data['number'].replace(' ', '_').replace('/', '_')}_{int(time.time())}.{image_file.filename.rsplit('.', 1)[1].lower()}") save_dir = database.truck_img_dir if not os.path.exists(save_dir): os.makedirs(save_dir) full_path = os.path.join(save_dir, filename) try: image_file.save(full_path) # Store relative path with forward slashes image_path = os.path.join("truck_images", filename).replace("\\", "/") print(f"Saved truck image to: {full_path}") form_data['image_path'] = image_path except Exception as e: print(f"Error saving truck image: {e}") return jsonify({"success": False, "message": f"خطأ في حفظ صورة الشاحنة: {e}"}), 500 else: form_data['image_path'] = None # Ensure it's None if no image uploaded # Accept new fields form_data['driver_phone'] = form_data.get('driver_phone') form_data['model'] = form_data.get('model') form_data['identification_number'] = form_data.get('identification_number') truck = database.add_truck(form_data) if truck: # Trigger reload (function to be added in video_processing) try: video_processing.load_known_trucks() # Call reload except AttributeError: print("Note: video_processing.load_known_trucks() not implemented yet.") except Exception as e: print(f"Error calling load_known_trucks after add: {e}") return jsonify({"success": True, "truck": truck}) else: # Clean up uploaded image if DB insert failed if full_path and os.path.exists(full_path): try: os.remove(full_path) except Exception as del_e: print(f"Error cleaning up truck image {full_path}: {del_e}") return jsonify({"success": False, "message": "فشل إضافة الشاحنة إلى قاعدة البيانات."}), 500 @app.route('/api/trucks/', methods=['GET', 'PUT', 'DELETE']) @login_required def api_truck_detail(truck_id): if request.method == 'GET': truck = database.get_truck_by_id(truck_id) if truck: return jsonify({"success": True, "truck": truck}) else: return jsonify({"success": False, "message": "الشاحنة غير موجودة"}), 404 elif request.method == 'PUT': form_data = request.form.to_dict() image_file = request.files.get('image') truck = database.get_truck_by_id(truck_id) if not truck: return jsonify({"success": False, "message": "الشاحنة غير موجودة"}), 404 image_path = truck.get('image_path') # Keep old path by default old_full_path = os.path.join(database.uploads_dir, image_path) if image_path else None new_full_path = None # Handle image update if image_file and allowed_file(image_file.filename): number_for_file = form_data.get('number', truck.get('number', 'unknown')).replace(' ', '_').replace('/', '_') filename = secure_filename(f"truck_{number_for_file}_{int(time.time())}.{image_file.filename.rsplit('.', 1)[1].lower()}") save_dir = database.truck_img_dir if not os.path.exists(save_dir): os.makedirs(save_dir) new_full_path = os.path.join(save_dir, filename) try: image_file.save(new_full_path) # Store relative path with forward slashes image_path = os.path.join("truck_images", filename).replace("\\", "/") print(f"Saved updated truck image to: {new_full_path}") # Delete old image if it exists and path changed if old_full_path and os.path.exists(old_full_path) and old_full_path != new_full_path: try: os.remove(old_full_path); print(f"Deleted old truck image: {old_full_path}") except Exception as del_e: print(f"Error deleting old truck image {old_full_path}: {del_e}") except Exception as e: print(f"Error saving updated truck image: {e}") # Don't fail the whole request, maybe just the image failed? # Potentially return a warning in the response? # For now, just log and continue without updating image path image_path = truck.get('image_path') # Revert to old path form_data['image_path'] = image_path # Update image path in data to be saved # Accept new fields form_data['driver_phone'] = form_data.get('driver_phone') form_data['model'] = form_data.get('model') form_data['identification_number'] = form_data.get('identification_number') if database.update_truck(truck_id, form_data): # Trigger reload try: video_processing.load_known_trucks() except AttributeError: print("Note: video_processing.load_known_trucks() not implemented yet.") except Exception as e: print(f"Error calling load_known_trucks after update: {e}") updated_truck = database.get_truck_by_id(truck_id) # Fetch updated data return jsonify({"success": True, "truck": updated_truck}) else: # Clean up newly saved image if DB update failed if new_full_path and os.path.exists(new_full_path): try: os.remove(new_full_path) except Exception as del_e: print(f"Error cleaning up new truck image {new_full_path}: {del_e}") return jsonify({"success": False, "message": "فشل تحديث بيانات الشاحنة."}), 500 elif request.method == 'DELETE': # Let database.delete_truck handle image file cleanup if database.delete_truck(truck_id): # Trigger reload try: video_processing.load_known_trucks() except AttributeError: print("Note: video_processing.load_known_trucks() not implemented yet.") except Exception as e: print(f"Error calling load_known_trucks after delete: {e}") return jsonify({"success": True, "message": "تم حذف الشاحنة بنجاح."}) else: return jsonify({"success": False, "message": "فشل حذف الشاحنة."}), 500 @app.route('/api/trucks/reload', methods=['POST']) @login_required def reload_trucks(): try: # This function needs to be created in video_processing.py video_processing.load_known_trucks() return jsonify({"success": True, "message": "تم تحديث بيانات الشاحنات المعروفة."}) except AttributeError: print("Error: video_processing.load_known_trucks() does not exist yet.") return jsonify({"success": False, "message": "وظيفة تحديث الشاحنات غير معرفة بعد."}), 501 # Not Implemented except Exception as e: print(f"Error reloading trucks: {e}") return jsonify({"success": False, "message": f"خطأ أثناء تحديث الشاحنات: {e}"}), 500 # --- Serve Uploaded Files (Refined) --- @app.route('/uploads/') @login_required def uploaded_file(filename): # Basic security: Prevent path traversal if '..' in filename or filename.startswith( ('/', '\\') ): return "Forbidden", 403 # Construct potential absolute path relative to project root's 'uploads' folder # This assumes UPLOAD_FOLDER is relative. If it can be absolute, logic needs adjustment. safe_filename = secure_filename(filename) # Basic sanitization if safe_filename != filename: # Check if secure_filename removed unsafe parts print(f"Warning: Potentially unsafe filename requested: {filename}, sanitized to: {safe_filename}") # Decide whether to reject or proceed with sanitized name. Rejecting is safer. # return "Forbidden", 403 # Let's proceed with the original for now, relying on the '..' check # Use app.config['UPLOAD_FOLDER'] as the base directory upload_directory = os.path.abspath(app.config['UPLOAD_FOLDER']) # The filename might contain subdirectories like 'people_images' or 'truck_images' file_path = os.path.join(upload_directory, filename) # Normalize path to handle mixed slashes and redundant separators file_path = os.path.normpath(file_path) # Final security check: ensure the final path is still within the upload directory if not file_path.startswith(upload_directory): print(f"Forbidden access attempt: {filename} resolved outside upload dir ({file_path})") return "Forbidden", 403 if os.path.exists(file_path) and os.path.isfile(file_path): print(f"Serving file: {file_path}") return send_file(file_path) # send_file handles mimetype and caching headers else: print(f"File not found: {filename} (Checked path: {file_path})") return "File Not Found", 404 # --- SocketIO Events for Chat --- @socketio.on('connect') @login_required # Ensure user is logged in to use chat def handle_connect(): print(f"Client connected: {request.sid}, User: {current_user.username}") # Join a room based on user ID? Or a general chat room? # For simplicity, maybe just emit to the specific client for now. emit('status', {'msg': 'تم الاتصال بخادم المحادثة.'}) @socketio.on('disconnect') def handle_disconnect(): print(f"Client disconnected: {request.sid}") @socketio.on('send_message') @login_required def handle_send_message(data): message_text = data.get('message') bot_id = data.get('bot_id') print(f"Received message from {current_user.username} for bot {bot_id}: {message_text}") if not message_text or not bot_id: emit('error', {'msg': 'الرسالة أو معرف البوت مفقود.'}) return bot = database.get_telegram_bot_by_id(bot_id) if not bot: emit('error', {'msg': 'لم يتم العثور على البوت المحدد.'}) return token = bot.get('token') chat_id = bot.get('chat_id') # Send to the bot's configured chat_id if not token or not chat_id: emit('error', {'msg': 'بيانات البوت غير مكتملة.'}) return # Emit the message back to the sender's UI immediately emit('new_message', { 'user': current_user.username, 'text': message_text, 'timestamp': time.time(), 'is_sender': True # Mark as sent by current user }) # Send the message to Telegram API telegram_api_url = f"https://api.telegram.org/bot{token}/sendMessage" payload = { 'chat_id': chat_id, 'text': f"({current_user.username}): {message_text}" } # Prepend username? try: response = requests.post(telegram_api_url, data=payload, timeout=10) response.raise_for_status() result = response.json() if not result.get('ok'): error_description = result.get('description', 'خطأ غير معروف من تليجرام') emit('error', {'msg': f"فشل إرسال الرسالة إلى تليجرام: {error_description}"}) except requests.exceptions.RequestException as e: print(f"Telegram send error: {e}") emit('error', {'msg': f"خطأ في الاتصال بـ Telegram: {e}"}) except Exception as e: print(f"Telegram send unexpected error: {e}") emit('error', {'msg': f"خطأ غير متوقع عند الإرسال: {e}"}) # --- TODO: Add Webhook Endpoint for Receiving Telegram Replies --- # @app.route('/webhook/telegram/', methods=['POST']) # def handle_telegram_webhook(secure_token): # # 1. Verify secure_token to ensure request is from Telegram # # 2. Parse the incoming update (request.json) # # 3. Identify the message, sender (chat_id), and text # # 4. Find the corresponding bot configuration based on the token used in the webhook URL # # 5. Emit the received message to the relevant client(s) via SocketIO # # socketio.emit('new_message', {'user': 'BotName/User', 'text': received_text, 'timestamp': time.time(), 'is_sender': False}, room=?) # # 6. Return a 200 OK response to Telegram quickly # pass # --- Application Context & Cleanup --- with app.app_context(): database.initialize_data() start_all_streams() import atexit atexit.register(stop_all_streams) # --- Run --- if __name__ == '__main__': font_dir = os.path.join(project_root, "static", "fonts") if not os.path.exists(font_dir): os.makedirs(font_dir) print(f"Ensure Arabic font (e.g., Amiri-Regular.ttf) is placed in: {font_dir}") # Use socketio.run() instead of app.run() print("Starting Flask-SocketIO server...") socketio.run(app, host='0.0.0.0', port=5000, debug=True, use_reloader=False, allow_unsafe_werkzeug=True) # allow_unsafe_werkzeug=True might be needed for debug mode with SocketIO + Werkzeug > 2.1 database.py import sqlite3 import os import uuid from werkzeug.security import generate_password_hash import appdirs import time import json # For storing list/dict in settings if needed, though specific columns are better # --- Define Database Location --- APP_NAME = "WebDVRPro" APP_AUTHOR = "DVRApp" user_data_dir = appdirs.user_data_dir(APP_NAME, APP_AUTHOR) if not os.path.exists(user_data_dir): try: os.makedirs(user_data_dir) except OSError as e: print(f"Error creating data directory {user_data_dir}: {e}"); user_data_dir = os.path.abspath('.') DB_FILE = os.path.join(user_data_dir, 'dvr_web_data.sqlite') # Changed extension print(f"Database location: {DB_FILE}") # --- Define Project-Relative Directories (Keep as before) --- project_root = os.path.dirname(os.path.abspath(__file__)) uploads_dir = os.path.join(project_root, "uploads") people_img_dir = os.path.join(uploads_dir, "people_images") truck_img_dir = os.path.join(uploads_dir, "truck_images") event_snippets_dir = os.path.join(uploads_dir, "event_snippets") default_recordings_dir = os.path.join(project_root, "recordings") default_backups_dir = os.path.join(project_root, "backups") # Ensure directories exist for dir_path in [uploads_dir, people_img_dir, truck_img_dir, event_snippets_dir, default_recordings_dir, default_backups_dir]: if not os.path.exists(dir_path): try: os.makedirs(dir_path) except OSError as e: print(f"Warning: Could not create directory {dir_path}: {e}") # --- Database Connection --- def connect_db(): """Connects to the SQLite database.""" conn = sqlite3.connect(DB_FILE) conn.row_factory = sqlite3.Row # Return rows as dictionary-like objects return conn # --- Initialization --- def initialize_data(): """Creates tables if they don't exist and adds default data with optimized settings.""" conn = connect_db() cursor = conn.cursor() try: # Users Table with enhanced security cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'user', full_name TEXT, last_login TIMESTAMP, failed_attempts INTEGER DEFAULT 0, last_failed_attempt TIMESTAMP, is_locked BOOLEAN DEFAULT FALSE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )''') # Videos Table with optimized settings cursor.execute(''' CREATE TABLE IF NOT EXISTS videos ( id TEXT PRIMARY KEY, name TEXT NOT NULL, path TEXT, url TEXT, telegram_enabled INTEGER DEFAULT 1, face_recognition_enabled INTEGER DEFAULT 1, suspicious_enabled INTEGER DEFAULT 0, motion_detection_enabled INTEGER DEFAULT 0, recording_enabled INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_access TIMESTAMP, frame_rate INTEGER DEFAULT 20, resolution_width INTEGER DEFAULT 1280, resolution_height INTEGER DEFAULT 720, motion_threshold INTEGER DEFAULT 30, motion_contour_area INTEGER DEFAULT 500, face_size_threshold INTEGER DEFAULT 50, recording_quality INTEGER DEFAULT 50, recording_interval INTEGER DEFAULT 5, recording_duration INTEGER DEFAULT 30 )''') # Add indexes for faster lookups cursor.execute('CREATE INDEX IF NOT EXISTS idx_video_id ON videos (id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_video_name ON videos (name)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_video_created ON videos (created_at)') # Cameras Table with optimized settings cursor.execute(''' CREATE TABLE IF NOT EXISTS cameras ( id TEXT PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL, username TEXT, password TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_access TIMESTAMP, frame_rate INTEGER DEFAULT 20, resolution_width INTEGER DEFAULT 1280, resolution_height INTEGER DEFAULT 720 )''') # People Table with optimized settings cursor.execute(''' CREATE TABLE IF NOT EXISTS people ( id TEXT PRIMARY KEY, name TEXT NOT NULL, image_path TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_detected TIMESTAMP, detection_count INTEGER DEFAULT 0 )''') # Trucks Table with optimized settings cursor.execute(''' CREATE TABLE IF NOT EXISTS trucks ( id TEXT PRIMARY KEY, number TEXT UNIQUE NOT NULL, image_path TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_detected TIMESTAMP, detection_count INTEGER DEFAULT 0 )''') # Telegram Bots Table with optimized settings cursor.execute(''' CREATE TABLE IF NOT EXISTS telegram_bots ( id TEXT PRIMARY KEY, token TEXT UNIQUE NOT NULL, chat_id TEXT NOT NULL, name TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_notification TIMESTAMP, notification_count INTEGER DEFAULT 0 )''') # Settings Table with optimized settings cursor.execute(''' CREATE TABLE IF NOT EXISTS settings ( id INTEGER PRIMARY KEY, frame_skip INTEGER DEFAULT 5, motion_threshold INTEGER DEFAULT 30, motion_contour_area INTEGER DEFAULT 500, face_size_threshold INTEGER DEFAULT 50, recording_quality INTEGER DEFAULT 50, recording_interval INTEGER DEFAULT 5, recording_duration INTEGER DEFAULT 30, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )''') # Insert default settings if not exists cursor.execute(''' INSERT OR IGNORE INTO settings (id, frame_skip, motion_threshold, motion_contour_area, face_size_threshold, recording_quality, recording_interval, recording_duration) VALUES (1, 5, 30, 500, 50, 50, 5, 30) ''') # Detection Events Table with optimized settings cursor.execute(''' CREATE TABLE IF NOT EXISTS detection_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, video_id TEXT NOT NULL, type TEXT NOT NULL, name TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, image_path TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY(video_id) REFERENCES videos(id) )''') # Create indexes for faster queries cursor.execute('CREATE INDEX IF NOT EXISTS idx_detection_video ON detection_events(video_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_detection_time ON detection_events(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_detection_type ON detection_events(type)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_detection_name ON detection_events(name)') # Add Default Data --- # Default Admin User cursor.execute("SELECT COUNT(*) FROM users WHERE username = 'admin'") if cursor.fetchone()[0] == 0: cursor.execute("INSERT INTO users (id, username, password_hash, role, full_name) VALUES (?, ?, ?, ?, ?)", (str(uuid.uuid4()), 'admin', generate_password_hash('admin'), 'admin', 'المسؤول')) # Default Settings Row cursor.execute("SELECT COUNT(*) FROM settings WHERE id = 1") if cursor.fetchone()[0] == 0: cursor.execute("INSERT INTO settings (id) VALUES (1)") # Insert default values defined in schema conn.commit() print("Database initialized successfully.") except sqlite3.Error as e: print(f"Database initialization error: {e}") conn.rollback() # Rollback changes on error finally: conn.close() # === Getters === def _get_all(table_name): """Generic function to get all rows from a table.""" try: conn = connect_db() cursor = conn.cursor() cursor.execute(f"SELECT * FROM {table_name}") rows = [dict(row) for row in cursor.fetchall()] # Convert to list of dicts conn.close() return rows except sqlite3.Error as e: print(f"Error getting all from {table_name}: {e}") return [] def _get_by_id(table_name, item_id): """Generic function to get a single row by ID.""" try: conn = connect_db() cursor = conn.cursor() cursor.execute(f"SELECT * FROM {table_name} WHERE id = ?", (item_id,)) row = cursor.fetchone() conn.close() return dict(row) if row else None except sqlite3.Error as e: print(f"Error getting {item_id} from {table_name}: {e}") return None def get_users(): return _get_all('users') def get_user_by_id(user_id): return _get_by_id('users', user_id) def get_videos(): return _get_all('videos') def get_video_by_id(video_id): return _get_by_id('videos', video_id) def get_cameras(): return _get_all('cameras') def get_camera_by_id(cam_id): return _get_by_id('cameras', cam_id) def get_people(): return _get_all('people') def get_person_by_id(person_id): return _get_by_id('people', person_id) def get_trucks(): return _get_all('trucks') def get_truck_by_id(truck_id): return _get_by_id('trucks', truck_id) def get_telegram_bots(): return _get_all('telegram_bots') def get_telegram_bot_by_id(bot_id): return _get_by_id('telegram_bots', bot_id) def get_user_by_username(username): try: conn = connect_db() cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE username = ?", (username,)) row = cursor.fetchone() conn.close() return dict(row) if row else None except sqlite3.Error as e: print(f"Error getting user by username {username}: {e}") return None def get_settings(): try: conn = connect_db() cursor = conn.cursor() cursor.execute("SELECT * FROM settings WHERE id = 1") row = cursor.fetchone() conn.close() if row: return dict(row) else: # Should not happen if initialize_data worked, but return defaults as fallback print("Warning: Settings row not found, returning defaults.") return { 'id': 1, 'use_cpu': 1, 'use_gpu': 1, 'auto_delete_days': 30, 'recording_path': 'recordings', 'live_view_grid_cols': 4, 'notification_timeout': 5000, 'frame_skip': 5 } except sqlite3.Error as e: print(f"Error getting settings: {e}") return {} # Return empty dict or defaults on error # === Setters / Adders / Updaters / Deleters === def save_settings(settings_data): """Updates the single settings row.""" # Convert boolean-like values if needed (e.g., from forms) use_cpu = 1 if settings_data.get('use_cpu') in [True, 'true', 'on', 1] else 0 use_gpu = 1 if settings_data.get('use_gpu') in [True, 'true', 'on', 1] else 0 try: auto_delete_days = int(settings_data.get('auto_delete_days', 30)) live_view_grid_cols = int(settings_data.get('live_view_grid_cols', 4)) notification_timeout = int(settings_data.get('notification_timeout', 5000)) frame_skip = int(settings_data.get('frame_skip', 5)) recording_path = settings_data.get('recording_path', 'recordings') # Validate path if '..' in recording_path or os.path.isabs(recording_path): print(f"Warning: Invalid recording path '{recording_path}'. Using default 'recordings'.") recording_path = 'recordings' conn = connect_db() cursor = conn.cursor() cursor.execute(''' UPDATE settings SET use_cpu = ?, use_gpu = ?, auto_delete_days = ?, recording_path = ?, live_view_grid_cols = ?, notification_timeout = ?, frame_skip = ? WHERE id = 1 ''', (use_cpu, use_gpu, auto_delete_days, recording_path, live_view_grid_cols, notification_timeout, frame_skip)) conn.commit() conn.close() return True except (sqlite3.Error, ValueError, TypeError) as e: print(f"Error saving settings: {e}") return False def add_user(user_data): user_id = str(uuid.uuid4()) hashed_pw = generate_password_hash(user_data['password']) role = user_data.get('role', 'user') full_name = user_data.get('full_name', user_data['username']) try: conn = connect_db() cursor = conn.cursor() cursor.execute("INSERT INTO users (id, username, password_hash, role, full_name) VALUES (?, ?, ?, ?, ?)", (user_id, user_data['username'], hashed_pw, role, full_name)) conn.commit() conn.close() # Return the added user data (without hash) return {'id': user_id, 'username': user_data['username'], 'role': role, 'full_name': full_name} except sqlite3.IntegrityError: # Handles UNIQUE constraint violation for username print(f"Error adding user: Username '{user_data['username']}' already exists.") conn.close() return "username_exists" except sqlite3.Error as e: print(f"Error adding user {user_data.get('username', 'N/A')}: {e}") conn.close() return False def update_user(user_id, user_data): try: conn = connect_db() cursor = conn.cursor() fields = [] params = [] if 'full_name' in user_data: fields.append("full_name = ?") params.append(user_data['full_name']) if 'role' in user_data: fields.append("role = ?") params.append(user_data['role']) if 'password' in user_data and user_data['password']: fields.append("password_hash = ?") params.append(generate_password_hash(user_data['password'])) if not fields: return None # Nothing to update params.append(user_id) sql = f"UPDATE users SET {', '.join(fields)} WHERE id = ?" cursor.execute(sql, tuple(params)) updated_rows = cursor.rowcount conn.commit() conn.close() return get_user_by_id(user_id) if updated_rows > 0 else None except sqlite3.Error as e: print(f"Error updating user {user_id}: {e}") return None def delete_item(table_name, item_id): """Generic function to delete an item by ID.""" try: conn = connect_db() cursor = conn.cursor() cursor.execute(f"DELETE FROM {table_name} WHERE id = ?", (item_id,)) deleted_rows = cursor.rowcount conn.commit() conn.close() return deleted_rows > 0 except sqlite3.Error as e: print(f"Error deleting {item_id} from {table_name}: {e}") return False def delete_user(user_id): return delete_item('users', user_id) def delete_video(video_id): return delete_item('videos', video_id) def delete_camera(cam_id): return delete_item('cameras', cam_id) def delete_telegram_bot(bot_id): return delete_item('telegram_bots', bot_id) # Specific delete functions for items with associated files def delete_person(person_id): person = get_person_by_id(person_id) if delete_item('people', person_id): if person and 'image_path' in person and person['image_path']: try: base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) full_image_path = os.path.join(base_dir, 'uploads', person['image_path']) if os.path.exists(full_image_path): os.remove(full_image_path); print(f"Deleted image: {full_image_path}") except Exception as img_e: print(f"Error deleting image for person {person_id}: {img_e}") return True return False def delete_truck(truck_id): truck = get_truck_by_id(truck_id) if delete_item('trucks', truck_id): if truck and 'image_path' in truck and truck['image_path']: try: base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) full_image_path = os.path.join(base_dir, 'uploads', truck['image_path']) if os.path.exists(full_image_path): os.remove(full_image_path); print(f"Deleted image: {full_image_path}") except Exception as img_e: print(f"Error deleting image for truck {truck_id}: {img_e}") return True return False # --- Add/Update for other models --- def add_video(video_data): video_id = str(uuid.uuid4()) name = video_data.get('name', video_data.get('path') or video_data.get('url', 'فيديو غير مسمى')) path = video_data.get('path') url = video_data.get('url') # Convert booleans to 0/1 for SQLite tg_enabled = 1 if video_data.get('telegram_enabled', True) else 0 face_enabled = 1 if video_data.get('face_recognition_enabled', True) else 0 susp_enabled = 1 if video_data.get('suspicious_enabled', False) else 0 motion_enabled = 1 if video_data.get('motion_detection_enabled', False) else 0 try: conn = connect_db() cursor = conn.cursor() cursor.execute('''INSERT INTO videos (id, name, path, url, telegram_enabled, face_recognition_enabled, suspicious_enabled, motion_detection_enabled) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''', (video_id, name, path, url, tg_enabled, face_enabled, susp_enabled, motion_enabled)) conn.commit() conn.close() return get_video_by_id(video_id) # Fetch the inserted row except sqlite3.Error as e: print(f"Error adding video {name}: {e}"); return None def update_video(video_id, updated_data): """Updates specific fields for a video, handling boolean conversion.""" if not updated_data: return None # Or False? Indicate no change or failure? # Define boolean fields that need conversion to integer (0 or 1) boolean_fields = [ 'telegram_enabled', 'face_recognition_enabled', 'suspicious_enabled', 'motion_detection_enabled', 'recording_enabled' # Add recording_enabled here! ] update_values = [] set_parts = [] for key, value in updated_data.items(): # Add validation or sanitation if needed for other fields like 'name' if key in boolean_fields: # Convert Python boolean or integer 1/0 to integer 1 or 0 db_value = 1 if value in [True, 1, '1', 'true', 'on'] else 0 set_parts.append(f"{key} = ?") update_values.append(db_value) elif key == 'name': # Example for handling other fields set_parts.append(f"{key} = ?") update_values.append(str(value) if value is not None else None) # Ensure name is string # Add elif for other valid fields like path, url if they become editable # else: print(f"Warning: Attempted to update invalid or non-editable field '{key}' in videos table.") if not set_parts: # If no valid fields were found to update print(f"Warning: No valid fields provided to update for video {video_id}") return None set_clause = ", ".join(set_parts) update_values.append(video_id) # Add video_id for the WHERE clause conn = connect_db() cursor = conn.cursor() try: query = f"UPDATE videos SET {set_clause} WHERE id = ?" # print(f"Executing SQL: {query} with values: {tuple(update_values)}") # Debug print cursor.execute(query, tuple(update_values)) conn.commit() rows_affected = cursor.rowcount if rows_affected > 0: updated_video = _get_by_id('videos', video_id) # Use internal getter conn.close() return updated_video else: conn.close() print(f"Video with ID {video_id} not found for update.") return None # Video ID not found except sqlite3.Error as e: print(f"Database error updating video {video_id}: {e}") conn.rollback() conn.close() return False # Indicate database error except Exception as e: # Catch other potential errors print(f"Unexpected error updating video {video_id}: {e}") if conn: conn.rollback(); conn.close() return False def add_camera(cam_data): cam_id = str(uuid.uuid4()) try: conn = connect_db() cursor = conn.cursor() cursor.execute("INSERT INTO cameras (id, name, url, username, password) VALUES (?, ?, ?, ?, ?)", (cam_id, cam_data['name'], cam_data['url'], cam_data.get('username'), cam_data.get('password'))) conn.commit() conn.close() return get_camera_by_id(cam_id) except sqlite3.Error as e: print(f"Error adding camera {cam_data.get('name', 'N/A')}: {e}"); return None def update_camera(cam_id, updated_data): try: conn = connect_db() cursor = conn.cursor() fields = [] params = [] for key in ['name', 'url', 'username', 'password']: if key in updated_data: fields.append(f"{key} = ?") params.append(updated_data[key]) if not fields: return None params.append(cam_id) sql = f"UPDATE cameras SET {', '.join(fields)} WHERE id = ?" cursor.execute(sql, tuple(params)) updated_rows = cursor.rowcount conn.commit() conn.close() return get_camera_by_id(cam_id) if updated_rows > 0 else None except sqlite3.Error as e: print(f"Error updating camera {cam_id}: {e}"); return None def add_person(person_data): person_id = str(uuid.uuid4()) try: conn = connect_db() cursor = conn.cursor() cursor.execute("INSERT INTO people (id, name, image_path) VALUES (?, ?, ?)", (person_id, person_data['name'], person_data.get('image_path'))) conn.commit() conn.close() return get_person_by_id(person_id) except sqlite3.IntegrityError: # Handles UNIQUE constraint violation for image_path? (If added) print(f"Error adding person: Image path '{person_data.get('image_path')}' might already exist.") conn.close(); return None except sqlite3.Error as e: print(f"Error adding person {person_data.get('name', 'N/A')}: {e}"); conn.close(); return None def update_person(person_id, updated_data): # Image update needs separate handling (upload new, delete old, update path) try: conn = connect_db() cursor = conn.cursor() fields = [] params = [] if 'name' in updated_data: fields.append("name = ?"); params.append(updated_data['name']) if 'image_path' in updated_data: fields.append("image_path = ?"); params.append(updated_data['image_path']) if not fields: return None params.append(person_id) sql = f"UPDATE people SET {', '.join(fields)} WHERE id = ?" cursor.execute(sql, tuple(params)) updated_rows = cursor.rowcount conn.commit() conn.close() return get_person_by_id(person_id) if updated_rows > 0 else None except sqlite3.Error as e: print(f"Error updating person {person_id}: {e}"); return None def add_truck(truck_data): truck_id = str(uuid.uuid4()) try: conn = connect_db() cursor = conn.cursor() cursor.execute("INSERT INTO trucks (id, number, type, driver_name, driver_phone, model, identification_number, image_path) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (truck_id, truck_data['number'], truck_data.get('type'), truck_data.get('driver_name'), truck_data.get('driver_phone'), truck_data.get('model'), truck_data.get('identification_number'), truck_data.get('image_path'))) conn.commit() conn.close() return get_truck_by_id(truck_id) except sqlite3.IntegrityError: # Handles UNIQUE constraint violation for number print(f"Error adding truck: Number '{truck_data.get('number')}' already exists.") conn.close(); return None except sqlite3.Error as e: print(f"Error adding truck {truck_data.get('number', 'N/A')}: {e}"); conn.close(); return None def update_truck(truck_id, updated_data): # Image update needs separate handling try: conn = connect_db() cursor = conn.cursor() fields = [] params = [] for key in ['number', 'type', 'driver_name', 'driver_phone', 'model', 'identification_number', 'image_path']: if key in updated_data: fields.append(f"{key} = ?") params.append(updated_data[key]) if not fields: return None params.append(truck_id) sql = f"UPDATE trucks SET {', '.join(fields)} WHERE id = ?" cursor.execute(sql, tuple(params)) updated_rows = cursor.rowcount conn.commit() conn.close() return get_truck_by_id(truck_id) if updated_rows > 0 else None except sqlite3.IntegrityError: print(f"Error updating truck {truck_id}: Truck number '{updated_data.get('number')}' might already exist.") return None except sqlite3.Error as e: print(f"Error updating truck {truck_id}: {e}"); return None def add_telegram_bot(bot_data): bot_id = str(uuid.uuid4()) try: conn = connect_db() cursor = conn.cursor() cursor.execute("INSERT INTO telegram_bots (id, token, chat_id) VALUES (?, ?, ?)", (bot_id, bot_data['token'], bot_data['chat_id'])) conn.commit() conn.close() return get_telegram_bot_by_id(bot_id) except sqlite3.Error as e: print(f"Error adding telegram bot: {e}"); return None def update_telegram_bot(bot_id, updated_data): try: conn = connect_db() cursor = conn.cursor() fields = [] params = [] if 'token' in updated_data: fields.append("token = ?"); params.append(updated_data['token']) if 'chat_id' in updated_data: fields.append("chat_id = ?"); params.append(updated_data['chat_id']) if not fields: return None params.append(bot_id) sql = f"UPDATE telegram_bots SET {', '.join(fields)} WHERE id = ?" cursor.execute(sql, tuple(params)) updated_rows = cursor.rowcount conn.commit() conn.close() return get_telegram_bot_by_id(bot_id) if updated_rows > 0 else None except sqlite3.Error as e: print(f"Error updating telegram bot {bot_id}: {e}"); return None # === Detection Events for Reports === def add_detection_event(event_data): event_id = str(uuid.uuid4()) timestamp = time.time() try: conn = connect_db() cursor = conn.cursor() cursor.execute('''INSERT INTO detection_events (id, timestamp, type, name, video_id, video_name, image_snippet) VALUES (?, ?, ?, ?, ?, ?, ?)''', (event_id, timestamp, event_data['type'], event_data['name'], event_data['video_id'], event_data['video_name'], event_data.get('image_snippet'))) conn.commit() conn.close() return True except sqlite3.Error as e: print(f"Error adding detection event: {e}") return False def get_detection_events(start_time=None, end_time=None, event_type=None, specific_name=None): try: conn = connect_db() cursor = conn.cursor() sql = "SELECT id, timestamp, type, name, video_id, video_name, image_snippet FROM detection_events" conditions = [] params = [] if start_time: conditions.append("timestamp >= ?") params.append(start_time) if end_time: conditions.append("timestamp <= ?") params.append(end_time) if event_type and event_type in ['face', 'truck']: conditions.append("type = ?") params.append(event_type) if specific_name: conditions.append("name LIKE ?") params.append(f"%{specific_name}%") if conditions: sql += " WHERE " + " AND ".join(conditions) sql += " ORDER BY timestamp DESC" # Newest first cursor.execute(sql, tuple(params)) rows = [dict(row) for row in cursor.fetchall()] conn.close() return rows except sqlite3.Error as e: print(f"Error getting detection events: {e}") return [] print("Database module (SQLite) loaded.") # initialize_data() # Call once at app startup models.py from flask_login import UserMixin from werkzeug.security import generate_password_hash, check_password_hash import database # Imports the updated database.py (using SQLite) import sqlite3 # Import sqlite3 for error handling class User(UserMixin): def __init__(self, id, username, password_hash, role='user', full_name=None): # Added full_name self.id = id self.username = username self.password_hash = password_hash self.role = role self.full_name = full_name if full_name else username # Default full_name to username @staticmethod def get(user_id): conn = None # Initialize conn to None try: conn = database.connect_db() # Use the correct function name cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,)) user_data = cursor.fetchone() # Fetch one row if user_data: # user_data is a sqlite3.Row object, access by column name return User(id=user_data['id'], username=user_data['username'], password_hash=user_data['password_hash'], role=user_data['role'], full_name=user_data['full_name']) # Added full_name return None except sqlite3.Error as e: print(f"Database error in User.get: {e}") return None finally: if conn: conn.close() # Ensure connection is closed @staticmethod def get_by_username(username): conn = None # Initialize conn to None try: conn = database.connect_db() # Use the correct function name cursor = conn.cursor() cursor.execute("SELECT * FROM users WHERE username = ?", (username,)) user_data = cursor.fetchone() # Fetch one row if user_data: # user_data is a sqlite3.Row object, access by column name return User(id=user_data['id'], username=user_data['username'], password_hash=user_data['password_hash'], role=user_data['role'], full_name=user_data['full_name']) # Added full_name return None except sqlite3.Error as e: print(f"Database error in User.get_by_username: {e}") return None finally: if conn: conn.close() # Ensure connection is closed def check_password(self, password): # Ensure password_hash is not None before checking if self.password_hash is None: return False return check_password_hash(self.password_hash, password) # No changes needed for other models if they are managed as dicts via database.py video_processing.py import cv2 import time import threading import numpy as np from PIL import Image, ImageDraw, ImageFont import face_recognition import arabic_reshaper from bidi.algorithm import get_display import os import database # Import the database module import requests # Ensure requests is imported if needed later for OCR/API calls import datetime # Needed for recording filenames import re import flask # --- Constants --- VIDEO_BOX_WIDTH = 320 VIDEO_BOX_HEIGHT = 240 FACE_SCALE = 0.5 # FRAME_SKIP = 5 # Now fetched from settings RECORDINGS_FOLDER = 'uploads/recordings' # Define folder for recordings # --- Global Variables --- video_streams = {} known_face_encodings = [] known_face_names = [] known_truck_numbers = [] # <<< Re-enable truck list >>> face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml') processing_lock = threading.Lock() recognition_lock = threading.Lock() # <<< Add a new lock specifically for recognition computation >>> # --- Font Loading --- ARABIC_FONT = None FONT_NAME = "Amiri-Regular.ttf" try: # Construct the path relative to the script's directory script_dir = os.path.dirname(__file__) font_path = os.path.join(script_dir, 'static', 'fonts', FONT_NAME) if os.path.exists(font_path): ARABIC_FONT = ImageFont.truetype(font_path, 15) print(f"Successfully loaded Arabic font: {font_path}") else: print(f"!!! Warning: Arabic font file not found at: {font_path}") print("!!! Warning: Please ensure 'Amiri-Regular.ttf' exists in the 'static/fonts' directory.") print("!!! Warning: Using default PIL font, Arabic text will not render correctly.") ARABIC_FONT = ImageFont.load_default() except Exception as e: print(f"!!! Error loading font: {e}. Using default.") ARABIC_FONT = ImageFont.load_default() # --- Data Loading --- def load_known_data(): global known_face_encodings, known_face_names print("--- Starting load_known_data --- Accessing DB...") people = database.get_people() if people is None: # Check if DB query failed print("!!! Error: Failed to retrieve people from database in load_known_data.") return print(f"Found {len(people)} people in database.") new_encodings = [] new_names = [] # Define the base upload folder path - Assuming UPLOAD_FOLDER is relative to project root # project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # upload_dir = os.path.join(project_root, 'uploads') # Simpler: Assume uploads is relative to current working directory upload_dir = os.path.abspath('uploads') print(f"Using upload directory: {upload_dir}") for person in people: name = person.get('name') image_path_relative = person.get('image_path') # e.g., people_images/name_ts.jpg if not name or not image_path_relative: print(f"Skipping person with missing name or image path: {person.get('id')}") continue # Construct the full path to the image file image_path_full = os.path.join(upload_dir, image_path_relative) # Replace backslashes just in case for consistency, though os.path.join handles it image_path_full = image_path_full.replace('\\', '/') print(f"Processing {name} - Image path: {image_path_full}") if not os.path.exists(image_path_full): print(f"!!! Warning: Image file not found for {name} at {image_path_full}. Skipping.") continue try: print(f"Loading image for {name}...") known_image = face_recognition.load_image_file(image_path_full) print(f"Generating encoding for {name}...") # Get face encodings - use [0] assuming only one face per image face_encodings = face_recognition.face_encodings(known_image) if face_encodings: encoding = face_encodings[0] new_encodings.append(encoding) new_names.append(name) print(f"Successfully generated encoding for {name}.") else: print(f"!!! Warning: No face found in the image for {name} at {image_path_full}. Skipping.") except Exception as e: print(f"!!! Error processing image for {name} at {image_path_full}: {e}") # Update global lists safely using the lock with processing_lock: print(f"--- Finished load_known_data: Loaded {len(known_face_names)} known faces ---") # Load known data initially when the module loads load_known_data() # <<< Re-enable truck data loading function >>> def load_known_trucks(): global known_truck_numbers print("--- Starting load_known_trucks --- Accessing DB...") try: trucks = database.get_trucks() if trucks is None: print("!!! Error: Failed to retrieve trucks from database in load_known_trucks.") # Ensure list is empty on failure with processing_lock: known_truck_numbers = [] return print(f"Found {len(trucks)} trucks in database.") new_truck_numbers = [] for truck in trucks: number = truck.get('number') if number: new_truck_numbers.append(str(number)) # Ensure it's a string print(f"Loaded truck number: {number}") else: print(f"Skipping truck with missing number: {truck.get('id')}") # Update global list safely using the lock with processing_lock: known_truck_numbers = new_truck_numbers print(f"--- Finished load_known_trucks: Loaded {len(known_truck_numbers)} known truck numbers ---") except Exception as e: print(f"!!! Exception in load_known_trucks: {e}") # Ensure list is empty on major exception with processing_lock: known_truck_numbers = [] # Load known data initially when the module loads load_known_trucks() # <<< Re-enable truck loading call >>> # --- Telegram Bot Handling Placeholder --- # ... # --- Video Stream Class --- class VideoStream: def __init__(self, video_config): self.video_config = video_config self.source = video_config.get('url') or os.path.abspath(os.path.join('uploads', video_config.get('path', ''))) if not self.source: print(f"[{video_config.get('id')}] Error: No valid source (URL or Path) provided in config.") self.cap = None return self.video_id = video_config.get('id') try: self.cap = cv2.VideoCapture(self.source) if not self.cap.isOpened(): print(f"[{self.video_id}] Error: Failed to open video source: {self.source}") self.cap = None return except Exception as e: print(f"[{self.video_id}] Error opening video source: {str(e)}") self.cap = None return # Memory management self.frame_buffer = [] self.max_buffer_size = 10 # Maximum frames to keep in memory # Performance optimization self.frame_skip = 5 # Skip frames to improve performance self.last_frame_time = time.time() self.fps = 0 self.is_running = False self.thread = None self.output_frame = None self.lock = threading.Lock() self.frame_count = 0 self.overlay_data = [] self.last_sent_notification = {} # Thread management self.processing_thread = None self.processing_queue = queue.Queue(maxsize=5) self.processing_lock = threading.Lock() # --- Recording State Variables - Added --- self.is_recording = False self.video_writer = None self.recording_filename = None # --- Motion Detection Settings --- self.motion_threshold = 30 self.motion_contour_area = 500 # --- Face Detection Settings --- self.face_size_threshold = 50 # --- Telegram Settings --- self.telegram_bot = None self.telegram_chat_id = None self.telegram_enabled = False # --- Feature Control Flags --- self.face_recognition_enabled = False self.truck_recognition_enabled = False self.notifications_enabled = False print(f"[{self.video_id}] Initializing stream for source: {self.source}") try: # Use the determined self.source self.cap = cv2.VideoCapture(self.source) if not self.cap or not self.cap.isOpened(): print(f"[{self.video_id}] Error: Could not open video source '{self.source}'. Check path and permissions.") self.cap = None # Ensure cap is None if failed else: print(f"[{self.video_id}] Successfully opened video source: {self.source}") except Exception as e: print(f"[{self.video_id}] Exception opening video source '{self.source}': {e}") self.cap = None def update_settings(self, new_settings): with self.lock: self.face_rec_enabled = new_settings.get('face_recognition_enabled', self.video_config.get('face_recognition_enabled', False)) self.telegram_enabled = new_settings.get('telegram_enabled', self.video_config.get('telegram_enabled', False)) self.motion_enabled = new_settings.get('motion_detection_enabled', self.video_config.get('motion_detection_enabled', False)) # Update frame skip if global setting changed (passed via API) if 'frame_skip' in new_settings: self.frame_skip = new_settings.get('frame_skip', 5) print(f"[{self.video_id}] Frame skip updated to: {self.frame_skip}") # Update internal config dict self.video_config.update(new_settings) def start(self): if not self.cap or not self.cap.isOpened(): print(f"[{self.video_id}] Cannot start stream, capture failed.") return False if not self.is_running: self.is_running = True self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() print(f"Started video stream thread for {self.video_id}") return True return False def stop(self): print(f"[{self.video_id}] Signaling stop...") # Stop recording if active if self.is_recording: self.stop_recording() self.is_running = False # Release the capture immediately, don't wait for the thread here if self.cap and self.cap.isOpened(): try: self.cap.release(); print(f"[{self.video_id}] Capture released.") except Exception as e: print(f"[{self.video_id}] Exception releasing capture: {e}") else: print(f"[{self.video_id}] Capture not open or already released during stop signal.") print(f"[{self.video_id}] Stop signal sent.") def _run(self): print(f"[{self.video_id}] Stream thread started for {self.source}") frame_count = 0 last_frame_time = time.time() while self.is_running: if not self.cap or not self.cap.isOpened(): print(f"[{self.video_id}] Stream closed unexpectedly") break ret, frame = self.cap.read() if not ret: print(f"[{self.video_id}] Failed to read frame") break frame_count += 1 current_time = time.time() elapsed = current_time - last_frame_time fps = frame_count / elapsed # Apply frame skip with optimized parameters if frame_count % (self.frame_skip + 1) != 0: # +1 because frame_count starts from 1 continue # Process frame with lock and optimized parameters with self.lock: # Resize frame for processing frame = cv2.resize(frame, (VIDEO_BOX_WIDTH, VIDEO_BOX_HEIGHT)) # Process motion detection with optimized parameters if self.motion_detection_enabled: gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) fgmask = self.motion_detector.apply(gray) # Apply morphological operations to improve detection kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (3, 3)) fgmask = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel) contours, _ = cv2.findContours(fgmask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) for contour in contours: if cv2.contourArea(contour) > self.motion_contour_area: (x, y, w, h) = cv2.boundingRect(contour) cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2) # Send telegram notification with optimized parameters if self.telegram_enabled: self._handle_detection_notification( "حركة مشبوهة", "حركة", frame, quality=50 ) # Process face recognition with optimized parameters if self.face_recognition_enabled: small_frame = cv2.resize(frame, (0, 0), fx=FACE_SCALE, fy=FACE_SCALE) rgb_small_frame = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB) with recognition_lock: # Protect face recognition computation face_locations = face_recognition.face_locations(rgb_small_frame) if face_locations: face_encodings = face_recognition.face_encodings(rgb_small_frame, face_locations) for (top, right, bottom, left), face_encoding in zip(face_locations, face_encodings): matches = face_recognition.compare_faces(known_face_encodings, face_encoding) name = "غير معروف" if True in matches: first_match_index = matches.index(True) name = known_face_names[first_match_index] # Scale back up face locations since the frame we detected in was scaled to 1/4 size top *= int(1/FACE_SCALE) right *= int(1/FACE_SCALE) bottom *= int(1/FACE_SCALE) left *= int(1/FACE_SCALE) # Draw rectangle and name cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 255), 2) cv2.rectangle(frame, (left, bottom - 35), (right, bottom), (0, 0, 255), cv2.FILLED) # Draw name with Arabic support reshaped_name = arabic_reshaper.reshape(name) bidi_name = get_display(reshaped_name) # Draw name with Arabic font if ARABIC_FONT: # Create a blank image for text txt_img = Image.new('RGB', (right - left, 35), (0, 0, 255)) draw = ImageDraw.Draw(txt_img) # Calculate text size txt_size = draw.textsize(bidi_name, font=ARABIC_FONT) # Draw text centered draw.text(((right - left - txt_size[0]) // 2, 0), bidi_name, font=ARABIC_FONT, fill=(255, 255, 255)) # Convert to OpenCV format txt_np = np.array(txt_img) txt_np = cv2.cvtColor(txt_np, cv2.COLOR_RGB2BGR) # Copy text onto frame frame[bottom - 35:bottom, left:right] = txt_np else: # Fallback to OpenCV text if Arabic font fails cv2.putText(frame, bidi_name, (left + 6, bottom - 6), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1) # Send telegram notification with optimized parameters if self.telegram_enabled: face_crop = frame[top:bottom, left:right] self._handle_detection_notification( name, "وجه", face_crop, quality=50 ) # Process truck recognition with optimized parameters if self.truck_recognition_enabled: # TODO: Implement truck recognition pass # Add FPS counter with optimized parameters cv2.putText(frame, f"FPS: {fps:.1f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1) # Add recording status indicator with optimized parameters if self.is_recording: cv2.putText(frame, "تسجيل مفعل", (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1) # Convert frame to JPEG with optimized parameters ret, jpeg = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) if ret: self.output_frame = jpeg.tobytes() # Sleep to maintain frame rate with optimized parameters time.sleep(0.03) # Adjust as needed pil_im = None draw = None with self.lock: # Access shared overlay_data safely overlays_to_draw = self.overlay_data[:] # Make a copy for ((x, y, w, h), name, obj_type) in overlays_to_draw: color = (0, 255, 0) if name != "مجهول" else (0, 0, 255) if obj_type == 'truck': color = (255, 0, 0) # Keep truck color logic just in case # Draw rectangle using OpenCV (simpler) cv2.rectangle(processed_frame, (x, y), (x + w, y + h), color, 2) # Draw text using PIL for Arabic support try: if pil_im is None: # Initialize PIL objects only if needed pil_im = Image.fromarray(cv2.cvtColor(processed_frame, cv2.COLOR_BGR2RGB)) draw = ImageDraw.Draw(pil_im) reshaped_text = arabic_reshaper.reshape(name) bidi_text = get_display(reshaped_text) text_y = y - 18 if y > 18 else y + 10 # Adjust text position draw.text((x + 6, text_y), bidi_text, font=ARABIC_FONT, fill=color[::-1]) # Use BGR color except Exception as text_e: # Fallback to OpenCV text drawing if PIL fails print(f"!!! Error drawing text with PIL for {name}: {text_e}") fallback_text = "Text Error" cv2.putText(processed_frame, fallback_text, (x, y - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2) # Convert back from PIL if text was drawn if pil_im is not None: processed_frame = cv2.cvtColor(np.array(pil_im), cv2.COLOR_RGB2BGR) # <<< Add Recording Logic >>> # Write the *processed* frame (with overlays) if recording is active if self.is_recording and self.video_writer is not None: try: # Ensure frame dimensions match VideoWriter dimensions writer_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)) # Re-get in case of change? Safer to use init value? writer_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) # Resize processed_frame if needed (though ideally it should match) if processed_frame.shape[1] != writer_width or processed_frame.shape[0] != writer_height: frame_to_write = cv2.resize(processed_frame, (writer_width, writer_height)) else: frame_to_write = processed_frame self.video_writer.write(frame_to_write) except Exception as write_e: print(f"[{self.video_id}] Error writing frame to {self.recording_filename}: {write_e}") # Consider stopping recording on write error? # self.stop_recording() # <<< End of Recording Logic >>> # --- Update Output Frame --- (This part had an error in the attached file, was copying processed_frame instead of encoding) # Replace: # with self.lock: self.output_frame = processed_frame.copy() # With correct encoding: resized_frame = cv2.resize(processed_frame, (VIDEO_BOX_WIDTH, VIDEO_BOX_HEIGHT)) ret, buffer = cv2.imencode('.jpg', resized_frame, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) if ret: with self.lock: self.output_frame = buffer.tobytes() else: print(f"[{self.video_id}] Error encoding frame.") # --- End of loop --- (Keep sleep) time.sleep(0.01) # <<< Add explicit deletion of large/intermediate objects >>> try: del small_frame del rgb_small_frame del face_locations del face_encodings # del processed_frame # Keep this as it might be the base for next loop if PIL wasn't used del resized_frame del buffer if 'pil_im' in locals(): del pil_im if 'draw' in locals(): del draw except NameError: # In case some variables weren't assigned due to errors/conditions pass except Exception as del_e: print(f"[{self.video_id}] Error during explicit deletion: {del_e}") # <<< Add print statement to confirm loop exit >>> print(f"[{self.video_id}] Video stream processing loop exited cleanly.") # --- Cleanup --- if self.cap and self.cap.isOpened(): try: self.cap.release(); print(f"[{self.video_id}] Capture released on exit.") except Exception as e: print(f"[{self.video_id}] Exception releasing capture on exit: {e}") print(f"Video stream thread finished for {self.video_id}") def _handle_detection_notification(self, name, obj_type, image_crop): """ Sends notifications and logs event if enabled and cooldown passed """ import re identifier = "unknown_face" if name == "مجهول" and obj_type == 'face' else f"{obj_type}_{name}" current_time = time.time() last_sent = self.last_sent_notification.get(identifier, 0) cooldown = 300 # 5 minutes cooldown def enable_face_recognition(self): """Enable face recognition feature.""" self.face_recognition_enabled = True print(f"[{self.video_id}] Face recognition enabled") def disable_face_recognition(self): """Disable face recognition feature.""" self.face_recognition_enabled = False print(f"[{self.video_id}] Face recognition disabled") def enable_truck_recognition(self): """Enable truck recognition feature.""" self.truck_recognition_enabled = True print(f"[{self.video_id}] Truck recognition enabled") def disable_truck_recognition(self): """Disable truck recognition feature.""" self.truck_recognition_enabled = False print(f"[{self.video_id}] Truck recognition disabled") def enable_notifications(self): """Enable notifications feature.""" self.notifications_enabled = True print(f"[{self.video_id}] Notifications enabled") def disable_notifications(self): """Disable notifications feature.""" self.notifications_enabled = False print(f"[{self.video_id}] Notifications disabled") if current_time - last_sent > cooldown: print(f"[{self.video_id}] Cooldown passed for {identifier}. Processing detection.") self.last_sent_notification[identifier] = current_time # --- Save Snippet for Report --- snippet_filename = None snippet_relative_path = None if image_crop is not None: try: # Check if image_crop is a numpy array (cv2 image) import numpy as np if not isinstance(image_crop, np.ndarray): print(f"[{self.video_id}] Warning: image_crop is not a numpy array, cannot save snippet.") raise ValueError("image_crop is not a numpy array") if image_crop.size == 0: print(f"[{self.video_id}] Warning: image_crop has zero size, skipping save.") raise ValueError("image_crop has zero size") # Get project root directory project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) event_dir = os.path.join(project_root, "uploads", "event_snippets") # Ensure event snippets directory exists try: os.makedirs(event_dir, exist_ok=True) print(f"[{self.video_id}] Created/verified event snippets directory: {event_dir}") except OSError as e: print(f"[{self.video_id}] Error creating event snippets directory: {e}") raise ts_str = time.strftime('%Y%m%d_%H%M%S') # Sanitize name for filename: replace non-alphanumeric with underscore safe_name = re.sub(r'[^a-zA-Z0-9_-]', '_', name) snippet_filename = f"{obj_type}_{safe_name}_{ts_str}_{self.video_id}.jpg" snippet_full_path = os.path.join(event_dir, snippet_filename) # Attempt to save image with multiple retries max_retries = 3 for attempt in range(max_retries): try: # Ensure we have write permissions if not os.access(os.path.dirname(snippet_full_path), os.W_OK): raise PermissionError(f"No write permission for directory: {os.path.dirname(snippet_full_path)}") # Save image using cv2.imwrite success = cv2.imwrite(snippet_full_path, image_crop) if success: snippet_relative_path = os.path.join("event_snippets", snippet_filename) print(f"[{self.video_id}] Saved event snippet: {snippet_relative_path}") break else: raise IOError(f"cv2.imwrite failed to save file at {snippet_full_path}") except Exception as save_e: if attempt == max_retries - 1: # Last attempt print(f"[{self.video_id}] Final save attempt failed: {save_e}") raise print(f"[{self.video_id}] Save attempt {attempt + 1} failed: {save_e}, retrying...") time.sleep(0.1) # Small delay before retry except Exception as e: print(f"[{self.video_id}] Error saving event snippet: {e}") # Log the error to database for debugging error_event = { 'type': 'error', 'name': 'image_save_failed', 'video_id': self.video_id, 'video_name': self.video_config.get('name', self.video_id), 'details': str(e) } database.add_detection_event(error_event) # --- Log Event to Database --- event_data = { 'type': obj_type, 'name': name, 'video_id': self.video_id, 'video_name': self.video_config.get('name', self.video_id), 'image_snippet': snippet_relative_path # Store relative path or None } database.add_detection_event(event_data) # Add timestamp inside this function # --- On-Screen Notification (via print/log for now) --- print(f"EVENT: {{'type': 'detection', 'video_id': '{self.video_id}', 'name': '{name}', 'object_type': '{obj_type}'}}") # --- Telegram Notification --- if self.telegram_enabled: print(f"[{self.video_id}] Sending Telegram notification for {name}...") message = f"اكتشاف في الفيديو {self.video_config.get('name', self.video_id)}:\nالنوع: {obj_type}\nالاسم: {name}" # Need a robust way to send this, e.g., queue or dedicated sender # notification_manager.send_telegram_with_image(message, image_crop) # --- Save Unknown Face Image (Separate from snippet) --- if name == "مجهول" and obj_type == 'face': try: base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # تعديل مسار حفظ الأشخاص المجهولين ليكون داخل uploads/people_images/مجهول unknown_folder = os.path.join(base_dir, "uploads", "people_images", "مجهول") if not os.path.exists(unknown_folder): os.makedirs(unknown_folder) ts = time.strftime('%Y%m%d_%H%M%S') unknown_filename = os.path.join(unknown_folder, f"unknown_{ts}_{self.video_id}.jpg") if image_crop.size > 0: cv2.imwrite(unknown_filename, image_crop) print(f"Saved unknown face: {unknown_filename}") except Exception as e: print(f"Error saving unknown face: {e}") def get_frame(self): with self.lock: if self.output_frame is None: # <<< Encode placeholder to JPEG bytes >>> placeholder = np.zeros((VIDEO_BOX_HEIGHT, VIDEO_BOX_WIDTH, 3), dtype=np.uint8) status_text = 'متوقف' if not self.is_running else 'جار التحميل...' color = (100, 100, 100) if not self.is_running else (200, 200, 200) cv2.putText(placeholder, status_text, (30, VIDEO_BOX_HEIGHT // 2), cv2.FONT_HERSHEY_SIMPLEX, 0.8, color, 2) ret, buffer = cv2.imencode('.jpg', placeholder) if ret: return buffer.tobytes() else: return b'' # Return empty bytes if placeholder encoding fails # <<< Remove .copy() as self.output_frame is bytes (immutable) >>> return self.output_frame # --- Recording Methods - Added --- def start_recording(self, filename): """Starts writing video frames to a file.""" if not self.cap or not self.cap.isOpened(): print(f"[{self.video_id}] Cannot start recording, video capture is not open.") return False if self.is_recording: print(f"[{self.video_id}] Already recording to {self.recording_filename}. Stop first.") return False try: # Get video properties # Note: Getting FPS can be unreliable for streams. Use a default or estimate. fps = self.cap.get(cv2.CAP_PROP_FPS) if fps <= 0: print(f"[{self.video_id}] Warning: Could not get FPS from source. Using default 20 FPS for recording.") fps = 20.0 # Default FPS # Get frame dimensions from the source directly width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) if width <= 0 or height <= 0: print(f"[{self.video_id}] Error: Could not get valid frame dimensions from source ({width}x{height}). Cannot start recording.") return False # Define the codec and create VideoWriter object # Using MP4V for .mp4 files. Other options: 'XVID' for .avi fourcc = cv2.VideoWriter_fourcc(*'mp4v') # Ensure the recordings directory exists recordings_dir = os.path.abspath(RECORDINGS_FOLDER) if not os.path.exists(recordings_dir): try: os.makedirs(recordings_dir) print(f"Created recordings directory: {recordings_dir}") except OSError as e: print(f"[{self.video_id}] Error creating recordings directory {recordings_dir}: {e}") return False full_filename = os.path.join(recordings_dir, filename) print(f"[{self.video_id}] Starting recording. Output: {full_filename}, FPS: {fps}, Size: {width}x{height}") self.video_writer = cv2.VideoWriter(full_filename, fourcc, fps, (width, height)) if not self.video_writer.isOpened(): print(f"[{self.video_id}] Error: Could not open VideoWriter for file {full_filename}. Check permissions and codec support.") self.video_writer = None return False self.recording_filename = full_filename # Store the full path self.is_recording = True return True except Exception as e: print(f"[{self.video_id}] Exception during start_recording: {e}") if self.video_writer: self.video_writer.release(); self.video_writer = None self.is_recording = False self.recording_filename = None return False def stop_recording(self): """Stops writing video frames and releases the writer.""" if not self.is_recording: # print(f"[{self.video_id}] Not currently recording.") return False print(f"[{self.video_id}] Stopping recording for file: {self.recording_filename}") self.is_recording = False if self.video_writer: try: self.video_writer.release() print(f"[{self.video_id}] VideoWriter released successfully.") except Exception as e: print(f"[{self.video_id}] Exception releasing VideoWriter: {e}") self.video_writer = None self.recording_filename = None return True # --------------------------------- # --- Management Functions --- def start_stream(video_config): video_id = video_config.get('id') if not video_id: return None if video_id not in video_streams: print(f"Attempting to create/start stream {video_id}...") try: stream = VideoStream(video_config) if not stream.cap: return None # Init failed if stream.start(): video_streams[video_id] = stream; return stream else: stream.stop(); return None # Start failed, cleanup except Exception as e: print(f"Exception starting stream {video_id}: {e}"); return None else: # Stream exists print(f"Updating settings for existing stream {video_id}.") global_settings = database.get_settings() # Get latest global settings video_config['frame_skip'] = global_settings.get('frame_skip', 5) # Pass frame skip video_streams[video_id].update_settings(video_config) if not video_streams[video_id].is_running: print(f"Restarting stopped stream {video_id}..."); video_streams[video_id].start() return video_streams.get(video_id) # <<< Add new signal_stop_stream function >>> def signal_stop_stream(video_id): """Signals a stream to stop and returns its thread object for joining later.""" print(f"Signaling stop for stream {video_id}...") thread = None if video_id in video_streams: stream = video_streams.get(video_id) if stream: stream.stop() # Signal the loop to stop and release cap thread = stream.thread # Get the thread object else: print(f"Stream {video_id} not found for stopping.") return thread def get_stream_frame(video_id): frame_bytes = b'' # Initialize with empty bytes if video_id in video_streams: # <<< get_frame now returns bytes directly >>> frame_bytes = video_streams[video_id].get_frame() # <<< Remove redundant encoding >>> # frame = video_streams[video_id].get_frame() # ret, buffer = cv2.imencode('.jpg', frame) # if ret: return buffer.tobytes() # If frame_bytes is still empty (stream not found or get_frame failed), return empty bytes if not frame_bytes: # Placeholder logic moved to get_frame, just return empty bytes here # placeholder = np.zeros((VIDEO_BOX_HEIGHT, VIDEO_BOX_WIDTH, 3), dtype=np.uint8) # cv2.putText(placeholder, 'غير متصل', (30, VIDEO_BOX_HEIGHT // 2), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (150, 150, 150), 2) # ret, buffer = cv2.imencode('.jpg', placeholder) # return buffer.tobytes() if ret else b'' return b'' return frame_bytes def update_stream_settings(video_id, settings): if video_id in video_streams: # Include global frame skip setting if it changed global_settings = database.get_settings() settings['frame_skip'] = global_settings.get('frame_skip', 5) video_streams[video_id].update_settings(settings) print(f"Pushed settings update to stream {video_id}") else: print(f"Warning: Stream {video_id} not running for settings update.") # <<< Add Global Recording Control Functions >>> def start_recording_process(video_id, video_name): """Finds the stream and tells it to start recording with a generated filename.""" print(f"Received request to start recording process for {video_id} ({video_name})") if video_id not in video_streams: print(f"Error: Stream {video_id} not found for starting recording.") return False stream = video_streams[video_id] # Generate filename (cleaned name + timestamp) # Clean the name for filesystem use base_name = "".join(c if c.isalnum() else '_' for c in video_name) # Replace non-alphanumeric with underscore base_name = base_name.strip('_') # Remove leading/trailing underscores if not base_name: base_name = f"video_{video_id}" # Fallback name timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{base_name}_{timestamp}.mp4" # Assuming mp4 format # Call the stream's internal start recording method return stream.start_recording(filename) def stop_recording_process(video_id): """Finds the stream and tells it to stop recording.""" print(f"Received request to stop recording process for {video_id}") if video_id not in video_streams: print(f"Error: Stream {video_id} not found for stopping recording.") return False stream = video_streams[video_id] return stream.stop_recording() # <<< End of Global Recording Control Functions >>> اصلاح الاخطاء والعيوب والمشاكل وتوحيد الدوال واضافة تحسينات بطريقة اكثر احترافية والبدء بتنفيذ المشروع على 5 مراحل