""" Content Tracker — Link redirect logging and quiz XP logic. Handles: - Logging content access when tracked links are clicked - Awarding partial XP on first content access (xp_open_pct%) - Scoring quiz attempts and awarding remaining XP on pass - Generating short codes for tracked links """ import secrets import string from datetime import datetime, timezone from uuid import UUID from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from app.models.activity import Activity, ActivityCompletion from app.models.content_tracking import ContentAccessLog, Quiz, QuizAttempt, TrackedLink from app.models.user import User def generate_short_code(length: int = 8) -> str: """Generate a URL-safe random short code.""" alphabet = string.ascii_lowercase + string.digits return "".join(secrets.choice(alphabet) for _ in range(length)) async def log_content_access( db: AsyncSession, short_code: str, user_id: UUID | None = None, user_email: str | None = None, user_agent: str | None = None, ip_address: str | None = None, ) -> TrackedLink | None: """ Log a content access event and award partial XP if first access. Returns the tracked link for redirect, or None if not found. """ # Look up the tracked link result = await db.execute( select(TrackedLink).where(TrackedLink.short_code == short_code, TrackedLink.is_active == True) # noqa: E712 ) link = result.scalar_one_or_none() if not link: return None # Resolve user by email if user_id not provided if not user_id and user_email: user_result = await db.execute(select(User).where(User.email == user_email.lower())) user = user_result.scalar_one_or_none() if user: user_id = user.id # Log the access access_log = ContentAccessLog( tracked_link_id=link.id, user_id=user_id, user_email=user_email, user_agent=user_agent, ip_address=ip_address, ) db.add(access_log) # Increment click count await db.execute( update(TrackedLink).where(TrackedLink.id == link.id).values(click_count=TrackedLink.click_count + 1) ) # If linked to an activity and we know the user, award partial XP if link.activity_id and user_id: await _award_content_open_xp(db, user_id, link.activity_id) return link async def _award_content_open_xp( db: AsyncSession, user_id: UUID, activity_id: UUID, ) -> None: """ Award partial XP for opening content (xp_open_pct% of total). Only on first access — idempotent via upsert. """ activity = await db.get(Activity, activity_id) if not activity: return # Check if completion already exists result = await db.execute( select(ActivityCompletion).where( ActivityCompletion.user_id == user_id, ActivityCompletion.activity_id == activity_id, ActivityCompletion.period_key == None, # noqa: E711 ) ) existing = result.scalar_one_or_none() if existing and existing.status in ("completed", "verified"): return # Already done, don't re-award open_xp = int(activity.xp_value * activity.xp_open_pct / 100) if existing: if existing.status == "not_started": existing.status = "in_progress" existing.started_at = datetime.now(timezone.utc) existing.xp_awarded = open_xp existing.source = "content_tracker" existing.trust_tier = 1 # System-verified access else: completion = ActivityCompletion( user_id=user_id, activity_id=activity_id, status="in_progress", progress_pct=activity.xp_open_pct, started_at=datetime.now(timezone.utc), xp_awarded=open_xp, source="content_tracker", source_detail="tracked_link_access", trust_tier=1, ) db.add(completion) # Update user XP await db.execute( update(User).where(User.id == user_id).values( total_xp=User.total_xp + open_xp, verified_xp=User.verified_xp + open_xp, ) ) async def score_quiz_attempt( db: AsyncSession, quiz_id: UUID, user_id: UUID, answers: list[int], ) -> QuizAttempt: """ Score a quiz attempt. If passed, award remaining XP for the activity. Returns the QuizAttempt record. """ quiz = await db.get(Quiz, quiz_id) if not quiz: raise ValueError(f"Quiz {quiz_id} not found") questions = quiz.questions if isinstance(quiz.questions, list) else quiz.questions.get("questions", []) # Score correct = 0 for i, q in enumerate(questions): if i < len(answers) and answers[i] == q.get("correct_index"): correct += 1 score_pct = int((correct / len(questions)) * 100) if questions else 0 passed = score_pct >= quiz.passing_score_pct # Calculate XP to award xp_awarded = 0 if passed and quiz.activity_id: activity = await db.get(Activity, quiz.activity_id) if activity: remaining_pct = 100 - activity.xp_open_pct xp_awarded = int(activity.xp_value * remaining_pct / 100) attempt = QuizAttempt( quiz_id=quiz_id, user_id=user_id, answers=answers, score_pct=score_pct, passed=passed, started_at=datetime.now(timezone.utc), xp_awarded=xp_awarded, ) db.add(attempt) # If passed, complete the activity and award remaining XP if passed and quiz.activity_id: result = await db.execute( select(ActivityCompletion).where( ActivityCompletion.user_id == user_id, ActivityCompletion.activity_id == quiz.activity_id, ActivityCompletion.period_key == None, # noqa: E711 ) ) completion = result.scalar_one_or_none() if completion: completion.status = "completed" completion.completed_at = datetime.now(timezone.utc) completion.progress_pct = 100 completion.xp_awarded = (completion.xp_awarded or 0) + xp_awarded else: # Edge case: quiz taken without clicking tracked link first completion = ActivityCompletion( user_id=user_id, activity_id=quiz.activity_id, status="completed", progress_pct=100, completed_at=datetime.now(timezone.utc), xp_awarded=xp_awarded, source="quiz", trust_tier=1, ) db.add(completion) # Award XP to user if xp_awarded > 0: await db.execute( update(User).where(User.id == user_id).values( total_xp=User.total_xp + xp_awarded, verified_xp=User.verified_xp + xp_awarded, ) ) return attempt