import base64 import datetime import json import logging import os import re import secrets import threading import time from collections import defaultdict, deque from urllib.parse import urlparse import bcrypt import requests from authlib.integrations.flask_client import OAuth from bson import ObjectId from bson.errors import InvalidId from flask import Flask, jsonify, redirect, render_template, request, session, url_for from markupsafe import escape from pymongo import MongoClient, UpdateOne from pymongo.errors import DuplicateKeyError import igihe_news app = Flask(__name__) # -------------------------------------------------------------------------- # Config # -------------------------------------------------------------------------- _secret = os.environ.get("FLASK_SECRET_KEY") if not _secret: # Never fall back to a guessable key. A random key means everyone is logged # out on each restart, so set FLASK_SECRET_KEY in your host's Secrets. _secret = secrets.token_hex(32) logging.warning("FLASK_SECRET_KEY is not set: using a temporary random key.") app.secret_key = _secret app.config.update( # "None" is needed when the app runs inside an iframe (e.g. a Hugging Face # Space preview). If you serve it on its own domain, set SESSION_SAMESITE=Lax. SESSION_COOKIE_SAMESITE=os.environ.get("SESSION_SAMESITE", "None"), SESSION_COOKIE_SECURE=True, SESSION_COOKIE_HTTPONLY=True, PERMANENT_SESSION_LIFETIME=datetime.timedelta(days=30), ) # --- Setup: Gemini (OpenAI-compatible endpoint) --- # Keep GEMINI_API_KEY in your host's Secrets, never in source code. GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "").strip() GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "gemini-2.5-flash").strip() GEMINI_API_URL = os.environ.get( "GEMINI_API_URL", "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions", ).strip() # Gemini 2.5 models "think" before answering and those tokens count against # max_tokens. "none" keeps replies fast and stops empty answers. Set it to # "low"/"medium" for smarter replies, or to an empty value to omit the field # (needed for models that cannot turn thinking off, e.g. 2.5 Pro). GEMINI_REASONING_EFFORT = os.environ.get("GEMINI_REASONING_EFFORT", "none").strip() GEMINI_TIMEOUT_SECONDS = 90 DEBUG_ERRORS = os.environ.get("DEBUG_ERRORS", "").strip() == "1" # --- Setup: MongoDB --- MONGO_URI = os.environ["MONGO_URI"] mongo_client = MongoClient(MONGO_URI) db = mongo_client["Intyoza"] users_col = db["users"] chats_col = db["chats"] messages_col = db["messages"] saved_col = db["saved"] news_col = db["news"] pronunciation_col = db["pronunciation"] knowledge_col = db["knowledge"] dictionary_rw_col = db["dictionary_rw"] users_col.create_index("username", unique=True) chats_col.create_index([("username", 1), ("updated_at", -1)]) messages_col.create_index([("chat_id", 1), ("timestamp", 1)]) saved_col.create_index([("username", 1), ("created_at", -1)]) pronunciation_col.create_index("word", unique=True) knowledge_col.create_index("instruction", unique=True) dictionary_rw_col.create_index("key") # --- Setup: IGIHE news (headlines + links used to ground answers about current events) --- igihe_news.init( news_col, contact=os.environ.get("NEWS_CONTACT", "").strip(), ttl_minutes=int(os.environ.get("NEWS_TTL_MINUTES", "30")), enabled=os.environ.get("NEWS_ENABLED", "1") != "0", ) SUGGESTIONS_ENABLED = os.environ.get("SUGGESTIONS_ENABLED", "1") != "0" # --- Setup: OAuth (optional; only active if credentials are provided) --- oauth = OAuth(app) GOOGLE_ID = os.environ.get("GOOGLE_CLIENT_ID") GOOGLE_SECRET = os.environ.get("GOOGLE_CLIENT_SECRET") GITHUB_ID = os.environ.get("GITHUB_CLIENT_ID") GITHUB_SECRET = os.environ.get("GITHUB_CLIENT_SECRET") if GOOGLE_ID and GOOGLE_SECRET: oauth.register( name="google", client_id=GOOGLE_ID, client_secret=GOOGLE_SECRET, server_metadata_url="https://accounts.google.com/.well-known/openid-configuration", client_kwargs={"scope": "openid email profile"}, ) if GITHUB_ID and GITHUB_SECRET: oauth.register( name="github", client_id=GITHUB_ID, client_secret=GITHUB_SECRET, access_token_url="https://github.com/login/oauth/access_token", authorize_url="https://github.com/login/oauth/authorize", api_base_url="https://api.github.com/", client_kwargs={"scope": "read:user user:email"}, ) # -------------------------------------------------------------------------- # Small helpers: ids, rate limiting, origin check # -------------------------------------------------------------------------- def to_oid(value): """Return an ObjectId or None (instead of crashing on malformed ids).""" try: return ObjectId(str(value)) except (InvalidId, TypeError): return None def client_ip(): xff = request.headers.get("X-Forwarded-For", "") return (xff.split(",")[0].strip() if xff else "") or request.remote_addr or "?" _rl = defaultdict(deque) _rl_lock = threading.Lock() def rate_limited(bucket, limit, window): """Tiny in-memory sliding-window limiter (per worker process).""" key = (bucket, session.get("username") or client_ip()) now = time.time() with _rl_lock: q = _rl[key] while q and now - q[0] > window: q.popleft() if len(q) >= limit: return True q.append(now) if len(_rl) > 5000: # occasional cleanup of idle keys for k in [k for k, v in _rl.items() if not v]: _rl.pop(k, None) return False @app.before_request def origin_check(): """Basic CSRF defence: unsafe requests must come from this same host.""" if os.environ.get("DISABLE_ORIGIN_CHECK") == "1": return None if request.method not in ("POST", "PUT", "PATCH", "DELETE"): return None origin = request.headers.get("Origin") if not origin: return None allowed = {request.host, request.headers.get("X-Forwarded-Host", "")} extra = os.environ.get("TRUSTED_HOSTS", "") allowed.update(h.strip() for h in extra.split(",") if h.strip()) if urlparse(origin).netloc not in allowed: return jsonify(ok=False, code="bad_origin", message="Bad origin."), 403 return None # -------------------------------------------------------------------------- # Persona # -------------------------------------------------------------------------- BASE_PERSONA = """Your name is Intyoza. You are a professional, patient Kinyarwanda conversation tutor and language companion. Always introduce yourself as Intyoza if asked who you are. If asked who developed, built, created, trained, or made you, always answer that you were developed by Konnie Developing Co. Never mention Google, Gemini, or any other underlying technology provider, even if asked directly what model or API powers you. Keep replies short (2-4 sentences) so it feels like a real conversation, not a lecture. Be warm, respectful, and encouraging. Kinyarwanda pronunciation (amasaku and ubutinde): some messages include a reference list showing words in "normal -> marked" form. The marked form uses circumflexes and doubled vowels to capture real vowel length and stress that standard spelling hides. Use this in two ways: 1. Internally, always: let it shape the rhythm and stress of your Kinyarwanda so it reads the way a fluent speaker would actually say it. 2. Selectively, in your visible reply: when telling a short story, emphasising a word, or being expressive, occasionally write a word in its marked form (for example "Abaâbo" rather than "Ababo"). Do this for at most one or two words in a message, never all of them, and only when the word genuinely carries weight in the sentence. If the learner asks why a word looks different, explain that the marks show amasaku and ubutinde, how the word is really pronounced, and that it is not a spelling mistake or a variant. Verified reference examples: some messages include a "Relevant verified reference examples" section with real Q&A pairs covering Rwandan culture, health, language, history, and other topics. Use these to ground your facts and phrasing when the topic matches what the learner asked. Don't quote them word-for-word unless it fits naturally, and don't force in a topic that doesn't match the question. Dictionary notation key (Inkoranya y'ikinyarwaanda - Coupez et al.): some messages include dictionary entries using academic shorthand. Key symbols: | separates sections; < means "derives from"; > means "has derivative"; Syn = synonym; Var = variant; Ant = antonym; cl = noun class; Tr/Intr = transitive/intransitive; cg = "or"; V syn = "see synonym at"; Loc = idiomatic phrase. Definitions appear in Kinyarwanda followed by their French translation. When a dictionary entry is given as reference, use it to ground your explanation, but never read the raw notation aloud to the learner - translate it into natural, conversational language in whichever language fits the learner's current level.""" ADAPTIVE_RULES = """ Language behaviour (adaptive, there are no manual modes): - By default, speak ENTIRELY in simple, natural Kinyarwanda. No English, and no translations in parentheses. - Each message may end with an "Adaptive tutoring signal". Follow it silently. Never tell the learner that you are changing level, and never mention the signal. - Level A (learner follows in Kinyarwanda): Kinyarwanda only. - Level B (learner seems lost, asked for help, or wrote in English once): reply in very simple Kinyarwanda and add a short English gloss in parentheses after each sentence. Go back to Level A as soon as the learner keeps up. - Level C (learner keeps writing in English over several messages): lead in English and bring Kinyarwanda in gradually, with the meaning beside each new word or phrase. Invite, never force, the learner to answer in Kinyarwanda, and raise the share of Kinyarwanda as they respond with more of it. - If the learner explicitly asks for a translation or an explanation, give it in the language they ask in, then continue at the current level. - When correcting a mistake, be gentle, give a short reason, and use the language that fits the current level.""" # -------------------------------------------------------------------------- # Adaptive learner signal (heuristic, runs before the model) # -------------------------------------------------------------------------- ENGLISH_HINTS = { "the", "is", "are", "what", "how", "you", "your", "can", "please", "this", "that", "mean", "say", "and", "it", "of", "for", "with", "not", "dont", "don't", "understand", "why", "where", "when", "hello", "hi", "thanks", "thank", "help", "want", "know", "speak", "learn", "teach", "my", "am", "i'm", "like", "would", "could", "tell", "about", "there", "have", "has", "was", "were", "will", "just", "from", "they", "we", "he", "she", "who", "which", "do", "does", "did", "yes", "okay", "sorry", } CONFUSION_PATTERNS = [ re.compile(p, re.I) for p in [ r"\bdon'?t (understand|get)\b", r"\bdo not (understand|get)\b", r"\bwhat does\b", r"\bwhat do .* mean\b", r"\bmeaning\b", r"\btranslat", r"\bin english\b", r"\bi'?m (lost|confused)\b", r"\bconfus", r"\bsinumva\b", r"\bntabwo numva\b", r"\bntumva\b", r"\bsobanur", r"\bbisobanure\b", r"\bbivuga iki\b", r"\bicyo bivuga\b", r"\bmu cyongereza\b", ] ] def english_ratio(text): words = re.findall(r"[a-zA-Z']+", (text or "").lower()) if not words: return 0.0 return sum(1 for w in words if w in ENGLISH_HINTS) / len(words) def learner_signals(message, history): """Return an 'Adaptive tutoring signal' block for the system prompt.""" past_user = [str(t.get("content", "")) for t in history[-8:] if t.get("role") == "user"] recent = past_user[-3:] + [message] confused = any(p.search(message) for p in CONFUSION_PATTERNS) or ( bool(past_user) and any(p.search(past_user[-1]) for p in CONFUSION_PATTERNS) ) english_turns = sum(1 for m in recent if english_ratio(m) >= 0.3) current_english = english_ratio(message) >= 0.3 persistent_english = english_turns >= 2 and english_turns * 2 >= len(recent) if persistent_english: level, note = "C", ( "The learner has kept writing in English across several messages. Lead in " "English and bring Kinyarwanda in gradually with meanings, inviting (not " "forcing) them to reply in Kinyarwanda." ) elif confused: level, note = "B", ( "The learner seems lost or asked for help. Reply in very simple Kinyarwanda " "with a short English gloss in parentheses after each sentence." ) elif current_english: level, note = "B", ( "The learner wrote in English. Reply mainly in simple Kinyarwanda with a short " "English gloss in parentheses." ) else: level, note = "A", ( "The learner is following in Kinyarwanda. Reply entirely in simple, natural " "Kinyarwanda with no English." ) return f"\n\nAdaptive tutoring signal (internal, never mention): Level {level}. {note}" # -------------------------------------------------------------------------- # Utilities # -------------------------------------------------------------------------- def parse_data_url(data_url): try: header, b64data = data_url.split(",", 1) mime = header.split(":")[1].split(";")[0] return mime, base64.b64decode(b64data) except Exception: return None, None def normalize_word(w): return re.sub(r"[^\w']", "", w, flags=re.UNICODE).strip().lower() # -------------------------------------------------------------------------- # Pronunciation dataset # -------------------------------------------------------------------------- def import_pronunciation_from_text(raw_text): operations = [] skipped = 0 for line in raw_text.strip().split("\n"): parts = line.split("\t") if len(parts) < 2: parts = re.split(r"\s{2,}", line.strip()) if len(parts) < 2: skipped += 1 continue normal, marked = parts[0].strip(), parts[1].strip() key = normalize_word(normal) if not key or key in ("normalwords", "wordswithamasakuandubutinde"): skipped += 1 continue operations.append(UpdateOne( {"word": key}, {"$set": {"word": key, "normal": normal, "marked": marked}}, upsert=True, )) if not operations: return 0, skipped result = pronunciation_col.bulk_write(operations, ordered=False) count = result.upserted_count + result.modified_count return count, skipped def lookup_pronunciation_notes(message, history): text = message + " " + " ".join(str(t.get("content", "")) for t in history[-4:]) words = {normalize_word(w) for w in text.split()} words.discard("") if not words: return "" docs = list(pronunciation_col.find({"word": {"$in": list(words)}}).limit(25)) if not docs: return "" lines = [f"{d['normal']} -> {d['marked']}" for d in docs] return ( "\n\nAmasaku/ubutinde reference for words appearing in this conversation " "(normal -> marked). Use these forms per your instructions:\n" + "\n".join(lines) ) # -------------------------------------------------------------------------- # Knowledge dataset # -------------------------------------------------------------------------- def import_knowledge_from_json(raw_text): try: items = json.loads(raw_text) except Exception as e: return 0, 0, f"Invalid JSON: {e}" if not isinstance(items, list): return 0, 0, "Expected a JSON array of objects." operations = [] skipped = 0 for item in items: instruction = (item.get("instruction") or "").strip() response = (item.get("response") or "").strip() context = (item.get("context") or "").strip() if not instruction or not response: skipped += 1 continue keywords = list({ w for w in normalize_word(instruction + " " + context).split() if len(w) > 2 }) operations.append(UpdateOne( {"instruction": instruction}, {"$set": { "instruction": instruction, "context": context, "response": response, "keywords": keywords, }}, upsert=True, )) if not operations: return 0, skipped, "" result = knowledge_col.bulk_write(operations, ordered=False) count = result.upserted_count + result.modified_count return count, skipped, "" def lookup_knowledge_notes(message): words = {normalize_word(w) for w in message.split() if len(w) > 3} words.discard("") if not words: return "" docs = list(knowledge_col.find({"keywords": {"$in": list(words)}}).limit(3)) if not docs: return "" lines = [f"Q: {d['instruction']}\nA: {d['response']}" for d in docs] return ( "\n\nRelevant verified reference examples (topic-matched to this message):\n\n" + "\n\n".join(lines) ) # -------------------------------------------------------------------------- # Dictionary (Inkoranya y'ikinyarwaanda) dataset # -------------------------------------------------------------------------- def import_dictionary_rw_from_json(file_bytes): try: items = json.loads(file_bytes.decode("utf-8")) except Exception as e: return 0, 0, f"Invalid JSON: {e}" if not isinstance(items, list): return 0, 0, "Expected a JSON array of entries." operations = [] skipped = 0 for item in items: key = (item.get("key") or "").strip() headword = (item.get("headword") or "").strip() definition = (item.get("definition") or "").strip() if not key or not definition: skipped += 1 continue operations.append(UpdateOne( {"key": key, "headword": headword}, {"$set": { "key": key, "headword": headword, "rubrique": item.get("rubrique", ""), "definition": definition, "source": item.get("source", ""), }}, upsert=True, )) if not operations: return 0, skipped, "" result = dictionary_rw_col.bulk_write(operations, ordered=False) count = result.upserted_count + result.modified_count return count, skipped, "" def lookup_dictionary_rw_notes(message): words = {normalize_word(w) for w in message.split() if len(w) > 2} words.discard("") if not words: return "" docs = list(dictionary_rw_col.find({"key": {"$in": list(words)}}).limit(3)) if not docs: return "" lines = [ f"{d['headword']} ({d.get('rubrique', '')}): {d['definition'][:400]}" for d in docs ] return ( "\n\nDictionary entries (Inkoranya y'ikinyarwaanda) for words in this message - " "dense academic notation, interpret and simplify naturally when explaining; " "the text after the French translation is cross-reference data, not meant to be " "read aloud to the learner:\n\n" + "\n\n".join(lines) ) def build_system_prompt(extra_notes=""): return BASE_PERSONA + ADAPTIVE_RULES + extra_notes # -------------------------------------------------------------------------- # Gemini # -------------------------------------------------------------------------- def _gemini_chat(messages, temperature=0.85, max_tokens=1200): """Send a chat-completions request to Gemini, with safe retries and useful logs.""" if not GEMINI_API_KEY: raise RuntimeError("GEMINI_API_KEY is not configured in the environment.") headers = { "Authorization": f"Bearer {GEMINI_API_KEY}", "Content-Type": "application/json", } payload = { "model": GEMINI_MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False, } if GEMINI_REASONING_EFFORT: payload["reasoning_effort"] = GEMINI_REASONING_EFFORT last_error = None for attempt in range(3): try: response = requests.post( GEMINI_API_URL, headers=headers, json=payload, timeout=GEMINI_TIMEOUT_SECONDS, ) if response.status_code in (429, 500, 502, 503, 504) and attempt < 2: app.logger.warning( "Temporary Gemini response %s (attempt %s/3).", response.status_code, attempt + 1, ) time.sleep(1.5 * (attempt + 1)) continue if not response.ok: # Never log request headers (they contain the API key). app.logger.error("Gemini error %s: %s", response.status_code, response.text[:2000]) response.raise_for_status() result = response.json() choices = result.get("choices") if not isinstance(choices, list) or not choices: app.logger.error("Gemini returned no choices: %s", json.dumps(result)[:2000]) raise RuntimeError("Gemini returned no completion choices.") content = choices[0].get("message", {}).get("content", "") if isinstance(content, list): content = "\n".join( str(p.get("text", "")) for p in content if isinstance(p, dict) and p.get("text") ) if not isinstance(content, str) or not content.strip(): raise RuntimeError("Gemini returned an empty assistant message.") return content.strip() except (requests.Timeout, requests.ConnectionError) as exc: last_error = exc if attempt == 2: raise time.sleep(1.5 * (attempt + 1)) raise RuntimeError("Gemini request failed.") from last_error def call_intyoza(message, history, image_data_url=None): """Return (text, ok, code, sources). On failure ok=False and text is NOT meant to be saved.""" news_context, sources = igihe_news.get_news_context(message) combined_notes = ( lookup_pronunciation_notes(message, history) + lookup_knowledge_notes(message) + lookup_dictionary_rw_notes(message) + news_context + learner_signals(message, history) ) system_prompt = build_system_prompt(combined_notes) messages = [{"role": "system", "content": system_prompt}] for turn in history[-20:]: role = "assistant" if turn.get("role") in ("assistant", "model") else "user" messages.append({"role": role, "content": str(turn.get("content", ""))}) user_text = message or "What is in this image?" if image_data_url: mime, raw = parse_data_url(image_data_url) if not raw or not mime or not mime.startswith("image/"): return "Bad image.", False, "bad_image", [] messages.append({ "role": "user", "content": [ {"type": "text", "text": user_text}, {"type": "image_url", "image_url": {"url": image_data_url}}, ], }) else: messages.append({"role": "user", "content": user_text}) try: return _gemini_chat(messages), True, "", sources except requests.HTTPError as exc: status = exc.response.status_code if exc.response is not None else None app.logger.exception("Gemini API request failed (%s).", status) if status == 429: return "Rate limited by the AI provider.", False, "ai_busy", [] if status in (400, 401, 402, 403, 404): return f"AI configuration problem (HTTP {status}).", False, "ai_config", [] return "AI service error.", False, "ai_error", [] except requests.Timeout: return "AI timeout.", False, "ai_timeout", [] except Exception: app.logger.exception("Unexpected error while generating an Intyoza reply.") return "Unexpected error.", False, "ai_error", [] def generate_title(history_snippet): try: prompt = ( "Write a short title of 3 to 5 words for this conversation, in Kinyarwanda " "if the conversation is mostly Kinyarwanda, otherwise in English. " "No quotes, no trailing punctuation, no prefix. Reply with the title only.\n\n" + history_snippet ) title = _gemini_chat( [ {"role": "system", "content": "You create concise conversation titles."}, {"role": "user", "content": prompt}, ], temperature=0.3, max_tokens=60, ) title = (title or "").strip().strip('"').strip("'").split("\n")[0] return title[:60] if title else None except Exception: app.logger.exception("Could not generate conversation title through Gemini.") return None def snippet_from(history, limit=6): parts = [] for t in history[-limit:]: who = "User" if t["role"] == "user" else "Intyoza" parts.append(f"{who}: {str(t['content'])[:200]}") return "\n".join(parts) # -------------------------------------------------------------------------- # Pages # -------------------------------------------------------------------------- @app.route("/") def landing(): return render_template("landing.html") @app.route("/chat") def chat_app(): return render_template("index.html") # -------------------------------------------------------------------------- # OAuth routes # -------------------------------------------------------------------------- @app.route("/auth/") def oauth_login(provider): if provider not in ("google", "github"): return redirect("/chat") if not hasattr(oauth, provider) or getattr(oauth, provider, None) is None: return redirect("/chat?oauth_error=not_configured") redirect_uri = url_for("oauth_callback", provider=provider, _external=True, _scheme="https") return getattr(oauth, provider).authorize_redirect(redirect_uri) @app.route("/auth//callback") def oauth_callback(provider): if provider not in ("google", "github"): return redirect("/chat") try: c = getattr(oauth, provider) token = c.authorize_access_token() if provider == "google": info = token.get("userinfo") or c.get("https://openidconnect.googleapis.com/v1/userinfo").json() handle = info.get("email", "").split("@")[0] or info.get("sub") avatar = info.get("picture") else: info = c.get("user").json() handle = info.get("login") avatar = info.get("avatar_url") if not handle: return redirect("/chat?oauth_error=no_identity") username = f"{provider}:{handle}" try: users_col.update_one( {"username": username}, {"$setOnInsert": { "username": username, "display_name": handle, "password_hash": None, "provider": provider, "avatar": avatar, "created_at": datetime.datetime.utcnow(), }}, upsert=True, ) except DuplicateKeyError: pass session.permanent = True session["username"] = username return redirect("/chat") except Exception: app.logger.exception("OAuth callback failed.") return redirect("/chat?oauth_error=failed") @app.route("/api/oauth-status") def oauth_status(): return jsonify( google=bool(GOOGLE_ID and GOOGLE_SECRET), github=bool(GITHUB_ID and GITHUB_SECRET), ) # -------------------------------------------------------------------------- # Admin: import datasets # -------------------------------------------------------------------------- @app.route("/admin/data", methods=["GET", "POST"]) def admin_data(): if request.args.get("key", "") != os.environ.get("ADMIN_KEY", "___never___"): return "Not authorized.", 403 note = "" if request.method == "POST": dataset = request.form.get("dataset", "pronunciation") try: if dataset == "pronunciation": count, skipped = import_pronunciation_from_text(request.form.get("data", "")) note = f"Pronunciation: imported/updated {count}, skipped {skipped}." elif dataset == "knowledge": count, skipped, err = import_knowledge_from_json(request.form.get("data", "")) note = err or f"Knowledge: imported/updated {count}, skipped {skipped}." elif dataset == "dictionary_rw": file = request.files.get("file") if not file: note = "Please choose a JSON file to upload." else: count, skipped, err = import_dictionary_rw_from_json(file.read()) note = err or f"Dictionary: imported/updated {count}, skipped {skipped}." except Exception as e: note = f"Import failed: {e}" pron_total = pronunciation_col.count_documents({}) know_total = knowledge_col.count_documents({}) dict_total = dictionary_rw_col.count_documents({}) return f""" Intyoza data import

