backend.components.workers.workerService

  1from datetime import datetime
  2import io
  3from cryptography.fernet import Fernet
  4import numpy as np
  5import face_recognition
  6from sqlalchemy import select
  7import random
  8import json
  9
 10from backend.components.camera_verification.qrcode.qrcodeService import ExpiredCodeError, InvalidCodeError, MultipleCodesError, NoCodeFoundError, decode_qr_image, generate_qr_code
 11from backend.config import QR_SECRET_KEY
 12from backend.database.models import Worker
 13from backend.app import db
 14
 15def get_worker_embedding(worker):
 16    """
 17    Decodes and returns the worker's face embedding from the database.
 18
 19    **Parameters**:
 20    - `worker` (Worker): The Worker object whose face embedding is to be retrieved.
 21
 22    **Returns**:
 23    - `np.array`: The decoded face embedding as a NumPy array.
 24    """
 25    blob = worker.face_embedding
 26    buffer = io.BytesIO(blob)
 27    arr = np.load(buffer)
 28    return arr
 29
 30
 31def generate_worker_entry_pass(worker: Worker):
 32    """
 33    Generates the worker entry pass as png image.
 34
 35    **Parameters**:
 36    - `worker` (Worker): Worker to generate an entry pass for.
 37
 38    **Returns**:
 39    - `bytes`: Bytes of PNG image containing the worker entry pass.
 40    """
 41    # TODO: Create a fun, pretty access card, not just bare qr code.
 42    # TODO 2: Should probably be returned as pdf for certainer printing.
 43    qrcode = generate_qr_code(worker.secret)
 44    return qrcode
 45    
 46
 47
 48def create_worker_embedding(img):
 49    """
 50    Creates and encodes a worker's face embedding into a BLOB format for storage in the database.
 51
 52    **Parameters**:
 53    - `img` (ndarray): The image of the worker's face.
 54
 55    **Returns**:
 56    - `bytes`: The encoded face embedding as a BLOB.
 57    """
 58    img_embedding = face_recognition.face_encodings(img)
 59
 60    buffer = io.BytesIO()
 61    np.save(buffer, img_embedding)
 62    blob = buffer.getvalue()
 63    return blob
 64
 65
 66def generate_worker_secret(worker: Worker) -> str:
 67    '''
 68    Generate a secret for the worker QR code.
 69
 70    **Parameters**:
 71    - `worker_id` (int): Unique numeric ID of the worker.
 72    - `name` (str): Worker display name; included to add entropy but not treated as a secret.
 73
 74    **Returns**:
 75    - `str`: Hex-encoded SHA-256 hash of the string "{worker_id}:{name}:{rand}",
 76        where `rand` is a 6-digit random nonce. 
 77    '''
 78
 79    rand_value = str(random.randint(100000, 999999))
 80    data = {
 81        "worker_id": worker.id,
 82        "name": worker.name,
 83        "rand_value": rand_value
 84    }
 85    json_data = json.dumps(data).encode('utf-8')
 86    fernet = Fernet(QR_SECRET_KEY)
 87    secret = fernet.encrypt(json_data)
 88    return secret.decode('utf-8')
 89
 90
 91def decrypt_worker_secret(encrypted_secret: str):
 92    try:
 93        fernet = Fernet(QR_SECRET_KEY)
 94        decrypted = fernet.decrypt(encrypted_secret.encode('utf-8'))
 95        data = json.loads(decrypted.decode('utf-8'))
 96        return data
 97    except Exception as e:
 98        print(f"Błąd deszyfrowania: {e}")
 99        return None
