import os import base64 import json import logging import time import hmac import secrets from pathlib import Path from collections import defaultdict, deque, OrderedDict from datetime import datetime from flask import Flask, request, jsonify, render_template, Response, session, redirect try: from flask_limiter import Limiter from flask_limiter.util import get_remote_address _LIMITER_AVAILABLE = True except Exception as _limiter_exc: _LIMITER_AVAILABLE = False _LIMITER_ERROR = str(_limiter_exc) logging.getLogger("bord").warning("flask-limiter tidak aktif: %s", _LIMITER_ERROR) from groq import Groq from supabase import create_client PROJECT_ROOT = Path(__file__).resolve().parent SYSTEM_PROMPT_PATH = PROJECT_ROOT / "system_prompt.txt" def load_system_prompt() -> str: try: return SYSTEM_PROMPT_PATH.read_text(encoding="utf-8").strip() except Exception as exc: logging.getLogger("bord").warning("Gagal membaca system prompt: %s", exc) return ( "Kamu adalah Bord AI, asisten belajar berbahasa Indonesia untuk mahasiswa. " "Jawab dengan jelas, jujur, dan ramah." ) def identity_bootstrap(user_text: str): value = (user_text or "").strip().lower() if not value: return None identity_questions = [ "kamu siapa", "namamu", "siapa kamu", "siapa dirimu", "pembuatmu", "siapa yang membuatmu", "penciptamu", "siapa developer kamu", "pembuat kamu", ] if any(q in value for q in identity_questions): return ( "Aku Bord AI, asisten belajar yang dibuat oleh NaDev. " "Aku ada untuk bantu kamu belajar, bukan ngobrol santai terus. " "Ada yang mau ditanyain?" ) return None import re def redact_pii(text: str) -> str: value = text or "" value = re.sub(r'\b\d{4}[-\s]\d{4}[-\s]\d{4}[-\s]\d{4}\b', '[KARTU_TERSENSITIF]', value) return value # FIX: "pelan" dihapus dari blocklist. Karena matching-nya substring, # "jelasin pelan-pelan dong" ikut keblok. Hati-hati nambah kata pendek di sini. _MODERATION_BLOCKED = [ "cara membuat bom", "cara membuat narkoba", "cara membuat senjata", "cara menyuap", "cara korupsi", "cara curi", "cara mencuri", "cara hack", "cara meretas", "cara bobol", "pornografi", "child porn", "eksploitasi anak", ] def moderate_input(text: str) -> tuple[bool, str]: value = (text or "").strip() if not value: return True, "" lowered = value.lower() for phrase in _MODERATION_BLOCKED: if phrase in lowered: return False, "Maaf, permintaan ini nggak bisa aku bantu karena bertentangan dengan aturan keamanan." return True, "" log = logging.getLogger("bord") logging.basicConfig(level=logging.INFO) app = Flask(__name__) # Buat session cookie login admin. Set FLASK_SECRET_KEY di env HF Spaces biar # session nggak ke-reset tiap restart. Kalau kosong, fallback random (login ulang tiap restart). app.secret_key = os.environ.get("FLASK_SECRET_KEY") or secrets.token_hex(32) app.config["SESSION_COOKIE_HTTPONLY"] = True app.config["SESSION_COOKIE_SAMESITE"] = "Lax" app.config["SESSION_COOKIE_SECURE"] = True # HF Spaces pakai https app.config["RATELIMIT_HEADERS_ENABLED"] = True app.config["RATELIMIT_DEFAULT_ERROR_MESSAGE"] = "Bord lagi sibuk nih ๐Ÿ™ Tunggu sebentar ya." if _LIMITER_AVAILABLE: limiter = Limiter( key_func=lambda: ( (request.get_json(silent=True) or {}).get("session_id") or request.remote_addr or "anon" ), app=app, default_limits=["60 per minute"], storage_uri="memory://", ) else: limiter = None STATE = {"model": "llama-3.1-8b-instant"} DEV_STATS = {"chats": 0, "tokens": 0, "foto": 0} SESSION_HISTORY = defaultdict(list) DEV_KEY = os.environ.get("DEV_KEY", "") ADMIN_PASSWORD = os.environ.get("BORD_ADMIN_PASSWORD", "") SYSTEM_PROMPT = load_system_prompt() # FIX: eviction sesi biar SESSION_HISTORY nggak numpuk selamanya. # Penting kalau nanti pindah ke server sendiri (HF Spaces ketolong restart, server sendiri enggak). SESSION_MAX = 500 # maksimal sesi disimpan (LRU) SESSION_TTL = 3600 # detik; sesi nganggur > 1 jam dibuang SESSION_SEEN = OrderedDict() # session_id -> epoch terakhir aktif def touch_session(session_id: str) -> None: """Tandai sesi aktif + buang yang kedaluwarsa / kelebihan kapasitas.""" now = time.time() SESSION_SEEN[session_id] = now SESSION_SEEN.move_to_end(session_id) # TTL: buang sesi yang udah lama nganggur expired = [sid for sid, ts in SESSION_SEEN.items() if now - ts > SESSION_TTL] for sid in expired: SESSION_SEEN.pop(sid, None) SESSION_HISTORY.pop(sid, None) # Cap: kalau kebanyakan, buang yang paling lama nggak kepake (LRU) while len(SESSION_SEEN) > SESSION_MAX: old_sid, _ = SESSION_SEEN.popitem(last=False) SESSION_HISTORY.pop(old_sid, None) EVENT_BUFFER = deque(maxlen=300) EVENT_BUFFER.append({ "ts": datetime.now().isoformat(), "type": "system", "session_id": "-", "summary": "Bot started", }) client = Groq(api_key=os.environ.get("GROQ_API_KEY")) # GEMINI GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "") gemini_client = None GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "gemini-2.5-flash") if GEMINI_API_KEY: try: from google import genai gemini_client = genai.Client(api_key=GEMINI_API_KEY) except Exception as exc: log.warning("Gemini gagal connect: %s", exc) OCR_PROMPT = ( "Kamu adalah pembaca teks dari gambar untuk membantu mahasiswa.\n" "Baca dan tuliskan ULANG semua teks, soal, atau materi yang ada di gambar ini dengan rapi dan selengkap mungkin, persis apa adanya.\n" "JANGAN menjawab soal atau memberi penjelasan โ€” cukup tuliskan ulang isinya.\n" "Kalau ada rumus, simbol, atau kode, tuliskan sebisamu.\n" "Kalau gambar buram, gelap, atau teks tidak terbaca, jawab PERSIS: [TIDAK TERBACA]" ) def baca_foto(image_bytes, mime="image/jpeg"): if not gemini_client: return None, "Gemini belum dikonfigurasi (GEMINI_API_KEY kosong)." try: if not image_bytes: return None, "Gambar kosong." from google.genai import types as gtypes resp = gemini_client.models.generate_content( model=GEMINI_MODEL, contents=[OCR_PROMPT, gtypes.Part.from_bytes(data=image_bytes, mime_type=mime)], ) text = (resp.text or "").strip() if not text or "[TIDAK TERBACA]" in text.upper(): return "[TIDAK TERBACA]", None return text, None except Exception as exc: log.warning("OCR error: %s", exc) return None, str(exc) # SUPABASE SUPABASE_URL = os.environ.get("SUPABASE_URL", "") SUPABASE_KEY = os.environ.get("SUPABASE_KEY", "") supabase = None if SUPABASE_URL and SUPABASE_KEY: try: supabase = create_client(SUPABASE_URL, SUPABASE_KEY) except Exception as exc: log.warning("Supabase gagal connect: %s", exc) def log_chat(question, answer, model, dev=False): if not supabase: return try: supabase.table("percakapan").insert({ "pertanyaan": redact_pii(question or ""), "jawaban": redact_pii(answer or ""), "model": redact_pii(model or ""), "dev": bool(dev), }).execute() except Exception as exc: log.warning("Gagal simpan percakapan: %s", exc) @app.route("/") def index(): return render_template("index.html") @app.route("/healthz", methods=["GET"]) def healthz(): EVENT_BUFFER.append({ "ts": datetime.now().isoformat(), "type": "access", "session_id": "-", "summary": "healthz hit", }) return jsonify({"status": "ok", "model": STATE["model"]}) @app.route("/metrics", methods=["GET"]) def metrics(): if not (DEV_KEY and request.args.get("dev_key") == DEV_KEY): return jsonify({"error": "unauthorized"}), 401 EVENT_BUFFER.append({ "ts": datetime.now().isoformat(), "type": "access", "session_id": "-", "summary": "metrics hit", }) return jsonify({"stats": DEV_STATS, "model": STATE["model"]}), 200 def require_admin() -> bool: """True kalau udah login lewat session cookie, ATAU header Authorization cocok (header tetep dipertahanin biar curl/script masih jalan). Pakai compare_digest biar nggak bisa ditebak lewat timing attack.""" if session.get("is_admin"): return True if not ADMIN_PASSWORD: return False auth = request.headers.get("Authorization", "") or "" if not auth: return False candidates = (ADMIN_PASSWORD, f"Bearer {ADMIN_PASSWORD}") return any(hmac.compare_digest(auth, c) for c in candidates) # Halaman login admin sederhana (self-contained, nggak butuh file template terpisah). _ADMIN_LOGIN_PAGE = """ Bord Admin โ€” Login

