import gradio as gr import time from utils.logger import Logger from utils.database import get_db from data.repository.annotator_repo import AnnotatorRepo from data.repository.annotator_workload_repo import AnnotatorWorkloadRepo from utils.security import verify_password from config import conf log = Logger() class AuthService: """ Authenticate users against DB and drive Gradio UI states. """ # ───────────── LOGIN ───────────── # @staticmethod def login(username: str, password: str, session: dict): """ Returns different UI states based on user role: - Phase 1 users: Normal annotation interface - Phase 2 users: Review interface Outputs depend on which interface is being used: For Phase 1: (message, login_container, dashboard_container, header_welcome, items_state, idx_state, tts_id, filename, sentence, ann_sentence) For Phase 2: (message, login_container, dashboard_container, review_dashboard_container, header_welcome, items_state, idx_state, tts_id, filename, sentence, ann_sentence) """ # ---------- اعتبارسنجی ---------- # log.info(f"Login attempt: username={username}") with get_db() as db: repo = AnnotatorRepo(db) annotator = repo.get_annotator_by_name(username) # ⬇️ توابع کمکی برای تولید خروجی خالی (درصورت خطا) def empty_dashboard_outputs_for_ui(): # Renamed and adjusted for UI outputs return ( [], # items_state 0, # idx_state "", # tts_id "", # filename "", # sentence "", # ann_sentence ) # --- کاربر موجود نیست / غیر فعال if annotator is None or not annotator.is_active: log.warning("Failed login (not found / inactive)") return ( "❌ Wrong username or password!", # message gr.update(), # login_container (no change) gr.update(visible=False), # dashboard_container (Phase 1) gr.update(visible=False), # review_dashboard_container (Phase 2) gr.update(), # review_dashboard_page.load_trigger gr.update(value=""), # header_welcome *empty_dashboard_outputs_for_ui(), ) # --- رمز عبور اشتباه if not verify_password(password, annotator.password): log.warning("Failed login (bad password)") return ( "❌ Wrong username or password!", # message gr.update(), # login_container (no change) gr.update(visible=False), # dashboard_container (Phase 1) gr.update(visible=False), # review_dashboard_container (Phase 2) gr.update(), # review_dashboard_page.load_trigger gr.update(value=""), # header_welcome *empty_dashboard_outputs_for_ui(), ) # ---------- ورود موفق ---------- # session["user_id"] = annotator.id session["username"] = annotator.name session["is_reviewer"] = annotator.name in conf.REVIEW_MAPPING.values() if session["is_reviewer"]: log.info(f"User '{username}' is a Phase 2 reviewer") return ( None, # 0: message gr.update(visible=False), # 1: login_container gr.update(visible=False), # 2: dashboard_container (Phase 1) gr.update(visible=True), # 3: review_dashboard_container (Phase 2) gr.update(value=time.time()), # 4: review_dashboard_page.load_trigger gr.update(value=f"👋 Welcome, {annotator.name}! (Phase 2 Review Mode)"), # 5: header_welcome # Dummy updates for dashboard_page components gr.update(value=[]), # 6: dashboard_page.items_state gr.update(value=0), # 7: dashboard_page.idx_state gr.update(value=""), # 8: dashboard_page.tts_id gr.update(value=""), # 9: dashboard_page.filename gr.update(value=""), # 10: dashboard_page.sentence gr.update(value=""), # 11: dashboard_page.ann_sentence ) else: # Phase 1 users - existing logic log.info(f"User '{username}' is a Phase 1 annotator") # بارگذاری داده‌های داشبورد workload_repo = AnnotatorWorkloadRepo(db) raw_items = workload_repo.get_tts_data_with_annotations(username) dashboard_items = [ { "id": row["tts_data"].id, "filename": row["tts_data"].filename, "sentence": row["tts_data"].sentence, "annotated_sentence": ( row["annotation"].annotated_sentence if row["annotation"] and row["annotation"].annotated_sentence is not None else "" ), "annotated_at": ( row["annotation"].annotated_at.isoformat() if row["annotation"] and row["annotation"].annotated_at else "" ), "validated": ( row["annotation"].validated if row["annotation"] is not None else False ), } for row in raw_items ] session["dashboard_items"] = dashboard_items # --- Resume Logic: Find first unannotated or default to first/last item --- initial_idx = 0 if dashboard_items: first_unannotated_idx = -1 for i, item_data in enumerate(dashboard_items): if not item_data["annotated_sentence"]: # Check if annotated_sentence is empty first_unannotated_idx = i break if first_unannotated_idx != -1: initial_idx = first_unannotated_idx log.info(f"User '{username}' resuming at first unannotated item, index: {initial_idx} (ID: {dashboard_items[initial_idx]['id']})") else: # All items are annotated or list is empty initial_idx = len(dashboard_items) - 1 if dashboard_items else 0 # Go to the last item if all annotated, else 0 if dashboard_items: log.info(f"User '{username}' has all items annotated or list is empty, starting at item index: {initial_idx} (ID: {dashboard_items[initial_idx]['id']})") else: # No items assigned log.info(f"User '{username}' has no items assigned, starting at index 0.") # مقداردهی فیلدهای رکورد بر اساس initial_idx if dashboard_items: # Check if list is not empty and initial_idx is valid current_item_for_ui = dashboard_items[initial_idx] first_vals_for_ui = ( current_item_for_ui["id"], current_item_for_ui["filename"], current_item_for_ui["sentence"], current_item_for_ui["annotated_sentence"], ) else: first_vals_for_ui = ("", "", "", "") # id, filename, sentence, ann_sentence log.info(f"User '{username}' logged in successfully. Initial index set to: {initial_idx}") # ---------- خروجی نهایی برای Gradio ---------- # return ( None, # 0: پیام خطا وجود ندارد gr.update(visible=False), # 1: فرم لاگین را مخفی کن gr.update(visible=True), # 2: داشبورد را نشان بده (Phase 1) gr.update(visible=False), # 3: review dashboard را مخفی کن (Phase 2) gr.update(), # 4: review_dashboard_page.load_trigger (no change) gr.update(value=f"👋 Welcome, {annotator.name}!"), # 5 dashboard_items, # 6: items_state initial_idx, # 7: idx_state *first_vals_for_ui, # 8-11: چهار فیلد نخست برای UI ) # ───────────── LOGOUT ───────────── # @staticmethod def logout(session: dict): username = session.get("username", "unknown") session.clear() log.info(f"User '{username}' logged out.") return ( gr.update(visible=True), # 1 → login_page.container gr.update(visible=False), # 2 → dashboard_page.container (Phase 1) gr.update(visible=False), # 3 → review_dashboard_page.container (Phase 2) gr.update(value=""), # 4 → self.welcome gr.update(value=""), # 5 → login_page.message )