100
101
102def create_worker(name, face_image, expiration_date):
103    """
104    Creates a new worker and stores their information in the database.
105
106    **Parameters**:
107    - `name` (str): The name of the worker.
108    - `face_image` (ndarray): The image of the worker's face.
109    - `expiration_date` (datetime): The expiration date for the worker's access.
110
111    **Returns**:
112    - `Worker`: The newly created Worker object.
113    """
114    face_embedding_blob = create_worker_embedding(face_image)
115    worker = Worker(
116        name=name,
117        face_embedding=face_embedding_blob,
118        expiration_date=expiration_date,
119        secret="TEMP_SECRET"
120    )
121    db.session.add(worker)
122    db.session.flush()
123    new_secret_value = generate_worker_secret(worker)
124    worker.secret = new_secret_value
125    db.session.commit()
126    return worker
127
128
129def extend_worker_expiration(worker: Worker, new_expiration_date):
130    """
131    Extends the expiration date of a worker entry permit.
132
133    **Parameters**:
134    - `worker` (Worker): The worker object whose expiration date is to be updated.
135    - `new_expiration_date` (datetime): The new expiration date.
136    """
137    worker.expiration_date = new_expiration_date
138    db.session.commit()
139    return worker
140
141
142def update_worker_name(worker: Worker, new_name: str):
143    """
144    Updates the name of a worker.
145
146    **Parameters**:
147    - `worker` (Worker): The worker object whose name is to be updated.
148    - `new_name` (str): The new name for the worker.
149    """
150    worker.name = new_name
151    db.session.commit()
152
153
154def update_worker_face_image(worker: Worker, new_face_image):
155    """
156    Updates the face image of a worker.
157
158    **Parameters**:
159    - `worker` (Worker): The worker object whose face image is to be updated.
160    - `new_face_image` (ndarray): The new face image of the worker.
161    """
162    new_face_embedding_blob = create_worker_embedding(new_face_image)
163    worker.face_embedding = new_face_embedding_blob
164    db.session.commit()
165
166
167def get_all_workers():
168    """
169    Retrieves all workers from the database.
170
171    **Returns**:
172    - `list[Worker]`: A list of all Worker objects.
173    """
174    workers = db.session.query(Worker).all()
175    return workers
176
177
178def get_worker_by_id(worker_id):
179    """
180    Retrieves a worker from the database by their ID.
181
182    **Parameters**:
183    - `worker_id` (int): The ID of the worker to retrieve.
184
185    **Returns**:
186    - `Worker`: The Worker object if found.
187
188    **Raises**:
189    - `ValueError`: If no worker with the given ID is found.
190    """
191    worker = db.session.get(Worker, worker_id)
192    if not worker:
193        raise ValueError(f"Worker with id {worker_id} not found")
194    return worker
195
196
197def get_worker_by_secret(secret: str):
198    '''
199    Get a Worker by the secret decoded from the QR code.
200
201    **Parameters**:
202    - `secret` (str): Secret extracted from the QR code. 
203
204    **Returns**:
205    - `Worker|None`: Worker belonging to the secret, if found.
206    '''
207    stmt = select(Worker).where(Worker.secret == secret)
208    result = db.session.execute(stmt).scalar_one_or_none()
209    return result
210
211
212def get_worker_from_qr_code(img) -> Worker:
213    '''
214    Method that reads the QR code and returns a Worker that belongs to the code.
215
216    **Parameters**:
217    - `img` (ndarray): Decoded image in ndarray.
218
219    **Returns**:
220    - `Worker` - Worker belonging to the scanned code.
221
222    **Raises**:
223    - `InvalidCodeError` - The code is invalid or no worker with the code was found.
224    - `MultipleCodesError` - Multiple QR codes were detected on the image.
225    - `NoCodeFoundError` - No QR codes were found on the image.
226    - `ExpiredCodeError` - The QR code is expired.
227    '''
228    try:
229        qr_secret = decode_qr_image(img)
230        worker = get_worker_by_secret(qr_secret)
231
232        if not worker:
233            raise InvalidCodeError("Wykryto niepoprawny kod QR")
234
235        # Check if the worker's expiration date has passed
236        if worker.expiration_date and worker.expiration_date < datetime.now():
237            raise ExpiredCodeError("Przepustka wygasła")
238
239        return worker
240
241    except (MultipleCodesError, NoCodeFoundError, InvalidCodeError, ExpiredCodeError) as e:
242        raise e
243
244    except Exception as e:
245        print(f"Internal Error in getWorkerFromQRCode: {e}")
246        raise e