mhilya
/
sista_mif_ta
mirror of https://github.com/mhilya/sista_mif_ta.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
sista_mif_ta/fastapi/app/main.py

422 lines
22 KiB
Python
Raw Blame History

from fastapi import FastAPI, UploadFile, File, HTTPException, Depends, Security, BackgroundTasks
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import JSONResponse
import pandas as pd
import numpy as np
import joblib
import re
import io
import json
import uuid
import subprocess
import sys
import os
import logging
from contextlib import asynccontextmanager
from pathlib import Path
from typing import List, Dict, Any
from Sastrawi.Stemmer.StemmerFactory import StemmerFactory
from filelock import FileLock, Timeout
# ──────────────────────────────────────────────────────────────
# KONFIGURASI & KONSTANTA
# ──────────────────────────────────────────────────────────────
BASE_DIR = Path(__file__).parent.parent
ML_DIR = BASE_DIR / "ml_assets"
PIPELINE_PATH = ML_DIR / "ml_pipeline_internal.pkl"
STATUS_PATH = ML_DIR / "retrain_status.json"
LOCK_PATH = ML_DIR / "retrain_status.lock"
RETRAIN_WORKER = Path(__file__).parent / "retrain_worker.py"
TEMP_DIR = ML_DIR / "tmp"
TEMP_DIR.mkdir(parents=True, exist_ok=True)
JOBS_DIR = ML_DIR / "jobs"
JOBS_DIR.mkdir(parents=True, exist_ok=True)
CONFIDENCE_THRESHOLD = 0.50
TARGET_CLASSES = ["Programmer", "Data Analyst", "Wirausaha Informatika", "Non-IT"]
KEYWORD_RULES = {
"Programmer": ["programmer", "developer", "engineer", "fullstack", "backend", "frontend", "mobile", "android", "ios", "software", "web dev", "coding", "it staff", "teknisi", "sistem informasi", "application", "network", "devops", "qa", "tester", "ui", "ux", "swe"],
"Data Analyst": ["data analyst", "analis data", "data science", "business analyst", "research", "statistik", "bi analyst", "reporting", "database", "sql", "etl", "data engineer", "big data", "analyst", "data mining", "machine learning", "data visual", "power bi", "tableau", "looker", "business intelligence", "bi developer", "data warehouse"],
"Wirausaha Informatika": ["founder", "owner", "ceo", "wiraswasta", "startup", "freelance", "freelancer", "wirausaha", "bisnis", "usaha mandiri", "konsultan", "co founder", "entrepreneur", "self employed", "owner toko", "usaha", "dagang online", "tokopedia", "shopee", "dropship", "reseller"]
}
F5C_MAP = {"1": "founder owner wirausaha startup", "2": "co-founder partner wirausaha", "3": "staff karyawan pegawai", "4": "freelance kerja lepas lepasan"}
F1101_MAP = {"1": "instansi pemerintah dinas kementerian", "2": "non-profit lsm yayasan", "3": "perusahaan swasta corporate", "4": "wiraswasta usaha mandiri", "6": "bumn bumd pemerintah", "7": "multilateral internasional"}
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("tracer_worker")
stemmer = StemmerFactory().create_stemmer()
STOPWORDS = {
"yang", "di", "ke", "dari", "dan", "atau", "dengan", "untuk", "pada", "dalam",
"adalah", "ini", "itu", "tidak", "juga", "sudah", "akan", "bisa", "ada", "oleh",
"karena", "secara", "serta", "sebagai", "bagi", "telah", "maka", "namun", "sehingga",
"jika", "agar", "ketika", "saat", "sebelum", "sesudah", "hingga", "sampai", "antara",
"sekitar", "hanya", "saja", "belum", "masih", "lagi", "pun", "justru", "walaupun",
"meskipun", "bahkan", "cukup", "sangat", "paling", "lebih", "kurang", "lain",
"macam", "cara", "hal", "tentang", "mengenai", "terhadap", "kepada", "menuju",
"kecuali", "selain", "tanpa", "demi", "guna", "khususnya", "umumnya", "kebanyakan",
"sebagian", "beberapa", "semua", "setiap", "tiap", "satu", "dua", "tiga", "empat",
"lima", "enam", "tujuh", "delapan", "sembilan", "sepuluh", "ratus", "ribu", "juta"
}
# ──────────────────────────────────────────────────────────────
# MODEL LOADING
# ──────────────────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
try:
app.state.pipeline = joblib.load(PIPELINE_PATH)
logger.info(f" ML Pipeline loaded successfully from {PIPELINE_PATH}")
except Exception as e:
logger.error(f" Failed to load ML pipeline: {e}")
app.state.pipeline = None
yield
app = FastAPI(title="Tracer Study Classification Worker", lifespan=lifespan)
# ──────────────────────────────────────────────────────────────
# SECURITY
# ──────────────────────────────────────────────────────────────
security = HTTPBearer()
EXPECTED_TOKEN = os.getenv("FASTAPI_SECRET_KEY", "")
def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
if not EXPECTED_TOKEN: return credentials.credentials
if credentials.credentials != EXPECTED_TOKEN:
raise HTTPException(status_code=401, detail="Unauthorized - Invalid Token")
return credentials.credentials
# ──────────────────────────────────────────────────────────────
# TEXT PROCESSING
# ──────────────────────────────────────────────────────────────
def clean_text(text: str) -> str:
if pd.isna(text) or not isinstance(text, str): return ""
text = text.lower().strip()
if text in {"nan", "none", "null", "-", "0", "", "tidak diisi"}: return ""
text = text.replace('-', ' ').replace('/', ' ').replace('_', ' ').replace(',', ' ')
text = re.sub(r'[^\w\s]', '', text)
text = re.sub(r'\d+', '', text)
words = [w for w in text.split() if w not in STOPWORDS and len(w) > 2]
return " ".join([stemmer.stem(w) for w in words])
def find_column_by_code(df_columns: list, code: str) -> str | None:
for col in df_columns:
if code.lower() in col.lower(): return col
return None
def safe_get(row: pd.Series, df_columns: list, code: str, default: str = "") -> str:
actual_col = find_column_by_code(df_columns, code)
if actual_col and actual_col in row:
val = row[actual_col]
if pd.isna(val) or val is None: return default
return str(val).strip()
return default
# ──────────────────────────────────────────────────────────────
# CLASSIFICATION LOGIC
# ──────────────────────────────────────────────────────────────
def classify_rule(job_text: str) -> dict:
clean = clean_text(job_text)
if not clean or len(clean) < 3:
return {"profile": "Tidak Diketahui", "confidence": 0.65, "method": "rule_based_fallback"}
for profile, keywords in KEYWORD_RULES.items():
if any(kw in clean for kw in keywords):
return {"profile": profile, "confidence": 0.65, "method": "rule_based"}
return None
def classify_ml(job_text: str, pipeline) -> dict:
clean = clean_text(job_text)
if not clean:
return {"profile": "Tidak Diketahui", "confidence": 0.0, "method": "ml_empty_input"}
try:
proba = pipeline.predict_proba([clean])[0]
max_conf = float(np.max(proba))
pred_class = pipeline.classes_[np.argmax(proba)]
method = "ml_fallback" if max_conf >= CONFIDENCE_THRESHOLD else "manual_review"
return {"profile": pred_class, "confidence": round(max_conf, 4), "method": method}
except Exception as e:
logger.error(f"ML_INFER_ERROR | job_text='{job_text[:50]}' | err={e}")
raise HTTPException(status_code=500, detail=f"ML inference failed: {str(e)}")
# [KEEP v3] Ekstraksi teks dari format Kemendikbud
def extract_kemendik_text(row: pd.Series, df_columns: list) -> str:
f5b_col = find_column_by_code(df_columns, "f5b")
f5c_col = find_column_by_code(df_columns, "f5c")
f1101_col = find_column_by_code(df_columns, "f1101")
f1102_col = find_column_by_code(df_columns, "f1102")
f5b = clean_text(row.get(f5b_col, "") if f5b_col else "")
f1102 = clean_text(row.get(f1102_col, "") if f1102_col else "")
f5c_raw = str(row.get(f5c_col, "")).strip() if f5c_col else ""
f5c_code = re.match(r'^(\d+)', f5c_raw)
f5c_text = F5C_MAP.get(f5c_code.group(1), "") if f5c_code else ""
f1101_raw = str(row.get(f1101_col, "")).strip() if f1101_col else ""
f1101_code = re.match(r'^(\d+)', f1101_raw)
f1101_text = F1101_MAP.get(f1101_code.group(1), "") if f1101_code else ""
return " ".join(p for p in [f5b, f5c_text, f1101_text, f1102] if p).strip()
# ──────────────────────────────────────────────────────────────
# SOURCE DETECTION
# ──────────────────────────────────────────────────────────────
def detect_source(df: pd.DataFrame) -> str:
cols_lower = [c.lower().strip() for c in df.columns]
kemendik_codes = ["f5b", "f5c", "f8", "f1101", "nimhsmsmh", "nmmhsmsmh"]
for col in cols_lower:
if any(code in col for code in kemendik_codes): return "kemendik"
if any("jabatan" in c for c in cols_lower): return "internal_mif"
return "unknown"
# ──────────────────────────────────────────────────────────────
# ROUTES
# ──────────────────────────────────────────────────────────────
@app.get("/health")
def health_check():
return {"status": "healthy", "pipeline_loaded": app.state.pipeline is not None}
@app.post("/api/v1/retrain", dependencies=[Depends(verify_token)])
async def trigger_retrain(file: UploadFile = File(None)):
# [v4] FileLock mencegah race condition dari concurrent requests
lock = FileLock(LOCK_PATH, timeout=2)
try:
with lock:
if STATUS_PATH.exists():
try:
current = json.loads(STATUS_PATH.read_text(encoding="utf-8"))
if current.get("stage") in ["started", "backup", "loading_data", "preprocessing", "training", "evaluating", "promoting"]:
raise HTTPException(status_code=409, detail="Re-training sedang berjalan. Tunggu hingga selesai.")
except json.JSONDecodeError:
pass
extra_csv_path = ""
if file and file.filename:
job_id = str(uuid.uuid4())[:8]
extra_csv_path = str(TEMP_DIR / f"extra_{job_id}.csv")
contents = await file.read()
Path(extra_csv_path).write_bytes(contents)
logger.info(f"Extra CSV disimpan: {extra_csv_path}")
STATUS_PATH.write_text(json.dumps({
"stage": "started",
"message": "Memulai proses re-training...",
"timestamp": __import__('datetime').datetime.now().isoformat(),
}, ensure_ascii=False), encoding="utf-8")
cmd = [sys.executable, str(RETRAIN_WORKER)]
if extra_csv_path: cmd.append(extra_csv_path)
subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
logger.info(f"Retrain subprocess spawned: {' '.join(cmd)}")
return JSONResponse(content={"status": "started", "message": "Re-training dimulai di background."})
except Timeout:
raise HTTPException(status_code=429, detail="Sistem sibuk. Permintaan retraining lain sedang diproses.")
except Exception as e:
if not isinstance(e, HTTPException):
raise HTTPException(status_code=500, detail=f"Gagal memulai training: {str(e)}")
raise e
@app.get("/api/v1/retrain/status", dependencies=[Depends(verify_token)])
def retrain_status():
if not STATUS_PATH.exists():
return JSONResponse(content={"stage": "idle", "message": "Belum ada proses re-training."})
try:
# [v4] Lock saat baca untuk menghindari partial write dari worker
lock = FileLock(LOCK_PATH, timeout=2)
with lock:
data = json.loads(STATUS_PATH.read_text(encoding="utf-8"))
terminal_stages = {"promoted", "rolled_back", "failed"}
if data.get("stage") in terminal_stages and not data.get("_reloaded"):
try:
app.state.pipeline = joblib.load(PIPELINE_PATH)
data["_reloaded"] = True
STATUS_PATH.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
logger.info(f"Pipeline di-reload setelah retrain (stage={data['stage']})")
except Exception as e:
logger.error(f"Gagal reload pipeline: {e}")
return JSONResponse(content=data)
except Timeout:
return JSONResponse(status_code=429, content={"stage": "locked", "message": "Status sedang diupdate sistem."})
except Exception:
return JSONResponse(content={"stage": "unknown", "message": "Status tidak terbaca."})
# [KEEP v3] Endpoint reload manual — berguna jika pipeline perlu di-reload tanpa restart server
@app.post("/api/v1/retrain/reload", dependencies=[Depends(verify_token)])
def reload_model():
try:
app.state.pipeline = joblib.load(PIPELINE_PATH)
logger.info("Pipeline di-reload secara manual.")
return JSONResponse(content={"status": "ok", "message": "Model berhasil di-reload."})
except Exception as e:
raise HTTPException(status_code=500, detail=f"Gagal reload model: {str(e)}")
def process_classification_bg(job_id: str, file_path: Path, filename: str, pipeline):
status_file = JOBS_DIR / f"{job_id}.json"
def update_status(status, message, results=None, processed_rows=0, total_rows=0, source_type="unknown"):
data = {
"status": status, "message": message,
"processed_rows": processed_rows, "total_rows": total_rows,
"source_type": source_type
}
if results is not None:
data["results"] = results
status_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
try:
update_status("processing", "Parsing file excel/csv...")
contents = file_path.read_bytes()
if filename.endswith(".xlsx") or filename.endswith(".xls"):
try:
df = pd.read_excel(io.BytesIO(contents), dtype=str, engine="openpyxl")
except ImportError:
update_status("error", "Server misconfiguration: openpyxl missing")
return
except Exception as excel_err:
logger.error(f"Excel parsing error: {excel_err}")
try:
df = pd.read_csv(io.BytesIO(contents), sep=";", encoding="utf-8-sig", dtype=str)
except Exception:
df = pd.read_csv(io.BytesIO(contents), sep=";", encoding="latin-1", dtype=str)
else:
try:
df = pd.read_csv(io.BytesIO(contents), sep=";", encoding="utf-8-sig", dtype=str)
except UnicodeDecodeError:
df = pd.read_csv(io.BytesIO(contents), sep=";", encoding="latin-1", dtype=str)
df.columns = df.columns.str.strip()
source_type = detect_source(df)
if source_type == "unknown":
update_status("error", "Unrecognized file format.")
return
if source_type == "internal_mif" and pipeline is None:
update_status("error", "ML model not available.")
return
update_status("processing", f"Processing {len(df)} rows...", total_rows=len(df), source_type=source_type)
results = []
logger.info(f"Processing {len(df)} rows | Source: {source_type}")
find_col_cached = lambda p: next((c for c in df.columns if p.lower() in c.lower()), None)
col_f5a1 = find_col_cached("f5a1")
col_f5c = find_col_cached("f5c")
col_f1101 = find_col_cached("f1101")
col_f5b = find_col_cached("f5b")
col_f1102 = find_col_cached("f1102")
col_nim_kem = find_col_cached("nimhsmsmh")
col_nama_kem = find_col_cached("nmmhsmsmh")
col_tahun_kem = find_col_cached("tahun_lulus")
col_nim_int = find_col_cached("nim")
col_nama_int = find_col_cached("nama_lengkap")
col_tahun_int = find_col_cached("tahun_lulus")
for _, row in df.iterrows():
nim = ""
try:
def safe_str(val):
if pd.isna(val) or val is None: return ""
return str(val).strip()
if source_type == "kemendik":
nim = safe_str(row.get(col_nim_kem))
nama = safe_str(row.get(col_nama_kem))
tahun = safe_str(row.get(col_tahun_kem))
else:
nim = safe_str(row.get(col_nim_int))
nama = safe_str(row.get(col_nama_int))
tahun = safe_str(row.get(col_tahun_int))
if source_type == "kemendik":
job_text = extract_kemendik_text(row, df.columns.tolist())
raw_data_payload = {str(k): safe_str(v) for k, v in row.to_dict().items()}
raw_data_payload.update({
"F5a1": safe_str(row.get(col_f5a1)),
"F5c": safe_str(row.get(col_f5c)),
"F1101": safe_str(row.get(col_f1101)),
"F5b": safe_str(row.get(col_f5b)),
"F1102": safe_str(row.get(col_f1102))
})
results.append({
"nim": nim, "nama": nama, "tahun_lulus": tahun,
"job_text_raw": job_text, "source_type": "kemendik",
"predicted_profile": None, "confidence_score": None,
"classification_method": "dashboard_only", "status": "processed",
"raw_data": raw_data_payload
})
else:
col_jabatan = find_col_cached("jabatan")
col_perusahaan = find_col_cached("perusahaan")
col_deskripsi = find_col_cached("deskripsi")
job_text = clean_text(safe_str(row.get(col_jabatan))) if col_jabatan else ""
if not job_text and col_perusahaan and col_deskripsi:
job_text = clean_text(f"{safe_str(row.get(col_perusahaan))} {safe_str(row.get(col_deskripsi))}".strip())
elif not job_text and col_perusahaan:
job_text = clean_text(safe_str(row.get(col_perusahaan)))
rule_res = classify_rule(job_text)
res = rule_res or (classify_ml(job_text, pipeline) if pipeline else {"profile": "Non-IT", "confidence": 0.0, "method": "ml_unavailable"})
status = "auto_classified" if res["method"] in ["rule_based", "ml_fallback"] or res["profile"] == "Tidak Diketahui" else "needs_review"
raw_data_payload = {str(k): safe_str(v) for k, v in row.to_dict().items()}
results.append({
"nim": nim, "nama": nama, "tahun_lulus": tahun,
"job_text_raw": job_text, "source_type": "internal_mif",
"predicted_profile": res["profile"], "confidence_score": res["confidence"],
"classification_method": res["method"], "status": status,
"raw_data": raw_data_payload
})
except Exception as row_err:
logger.warning(f"Row error: {row_err}")
results.append({"nim": nim, "status": "failed", "error_detail": str(row_err)[:100]})
update_status("success", "Selesai", results=results, processed_rows=len(results), total_rows=len(df), source_type=source_type)
except Exception as e:
logger.error(f"Job {job_id} error: {e}")
update_status("error", f"Processing failed: {str(e)}")
finally:
if file_path.exists():
try:
file_path.unlink()
except Exception as unlink_err:
logger.warning(f"Failed to delete temp file {file_path}: {unlink_err}")
@app.post("/api/v1/classify", dependencies=[Depends(verify_token)])
async def classify_tracer(background_tasks: BackgroundTasks, file: UploadFile = File(...)):
if not file.filename:
raise HTTPException(status_code=400, detail="No file provided")
job_id = str(uuid.uuid4())
temp_file = JOBS_DIR / f"raw_{job_id}_{file.filename}"
contents = await file.read()
temp_file.write_bytes(contents)
status_file = JOBS_DIR / f"{job_id}.json"
status_file.write_text(json.dumps({"status": "pending", "message": "Job masuk antrean"}), encoding="utf-8")
background_tasks.add_task(process_classification_bg, job_id, temp_file, file.filename, app.state.pipeline)
return JSONResponse(content={"job_id": job_id, "status": "pending", "message": "Classification started in background."})
@app.get("/api/v1/classify/status/{job_id}", dependencies=[Depends(verify_token)])
def classify_status(job_id: str):
status_file = JOBS_DIR / f"{job_id}.json"
if not status_file.exists():
raise HTTPException(status_code=404, detail="Job not found")
try:
data = json.loads(status_file.read_text(encoding="utf-8"))
return JSONResponse(content=data)
except json.JSONDecodeError:
return JSONResponse(content={"status": "processing", "message": "Updating status..."})
if __name__ == "__main__":
import uvicorn
uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)
Reference in New Issue View Git Blame Copy Permalink