fastapi uvicorn[standard] sqlalchemy pydantic python-multipart jinja2 opencv-python face_recognition python-telegram-bot reportlab pandas aiofiles python-jose[cryptography] passlib[bcrypt] import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # Determine the base directory for the database file BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATABASE_URL = f"sqlite:///{os.path.join(BASE_DIR, 'security.db')}" engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() def get_db(): db = SessionLocal() try: yield db finally: db.close() # Import models here so that Base.metadata.create_all(engine) works # from . import models # This will be done in main.py or init from sqlalchemy import Column, Integer, String, Boolean, Float, DateTime, Text, ForeignKey from sqlalchemy.orm import relationship from sqlalchemy.sql import func from database import Base class DBSettings(Base): __tablename__ = "settings" id = Column(Integer, primary_key=True, index=True) recordings_path = Column(String, default="recordings") backup_storage_days = Column(Integer, default=30) face_rec_library_version = Column(String, default="default") # Placeholder truck_rec_library_version = Column(String, default="default") # Placeholder frame_rate = Column(Integer, default=30) live_view_grid_size = Column(Integer, default=9) # e.g., 3x3 motion_detection_enabled = Column(Boolean, default=True) motion_stop_timeout = Column(Integer, default=10) # seconds old_recordings_delete_days = Column(Integer, default=30) scheduled_backup_enabled = Column(Boolean, default=False) scheduled_backup_frequency = Column(String, default="weekly") # daily, weekly scheduled_backup_time = Column(String, default="02:00") # HH:MM notify_settings_update = Column(Boolean, default=True) notify_recording_deletion = Column(Boolean, default=True) notify_low_storage = Column(Boolean, default=True) face_known_color = Column(String, default="#4CAF50") # Green face_unknown_color = Column(String, default="#FF5722") # Red box_line_thickness = Column(Integer, default=2) font_size = Column(Integer, default=16) class DBCamera(Base): __tablename__ = "cameras" id = Column(Integer, primary_key=True, index=True) name = Column(String, index=True) ip_address = Column(String, unique=True) username = Column(String) password = Column(String) # Should be encrypted in production is_connected = Column(Boolean, default=False) last_updated = Column(DateTime, server_default=func.now(), onupdate=func.now()) video_quality = Column(String, nullable=True) # e.g., 1080p signal_level = Column(Integer, nullable=True) # 0-100 # Relationship to camera control settings control_settings = relationship("DBCameraControlSettings", back_populates="camera", uselist=False) class DBVideo(Base): __tablename__ = "videos" id = Column(Integer, primary_key=True, index=True) name = Column(String, index=True) file_path = Column(String, unique=True) source_camera_id = Column(Integer, ForeignKey("cameras.id"), nullable=True) # Link to camera if recorded from one date_created = Column(DateTime, server_default=func.now()) duration_seconds = Column(Integer, nullable=True) file_size_bytes = Column(Integer, nullable=True) source_camera = relationship("DBCamera") # Relationship to camera control settings (if video has specific settings) control_settings = relationship("DBCameraControlSettings", back_populates="video", uselist=False) class DBCameraControlSettings(Base): __tablename__ = "camera_control_settings" id = Column(Integer, primary_key=True, index=True) camera_id = Column(Integer, ForeignKey("cameras.id"), unique=True, nullable=True) video_id = Column(Integer, ForeignKey("videos.id"), unique=True, nullable=True) motion_regions_json = Column(Text, default="[]") # Store list of regions as JSON string motion_recording_enabled = Column(Boolean, default=True) face_sensitivity = Column(Float, default=0.7) # 0.0 to 1.0 truck_sensitivity = Column(Float, default=0.7) # 0.0 to 1.0 camera = relationship("DBCamera", back_populates="control_settings") video = relationship("DBVideo", back_populates="control_settings") class DBTelegramBot(Base): __tablename__ = "telegram_bots" id = Column(Integer, primary_key=True, index=True) name = Column(String, index=True) token = Column(String, unique=True) # Should be encrypted username = Column(String, unique=True) # e.g., @SecurityBot is_active = Column(Boolean, default=True) last_updated = Column(DateTime, server_default=func.now(), onupdate=func.now()) # Response settings send_reports = Column(Boolean, default=True) send_recordings_notifications = Column(Boolean, default=True) send_event_notifications = Column(Boolean, default=True) send_camera_status_updates = Column(Boolean, default=True) camera_status_update_frequency_hours = Column(Integer, default=1) class DBPerson(Base): __tablename__ = "persons" id = Column(Integer, primary_key=True, index=True) name = Column(String, index=True) image_path = Column(String) # Path to the stored image identity_number = Column(String, nullable=True, unique=True) workplace = Column(String, nullable=True) telegram_notify_enabled = Column(Boolean, default=True) # Placeholder for face encoding data (e.g., stored separately or as blob) # face_encoding = Column(LargeBinary) class DBTruck(Base): __tablename__ = "trucks" id = Column(Integer, primary_key=True, index=True) driver_name = Column(String, index=True) plate_number = Column(String, unique=True) truck_type = Column(String, nullable=True) image_path = Column(String) # Path to the stored image (plate or truck front) telegram_notify_enabled = Column(Boolean, default=True) # Placeholder for truck/plate encoding data # truck_encoding = Column(LargeBinary) class DBReport(Base): __tablename__ = "reports" id = Column(Integer, primary_key=True, index=True) report_type = Column(String) # e.g., "all_events", "known_persons" start_time = Column(DateTime) end_time = Column(DateTime) generated_at = Column(DateTime, server_default=func.now()) file_path = Column(String, nullable=True) # Path to the generated report file summary = Column(Text, nullable=True) class DBUser(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String, unique=True, index=True) hashed_password = Column(String) role = Column(String, default="read") # e.g., "read", "write", "admin" last_login = Column(DateTime, nullable=True) is_active = Column(Boolean, default=True) # Placeholder for 2FA secret # two_factor_secret = Column(String, nullable=True) class DBLog(Base): __tablename__ = "logs" id = Column(Integer, primary_key=True, index=True) timestamp = Column(DateTime, server_default=func.now()) level = Column(String) # e.g., "INFO", "WARNING", "ERROR" message = Column(Text) source = Column(String, nullable=True) # e.g., "system", "camera:1", "user:admin" # Placeholder for event logs (person detected, truck detected, etc.) class DBEvent(Base): __tablename__ = "events" id = Column(Integer, primary_key=True, index=True) timestamp = Column(DateTime, server_default=func.now()) event_type = Column(String) # e.g., "person_detected", "truck_detected", "motion_detected" source_id = Column(Integer) # camera_id or video_id source_type = Column(String) # "camera" or "video" person_id = Column(Integer, ForeignKey("persons.id"), nullable=True) truck_id = Column(Integer, ForeignKey("trucks.id"), nullable=True) is_known = Column(Boolean, nullable=True) # True if person/truck is known details = Column(Text, nullable=True) # JSON string for bounding box, confidence, etc. snapshot_path = Column(String, nullable=True) # Path to snapshot image person = relationship("DBPerson") truck = relationship("DBTruck") from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime # Pydantic models for request/response data validation class SettingsBase(BaseModel): recordings_path: str = Field(default="recordings", description="مسار حفظ التسجيلات") backup_storage_days: int = Field(default=30, ge=1, description="مدة تخزين النسخ الاحتياطية بالأيام") # Placeholder for library versions - might be dynamic frame_rate: int = Field(default=30, ge=10, le=60, description="عدد الإطارات في الثانية") live_view_grid_size: int = Field(default=9, ge=1, le=32, description="عدد الخلايا في العرض المباشر") motion_detection_enabled: bool = Field(default=True, description="تفعيل التسجيل عند الحركة") motion_stop_timeout: int = Field(default=10, ge=5, le=30, description="فترة التوقف بعد توقف الحركة (ثواني)") old_recordings_delete_days: int = Field(default=30, ge=1, description="مدة حذف التسجيلات القديمة تلقائيًا (أيام)") scheduled_backup_enabled: bool = Field(default=False, description="تفعيل جدولة النسخ الاحتياطي") scheduled_backup_frequency: str = Field(default="weekly", description="تكرار النسخ الاحتياطي (daily, weekly)") scheduled_backup_time: str = Field(default="02:00", description="وقت النسخ الاحتياطي (HH:MM)") notify_settings_update: bool = Field(default=True, description="إشعارات تحديث الإعدادات") notify_recording_deletion: bool = Field(default=True, description="إشعارات حذف التسجيلات") notify_low_storage: bool = Field(default=True, description="إشعارات انخفاض مساحة التخزين") face_known_color: str = Field(default="#4CAF50", description="لون مستطيل الوجوه المعروفة") face_unknown_color: str = Field(default="#FF5722", description="لون مستطيل الوجوه الغير معروفة") box_line_thickness: int = Field(default=2, ge=1, le=5, description="سمك خطوط المستطيل") font_size: int = Field(default=16, ge=10, le=30, description="حجم الخط لاسم الشخص") class SettingsCreate(SettingsBase): pass # Only one settings entry expected class Settings(SettingsBase): id: int class Config: orm_mode = True class CameraBase(BaseModel): name: str = Field(description="اسم الكاميرا") ip_address: str = Field(description="عنوان IP") username: str = Field(description="اسم المستخدم") password: str = Field(description="كلمة المرور") # Note: Should be handled securely class CameraCreate(CameraBase): pass class CameraUpdate(CameraBase): name: Optional[str] = Field(None, description="اسم الكاميرا") ip_address: Optional[str] = Field(None, description="عنوان IP") username: Optional[str] = Field(None, description="اسم المستخدم") password: Optional[str] = Field(None, description="كلمة المرور") # Note: Should be handled securely class Camera(CameraBase): id: int is_connected: bool last_updated: datetime video_quality: Optional[str] signal_level: Optional[int] class Config: orm_mode = True class VideoBase(BaseModel): name: str = Field(description="اسم الفيديو") file_path: str = Field(description="مسار الملف") source_camera_id: Optional[int] = Field(None, description="معرف الكاميرا المصدر (إذا كان مسجل منها)") class VideoCreate(VideoBase): pass class VideoUpdate(VideoBase): name: Optional[str] = Field(None, description="اسم الفيديو") file_path: Optional[str] = Field(None, description="مسار الملف") source_camera_id: Optional[int] = Field(None, description="معرف الكاميرا المصدر (إذا كان مسجل منها)") class Video(VideoBase): id: int date_created: datetime duration_seconds: Optional[int] file_size_bytes: Optional[int] class Config: orm_mode = True class CameraControlSettingsBase(BaseModel): motion_regions_json: str = Field(default="[]", description="مناطق الحركة المحددة (JSON)") motion_recording_enabled: bool = Field(default=True, description="تفعيل التسجيل عند الحركة لهذه المصدر") face_sensitivity: float = Field(default=0.7, ge=0.0, le=1.0, description="حساسية التعرف على الوجوه") truck_sensitivity: float = Field(default=0.7, ge=0.0, le=1.0, description="حساسية التعرف على الشاحنات") class CameraControlSettingsCreate(CameraControlSettingsBase): camera_id: Optional[int] = Field(None, description="معرف الكاميرا المرتبطة") video_id: Optional[int] = Field(None, description="معرف الفيديو المرتبط") class CameraControlSettingsUpdate(CameraControlSettingsBase): motion_regions_json: Optional[str] = Field(None, description="مناطق الحركة المحددة (JSON)") motion_recording_enabled: Optional[bool] = Field(None, description="تفعيل التسجيل عند الحركة لهذه المصدر") face_sensitivity: Optional[float] = Field(None, ge=0.0, le=1.0, description="حساسية التعرف على الوجوه") truck_sensitivity: Optional[float] = Field(None, ge=0.0, le=1.0, description="حساسية التعرف على الشاحنات") class CameraControlSettings(CameraControlSettingsBase): id: int camera_id: Optional[int] video_id: Optional[int] class Config: orm_mode = True class TelegramBotBase(BaseModel): name: str = Field(description="اسم البوت") token: str = Field(description="توكن البوت") # Note: Should be handled securely username: str = Field(description="رمز البوت (@...)") class TelegramBotCreate(TelegramBotBase): pass class TelegramBotUpdate(TelegramBotBase): name: Optional[str] = Field(None, description="اسم البوت") token: Optional[str] = Field(None, description="توكن البوت") # Note: Should be handled securely username: Optional[str] = Field(None, description="رمز البوت (@...)") is_active: Optional[bool] = Field(None, description="حالة التفعيل") send_reports: Optional[bool] = Field(None, description="إرسال تقارير") send_recordings_notifications: Optional[bool] = Field(None, description="إرسال إشعارات التسجيلات") send_event_notifications: Optional[bool] = Field(None, description="إرسال إشعارات الأحداث") send_camera_status_updates: Optional[bool] = Field(None, description="إرسال تحديثات حالة الكاميرات") camera_status_update_frequency_hours: Optional[int] = Field(None, ge=1, description="تكرار تحديث حالة الكاميرات (ساعات)") class TelegramBot(TelegramBotBase): id: int is_active: bool last_updated: datetime send_reports: bool send_recordings_notifications: bool send_event_notifications: bool send_camera_status_updates: bool camera_status_update_frequency_hours: int class Config: orm_mode = True class PersonBase(BaseModel): name: str = Field(description="اسم الشخص") identity_number: Optional[str] = Field(None, description="رقم الهوية") workplace: Optional[str] = Field(None, description="جهة العمل") telegram_notify_enabled: bool = Field(default=True, description="تفعيل إشعارات Telegram") class PersonCreate(PersonBase): # For creation, image is uploaded separately pass class PersonUpdate(PersonBase): name: Optional[str] = Field(None, description="اسم الشخص") identity_number: Optional[str] = Field(None, description="رقم الهوية") workplace: Optional[str] = Field(None, description="جهة العمل") telegram_notify_enabled: Optional[bool] = Field(None, description="تفعيل إشعارات Telegram") # Image update handled separately class Person(PersonBase): id: int image_path: str # Path to the stored image class Config: orm_mode = True class TruckBase(BaseModel): driver_name: str = Field(description="اسم السائق") plate_number: str = Field(description="رقم اللوحة") truck_type: Optional[str] = Field(None, description="نوع الشاحنة") telegram_notify_enabled: bool = Field(default=True, description="تفعيل إشعارات Telegram") class TruckCreate(TruckBase): # For creation, image is uploaded separately pass class TruckUpdate(TruckBase): driver_name: Optional[str] = Field(None, description="اسم السائق") plate_number: Optional[str] = Field(None, description="رقم اللوحة") truck_type: Optional[str] = Field(None, description="نوع الشاحنة") telegram_notify_enabled: Optional[bool] = Field(None, description="تفعيل إشعارات Telegram") # Image update handled separately class Truck(TruckBase): id: int image_path: str # Path to the stored image class Config: orm_mode = True class ReportRequest(BaseModel): report_type: str = Field(description="نوع التقرير (all_events, known_persons, unknown_persons, known_trucks, unknown_trucks)") start_time: datetime = Field(description="وقت البدء") end_time: datetime = Field(description="وقت الانتهاء") class Report(BaseModel): id: int report_type: str start_time: datetime end_time: datetime generated_at: datetime file_path: Optional[str] summary: Optional[str] class Config: orm_mode = True class UserBase(BaseModel): username: str = Field(description="اسم المستخدم") role: str = Field(default="read", description="الصلاحية (read, write, admin)") is_active: bool = Field(default=True, description="مفعل") class UserCreate(UserBase): password: str = Field(description="كلمة المرور") class UserUpdate(UserBase): username: Optional[str] = Field(None, description="اسم المستخدم") role: Optional[str] = Field(None, description="الصلاحية (read, write, admin)") is_active: Optional[bool] = Field(None, description="مفعل") password: Optional[str] = Field(None, description="كلمة المرور (للتغيير)") class User(UserBase): id: int last_login: Optional[datetime] class Config: orm_mode = True class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): username: Optional[str] = None class Event(BaseModel): id: int timestamp: datetime event_type: str source_id: int source_type: str person_id: Optional[int] truck_id: Optional[int] is_known: Optional[bool] details: Optional[str] # JSON string snapshot_path: Optional[str] class Config: orm_mode = True from sqlalchemy.orm import Session from sqlalchemy import func from models import ( DBSettings, DBCamera, DBVideo, DBCameraControlSettings, DBTelegramBot, DBPerson, DBTruck, DBReport, DBUser, DBLog, DBEvent ) from schemas import ( SettingsCreate, SettingsUpdate, CameraCreate, CameraUpdate, VideoCreate, VideoUpdate, CameraControlSettingsCreate, CameraControlSettingsUpdate, TelegramBotCreate, TelegramBotUpdate, PersonCreate, PersonUpdate, TruckCreate, TruckUpdate, UserCreate, UserUpdate ) from typing import List, Optional from datetime import datetime import os import shutil import zipfile from passlib.context import CryptContext # Password Hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def get_password_hash(password): return pwd_context.hash(password) def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) # --- Settings CRUD --- def get_settings(db: Session): # Assuming only one settings entry return db.query(DBSettings).first() def create_settings(db: Session, settings: SettingsCreate): db_settings = DBSettings(**settings.dict()) db.add(db_settings) db.commit() db.refresh(db_settings) return db_settings def update_settings(db: Session, settings_id: int, settings: SettingsUpdate): db_settings = db.query(DBSettings).filter(DBSettings.id == settings_id).first() if db_settings: update_data = settings.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_settings, key, value) db.commit() db.refresh(db_settings) return db_settings # --- Camera CRUD --- def get_cameras(db: Session, skip: int = 0, limit: int = 100): return db.query(DBCamera).offset(skip).limit(limit).all() def get_camera(db: Session, camera_id: int): return db.query(DBCamera).filter(DBCamera.id == camera_id).first() def create_camera(db: Session, camera: CameraCreate): # Note: Password should be encrypted before storing db_camera = DBCamera(**camera.dict()) db.add(db_camera) db.commit() db.refresh(db_camera) # Create default control settings for the new camera create_camera_control_settings(db, CameraControlSettingsCreate(camera_id=db_camera.id)) return db_camera def update_camera(db: Session, camera_id: int, camera: CameraUpdate): db_camera = db.query(DBCamera).filter(DBCamera.id == camera_id).first() if db_camera: update_data = camera.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_camera, key, value) db.commit() db.refresh(db_camera) return db_camera def delete_camera(db: Session, camera_id: int): db_camera = db.query(DBCamera).filter(DBCamera.id == camera_id).first() if db_camera: # Delete associated control settings db_control_settings = db.query(DBCameraControlSettings).filter(DBCameraControlSettings.camera_id == camera_id).first() if db_control_settings: db.delete(db_control_settings) db.delete(db_camera) db.commit() return db_camera # --- Video CRUD --- def get_videos(db: Session, skip: int = 0, limit: int = 100): return db.query(DBVideo).offset(skip).limit(limit).all() def get_video(db: Session, video_id: int): return db.query(DBVideo).filter(DBVideo.id == video_id).first() def create_video(db: Session, video: VideoCreate): db_video = DBVideo(**video.dict()) db.add(db_video) db.commit() db.refresh(db_video) # Create default control settings for the new video create_camera_control_settings(db, CameraControlSettingsCreate(video_id=db_video.id)) return db_video def update_video(db: Session, video_id: int, video: VideoUpdate): db_video = db.query(DBVideo).filter(DBVideo.id == video_id).first() if db_video: update_data = video.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_video, key, value) db.commit() db.refresh(db_video) return db_video def delete_video(db: Session, video_id: int): db_video = db.query(DBVideo).filter(DBVideo.id == video_id).first() if db_video: # Delete associated control settings db_control_settings = db.query(DBCameraControlSettings).filter(DBCameraControlSettings.video_id == video_id).first() if db_control_settings: db.delete(db_control_settings) # Optionally delete the video file itself if os.path.exists(db_video.file_path): os.remove(db_video.file_path) db.delete(db_video) db.commit() return db_video # --- Camera Control Settings CRUD --- def get_camera_control_settings_by_source(db: Session, camera_id: Optional[int] = None, video_id: Optional[int] = None): query = db.query(DBCameraControlSettings) if camera_id is not None: query = query.filter(DBCameraControlSettings.camera_id == camera_id) elif video_id is not None: query = query.filter(DBCameraControlSettings.video_id == video_id) else: return None # Must specify camera or video return query.first() def create_camera_control_settings(db: Session, settings: CameraControlSettingsCreate): db_settings = DBCameraControlSettings(**settings.dict()) db.add(db_settings) db.commit() db.refresh(db_settings) return db_settings def update_camera_control_settings(db: Session, settings_id: int, settings: CameraControlSettingsUpdate): db_settings = db.query(DBCameraControlSettings).filter(DBCameraControlSettings.id == settings_id).first() if db_settings: update_data = settings.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_settings, key, value) db.commit() db.refresh(db_settings) return db_settings # --- Telegram Bot CRUD --- def get_telegram_bots(db: Session, skip: int = 0, limit: int = 100): return db.query(DBTelegramBot).offset(skip).limit(limit).all() def get_telegram_bot(db: Session, bot_id: int): return db.query(DBTelegramBot).filter(DBTelegramBot.id == bot_id).first() def create_telegram_bot(db: Session, bot: TelegramBotCreate): # Note: Token should be encrypted before storing db_bot = DBTelegramBot(**bot.dict()) db.add(db_bot) db.commit() db.refresh(db_bot) return db_bot def update_telegram_bot(db: Session, bot_id: int, bot: TelegramBotUpdate): db_bot = db.query(DBTelegramBot).filter(DBTelegramBot.id == bot_id).first() if db_bot: update_data = bot.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_bot, key, value) db.commit() db.refresh(db_bot) return db_bot def delete_telegram_bot(db: Session, bot_id: int): db_bot = db.query(DBTelegramBot).filter(DBTelegramBot.id == bot_id).first() if db_bot: db.delete(db_bot) db.commit() return db_bot # --- Person CRUD --- def get_persons(db: Session, skip: int = 0, limit: int = 100): return db.query(DBPerson).offset(skip).limit(limit).all() def get_person(db: Session, person_id: int): return db.query(DBPerson).filter(DBPerson.id == person_id).first() def create_person(db: Session, person: PersonCreate, image_path: str): db_person = DBPerson(**person.dict(), image_path=image_path) db.add(db_person) db.commit() db.refresh(db_person) # TODO: Process image for face encoding and store it return db_person def update_person(db: Session, person_id: int, person: PersonUpdate, image_path: Optional[str] = None): db_person = db.query(DBPerson).filter(DBPerson.id == person_id).first() if db_person: update_data = person.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_person, key, value) if image_path: # Optionally delete old image file if db_person.image_path and os.path.exists(db_person.image_path): os.remove(db_person.image_path) db_person.image_path = image_path # TODO: Re-process image for face encoding db.commit() db.refresh(db_person) return db_person def delete_person(db: Session, person_id: int): db_person = db.query(DBPerson).filter(DBPerson.id == person_id).first() if db_person: # Optionally delete image file if db_person.image_path and os.path.exists(db_person.image_path): os.remove(db_person.image_path) db.delete(db_person) db.commit() return db_person # --- Truck CRUD --- def get_trucks(db: Session, skip: int = 0, limit: int = 100): return db.query(DBTruck).offset(skip).limit(limit).all() def get_truck(db: Session, truck_id: int): return db.query(DBTruck).filter(DBTruck.id == truck_id).first() def create_truck(db: Session, truck: TruckCreate, image_path: str): db_truck = DBTruck(**truck.dict(), image_path=image_path) db.add(db_truck) db.commit() db.refresh(db_truck) # TODO: Process image for truck/plate encoding return db_truck def update_truck(db: Session, truck_id: int, truck: TruckUpdate, image_path: Optional[str] = None): db_truck = db.query(DBTruck).filter(DBTruck.id == truck_id).first() if db_truck: update_data = truck.dict(exclude_unset=True) for key, value in update_data.items(): setattr(db_truck, key, value) if image_path: # Optionally delete old image file if db_truck.image_path and os.path.exists(db_truck.image_path): os.remove(db_truck.image_path) db_truck.image_path = image_path # TODO: Re-process image for truck/plate encoding db.commit() db.refresh(db_truck) return db_truck def delete_truck(db: Session, truck_id: int): db_truck = db.query(DBTruck).filter(DBTruck.id == truck_id).first() if db_truck: # Optionally delete image file if db_truck.image_path and os.path.exists(db_truck.image_path): os.remove(db_truck.image_path) db.delete(db_truck) db.commit() return db_truck # --- Report CRUD --- def create_report(db: Session, report_type: str, start_time: datetime, end_time: datetime, file_path: Optional[str] = None, summary: Optional[str] = None): db_report = DBReport( report_type=report_type, start_time=start_time, end_time=end_time, file_path=file_path, summary=summary ) db.add(db_report) db.commit() db.refresh(db_report) return db_report def get_reports(db: Session, skip: int = 0, limit: int = 100): return db.query(DBReport).offset(skip).limit(limit).all() def get_report(db: Session, report_id: int): return db.query(DBReport).filter(DBReport.id == report_id).first() # --- User CRUD --- def get_users(db: Session, skip: int = 0, limit: int = 100): return db.query(DBUser).offset(skip).limit(limit).all() def get_user(db: Session, user_id: int): return db.query(DBUser).filter(DBUser.id == user_id).first() def get_user_by_username(db: Session, username: str): return db.query(DBUser).filter(DBUser.username == username).first() def create_user(db: Session, user: UserCreate): hashed_password = get_password_hash(user.password) db_user = DBUser(username=user.username, hashed_password=hashed_password, role=user.role, is_active=user.is_active) db.add(db_user) db.commit() db.refresh(db_user) return db_user def update_user(db: Session, user_id: int, user: UserUpdate): db_user = db.query(DBUser).filter(DBUser.id == user_id).first() if db_user: update_data = user.dict(exclude_unset=True) if "password" in update_data and update_data["password"] is not None: db_user.hashed_password = get_password_hash(update_data.pop("password")) for key, value in update_data.items(): setattr(db_user, key, value) db.commit() db.refresh(db_user) return db_user def delete_user(db: Session, user_id: int): db_user = db.query(DBUser).filter(DBUser.id == user_id).first() if db_user: db.delete(db_user) db.commit() return db_user # --- Log CRUD --- def create_log(db: Session, level: str, message: str, source: Optional[str] = None): db_log = DBLog(level=level, message=message, source=source) db.add(db_log) db.commit() db.refresh(db_log) return db_log def get_logs(db: Session, skip: int = 0, limit: int = 100): return db.query(DBLog).order_by(DBLog.timestamp.desc()).offset(skip).limit(limit).all() # --- Event CRUD --- def create_event(db: Session, event_type: str, source_id: int, source_type: str, person_id: Optional[int] = None, truck_id: Optional[int] = None, is_known: Optional[bool] = None, details: Optional[str] = None, snapshot_path: Optional[str] = None): db_event = DBEvent( event_type=event_type, source_id=source_id, source_type=source_type, person_id=person_id, truck_id=truck_id, is_known=is_known, details=details, snapshot_path=snapshot_path ) db.add(db_event) db.commit() db.refresh(db_event) return db_event def get_events(db: Session, skip: int = 0, limit: int = 100, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, event_type: Optional[str] = None, is_known: Optional[bool] = None): query = db.query(DBEvent) if start_time: query = query.filter(DBEvent.timestamp >= start_time) if end_time: query = query.filter(DBEvent.timestamp <= end_time) if event_type: query = query.filter(DBEvent.event_type == event_type) if is_known is not None: query = query.filter(DBEvent.is_known == is_known) return query.order_by(DBEvent.timestamp.desc()).offset(skip).limit(limit).all() # --- Backup/Restore Utility (Basic Implementation) --- def create_backup(db: Session, backup_path: str): """Creates a zip backup of the database and relevant files.""" if not os.path.exists(backup_path): os.makedirs(backup_path) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_filename = f"security_backup_{timestamp}.zip" backup_filepath = os.path.join(backup_path, backup_filename) db_path = os.path.join(os.path.dirname(__file__), 'security.db') recordings_dir = get_settings(db).recordings_path # Get path from settings persons_images_dir = "static/persons_images" # Assuming images are here trucks_images_dir = "static/trucks_images" # Assuming images are here snapshots_dir = "static/event_snapshots" # Assuming snapshots are here files_to_backup = [db_path] dirs_to_backup = [recordings_dir, persons_images_dir, trucks_images_dir, snapshots_dir] with zipfile.ZipFile(backup_filepath, 'w', zipfile.ZIP_DEFLATED) as zipf: # Add database file if os.path.exists(db_path): zipf.write(db_path, os.path.basename(db_path)) # Add directories recursively for dir_path in dirs_to_backup: if os.path.exists(dir_path): for root, _, files in os.walk(dir_path): for file in files: full_path = os.path.join(root, file) arcname = os.path.relpath(full_path, os.path.dirname(__file__)) # Path inside zip zipf.write(full_path, arcname) # TODO: Implement cleanup of old backups based on settings.backup_storage_days return backup_filepath def restore_backup(db: Session, backup_filepath: str): """Restores from a zip backup.""" if not os.path.exists(backup_filepath): return False, "Backup file not found." # Stop any active video processing/recording before restoring # TODO: Implement stopping processes # Create a temporary directory for extraction temp_restore_dir = f"temp_restore_{datetime.now().strftime('%Y%m%d_%H%M%S')}" os.makedirs(temp_restore_dir) try: with zipfile.ZipFile(backup_filepath, 'r') as zipf: zipf.extractall(temp_restore_dir) # Paths in the extracted temp directory extracted_db_path = os.path.join(temp_restore_dir, 'security.db') extracted_recordings_dir = os.path.join(temp_restore_dir, get_settings(db).recordings_path) extracted_persons_images_dir = os.path.join(temp_restore_dir, "static/persons_images") extracted_trucks_images_dir = os.path.join(temp_restore_dir, "static/trucks_images") extracted_snapshots_dir = os.path.join(temp_restore_dir, "static/event_snapshots") # Check if the database file exists in the backup if not os.path.exists(extracted_db_path): return False, "Database file not found in backup." # Close current DB session before replacing the file db.close() # Dispose the engine to release the file lock engine = db.get_bind() engine.dispose() # Replace the current database file current_db_path = os.path.join(os.path.dirname(__file__), 'security.db') if os.path.exists(current_db_path): os.rename(current_db_path, f"{current_db_path}.bak_{datetime.now().strftime('%Y%m%d_%H%M%S')}") # Backup current DB shutil.move(extracted_db_path, current_db_path) # Replace directories (careful with existing files) current_recordings_dir = get_settings(db).recordings_path current_persons_images_dir = "static/persons_images" current_trucks_images_dir = "static/trucks_images" current_snapshots_dir = "static/event_snapshots" # Simple replace: remove current, move extracted if os.path.exists(current_recordings_dir): shutil.rmtree(current_recordings_dir) if os.path.exists(extracted_recordings_dir): shutil.move(extracted_recordings_dir, current_recordings_dir) if os.path.exists(current_persons_images_dir): shutil.rmtree(current_persons_images_dir) if os.path.exists(extracted_persons_images_dir): shutil.move(extracted_persons_images_dir, current_persons_images_dir) if os.path.exists(current_trucks_images_dir): shutil.rmtree(current_trucks_images_dir) if os.path.exists(extracted_trucks_images_dir): shutil.move(extracted_trucks_images_dir, current_trucks_images_dir) if os.path.exists(current_snapshots_dir): shutil.rmtree(current_snapshots_dir) if os.path.exists(extracted_snapshots_dir): shutil.move(extracted_snapshots_dir, current_snapshots_dir) # TODO: Implement conflict detection/handling (e.g., if backup has conflicting camera IPs) # This would require reading data from the extracted DB before replacing the current one. # Re-initialize the database engine and session from database import engine, SessionLocal, Base # Base.metadata.create_all(bind=engine) # Ensure tables exist if DB was empty # db = SessionLocal() # Get a new session # TODO: Restart video processing/recording processes return True, "Restore successful." except Exception as e: # If restore fails, try to revert the database file from backup # This is a simplified rollback current_db_path = os.path.join(os.path.dirname(__file__), 'security.db') backup_db_path = f"{current_db_path}.bak_{datetime.now().strftime('%Y%m%d_%H%M%S')}" if os.path.exists(backup_db_path): if os.path.exists(current_db_path): os.remove(current_db_path) os.rename(backup_db_path, current_db_path) # Re-initialize DB connection from database import engine, SessionLocal, Base # Base.metadata.create_all(bind=engine) # db = SessionLocal() return False, f"Restore failed: {e}" finally: # Clean up the temporary directory if os.path.exists(temp_restore_dir): shutil.rmtree(temp_restore_dir) # --- Report Generation Utility (Placeholder) --- def generate_report_data(db: Session, report_type: str, start_time: datetime, end_time: datetime): """Fetches data for a report based on type and time range.""" # This is a simplified example. Real reports would involve more complex queries and data processing. if report_type == "all_events": events = get_events(db, start_time=start_time, end_time=end_time) # Format events for reporting report_data = [{"timestamp": e.timestamp, "type": e.event_type, "source": f"{e.source_type}:{e.source_id}", "details": e.details} for e in events] summary = f"Total events: {len(events)}" return report_data, summary elif report_type == "known_persons": events = get_events(db, start_time=start_time, end_time=end_time, event_type="person_detected", is_known=True) # Fetch person details for each event report_data = [] for event in events: person = get_person(db, event.person_id) if event.person_id else None report_data.append({ "timestamp": event.timestamp, "person_name": person.name if person else "N/A", "source": f"{event.source_type}:{event.source_id}", "details": event.details }) summary = f"Total known person detections: {len(events)}" return report_data, summary # Add other report types (unknown persons, known/unknown trucks) else: return [], "Unsupported report type." def export_report_to_pdf(report_data, summary, output_path): """Generates a simple PDF report (Placeholder).""" from reportlab.lib.pagesizes import letter from reportlab.pdfgen import canvas c = canvas.Canvas(output_path, pagesize=letter) c.drawString(100, 750, "تقرير النظام الأمني") # Report Title (Arabic) c.drawString(100, 730, f"ملخص: {summary}") # Summary (Arabic) c.drawString(100, 710, f"عدد السجلات: {len(report_data)}") # Record Count (Arabic) y_position = 690 for item in report_data: if y_position < 100: c.showPage() y_position = 750 # Simple text representation of data c.drawString(100, y_position, str(item)) y_position -= 15 c.save() return output_path def export_report_to_csv(report_data, output_path): """Generates a simple CSV report (Placeholder).""" import pandas as pd df = pd.DataFrame(report_data) df.to_csv(output_path, index=False) return output_path # --- User Authentication Utility --- # Handled in routers/auth.py