from fastapi import FastAPI, UploadFile, File, HTTPException, BackgroundTasks, Depends, Security 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 shutil 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 # PENAMBAHAN KUNCI MUTEX # 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) 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" } @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 = 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 def clean_text(text: str) -> str: # Stemmer HANYA berjalan pada saat ada permintaan inferensi baru 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 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)}") 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" @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)): lock = FileLock(LOCK_PATH, timeout=2) # Timeout 2 detik jika bentrok try: with lock: if STATUS_PATH.exists(): try: current = json.loads(STATUS_PATH.read_text()) 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)) 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: lock = FileLock(LOCK_PATH, timeout=2) with lock: data = json.loads(STATUS_PATH.read_text()) 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)) logger.info("Pipeline di-reload setelah retrain selesai.") 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."}) @app.post("/api/v1/classify", dependencies=[Depends(verify_token)]) async def classify_tracer(file: UploadFile = File(...)): if not file.filename: raise HTTPException(status_code=400, detail="No file provided") try: contents = await file.read() if file.filename.endswith(".xlsx") or file.filename.endswith(".xls"): try: df = pd.read_excel(io.BytesIO(contents), dtype=str, engine="openpyxl") except ImportError: logger.error("openpyxl not installed") raise HTTPException(status_code=500, detail="Server misconfiguration: openpyxl missing") 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: 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() except Exception as e: logger.error(f"File parsing error: {type(e).__name__}: {str(e)}") raise HTTPException(status_code=400, detail=f"Failed to parse file: {type(e).__name__}: {str(e)[:200]}") source_type = detect_source(df) if source_type == "unknown": raise HTTPException(status_code=400, detail="Unrecognized file format.") if source_type == "internal_mif" and app.state.pipeline is None: logger.error("ML pipeline not loaded") raise HTTPException(status_code=503, detail="ML model not available.") results = [] pipeline = app.state.pipeline 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(): 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 if 'nim' in locals() else "", "status": "failed", "error_detail": str(row_err)[:100]}) return JSONResponse(content={"status": "success", "total_rows": len(df), "processed_rows": len(results), "source_type": source_type, "results": results}) if __name__ == "__main__": import uvicorn uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)