tts_labeling / utils /auth.py
vargha's picture
revert authentication fix
ae8975a
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
)