Intyoza data import

Text-based datasets




Dictionary file upload (large JSON files)


{escape(note)}

{pron_total} pronunciation · {know_total} knowledge · {dict_total} dictionary entries

""" # -------------------------------------------------------------------------- # Auth # -------------------------------------------------------------------------- def require_login(): return session.get("username") @app.route("/api/signup", methods=["POST"]) def signup(): if rate_limited("signup", 5, 600): return jsonify(ok=False, code="rate_limited", message="Too many attempts."), 429 data = request.json or {} username = (data.get("username") or "").strip() password = data.get("password") or "" if not username or not password: return jsonify(ok=False, code="missing_fields", message="Please enter a username and password."), 400 if len(username) < 3 or len(username) > 40: return jsonify(ok=False, code="short_username", message="Username should be 3 to 40 characters."), 400 if len(password) < 6: return jsonify(ok=False, code="short_password", message="Password should be at least 6 characters."), 400 hashed = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()) try: users_col.insert_one({ "username": username, "display_name": username, "password_hash": hashed, "provider": "local", "avatar": None, "created_at": datetime.datetime.utcnow(), }) except DuplicateKeyError: return jsonify(ok=False, code="taken", message="That username is already taken."), 409 return jsonify(ok=True, message="Account created. You can log in now.") @app.route("/api/login", methods=["POST"]) def login(): if rate_limited("login", 10, 60): return jsonify(ok=False, code="rate_limited", message="Too many attempts."), 429 data = request.json or {} username = (data.get("username") or "").strip() password = data.get("password") or "" user = users_col.find_one({"username": username}) bad = jsonify(ok=False, code="bad_credentials", message="Incorrect username or password.") if not user or not user.get("password_hash"): return bad, 401 if not bcrypt.checkpw(password.encode("utf-8"), user["password_hash"]): return bad, 401 session.permanent = True session["username"] = username return jsonify(ok=True, username=username) @app.route("/api/logout", methods=["POST"]) def logout(): session.clear() return jsonify(ok=True) @app.route("/api/session") def get_session(): username = session.get("username") if not username: return jsonify(username=None) user = users_col.find_one({"username": username}) if not user: session.clear() return jsonify(username=None) created = user.get("created_at") return jsonify( username=username, display_name=user.get("display_name") or username, avatar=user.get("avatar"), provider=user.get("provider", "local"), has_password=bool(user.get("password_hash")), created_at=(created.isoformat() + "Z") if created else None, prefs=user.get("prefs") or {}, ) # -------------------------------------------------------------------------- # Profile + dashboard # -------------------------------------------------------------------------- @app.route("/api/profile/avatar", methods=["POST"]) def update_avatar(): username = require_login() if not username: return jsonify(ok=False, message="Not logged in."), 401 data = request.json or {} avatar = data.get("avatar") if not avatar or not isinstance(avatar, str) or len(avatar) > 2_000_000: return jsonify(ok=False, message="Image is missing or too large (2MB max)."), 400 if not avatar.startswith("data:image/"): return jsonify(ok=False, message="Not an image."), 400 users_col.update_one({"username": username}, {"$set": {"avatar": avatar}}) return jsonify(ok=True, avatar=avatar) @app.route("/api/profile", methods=["PATCH"]) def update_profile(): username = require_login() if not username: return jsonify(ok=False), 401 data = request.json or {} updates = {} if "display_name" in data: name = str(data.get("display_name") or "").strip()[:40] if not name: return jsonify(ok=False, code="missing_fields", message="Name can't be empty."), 400 updates["display_name"] = name prefs = data.get("prefs") if isinstance(prefs, dict): if prefs.get("theme") in ("light", "dark"): updates["prefs.theme"] = prefs["theme"] if prefs.get("lang") in ("rw", "en"): updates["prefs.lang"] = prefs["lang"] if updates: users_col.update_one({"username": username}, {"$set": updates}) return jsonify(ok=True, display_name=updates.get("display_name")) @app.route("/api/profile/password", methods=["POST"]) def change_password(): username = require_login() if not username: return jsonify(ok=False), 401 if rate_limited("pw", 6, 600): return jsonify(ok=False, code="rate_limited", message="Too many attempts."), 429 data = request.json or {} current = data.get("current") or "" new = data.get("new") or "" user = users_col.find_one({"username": username}) if not user or not user.get("password_hash"): return jsonify(ok=False, code="no_password", message="This account has no password."), 400 if not bcrypt.checkpw(current.encode("utf-8"), user["password_hash"]): return jsonify(ok=False, code="bad_credentials", message="Current password is incorrect."), 401 if len(new) < 6: return jsonify(ok=False, code="short_password", message="Password should be at least 6 characters."), 400 users_col.update_one( {"username": username}, {"$set": {"password_hash": bcrypt.hashpw(new.encode("utf-8"), bcrypt.gensalt())}}, ) return jsonify(ok=True) @app.route("/api/dashboard") def dashboard(): username = require_login() if not username: return jsonify(ok=False), 401 try: tz = max(-840, min(840, int(request.args.get("tz", "0")))) # JS getTimezoneOffset() except ValueError: tz = 0 sign = "+" if tz <= 0 else "-" tz_str = f"{sign}{abs(tz) // 60:02d}:{abs(tz) % 60:02d}" chat_ids = [str(c["_id"]) for c in chats_col.find({"username": username}, {"_id": 1})] per_day = {} if chat_ids: pipeline = [ {"$match": {"chat_id": {"$in": chat_ids}, "role": "user"}}, {"$group": { "_id": {"$dateToString": {"format": "%Y-%m-%d", "date": "$timestamp", "timezone": tz_str}}, "n": {"$sum": 1}, }}, ] for row in messages_col.aggregate(pipeline): per_day[row["_id"]] = row["n"] local_today = (datetime.datetime.utcnow() - datetime.timedelta(minutes=tz)).date() one_day = datetime.timedelta(days=1) activity = [] for i in range(6, -1, -1): d = local_today - datetime.timedelta(days=i) activity.append({"date": d.isoformat(), "count": per_day.get(d.isoformat(), 0)}) streak = 0 d = local_today if per_day.get(d.isoformat(), 0) == 0: d -= one_day # a streak stays alive until the day ends while per_day.get(d.isoformat(), 0) > 0: streak += 1 d -= one_day return jsonify( ok=True, stats={ "chats": len(chat_ids), "messages": sum(per_day.values()), "active_days": len(per_day), "streak": streak, "saved": saved_col.count_documents({"username": username}), }, activity=activity, ) @app.route("/api/export") def export_data(): username = require_login() if not username: return jsonify(ok=False), 401 out = [] for c in chats_col.find({"username": username}).sort("created_at", 1): msgs = messages_col.find({"chat_id": str(c["_id"])}).sort("timestamp", 1) out.append({ "title": c.get("title"), "created_at": c["created_at"].isoformat() + "Z" if c.get("created_at") else None, "messages": [ {"role": m["role"], "content": m["content"], "timestamp": m["timestamp"].isoformat() + "Z" if m.get("timestamp") else None} for m in msgs ], }) saved = [ {"text": s["text"], "created_at": s["created_at"].isoformat() + "Z"} for s in saved_col.find({"username": username}).sort("created_at", 1) ] resp = jsonify(user=username, exported_at=datetime.datetime.utcnow().isoformat() + "Z", chats=out, saved=saved) resp.headers["Content-Disposition"] = "attachment; filename=intyoza-export.json" return resp @app.route("/api/chats/clear", methods=["POST"]) def clear_chats(): username = require_login() if not username: return jsonify(ok=False), 401 ids = [str(c["_id"]) for c in chats_col.find({"username": username}, {"_id": 1})] if ids: messages_col.delete_many({"chat_id": {"$in": ids}}) chats_col.delete_many({"username": username}) return jsonify(ok=True) @app.route("/api/account", methods=["DELETE"]) def delete_account(): username = require_login() if not username: return jsonify(ok=False), 401 if (request.json or {}).get("confirm") != "DELETE": return jsonify(ok=False, message="Confirmation missing."), 400 ids = [str(c["_id"]) for c in chats_col.find({"username": username}, {"_id": 1})] if ids: messages_col.delete_many({"chat_id": {"$in": ids}}) chats_col.delete_many({"username": username}) saved_col.delete_many({"username": username}) users_col.delete_one({"username": username}) session.clear() return jsonify(ok=True) # -------------------------------------------------------------------------- # Saved phrases # -------------------------------------------------------------------------- @app.route("/api/saved", methods=["GET"]) def list_saved(): username = require_login() if not username: return jsonify(saved=[]) docs = saved_col.find({"username": username}).sort("created_at", -1).limit(200) return jsonify(saved=[ {"id": str(d["_id"]), "text": d["text"], "created_at": d["created_at"].isoformat() + "Z"} for d in docs ]) @app.route("/api/saved", methods=["POST"]) def add_saved(): username = require_login() if not username: return jsonify(ok=False, code="login"), 401 text = str((request.json or {}).get("text") or "").strip()[:1000] if not text: return jsonify(ok=False, message="Nothing to save."), 400 if saved_col.count_documents({"username": username}) >= 500: return jsonify(ok=False, code="saved_full", message="Saved list is full."), 400 saved_col.update_one( {"username": username, "text": text}, {"$setOnInsert": {"username": username, "text": text, "created_at": datetime.datetime.utcnow()}}, upsert=True, ) return jsonify(ok=True) @app.route("/api/saved/", methods=["DELETE"]) def delete_saved(saved_id): username = require_login() oid = to_oid(saved_id) if not username or not oid: return jsonify(ok=False), 400 saved_col.delete_one({"_id": oid, "username": username}) return jsonify(ok=True) # -------------------------------------------------------------------------- # Chats # -------------------------------------------------------------------------- def owned_chat(chat_id, username): oid = to_oid(chat_id) if not oid or not username: return None return chats_col.find_one({"_id": oid, "username": username}) @app.route("/api/chats", methods=["GET"]) def list_chats(): username = require_login() if not username: return jsonify(chats=[]) docs = chats_col.find({"username": username}).sort("updated_at", -1) return jsonify(chats=[{"id": str(d["_id"]), "title": d["title"]} for d in docs]) @app.route("/api/chats", methods=["POST"]) def new_chat(): username = require_login() if not username: return jsonify(ok=False, message="Log in to save chats."), 401 now = datetime.datetime.utcnow() result = chats_col.insert_one({ "username": username, "title": "New chat", "auto_titled_at": 0, "created_at": now, "updated_at": now, }) return jsonify(ok=True, id=str(result.inserted_id), title="New chat") @app.route("/api/chats//messages", methods=["GET"]) def get_messages(chat_id): username = require_login() if not owned_chat(chat_id, username): return jsonify(messages=[]), 404 docs = messages_col.find({"chat_id": chat_id}).sort("timestamp", 1) return jsonify(messages=[ {"id": str(d["_id"]), "role": d["role"], "content": d["content"], "sources": d.get("sources") or [], "suggestions": d.get("suggestions") or []} for d in docs ]) @app.route("/api/chats/", methods=["PATCH"]) def rename_chat(chat_id): username = require_login() new_title = (((request.json or {}).get("title") or "").strip() or "New chat")[:80] chat = owned_chat(chat_id, username) if chat: chats_col.update_one( {"_id": chat["_id"]}, {"$set": {"title": new_title, "auto_titled_at": 999}}, ) return jsonify(ok=True, title=new_title) @app.route("/api/chats/", methods=["DELETE"]) def delete_chat(chat_id): username = require_login() chat = owned_chat(chat_id, username) if chat: chats_col.delete_one({"_id": chat["_id"]}) messages_col.delete_many({"chat_id": chat_id}) return jsonify(ok=True) @app.route("/api/messages/", methods=["DELETE"]) def delete_message(message_id): username = require_login() if not username: return jsonify(ok=False), 401 oid = to_oid(message_id) if not oid: return jsonify(ok=False), 400 msg = messages_col.find_one({"_id": oid}) if msg and owned_chat(msg.get("chat_id"), username): messages_col.delete_one({"_id": oid}) return jsonify(ok=True) # -------------------------------------------------------------------------- # Messaging # -------------------------------------------------------------------------- MAX_MESSAGE_CHARS = 4000 @app.route("/api/message", methods=["POST"]) def send_message(): data = request.json or {} message = (data.get("message") or "").strip() chat_id = data.get("chat_id") image_data_url = data.get("image") # Note: the old "mode" field is ignored on purpose. The tutor adapts by itself. if not message and not image_data_url: return jsonify(ok=False, message="Empty message."), 400 if len(message) > MAX_MESSAGE_CHARS: return jsonify(ok=False, code="too_long", message="Message is too long."), 400 if image_data_url and (not isinstance(image_data_url, str) or len(image_data_url) > 4_000_000): return jsonify(ok=False, message="Image is too large (4MB max)."), 400 username = require_login() limit = int(os.environ.get("MSG_LIMIT_USER", "25")) if username else int(os.environ.get("MSG_LIMIT_GUEST", "10")) if rate_limited("msg", limit, 60): return jsonify(ok=False, code="rate_limited", message="Too many messages."), 429 if not (username and chat_id): guest_history = data.get("history", []) if not isinstance(guest_history, list): guest_history = [] guest_history = [h for h in guest_history if isinstance(h, dict)][-20:] reply, ok, code, sources = call_intyoza(message, guest_history, image_data_url) if not ok: return jsonify(ok=False, code=code, **({"detail": reply} if DEBUG_ERRORS else {})), 502 return jsonify(ok=True, reply=reply, sources=sources) chat_doc = owned_chat(chat_id, username) if not chat_doc: return jsonify(ok=False, message="Chat not found."), 404 history = [ {"role": d["role"], "content": d["content"]} for d in messages_col.find({"chat_id": chat_id}).sort("timestamp", 1) ] reply, ok, code, sources = call_intyoza(message, history, image_data_url) if not ok: # Errors are never saved into the conversation or fed back to the model. return jsonify(ok=False, code=code, **({"detail": reply} if DEBUG_ERRORS else {})), 502 now = datetime.datetime.utcnow() stored_message = message or "[Sent an image]" user_doc = messages_col.insert_one( {"chat_id": chat_id, "role": "user", "content": stored_message, "timestamp": now}) bot_record = {"chat_id": chat_id, "role": "assistant", "content": reply, "timestamp": now + datetime.timedelta(milliseconds=1)} if sources: bot_record["sources"] = sources bot_doc = messages_col.insert_one(bot_record) full_history = history + [ {"role": "user", "content": stored_message}, {"role": "assistant", "content": reply}, ] exchanges = len(full_history) // 2 stage = chat_doc.get("auto_titled_at", 0) new_title = None if stage < 999 and (exchanges == 1 or (exchanges >= 4 and stage < 4)): candidate = generate_title(snippet_from(full_history)) if candidate: new_title = candidate chats_col.update_one( {"_id": chat_doc["_id"]}, {"$set": {"title": candidate, "auto_titled_at": exchanges}}, ) chats_col.update_one({"_id": chat_doc["_id"]}, {"$set": {"updated_at": now}}) return jsonify( ok=True, reply=reply, sources=sources, user_message_id=str(user_doc.inserted_id), assistant_message_id=str(bot_doc.inserted_id), new_title=new_title, ) # -------------------------------------------------------------------------- # Follow-up suggestions (called by the page right after a reply is shown) # -------------------------------------------------------------------------- def parse_suggestions(raw): try: start, end = raw.index("["), raw.rindex("]") + 1 items = json.loads(raw[start:end]) except Exception: return [] out = [] for it in items if isinstance(items, list) else []: if isinstance(it, str) and 3 <= len(it.strip()) <= 140: out.append(it.strip()) return out[:3] @app.route("/api/suggest", methods=["POST"]) def suggest(): if not SUGGESTIONS_ENABLED: return jsonify(ok=False, suggestions=[]) if rate_limited("suggest", 30, 60): return jsonify(ok=False, code="rate_limited", suggestions=[]), 429 data = request.json or {} learner = str(data.get("user") or "")[:600] tutor = str(data.get("assistant") or "")[:1500] if not tutor.strip(): return jsonify(ok=False, suggestions=[]) prompt = ( f"The learner wrote:\n{learner or '(an image)'}\n\nThe tutor (Intyoza) replied:\n{tutor}\n\n" "Write exactly 3 short follow-up questions or requests the learner could send next " "(max 12 words each). Write them in the same language as the tutor's reply: simple " "Kinyarwanda if the reply is mostly Kinyarwanda, English if it is mostly English. Make them " "different from each other and a natural next step (meaning of a word, an example, a " "correction, more detail, a related topic). Reply with ONLY a JSON array of 3 strings." ) try: raw = _gemini_chat( [{"role": "system", "content": "You suggest follow-up questions for a Kinyarwanda learning chat."}, {"role": "user", "content": prompt}], temperature=0.7, max_tokens=220, ) except Exception: app.logger.exception("Suggestion generation failed.") return jsonify(ok=False, suggestions=[]) suggestions = parse_suggestions(raw) username = require_login() msg_id = to_oid(data.get("message_id")) if suggestions and username and msg_id: msg = messages_col.find_one({"_id": msg_id}, {"chat_id": 1}) if msg and owned_chat(msg.get("chat_id"), username): messages_col.update_one({"_id": msg_id}, {"$set": {"suggestions": suggestions}}) return jsonify(ok=bool(suggestions), suggestions=suggestions) # -------------------------------------------------------------------------- # Admin: check what the IGIHE reader sees (/admin/news?key=ADMIN_KEY[&refresh=1][&q=amakuru ubukungu]) # -------------------------------------------------------------------------- @app.route("/admin/news") def admin_news(): if request.args.get("key", "") != os.environ.get("ADMIN_KEY", "___never___"): return "Not authorized.", 403 found = None if request.args.get("refresh"): found = igihe_news.refresh() q = request.args.get("q", "") ctx, sources = igihe_news.get_news_context(q) if q else ("", []) docs = list(news_col.find({}, sort=[("seen_at", -1), ("rank", 1)]).limit(60)) rows = "".join( f"{escape(d.get('section') or '')}" f"{escape(d.get('title') or d.get('list_title', ''))}" f"{'yes' if d.get('excerpt') else ''}" for d in docs ) return f"""IGIHE news check

IGIHE news check

Stored headlines: {news_col.count_documents({})} {f'· refreshed now: {found} found' if found is not None else ''} · add &refresh=1 to fetch now, or &q=... to test a message.

{f'
{escape(ctx)}
' if q else ''} {rows}
SectionHeadlineExcerpt
""" if __name__ == "__main__": app.run(host="0.0.0.0", port=7860, debug=False)