258 lines
8.0 KiB
Python
258 lines
8.0 KiB
Python
import os
|
|
import json
|
|
import time
|
|
import logging
|
|
from typing import Any, Dict, List
|
|
|
|
import requests
|
|
from flask import Flask, jsonify, request
|
|
from dotenv import load_dotenv
|
|
|
|
BASE_DIR = os.path.dirname(__file__)
|
|
ROOT_ENV = os.path.abspath(os.path.join(BASE_DIR, "..", "..", ".env"))
|
|
load_dotenv(ROOT_ENV)
|
|
load_dotenv(os.path.join(BASE_DIR, ".env"))
|
|
|
|
app = Flask(__name__)
|
|
|
|
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "")
|
|
BACKEND_TOKEN = os.getenv("BACKEND_TOKEN", "")
|
|
GEMINI_BASE_URL = os.getenv("GEMINI_BASE_URL", "https://generativelanguage.googleapis.com/v1beta/models")
|
|
TIMEOUT_SECONDS = int(os.getenv("GEMINI_TIMEOUT", "30"))
|
|
MAJORS_FILE_PATH = os.getenv("MAJORS_FILE_PATH", os.path.join(BASE_DIR, "majors_data.json"))
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s",
|
|
handlers=[
|
|
logging.StreamHandler(),
|
|
],
|
|
)
|
|
logger = logging.getLogger("python_backend")
|
|
|
|
|
|
def _log(message: str) -> None:
|
|
logger.info(message)
|
|
|
|
|
|
def _is_authorized() -> bool:
|
|
if not BACKEND_TOKEN:
|
|
return True
|
|
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Bearer "):
|
|
return False
|
|
|
|
token = auth_header.replace("Bearer ", "", 1).strip()
|
|
return token == BACKEND_TOKEN
|
|
|
|
|
|
def _http_error(message: str, status_code: int, detail: str = "") -> Any:
|
|
payload = {
|
|
"success": False,
|
|
"message": message,
|
|
}
|
|
if detail:
|
|
payload["error"] = detail
|
|
return jsonify(payload), status_code
|
|
|
|
|
|
def _extract_text(data: Dict[str, Any]) -> str:
|
|
return (
|
|
data.get("candidates", [{}])[0]
|
|
.get("content", {})
|
|
.get("parts", [{}])[0]
|
|
.get("text", "")
|
|
)
|
|
|
|
|
|
def _load_majors_file() -> Dict[str, Any]:
|
|
try:
|
|
with open(MAJORS_FILE_PATH, "r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except FileNotFoundError:
|
|
return {
|
|
"ok": False,
|
|
"message": "Data jurusan belum tersedia.",
|
|
"data": {"majors": []},
|
|
}
|
|
except json.JSONDecodeError as exc:
|
|
return {
|
|
"ok": False,
|
|
"message": "Data jurusan tidak dapat dibaca.",
|
|
"data": {"majors": []},
|
|
}
|
|
|
|
majors = data.get("majors") if isinstance(data, dict) else []
|
|
if not isinstance(majors, list):
|
|
majors = []
|
|
|
|
return {
|
|
"ok": True,
|
|
"message": "ok",
|
|
"data": {
|
|
"schema_version": data.get("schema_version", "unknown") if isinstance(data, dict) else "unknown",
|
|
"last_updated": data.get("last_updated", "unknown") if isinstance(data, dict) else "unknown",
|
|
"notes": data.get("notes", "") if isinstance(data, dict) else "",
|
|
"majors": majors,
|
|
},
|
|
}
|
|
|
|
|
|
def _build_majors_context_text(majors_data: Dict[str, Any]) -> str:
|
|
majors = majors_data.get("majors", [])
|
|
if not majors:
|
|
return ""
|
|
|
|
lines = [
|
|
"DATA JURUSAN RESMI (WAJIB JADI ACUAN UTAMA)",
|
|
f"schema_version: {majors_data.get('schema_version', 'unknown')}",
|
|
f"last_updated: {majors_data.get('last_updated', 'unknown')}",
|
|
"Gunakan data berikut untuk menjawab informasi jurusan secara konsisten.",
|
|
"Jika ada konflik dengan asumsi model, utamakan data ini.",
|
|
]
|
|
|
|
for index, major in enumerate(majors, start=1):
|
|
name = major.get("name", "-")
|
|
description = major.get("description", "-")
|
|
prospects = major.get("career_prospects", "-")
|
|
preferences = ", ".join(major.get("study_preferences", [])) or "-"
|
|
keywords = ", ".join(major.get("keywords", [])) or "-"
|
|
|
|
lines.append(
|
|
f"{index}. {name} | deskripsi: {description} | preferensi studi: {preferences} | prospek: {prospects} | kata kunci: {keywords}"
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _inject_majors_context(payload: Dict[str, Any], majors_context: str) -> Dict[str, Any]:
|
|
if not majors_context:
|
|
return payload
|
|
|
|
contents = payload.get("contents", [])
|
|
if not isinstance(contents, list) or not contents:
|
|
return payload
|
|
|
|
first = contents[0]
|
|
if isinstance(first, dict) and first.get("role") == "user":
|
|
parts = first.get("parts", [])
|
|
if isinstance(parts, list) and parts and isinstance(parts[0], dict):
|
|
original_text = str(parts[0].get("text", ""))
|
|
parts[0]["text"] = f"{majors_context}\n\n{original_text}"
|
|
return payload
|
|
|
|
contents.insert(0, {"role": "user", "parts": [{"text": majors_context}]})
|
|
return payload
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> Any:
|
|
majors_result = _load_majors_file()
|
|
majors_count = len(majors_result["data"].get("majors", []))
|
|
_log("[PY-BACKEND] GET /health")
|
|
return jsonify(
|
|
{
|
|
"ok": True,
|
|
"gemini_key_configured": bool(GEMINI_API_KEY),
|
|
"majors_file": MAJORS_FILE_PATH,
|
|
"majors_loaded": majors_count,
|
|
"majors_valid": majors_result["ok"],
|
|
}
|
|
)
|
|
|
|
|
|
@app.get("/api/majors")
|
|
def majors() -> Any:
|
|
result = _load_majors_file()
|
|
status = 200 if result["ok"] else 500
|
|
return jsonify(result), status
|
|
|
|
|
|
@app.post("/api/chat")
|
|
def chat() -> Any:
|
|
_log("[PY-BACKEND] POST /api/chat")
|
|
if not _is_authorized():
|
|
_log("[PY-BACKEND] Unauthorized request")
|
|
return _http_error(
|
|
"Akses ke layanan chatbot ditolak.",
|
|
401,
|
|
)
|
|
|
|
if not GEMINI_API_KEY:
|
|
return _http_error(
|
|
"Layanan chatbot belum siap digunakan.",
|
|
500,
|
|
)
|
|
|
|
body = request.get_json(silent=True) or {}
|
|
payload = body.get("payload")
|
|
models = body.get("models", ["gemini-2.5-flash", "gemini-2.0-flash", "gemini-2.0-flash-lite"])
|
|
|
|
if not isinstance(payload, dict) or not payload.get("contents"):
|
|
_log("[PY-BACKEND] Invalid payload")
|
|
return _http_error(
|
|
"Pesan belum dapat diproses. Silakan perjelas pertanyaan Anda.",
|
|
422,
|
|
)
|
|
|
|
majors_result = _load_majors_file()
|
|
if majors_result["ok"]:
|
|
majors_context = _build_majors_context_text(majors_result["data"])
|
|
payload = _inject_majors_context(payload, majors_context)
|
|
_log(f"[PY-BACKEND] Majors context loaded: {len(majors_result['data'].get('majors', []))} jurusan")
|
|
else:
|
|
_log(f"[PY-BACKEND] Warning: {majors_result['message']}")
|
|
|
|
last_error = ""
|
|
for model in models:
|
|
url = f"{GEMINI_BASE_URL}/{model}:generateContent?key={GEMINI_API_KEY}"
|
|
try:
|
|
resp = requests.post(
|
|
url,
|
|
json=payload,
|
|
headers={"Content-Type": "application/json"},
|
|
timeout=TIMEOUT_SECONDS,
|
|
)
|
|
except requests.RequestException as exc:
|
|
last_error = str(exc)
|
|
continue
|
|
|
|
if resp.ok:
|
|
data = resp.json()
|
|
text = _extract_text(data)
|
|
if text:
|
|
_log(f"[PY-BACKEND] Success using model: {model}")
|
|
return jsonify({"success": True, "message": text, "model": model})
|
|
last_error = "Jawaban dari layanan tidak ditemukan."
|
|
continue
|
|
|
|
if resp.status_code in (404, 429):
|
|
if resp.status_code == 429:
|
|
time.sleep(1)
|
|
if resp.status_code == 429:
|
|
last_error = (
|
|
"Layanan chatbot sedang ramai digunakan. Silakan tunggu sebentar lalu coba lagi."
|
|
)
|
|
else:
|
|
last_error = "Layanan chatbot sementara tidak tersedia."
|
|
continue
|
|
|
|
last_error = (
|
|
"Terjadi gangguan pada layanan chatbot."
|
|
)
|
|
|
|
_log(f"[PY-BACKEND] Failed all models: {last_error}")
|
|
return _http_error(
|
|
"Maaf, layanan chatbot sedang mengalami gangguan. Silakan coba kembali beberapa saat lagi.",
|
|
502,
|
|
last_error,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
host = os.getenv("PY_BACKEND_HOST", "0.0.0.0")
|
|
port = int(os.getenv("PY_BACKEND_PORT", "5000"))
|
|
debug = os.getenv("PY_BACKEND_DEBUG", "true").lower() == "true"
|
|
app.run(host=host, port=port, debug=debug)
|