Spaces:
Running
Running
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 ───────────── # | |
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 ───────────── # | |
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 | |
) | |