๐Ÿค– Bord Admin

Masukin password buat buka dashboard.

Password salah, coba lagi.
""" def _login_page(show_error: bool = False) -> str: return _ADMIN_LOGIN_PAGE.replace( "__ERR_DISPLAY__", "" if show_error else "display:none" ) @app.route("/admin/login", methods=["GET", "POST"]) def admin_login(): if session.get("is_admin"): return redirect("/admin") if request.method == "POST": pw = request.form.get("password", "") or "" if ADMIN_PASSWORD and hmac.compare_digest(pw, ADMIN_PASSWORD): session["is_admin"] = True return redirect("/admin") return _login_page(show_error=True), 401 return _login_page() @app.route("/admin/logout") def admin_logout(): session.pop("is_admin", None) return redirect("/admin/login") @app.route("/admin") def admin_dashboard(): # Belum login -> lempar ke halaman login (bukan curl lagi). if not require_admin(): return redirect("/admin/login") return render_template("admin.html") @app.route("/admin/json") def admin_events_json(): # FIX: dulu siapa pun bisa narik event log + session_id. Sekarang dijaga. if not require_admin(): return jsonify({"error": "unauthorized"}), 401 ordered = list(EVENT_BUFFER) ordered.reverse() now = datetime.now() enriched = [] for item in ordered: try: ts = datetime.fromisoformat(item["ts"]) except Exception: ts = now enriched.append({ "ts": item.get("ts"), "type": item.get("type"), "session_id": item.get("session_id"), "summary": item.get("summary"), "elapsed": _ago(ts, now), }) return jsonify({ "stats": DEV_STATS, "model": STATE["model"], "events": enriched, "active_sessions": len(SESSION_HISTORY), }) def _ago(start, end): diff = int((end - start).total_seconds()) if diff < 60: return f"{diff}s" if diff < 3600: m = diff // 60 s = diff % 60 return f"{m}m {s}s" h = diff // 3600 m = (diff % 3600) // 60 return f"{h}h {m}m" @app.route("/admin/events") def admin_events_stream(): # FIX: stream SSE juga dijaga, dulu kebuka. if not require_admin(): return jsonify({"error": "unauthorized"}), 401 def event_stream(): snapshots = list(EVENT_BUFFER) for ev in snapshots: yield _format_sse(ev) last_ts = datetime.now().isoformat() while True: time.sleep(1) any_new = False for ev in EVENT_BUFFER: ts = ev.get("ts", "") if ts > last_ts: yield _format_sse(ev) last_ts = ts any_new = True if not any_new: yield ": ping\n\n" return Response(event_stream(), mimetype="text/event-stream") def _format_sse(ev: dict) -> str: data = json.dumps(ev, ensure_ascii=False) return f"data: {data}\n\n" @app.route("/chat", methods=["POST"]) def chat(): data = request.get_json(silent=True) or {} user_input = (data.get("message", "") or "").strip() image_b64 = data.get("image", "") or "" is_dev = bool(DEV_KEY) and ((data.get("dev_key", "") or "") == DEV_KEY) if not user_input and not image_b64: return jsonify({"reply": ""}), 200 if is_dev and not image_b64: cmd = (user_input or "").split() name = (cmd[0] or "").lower() if cmd else "" if name == "/stats": reply = ( f"๐Ÿ“Š STATS\n" f"Chat: {DEV_STATS['chats']}\n" f"Token: {DEV_STATS['tokens']:,}\n" f"Foto (Gemini): {DEV_STATS['foto']}\n" f"Model: {STATE['model']}\n" f"Gemini: {'ON' if gemini_client else 'OFF'}\n" f"Supabase: {'ON' if supabase else 'OFF'}" ) return jsonify({"reply": reply, "debug": {"cmd": "stats"}}), 200 if name == "/model": if len(cmd) > 1: STATE["model"] = cmd[1] reply = f"โœ… Model diganti ke: {cmd[1]}" else: reply = f"Model sekarang: {STATE['model']}\nContoh: /model llama-3.3-70b-versatile" return jsonify({"reply": reply, "debug": {"cmd": "model"}}), 200 if name == "/whoami": return jsonify({"reply": "๐Ÿ› ๏ธ DEV MODE aktif.", "debug": {"cmd": "whoami"}}), 200 if name == "/clear": DEV_STATS.update({"chats": 0, "tokens": 0, "foto": 0}) SESSION_HISTORY.clear() SESSION_SEEN.clear() # FIX: ikut bersihin tracker eviction return jsonify({"reply": "๐Ÿงน Stats + sesi direset.", "debug": {"cmd": "clear"}}), 200 if name == "/prompt": return jsonify({"reply": SYSTEM_PROMPT, "debug": {"cmd": "prompt"}}), 200 if name == "/help": return jsonify({"reply": ( "๐Ÿ› ๏ธ DEV COMMANDS\n" "/stats\n/model \n/whoami\n/clear\n/prompt\n/help" ), "debug": {"cmd": "help"}}), 200 return jsonify({"reply": "Perintah dev tidak dikenali.", "debug": {"cmd": "unknown"}}), 200 identity_reply = identity_bootstrap(user_input) if identity_reply: return jsonify({"reply": identity_reply}), 200 if not image_b64: allowed, moderation_msg = moderate_input(user_input) if not allowed: return jsonify({"reply": moderation_msg}), 200 model_label = STATE["model"] session_id = data.get("session_id") or request.remote_addr or "anon" # FIX: history satu sumber kebenaran (server-side). Dulu di-merge sama # history kiriman client -> turn kedobel -> model baca u1,a1,u1,a1,... touch_session(session_id) combined = SESSION_HISTORY[session_id][-8:] if image_b64: base64_data = image_b64.split(",", 1)[1] if "," in image_b64 else image_b64 try: image_bytes = base64.b64decode(base64_data) except Exception: return jsonify({"reply": "Fotonya gagal dibaca, coba kirim ulang ya ๐Ÿ™"}), 200 teks_soal, err = baca_foto(image_bytes) if err or teks_soal == "": result = {"reply": "Aduh, kakak nggak bisa baca fotonya nih ๐Ÿ™ Coba foto lebih terang & fokus, atau ketik soalnya langsung ya."} if is_dev: result["debug"] = {"error": err or "hasil kosong"} return jsonify(result), 200 if teks_soal == "[TIDAK TERBACA]": return jsonify({"reply": "Fotonya kurang jelas nih ๐Ÿ™ Coba foto ulang yang lebih terang dan nggak buram ya, biar kakak bisa baca soalnya."}), 200 DEV_STATS["foto"] += 1 model_label = f"gemini+{STATE['model']}" user_input = f"[Mahasiswa mengirim foto soal. Hasil bacanya:]\n{teks_soal}" + (f"\n\n[Pesan tambahan]: {user_input}" if user_input else "") EVENT_BUFFER.append({ "ts": datetime.now().isoformat(), "type": "chat_start", "session_id": session_id, "summary": f"msg len={len(user_input)} image={bool(image_b64)}", }) messages = [{"role": "system", "content": SYSTEM_PROMPT}] messages.extend(combined) messages.append({"role": "user", "content": user_input}) try: response = client.chat.completions.create( model=STATE["model"], messages=messages, max_tokens=1024, timeout=15, ) except Exception as exc: EVENT_BUFFER.append({ "ts": datetime.now().isoformat(), "type": "chat_error", "session_id": session_id, "summary": str(exc), }) lower = str(exc).lower() if any(token in lower for token in ["rate", "429", "limit"]): msg = "Bord lagi sibuk nih ๐Ÿ™ Tunggu sebentar ya." elif "invalid_api_key" in lower or "401" in lower: msg = "Konfigurasi API bermasalah. Hubungi admin untuk cek kunci API dulu ya." log.error("Groq API key error: %s", exc) else: msg = "Aduh, ada gangguan di sisi Bord nih ๐Ÿ™ Coba lagi sebentar ya." result = {"reply": msg} if is_dev: result["debug"] = {"error": str(exc)} return jsonify(result), 200 reply = (response.choices[0].message.content or "").strip() usage = getattr(response, "usage", None) DEV_STATS["chats"] += 1 DEV_STATS["tokens"] += (getattr(usage, "total_tokens", 0) or 0) if usage else 0 # Simpan turn ini ke history server-side biar konteks ngalir antar request. turn = [ {"role": "user", "content": user_input}, {"role": "assistant", "content": reply}, ] SESSION_HISTORY[session_id] = (combined + turn)[-8:] log_chat(user_input, reply, model_label, dev=is_dev) EVENT_BUFFER.append({ "ts": datetime.now().isoformat(), "type": "chat_success", "session_id": session_id, "summary": f"reply len={len(reply)}", }) result = {"reply": reply} if is_dev and usage: result["debug"] = { "model": model_label, "prompt_tokens": getattr(usage, "prompt_tokens", 0) or 0, "completion_tokens": getattr(usage, "completion_tokens", 0) or 0, "total_tokens": getattr(usage, "total_tokens", 0) or 0, } return jsonify(result), 200 @app.errorhandler(429) def handle_429(exc): return jsonify({"reply": "Bord lagi rame banget nih ๐Ÿ™ Coba lagi dalam beberapa detik."}), 429 @app.errorhandler(Exception) def handle_generic_exception(exc): log.exception("Unhandled error: %s", exc) return jsonify({"reply": "Aduh, ada gangguan nih ๐Ÿ™ Coba kirim lagi sebentar ya."}), 200 if __name__ == "__main__": port = int(os.environ.get("PORT", "7860")) app.run(host="0.0.0.0", port=port)