Python 64.6%
TypeScript 33.7%
CSS 0.8%
1"""User accounts, magic links, consent, data deletion (Loi 25)."""23from __future__ import annotations45from datetime import timedelta67from sqlalchemy import delete, func, select89from app.core.config import Settings10from app.core.security import hash_password, new_magic_token, role_for_email, verify_password11from app.db import SessionLocal12from app.models import (13 AppSetting,14 Conversation,15 MagicLink,16 Message,17 Quiz,18 QuizAttempt,19 ToolCall,20 User,21 utcnow,22)23from app.services import files as file_service242526async def get_or_create_user(email: str, settings: Settings) -> User:27 email = email.lower().strip()28 async with SessionLocal() as session:29 user = await session.scalar(select(User).where(User.email == email))30 role = role_for_email(email, settings)31 if user is None:32 user = User(email=email, role=role, display_name=email.split("@")[0],33 preferences={"tutoiement": True, "course": settings.courses[0],34 "deep": False, "locale": "fr-CA"})35 session.add(user)36 elif role != "student" and user.role == "student":37 user.role = role # promoted via env38 user.last_seen_at = utcnow()39 await session.commit()40 await session.refresh(user)41 return user424344async def get_user(user_id: str) -> User | None:45 async with SessionLocal() as session:46 return await session.get(User, user_id)474849async def touch(user_id: str) -> None:50 async with SessionLocal() as session:51 user = await session.get(User, user_id)52 if user:53 user.last_seen_at = utcnow()54 await session.commit()555657async def update_preferences(user_id: str, prefs: dict, display_name: str | None = None) -> User:58 async with SessionLocal() as session:59 user = await session.get(User, user_id)60 if user is None:61 raise ValueError("user not found")62 merged = dict(user.preferences or {})63 merged.update({k: v for k, v in prefs.items() if k in64 {"tutoiement", "course", "deep", "locale", "font_scale"}})65 user.preferences = merged66 if display_name is not None:67 user.display_name = display_name[:120]68 await session.commit()69 await session.refresh(user)70 return user717273async def set_consent(user_id: str) -> None:74 async with SessionLocal() as session:75 user = await session.get(User, user_id)76 if user and not user.consent_at:77 user.consent_at = utcnow()78 await session.commit()798081async def create_magic_link(email: str, ttl_minutes: int = 20, purpose: str = "login") -> str:82 """Single-use token. purpose: "login" (direct sign-in), "invite" / "reset" (choose a password)."""83 token = new_magic_token()84 async with SessionLocal() as session:85 session.add(MagicLink(token=token, email=email.lower().strip(), purpose=purpose,86 expires_at=utcnow() + timedelta(minutes=ttl_minutes)))87 await session.commit()88 return token899091def _link_valid(link: MagicLink | None) -> bool:92 return bool(link and not link.used_at and link.expires_at >= utcnow())939495async def peek_magic_link(token: str) -> MagicLink | None:96 """Validate without consuming (used by the set-password page to show the e-mail)."""97 async with SessionLocal() as session:98 link = await session.get(MagicLink, token)99 return link if _link_valid(link) else None100101102async def consume_magic_link(token: str, purposes: set[str] | None = None) -> str | None:103 async with SessionLocal() as session:104 link = await session.get(MagicLink, token)105 if not _link_valid(link):106 return None107 if purposes is not None and (link.purpose or "login") not in purposes:108 return None109 link.used_at = utcnow()110 await session.commit()111 return link.email112113114async def invalidate_links(email: str, purposes: set[str] | None = None) -> int:115 """Burn every outstanding token of an address (after a password was set)."""116 async with SessionLocal() as session:117 rows = (await session.execute(select(MagicLink).where(118 MagicLink.email == email.lower().strip(), MagicLink.used_at.is_(None)))).scalars().all()119 n = 0120 for link in rows:121 if purposes is None or (link.purpose or "login") in purposes:122 link.used_at = utcnow()123 n += 1124 await session.commit()125 return n126127128async def get_user_by_email(email: str) -> User | None:129 async with SessionLocal() as session:130 return await session.scalar(select(User).where(User.email == email.lower().strip()))131132133async def mark_invited(user_ids: list[str]) -> None:134 if not user_ids:135 return136 async with SessionLocal() as session:137 rows = (await session.execute(select(User).where(User.id.in_(user_ids)))).scalars().all()138 now = utcnow()139 for u in rows:140 u.invited_at = now141 await session.commit()142143144async def users_without_password(only_never_invited: bool = False) -> list[User]:145 async with SessionLocal() as session:146 q = select(User).where(User.password_hash.is_(None)).order_by(User.email)147 if only_never_invited:148 q = q.where(User.invited_at.is_(None))149 return list((await session.execute(q)).scalars().all())150151152async def set_role(email: str, role: str) -> bool:153 async with SessionLocal() as session:154 user = await session.scalar(select(User).where(User.email == email.lower().strip()))155 if not user:156 return False157 user.role = role158 await session.commit()159 return True160161162async def delete_user_data(user_id: str) -> None:163 """Purge everything about a user (conversations, files, quizzes) then the account."""164 await file_service.purge_user(user_id)165 async with SessionLocal() as session:166 conv_ids = [c for c in (await session.execute(167 select(Conversation.id).where(Conversation.user_id == user_id))).scalars()]168 if conv_ids:169 msg_ids = [m for m in (await session.execute(170 select(Message.id).where(Message.conversation_id.in_(conv_ids)))).scalars()]171 if msg_ids:172 await session.execute(delete(ToolCall).where(ToolCall.message_id.in_(msg_ids)))173 await session.execute(delete(Message).where(Message.id.in_(msg_ids)))174 await session.execute(delete(Conversation).where(Conversation.id.in_(conv_ids)))175 quiz_ids = [q for q in (await session.execute(176 select(Quiz.id).where(Quiz.user_id == user_id))).scalars()]177 if quiz_ids:178 await session.execute(delete(QuizAttempt).where(QuizAttempt.quiz_id.in_(quiz_ids)))179 await session.execute(delete(Quiz).where(Quiz.id.in_(quiz_ids)))180 await session.execute(delete(User).where(User.id == user_id))181 await session.commit()182183184# ------------------------------------------------------------------ professor: students185async def list_users() -> list[dict]:186 async with SessionLocal() as session:187 counts = dict((await session.execute(188 select(Conversation.user_id, func.count(Message.id)).join(Message)189 .where(Message.role == "user").group_by(Conversation.user_id))).all())190 rows = (await session.execute(select(User).order_by(User.role.desc(), User.email))).scalars()191 return [{"id": u.id, "email": u.email, "role": u.role, "display_name": u.display_name,192 "has_password": bool(u.password_hash),193 "invited_at": u.invited_at.isoformat() if u.invited_at else None,194 "created_at": u.created_at.isoformat(),195 "last_seen_at": u.last_seen_at.isoformat() if u.last_seen_at else None,196 "consent_at": u.consent_at.isoformat() if u.consent_at else None,197 "messages": int(counts.get(u.id, 0))} for u in rows]198199200async def create_users(emails: list[str], settings: Settings, role: str = "student",201 display_names: dict[str, str] | None = None) -> dict:202 created, existing, invalid = [], [], []203 created_ids: list[str] = []204 async with SessionLocal() as session:205 for raw in emails:206 email = raw.strip().lower()207 if not email or "@" not in email or " " in email:208 if raw.strip():209 invalid.append(raw.strip())210 continue211 if email in created:212 continue213 user = await session.scalar(select(User).where(User.email == email))214 if user:215 existing.append(email)216 continue217 r = role_for_email(email, settings)218 u = User(email=email, role=r if r != "student" else role,219 display_name=(display_names or {}).get(email) or email.split("@")[0],220 preferences={"tutoiement": True, "course": settings.courses[0],221 "deep": False, "locale": "fr-CA"})222 session.add(u)223 created.append(email)224 await session.flush()225 created_ids.append(u.id)226 await session.commit()227 return {"created": created, "existing": existing, "invalid": invalid, "created_ids": created_ids}228229230async def email_registered(email: str) -> bool:231 async with SessionLocal() as session:232 return (await session.scalar(select(User.id).where(User.email == email.lower().strip()))) is not None233234235async def get_setting(key: str) -> str | None:236 async with SessionLocal() as session:237 row = await session.get(AppSetting, key)238 return row.value if row else None239240241async def set_setting(key: str, value: str) -> None:242 async with SessionLocal() as session:243 row = await session.get(AppSetting, key)244 if row:245 row.value = value246 else:247 session.add(AppSetting(key=key, value=value))248 await session.commit()249250251# ------------------------------------------------------------------ passwords252async def set_password(user_id: str, password: str) -> bool:253 async with SessionLocal() as session:254 user = await session.get(User, user_id)255 if not user:256 return False257 user.password_hash = hash_password(password)258 await session.commit()259 return True260261262async def authenticate(email: str, password: str) -> User | None:263 async with SessionLocal() as session:264 user = await session.scalar(select(User).where(User.email == email.lower().strip()))265 if user and verify_password(password, user.password_hash):266 user.last_seen_at = utcnow()267 await session.commit()268 await session.refresh(user)269 return user270 return None271