import os import json import urllib.parse from pathlib import Path from contextlib import asynccontextmanager from typing import Optional, List, Dict, Any from datetime import datetime, timezone, date from fastapi import FastAPI, Depends, HTTPException, Request, Response, status, Query from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.middleware.gzip import GZipMiddleware from pydantic import BaseModel, Field import httpx from app import config from app.database import init_db, get_db from app.auth import ( ensure_admin_user, authenticate_user, create_session, get_user_from_session, delete_session, check_login_rate_limit, record_login_failure, record_login_success, change_user_password, ) from app.dictionary import lookup_term, USER_AGENT from app.export import export_csv, export_json, export_anki_tsv, export_markdown from app.srs import calculate_sm2, generate_quiz_options, build_activity_heatmap, calculate_cefr_level from app.reader import parse_epub, parse_text, count_words @asynccontextmanager async def lifespan(app: FastAPI): init_db() ensure_admin_user() yield app = FastAPI( title="Notes App", description="Personal Vocabulary & Language Notebook (notes.qpid.pro)", version="2.1.0", lifespan=lifespan, ) # GZip Compression Middleware (compresses responses >= 1000 bytes) app.add_middleware(GZipMiddleware, minimum_size=1000) # Mount static files app.mount("/static", StaticFiles(directory=str(config.BASE_DIR / "app" / "static")), name="static") # Security Headers Middleware @app.middleware("http") async def add_security_headers(request: Request, call_next): response = await call_next(request) response.headers["X-Frame-Options"] = "SAMEORIGIN" response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "same-origin" return response # --------------------------------------------------------------------------- # Auth dependencies # --------------------------------------------------------------------------- def get_optional_user(request: Request) -> Optional[dict]: token = request.cookies.get(config.SESSION_COOKIE_NAME) if not token: auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): token = auth_header[7:].strip() if token: return get_user_from_session(token) return None def require_user(request: Request) -> dict: user = get_optional_user(request) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required", ) return user # --------------------------------------------------------------------------- # Request / Response models # --------------------------------------------------------------------------- class LoginRequest(BaseModel): username: str = Field(..., min_length=1) password: str = Field(..., min_length=1) class ChangePasswordRequest(BaseModel): old_password: str = Field(..., min_length=1) new_password: str = Field(..., min_length=6) class CardCreateRequest(BaseModel): term: str = Field(..., min_length=1) translation: str = Field(..., min_length=1) context_sentence: Optional[str] = "" phonetic: Optional[str] = "" audio_url: Optional[str] = "" part_of_speech: Optional[str] = "" definition_en: Optional[str] = "" tags: Optional[str] = "general" notes: Optional[str] = "" collection: Optional[str] = "General" class CardUpdateRequest(BaseModel): term: Optional[str] = None translation: Optional[str] = None context_sentence: Optional[str] = None phonetic: Optional[str] = None audio_url: Optional[str] = None part_of_speech: Optional[str] = None definition_en: Optional[str] = None tags: Optional[str] = None notes: Optional[str] = None collection: Optional[str] = None class BatchActionRequest(BaseModel): card_ids: List[int] = Field(..., min_length=1) action: str = Field(..., pattern="^(set_collection|add_tags|delete)$") value: Optional[str] = "" class SrsReviewRequest(BaseModel): card_id: int rating: int = Field(..., ge=1, le=4) class BookCreateRequest(BaseModel): title: Optional[str] = "Untitled" author: Optional[str] = "" content: Optional[str] = "" source_type: Optional[str] = "text" class BookProgressRequest(BaseModel): current_chapter: Optional[int] = None current_position: Optional[float] = None chapter: Optional[int] = None position: Optional[float] = None # --------------------------------------------------------------------------- # PWA Static Endpoints # --------------------------------------------------------------------------- @app.get("/manifest.json", include_in_schema=False) async def manifest_endpoint(): manifest_path = config.BASE_DIR / "app" / "static" / "manifest.json" if manifest_path.exists(): return Response( content=manifest_path.read_text(encoding="utf-8"), media_type="application/manifest+json; charset=utf-8", headers={"Cache-Control": "public, max-age=86400"}, ) raise HTTPException(status_code=404, detail="Manifest not found") @app.get("/favicon.ico", include_in_schema=False) async def favicon_endpoint(): icon_path = config.BASE_DIR / "app" / "static" / "icons" / "icon-192.png" if icon_path.exists(): return Response( content=icon_path.read_bytes(), media_type="image/png", headers={"Cache-Control": "public, max-age=86400"}, ) raise HTTPException(status_code=404, detail="Favicon not found") @app.get("/sw.js", include_in_schema=False) async def service_worker_endpoint(): sw_path = config.BASE_DIR / "app" / "static" / "sw.js" if sw_path.exists(): return Response( content=sw_path.read_text(encoding="utf-8"), media_type="application/javascript; charset=utf-8", headers={ "Cache-Control": "no-cache, no-store, must-revalidate", "Service-Worker-Allowed": "/", }, ) raise HTTPException(status_code=404, detail="Service worker not found") # --------------------------------------------------------------------------- # HTML entry point # --------------------------------------------------------------------------- @app.api_route("/", methods=["GET", "HEAD"], response_class=HTMLResponse) async def index_page(request: Request): index_file = config.BASE_DIR / "app" / "static" / "index.html" if not index_file.exists(): raise HTTPException(status_code=404, detail="Index file not found") with open(index_file, "r", encoding="utf-8") as f: html_content = f.read() return HTMLResponse(content=html_content, status_code=200) # --------------------------------------------------------------------------- # Auth API # --------------------------------------------------------------------------- @app.post("/api/login") async def api_login(req: LoginRequest, request: Request, response: Response): ip = request.client.host if request.client else "unknown" if not check_login_rate_limit(ip): raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many login attempts. Please try again in a minute.", ) user = authenticate_user(req.username, req.password) if not user: record_login_failure(ip) raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid username or password", ) record_login_success(ip) user_agent = request.headers.get("User-Agent", "") session_token = create_session(user["id"], user_agent=user_agent, ip=ip) response.set_cookie( key=config.SESSION_COOKIE_NAME, value=session_token, max_age=config.SESSION_MAX_AGE, httponly=True, samesite="lax", secure=False, ) return {"status": "ok", "user": {"id": user["id"], "username": user["username"]}} @app.post("/api/logout") async def api_logout(request: Request, response: Response): token = request.cookies.get(config.SESSION_COOKIE_NAME) if token: delete_session(token) response.delete_cookie(key=config.SESSION_COOKIE_NAME) return {"status": "ok"} @app.get("/api/me") async def api_me(user: dict = Depends(require_user)): return {"status": "ok", "user": {"id": user["id"], "username": user["username"]}} @app.post("/api/change-password") @app.post("/api/auth/change-password") async def api_change_password(req: ChangePasswordRequest, user: dict = Depends(require_user)): ok, msg = change_user_password(user["id"], req.old_password, req.new_password) if not ok: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=msg or "Current password does not match", ) return {"status": "ok", "message": "Password changed successfully"} # --------------------------------------------------------------------------- # Dictionary Lookup & Audio Proxy # --------------------------------------------------------------------------- @app.get("/api/lookup") async def api_lookup( term: Optional[str] = None, q: Optional[str] = None, user: dict = Depends(require_user) ): query_str = (term or q or "").strip() if not query_str: raise HTTPException(status_code=400, detail="Query term cannot be empty") data = await lookup_term(query_str) # Check if this term already exists in user's dictionary with get_db() as conn: existing = conn.execute( "SELECT id, term, translation, context_sentence, tags, collection, srs_state, srs_reps FROM vocabulary_cards WHERE user_id = ? AND lower(trim(term)) = lower(trim(?))", (user["id"], query_str), ).fetchone() if existing: data["in_library"] = True data["library_card"] = dict(existing) else: data["in_library"] = False data["library_card"] = None return data @app.api_route("/api/audio", methods=["GET", "HEAD"]) async def api_audio_proxy(): # Audio feature disabled return Response(status_code=204) # --------------------------------------------------------------------------- # Collections # --------------------------------------------------------------------------- @app.get("/api/collections") async def api_get_collections(user: dict = Depends(require_user)): """Returns all collections for the current user with card count.""" with get_db() as conn: rows = conn.execute( """ SELECT collection, COUNT(*) as count FROM vocabulary_cards WHERE user_id = ? GROUP BY collection ORDER BY count DESC, collection ASC """, (user["id"],), ).fetchall() return [{"name": r["collection"], "collection": r["collection"], "count": r["count"]} for r in rows] # --------------------------------------------------------------------------- # Cards CRUD & Duplicate Check # --------------------------------------------------------------------------- @app.get("/api/cards/check-duplicate") async def api_check_duplicate(term: str = Query(..., min_length=1), user: dict = Depends(require_user)): clean_term = term.strip().lower() with get_db() as conn: row = conn.execute( "SELECT * FROM vocabulary_cards WHERE user_id = ? AND lower(trim(term)) = ?", (user["id"], clean_term), ).fetchone() if row: return {"exists": True, "card": dict(row)} return {"exists": False, "card": None} @app.get("/api/cards") async def api_get_cards( search: Optional[str] = None, tag: Optional[str] = None, collection: Optional[str] = None, srs_state: Optional[str] = None, limit: int = 100, offset: int = 0, user: dict = Depends(require_user), ): with get_db() as conn: query = """ SELECT id, term, translation, context_sentence, phonetic, audio_url, part_of_speech, definition_en, tags, notes, collection, srs_reps, srs_interval, srs_ease, srs_next_review, srs_last_review, srs_state, error_count, created_at, updated_at FROM vocabulary_cards WHERE user_id = ? """ params = [user["id"]] if search: s_clean = f"%{search.strip()}%" query += " AND (term LIKE ? OR translation LIKE ? OR context_sentence LIKE ? OR notes LIKE ?)" params.extend([s_clean, s_clean, s_clean, s_clean]) if tag: t_clean = f"%{tag.strip()}%" query += " AND tags LIKE ?" params.append(t_clean) if collection: query += " AND collection = ?" params.append(collection.strip()) if srs_state: query += " AND srs_state = ?" params.append(srs_state.strip()) # Total count before limit count_query = f"SELECT COUNT(*) FROM ({query})" total = conn.execute(count_query, params).fetchone()[0] query += " ORDER BY created_at DESC LIMIT ? OFFSET ?" params.extend([limit, offset]) rows = conn.execute(query, params).fetchall() cards = [dict(r) for r in rows] return {"total": total, "cards": cards} @app.post("/api/cards") async def api_create_card(card: CardCreateRequest, user: dict = Depends(require_user)): clean_term = card.term.strip() with get_db() as conn: existing = conn.execute( "SELECT * FROM vocabulary_cards WHERE user_id = ? AND lower(trim(term)) = lower(trim(?))", (user["id"], clean_term), ).fetchone() if existing: # Word already exists in library! existing_row = dict(existing) new_ctx = (card.context_sentence or "").strip() old_ctx = (existing_row.get("context_sentence") or "").strip() updated_ctx = old_ctx if new_ctx and new_ctx not in old_ctx: updated_ctx = f"{old_ctx}\n\n{new_ctx}" if old_ctx else new_ctx # Merge tags old_tags = set([t.strip() for t in (existing_row.get("tags") or "").split(",") if t.strip()]) new_tags = [t.strip() for t in (card.tags or "").split(",") if t.strip()] for t in new_tags: old_tags.add(t) updated_tags = ", ".join(sorted(old_tags)) if old_tags else "general" conn.execute( """ UPDATE vocabulary_cards SET context_sentence = ?, tags = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, (updated_ctx, updated_tags, existing_row["id"]), ) updated_row = conn.execute("SELECT * FROM vocabulary_cards WHERE id = ?", (existing_row["id"],)).fetchone() res = dict(updated_row) res["already_exists"] = True return res cursor = conn.execute( """ INSERT INTO vocabulary_cards ( user_id, term, translation, context_sentence, phonetic, audio_url, part_of_speech, definition_en, tags, notes, collection, srs_reps, srs_interval, srs_ease, srs_state, error_count ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, 2.5, 'new', 0) """, ( user["id"], clean_term, card.translation.strip(), (card.context_sentence or "").strip(), (card.phonetic or "").strip(), (card.audio_url or "").strip(), (card.part_of_speech or "").strip(), (card.definition_en or "").strip(), (card.tags or "general").strip(), (card.notes or "").strip(), (card.collection or "General").strip(), ), ) new_id = cursor.lastrowid row = conn.execute("SELECT * FROM vocabulary_cards WHERE id = ?", (new_id,)).fetchone() return dict(row) @app.put("/api/cards/{card_id}") async def api_update_card(card_id: int, update: CardUpdateRequest, user: dict = Depends(require_user)): with get_db() as conn: row = conn.execute( "SELECT * FROM vocabulary_cards WHERE id = ? AND user_id = ?", (card_id, user["id"]), ).fetchone() if not row: raise HTTPException(status_code=404, detail="Card not found") current = dict(row) def pick(new_val, current_val): return new_val.strip() if new_val is not None else current_val conn.execute( """ UPDATE vocabulary_cards SET term = ?, translation = ?, context_sentence = ?, phonetic = ?, audio_url = ?, part_of_speech = ?, definition_en = ?, tags = ?, notes = ?, collection = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND user_id = ? """, ( pick(update.term, current["term"]), pick(update.translation, current["translation"]), pick(update.context_sentence, current.get("context_sentence", "")), pick(update.phonetic, current.get("phonetic", "")), pick(update.audio_url, current.get("audio_url", "")), pick(update.part_of_speech, current.get("part_of_speech", "")), pick(update.definition_en, current.get("definition_en", "")), pick(update.tags, current.get("tags", "")), pick(update.notes, current.get("notes", "")), pick(update.collection, current.get("collection", "General")), card_id, user["id"], ), ) updated_row = conn.execute("SELECT * FROM vocabulary_cards WHERE id = ?", (card_id,)).fetchone() return dict(updated_row) @app.delete("/api/cards/{card_id}") async def api_delete_card(card_id: int, user: dict = Depends(require_user)): with get_db() as conn: cursor = conn.execute( "DELETE FROM vocabulary_cards WHERE id = ? AND user_id = ?", (card_id, user["id"]), ) if cursor.rowcount == 0: raise HTTPException(status_code=404, detail="Card not found") return {"status": "ok", "deleted_id": card_id} # --------------------------------------------------------------------------- # Batch operations # --------------------------------------------------------------------------- @app.post("/api/cards/batch") async def api_batch_action(req: BatchActionRequest, user: dict = Depends(require_user)): if not req.card_ids: raise HTTPException(status_code=400, detail="No card IDs provided") with get_db() as conn: placeholders = ",".join("?" * len(req.card_ids)) params_base = list(req.card_ids) + [user["id"]] if req.action == "delete": cursor = conn.execute( f"DELETE FROM vocabulary_cards WHERE id IN ({placeholders}) AND user_id = ?", params_base, ) return {"status": "ok", "action": "delete", "affected": cursor.rowcount} elif req.action == "set_collection": collection_val = (req.value or "General").strip() cursor = conn.execute( f"UPDATE vocabulary_cards SET collection = ?, updated_at = CURRENT_TIMESTAMP WHERE id IN ({placeholders}) AND user_id = ?", [collection_val] + params_base, ) return {"status": "ok", "action": "set_collection", "affected": cursor.rowcount} elif req.action == "add_tags": new_tags_raw = req.value or "" new_tags = [t.strip() for t in new_tags_raw.split(",") if t.strip()] if not new_tags: return {"status": "ok", "action": "add_tags", "affected": 0} rows = conn.execute( f"SELECT id, tags FROM vocabulary_cards WHERE id IN ({placeholders}) AND user_id = ?", params_base, ).fetchall() affected = 0 for r in rows: existing = [t.strip() for t in (r["tags"] or "").split(",") if t.strip()] combined = existing + [t for t in new_tags if t not in existing] conn.execute( "UPDATE vocabulary_cards SET tags = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", (", ".join(combined), r["id"]), ) affected += 1 return {"status": "ok", "action": "add_tags", "affected": affected} else: raise HTTPException(status_code=422, detail=f"Unknown action: {req.action}") # --------------------------------------------------------------------------- # SRS Spaced Repetition Endpoints & Smart Quiz # --------------------------------------------------------------------------- @app.get("/api/srs/queue") @app.get("/api/srs/due") async def api_srs_queue( collection: Optional[str] = None, tag: Optional[str] = None, limit: int = 25, mix: bool = False, user: dict = Depends(require_user), ): """ Returns queue of cards ready for study. If mix=True (Smart Daily Mix): - up to 10 new cards - up to 10 review due cards - up to 5 leeches / problematic cards (error_count > 0 or low ease) Shuffled for interleaved recall. """ with get_db() as conn: now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") col_filter = "" col_params = [] if collection: col_filter += " AND collection = ?" col_params.append(collection.strip()) if tag: col_filter += " AND tags LIKE ?" col_params.append(f"%{tag.strip()}%") if mix: # 1. New cards (up to 10) new_rows = conn.execute( f""" SELECT * FROM vocabulary_cards WHERE user_id = ? AND srs_state = 'new'{col_filter} ORDER BY created_at DESC LIMIT 10 """, [user["id"]] + col_params, ).fetchall() # 2. Due review cards (up to 10) due_rows = conn.execute( f""" SELECT * FROM vocabulary_cards WHERE user_id = ? AND srs_state IN ('learning', 'review', 'mastered') AND (srs_next_review <= ? OR srs_next_review IS NULL){col_filter} ORDER BY srs_next_review ASC LIMIT 10 """, [user["id"], now_str] + col_params, ).fetchall() # 3. Problematic leeches (up to 5 with highest error count) used_ids = {r["id"] for r in new_rows} | {r["id"] for r in due_rows} leech_rows = conn.execute( f""" SELECT * FROM vocabulary_cards WHERE user_id = ? AND (error_count > 0 OR srs_ease < 2.0){col_filter} ORDER BY error_count DESC, srs_ease ASC LIMIT 10 """, [user["id"]] + col_params, ).fetchall() leech_filtered = [r for r in leech_rows if r["id"] not in used_ids][:5] combined = [dict(r) for r in new_rows] + [dict(r) for r in due_rows] + [dict(r) for r in leech_filtered] import random random.shuffle(combined) cards = combined[:limit] else: # Standard SRS due queue: Due cards first, then new cards query = f""" SELECT * FROM vocabulary_cards WHERE user_id = ?{col_filter} AND (srs_state = 'new' OR srs_next_review <= ? OR srs_next_review IS NULL) ORDER BY CASE WHEN srs_state != 'new' AND (srs_next_review <= ? OR srs_next_review IS NULL) THEN 0 ELSE 1 END, srs_next_review ASC, created_at DESC LIMIT ? """ params = [user["id"]] + col_params + [now_str, now_str, limit] rows = conn.execute(query, params).fetchall() cards = [dict(r) for r in rows] return {"count": len(cards), "cards": cards} @app.get("/api/srs/quiz") async def api_srs_quiz( collection: Optional[str] = None, tag: Optional[str] = None, count: int = 10, user: dict = Depends(require_user), ): """ Generates a Multiple-Choice Quiz (4 options per question). """ with get_db() as conn: # Fetch target cards for user col_filter = "" col_params = [] if collection: col_filter += " AND collection = ?" col_params.append(collection.strip()) if tag: col_filter += " AND tags LIKE ?" col_params.append(f"%{tag.strip()}%") rows = conn.execute( f"SELECT * FROM vocabulary_cards WHERE user_id = ?{col_filter} ORDER BY RANDOM() LIMIT ?", [user["id"]] + col_params + [count], ).fetchall() target_cards = [dict(r) for r in rows] # Fetch pool of all user cards for generating smart distractors all_rows = conn.execute( "SELECT id, term, translation, part_of_speech FROM vocabulary_cards WHERE user_id = ?", (user["id"],), ).fetchall() all_cards = [dict(r) for r in all_rows] quiz_items = [] for card in target_cards: options = generate_quiz_options(card, all_cards, count=4) quiz_items.append({ "card": card, "options": options, }) return {"count": len(quiz_items), "quiz": quiz_items} @app.post("/api/srs/review") @app.post("/api/srs/review/{card_id_path}") async def api_srs_review( req: Optional[SrsReviewRequest] = None, card_id_path: Optional[int] = None, rating_param: Optional[int] = Query(None, alias="rating"), request: Request = None, user: dict = Depends(require_user), ): """ Records a review rating (1..4) and updates SM-2 state & activity logs. """ target_card_id = req.card_id if req else card_id_path target_rating = req.rating if req else rating_param if target_card_id is None: try: body = await request.json() target_card_id = body.get("card_id") target_rating = body.get("rating") except Exception: pass if target_card_id is None or target_rating is None: raise HTTPException(status_code=400, detail="Missing card_id or rating") if target_rating not in (1, 2, 3, 4): raise HTTPException(status_code=422, detail="Rating must be between 1 and 4") with get_db() as conn: row = conn.execute( "SELECT * FROM vocabulary_cards WHERE id = ? AND user_id = ?", (target_card_id, user["id"]), ).fetchone() if not row: raise HTTPException(status_code=404, detail="Card not found") current = dict(row) new_srs = calculate_sm2( rating=target_rating, reps=current.get("srs_reps", 0), interval=current.get("srs_interval", 0), ease=current.get("srs_ease", 2.5), state=current.get("srs_state", "new"), error_count=current.get("error_count", 0), ) conn.execute( """ UPDATE vocabulary_cards SET srs_reps = ?, srs_interval = ?, srs_ease = ?, srs_state = ?, srs_last_review = ?, srs_next_review = ?, error_count = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND user_id = ? """, ( new_srs["srs_reps"], new_srs["srs_interval"], new_srs["srs_ease"], new_srs["srs_state"], new_srs["srs_last_review"], new_srs["srs_next_review"], new_srs["error_count"], target_card_id, user["id"], ), ) # Record review log conn.execute( """ INSERT INTO srs_logs (user_id, card_id, rating, interval, ease, reviewed_at) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP) """, (user["id"], target_card_id, target_rating, new_srs["srs_interval"], new_srs["srs_ease"]), ) updated_row = conn.execute("SELECT * FROM vocabulary_cards WHERE id = ?", (target_card_id,)).fetchone() return {"status": "ok", "card": dict(updated_row)} @app.get("/api/srs/stats") async def api_srs_stats(user: dict = Depends(require_user)): """ Returns full learning analytics, SRS states, retention rate, streaks. """ with get_db() as conn: now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") # Total cards total_cards = conn.execute( "SELECT COUNT(*) FROM vocabulary_cards WHERE user_id = ?", (user["id"],), ).fetchone()[0] # By state states = conn.execute( """ SELECT srs_state, COUNT(*) as count FROM vocabulary_cards WHERE user_id = ? GROUP BY srs_state """, (user["id"],), ).fetchall() state_counts = {r["srs_state"]: r["count"] for r in states} new_cards = state_counts.get("new", 0) learning = state_counts.get("learning", 0) review = state_counts.get("review", 0) mastered = state_counts.get("mastered", 0) # Due now due_cards = conn.execute( """ SELECT COUNT(*) FROM vocabulary_cards WHERE user_id = ? AND (srs_state = 'new' OR srs_next_review <= ? OR srs_next_review IS NULL) """, (user["id"], now_str), ).fetchone()[0] # Leeches leech_count = conn.execute( """ SELECT COUNT(*) FROM vocabulary_cards WHERE user_id = ? AND (error_count > 0 OR srs_ease < 2.0) """, (user["id"],), ).fetchone()[0] # Total reviews and retention review_logs = conn.execute( "SELECT rating, date(reviewed_at) as rev_date FROM srs_logs WHERE user_id = ?", (user["id"],), ).fetchall() total_reviews = len(review_logs) good_reviews = sum(1 for r in review_logs if r["rating"] in (3, 4)) retention_rate = round((good_reviews / total_reviews * 100), 1) if total_reviews > 0 else 100.0 # Daily review aggregation for streak date_counts: Dict[str, int] = {} for r in review_logs: d = r["rev_date"] if d: date_counts[d] = date_counts.get(d, 0) + 1 heatmap_info = build_activity_heatmap(date_counts, days=180) # Today's activity today_str = date.today().strftime("%Y-%m-%d") today_added = conn.execute( "SELECT COUNT(*) FROM vocabulary_cards WHERE user_id = ? AND date(created_at) = date('now')", (user["id"],), ).fetchone()[0] today_reviewed = date_counts.get(today_str, 0) cefr_info = calculate_cefr_level(mastered, review, retention_rate, total_cards) return { "total_cards": total_cards, "new_cards": new_cards, "learning": learning, "review": review, "mastered": mastered, "due_cards": due_cards, "leech_count": leech_count, "total_reviews": total_reviews, "retention_rate": retention_rate, "streak_days": heatmap_info["current_streak"], "longest_streak": heatmap_info["longest_streak"], "today_added": today_added, "today_reviewed": today_reviewed, "cefr": cefr_info, } @app.get("/api/srs/activity-heatmap") async def api_srs_activity_heatmap(days: int = 180, user: dict = Depends(require_user)): """ Returns GitHub-style daily activity heatmap data. """ with get_db() as conn: rows = conn.execute( "SELECT date(reviewed_at) as rev_date, COUNT(*) as cnt FROM srs_logs WHERE user_id = ? GROUP BY date(reviewed_at)", (user["id"],), ).fetchall() date_counts = {r["rev_date"]: r["cnt"] for r in rows if r["rev_date"]} heatmap_data = build_activity_heatmap(date_counts, days=days) return heatmap_data # --------------------------------------------------------------------------- # Tags & Stats # --------------------------------------------------------------------------- @app.get("/api/tags") async def api_get_tags(user: dict = Depends(require_user)): with get_db() as conn: rows = conn.execute( "SELECT tags FROM vocabulary_cards WHERE user_id = ?", (user["id"],), ).fetchall() tag_counts: dict[str, int] = {} for r in rows: raw = r["tags"] if raw: for t in raw.split(","): tag_clean = t.strip() if tag_clean: tag_counts[tag_clean] = tag_counts.get(tag_clean, 0) + 1 sorted_tags = sorted( [{"tag": k, "count": v} for k, v in tag_counts.items()], key=lambda x: x["count"], reverse=True, ) return sorted_tags @app.get("/api/stats") async def api_stats(user: dict = Depends(require_user)): with get_db() as conn: total = conn.execute( "SELECT COUNT(*) FROM vocabulary_cards WHERE user_id = ?", (user["id"],), ).fetchone()[0] today_added = conn.execute( "SELECT COUNT(*) FROM vocabulary_cards WHERE user_id = ? AND date(created_at) = date('now')", (user["id"],), ).fetchone()[0] return {"total_cards": total, "today_added": today_added} # --------------------------------------------------------------------------- # Export endpoints # --------------------------------------------------------------------------- def _fetch_all_cards(user_id: int, collection: Optional[str] = None) -> list: with get_db() as conn: if collection: rows = conn.execute( """ SELECT id, term, translation, context_sentence, phonetic, audio_url, part_of_speech, definition_en, tags, notes, collection, created_at, updated_at FROM vocabulary_cards WHERE user_id = ? AND collection = ? ORDER BY collection ASC, created_at DESC """, (user_id, collection), ).fetchall() else: rows = conn.execute( """ SELECT id, term, translation, context_sentence, phonetic, audio_url, part_of_speech, definition_en, tags, notes, collection, created_at, updated_at FROM vocabulary_cards WHERE user_id = ? ORDER BY collection ASC, created_at DESC """, (user_id,), ).fetchall() return [dict(r) for r in rows] @app.get("/api/export/csv") async def api_export_csv(collection: Optional[str] = None, user: dict = Depends(require_user)): cards = _fetch_all_cards(user["id"], collection) data = export_csv(cards) return Response( content=data, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": "attachment; filename=\"vocabulary.csv\""}, ) @app.get("/api/export/json") async def api_export_json(collection: Optional[str] = None, user: dict = Depends(require_user)): cards = _fetch_all_cards(user["id"], collection) data = export_json(cards) return Response( content=data, media_type="application/json; charset=utf-8", headers={"Content-Disposition": "attachment; filename=\"vocabulary.json\""}, ) @app.get("/api/export/anki") async def api_export_anki(collection: Optional[str] = None, user: dict = Depends(require_user)): cards = _fetch_all_cards(user["id"], collection) data = export_anki_tsv(cards) return Response( content=data, media_type="text/plain; charset=utf-8", headers={"Content-Disposition": "attachment; filename=\"vocabulary_anki.txt\""}, ) @app.get("/api/export/markdown") async def api_export_markdown(collection: Optional[str] = None, user: dict = Depends(require_user)): cards = _fetch_all_cards(user["id"], collection) data = export_markdown(cards) return Response( content=data, media_type="text/markdown; charset=utf-8", headers={"Content-Disposition": "attachment; filename=\"vocabulary.md\""}, ) # --------------------------------------------------------------------------- # Bookshelf & Interactive Reader Endpoints (Stage 5) # --------------------------------------------------------------------------- @app.get("/api/books") async def api_get_books(user: dict = Depends(require_user)): """ Lists books for current user (id, title, author, source_type, chapters_count, total_words, current_chapter, current_position, created_at, updated_at). Excludes full chapters_json for efficiency. """ with get_db() as conn: rows = conn.execute( """ SELECT id, user_id, title, author, source_type, json_array_length(chapters_json) AS chapters_count, total_words, current_chapter, current_position, created_at, updated_at FROM books WHERE user_id = ? ORDER BY updated_at DESC, id DESC """, (user["id"],), ).fetchall() return [dict(r) for r in rows] @app.post("/api/books") async def api_create_book( request: Request, user: dict = Depends(require_user), ): """ Creates or uploads a book. Accepts: - JSON: {"title": "...", "author": "...", "content": "...", "source_type": "text"} - Multipart file upload: file (UploadFile for .epub, .txt, .md) and optional title, author """ content_type = request.headers.get("content-type", "") title = "Untitled" author = "" source_type = "text" chapters = [] if "multipart/form-data" in content_type: form = await request.form() file_obj = form.get("file") if not file_obj or not hasattr(file_obj, "filename") or not file_obj.filename: raise HTTPException(status_code=400, detail="No file uploaded") filename = file_obj.filename file_bytes = await file_obj.read() if not file_bytes: raise HTTPException(status_code=400, detail="Uploaded file is empty") form_title = (form.get("title") or "").strip() form_author = (form.get("author") or "").strip() stem = Path(filename).stem lower_name = filename.lower() if lower_name.endswith(".epub"): source_type = "epub" parsed_title, parsed_author, chapters = parse_epub(file_bytes) title = form_title or (parsed_title if parsed_title != "Untitled" else "") or stem author = form_author or parsed_author elif lower_name.endswith(".md"): source_type = "md" text_str = file_bytes.decode("utf-8", errors="replace") parsed_title, parsed_author, chapters = parse_text(text_str, title=form_title or "Untitled", author=form_author) title = form_title or (parsed_title if parsed_title != "Untitled" else "") or stem author = form_author or parsed_author elif lower_name.endswith(".txt"): source_type = "txt" text_str = file_bytes.decode("utf-8", errors="replace") parsed_title, parsed_author, chapters = parse_text(text_str, title=form_title or "Untitled", author=form_author) title = form_title or (parsed_title if parsed_title != "Untitled" else "") or stem author = form_author or parsed_author else: source_type = "text" text_str = file_bytes.decode("utf-8", errors="replace") parsed_title, parsed_author, chapters = parse_text(text_str, title=form_title or "Untitled", author=form_author) title = form_title or (parsed_title if parsed_title != "Untitled" else "") or stem author = form_author or parsed_author else: try: body = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid JSON or missing body") raw_content = body.get("content") or "" title_in = (body.get("title") or "").strip() author_in = (body.get("author") or "").strip() source_type = body.get("source_type") or "text" if not raw_content and not title_in: raise HTTPException(status_code=400, detail="Book content is required") parsed_title, parsed_author, chapters = parse_text( raw_content, title=title_in or "Untitled", author=author_in ) title = title_in or parsed_title author = author_in or parsed_author total_words = sum(count_words(ch.get("content", "")) for ch in chapters) chapters_json = json.dumps(chapters, ensure_ascii=False) with get_db() as conn: cursor = conn.execute( """ INSERT INTO books ( user_id, title, author, source_type, chapters_json, current_chapter, current_position, total_words ) VALUES (?, ?, ?, ?, ?, 0, 0.0, ?) """, (user["id"], title, author, source_type, chapters_json, total_words), ) new_id = cursor.lastrowid row = conn.execute( """ SELECT id, user_id, title, author, source_type, chapters_json, json_array_length(chapters_json) AS chapters_count, current_chapter, current_position, total_words, created_at, updated_at FROM books WHERE id = ? """, (new_id,), ).fetchone() res = dict(row) res["chapters"] = chapters return res @app.get("/api/books/{book_id}") async def api_get_book(book_id: int, user: dict = Depends(require_user)): """ Returns full book details including chapters_json. """ with get_db() as conn: row = conn.execute( """ SELECT id, user_id, title, author, source_type, chapters_json, json_array_length(chapters_json) AS chapters_count, current_chapter, current_position, total_words, created_at, updated_at FROM books WHERE id = ? AND user_id = ? """, (book_id, user["id"]), ).fetchone() if not row: raise HTTPException(status_code=404, detail="Book not found") data = dict(row) try: data["chapters"] = json.loads(data["chapters_json"]) except Exception: data["chapters"] = [] return data @app.put("/api/books/{book_id}/progress") @app.patch("/api/books/{book_id}/progress") async def api_update_book_progress( book_id: int, req: Optional[BookProgressRequest] = None, current_chapter: Optional[int] = Query(None), current_position: Optional[float] = Query(None), request: Request = None, user: dict = Depends(require_user), ): """ Updates reading progress (current_chapter, current_position) for a book. """ ch = None pos = None if req is not None: ch = req.current_chapter if req.current_chapter is not None else req.chapter pos = req.current_position if req.current_position is not None else req.position if ch is None: ch = current_chapter if pos is None: pos = current_position if ch is None or pos is None: if request: try: body = await request.json() if ch is None: ch = body.get("current_chapter", body.get("chapter")) if pos is None: pos = body.get("current_position", body.get("position")) except Exception: pass if ch is None and pos is None: raise HTTPException(status_code=400, detail="Missing current_chapter or current_position") with get_db() as conn: row = conn.execute( "SELECT id, current_chapter, current_position FROM books WHERE id = ? AND user_id = ?", (book_id, user["id"]), ).fetchone() if not row: raise HTTPException(status_code=404, detail="Book not found") new_ch = int(ch) if ch is not None else row["current_chapter"] new_pos = float(pos) if pos is not None else row["current_position"] conn.execute( """ UPDATE books SET current_chapter = ?, current_position = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND user_id = ? """, (new_ch, new_pos, book_id, user["id"]), ) updated_row = conn.execute( """ SELECT id, user_id, title, author, source_type, chapters_json, json_array_length(chapters_json) AS chapters_count, current_chapter, current_position, total_words, created_at, updated_at FROM books WHERE id = ? """, (book_id,), ).fetchone() res = dict(updated_row) try: res["chapters"] = json.loads(res["chapters_json"]) except Exception: res["chapters"] = [] return res @app.delete("/api/books/{book_id}") async def api_delete_book(book_id: int, user: dict = Depends(require_user)): """ Deletes a book for the current user. """ with get_db() as conn: cursor = conn.execute( "DELETE FROM books WHERE id = ? AND user_id = ?", (book_id, user["id"]), ) if cursor.rowcount == 0: raise HTTPException(status_code=404, detail="Book not found") return {"status": "ok", "deleted_id": book_id}