File size: 5,780 Bytes
9b5e433
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
"""
database.py
-----------
SQLAlchemy models and engine for the Face Recognition Attendance System.
Uses SQLite stored at data/attendance.db.
"""

import os
import json
from datetime import datetime

from sqlalchemy import (
    create_engine, Column, Integer, String, Float, DateTime, Date,
    UniqueConstraint, func
)
from sqlalchemy.orm import declarative_base, sessionmaker, Session

# ─────────────────────────────────────────────────────────────────────────────
# Paths
# ─────────────────────────────────────────────────────────────────────────────
_BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA_DIR = os.path.join(_BASE_DIR, "data")
DB_PATH = os.path.join(DATA_DIR, "attendance.db")
JSON_PATH = os.path.join(DATA_DIR, "attendance.json")

os.makedirs(DATA_DIR, exist_ok=True)

# ─────────────────────────────────────────────────────────────────────────────
# Engine & Session
# ─────────────────────────────────────────────────────────────────────────────
engine = create_engine(f"sqlite:///{DB_PATH}", echo=False)
SessionLocal = sessionmaker(bind=engine)

Base = declarative_base()


# ─────────────────────────────────────────────────────────────────────────────
# Models
# ─────────────────────────────────────────────────────────────────────────────
class Student(Base):
    __tablename__ = "students"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.now)
    image_count = Column(Integer, default=0)

    def __repr__(self):
        return f"<Student(name='{self.name}', images={self.image_count})>"


class AttendanceRecord(Base):
    __tablename__ = "attendance"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False, index=True)
    date = Column(String, nullable=False, index=True)   # YYYY-MM-DD
    time = Column(String, nullable=False)                # HH:MM:SS
    confidence = Column(Float, default=0.0)

    # Prevent duplicate (name, date) entries
    __table_args__ = (
        UniqueConstraint("name", "date", name="uq_name_date"),
    )

    def __repr__(self):
        return f"<Attendance(name='{self.name}', date='{self.date}')>"

    def to_dict(self):
        return {
            "name": self.name,
            "date": self.date,
            "time": self.time,
            "confidence": self.confidence,
        }


# ─────────────────────────────────────────────────────────────────────────────
# Create tables
# ─────────────────────────────────────────────────────────────────────────────
Base.metadata.create_all(engine)


# ─────────────────────────────────────────────────────────────────────────────
# Auto-migrate from JSON
# ─────────────────────────────────────────────────────────────────────────────
def migrate_from_json():
    """
    One-time migration: import records from attendance.json into SQLite.
    Only runs if the JSON file exists and the DB table is empty.
    """
    if not os.path.exists(JSON_PATH):
        return

    session = SessionLocal()
    try:
        count = session.query(AttendanceRecord).count()
        if count > 0:
            return  # DB already has data, skip migration

        with open(JSON_PATH, "r") as f:
            records = json.load(f)

        if not isinstance(records, list):
            return

        migrated = 0
        for rec in records:
            try:
                ar = AttendanceRecord(
                    name=rec.get("name", ""),
                    date=rec.get("date", ""),
                    time=rec.get("time", ""),
                    confidence=float(rec.get("confidence", 0.0)),
                )
                session.add(ar)
                session.commit()
                migrated += 1
            except Exception:
                session.rollback()

        if migrated > 0:
            # Rename the old JSON to mark it as migrated
            backup = JSON_PATH + ".migrated"
            os.rename(JSON_PATH, backup)
            print(f"βœ… Migrated {migrated} records from JSON β†’ SQLite. Old file renamed to {backup}")
    finally:
        session.close()


# Run migration on import
migrate_from_json()