260 lines
13 KiB
Python
260 lines
13 KiB
Python
"""Versioned local model store. Run one Uvicorn worker (see PANDUAN_REVISI.md)."""
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import unicodedata
|
|
from datetime import datetime, timezone
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import joblib
|
|
import pandas as pd
|
|
from sklearn.base import clone
|
|
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix
|
|
from sklearn.model_selection import train_test_split
|
|
|
|
BASE = Path(__file__).resolve().parent
|
|
ARTIFACTS = BASE / 'artifacts_slearn'
|
|
STORE = Path(os.getenv('SLEARN_MODEL_STORE', str(BASE / 'model_store')))
|
|
STORE.mkdir(parents=True, exist_ok=True)
|
|
DB = STORE / 'registry.sqlite3'
|
|
_cache = {}
|
|
_cache_lock = threading.RLock()
|
|
_seed_cache = None
|
|
|
|
|
|
def now():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@contextmanager
|
|
def connect():
|
|
db = sqlite3.connect(DB, timeout=30)
|
|
db.row_factory = sqlite3.Row
|
|
try:
|
|
with db:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def key(text):
|
|
return ' '.join(unicodedata.normalize('NFKC', str(text)).casefold().split())
|
|
|
|
|
|
def initialize():
|
|
with connect() as db:
|
|
db.executescript('''
|
|
CREATE TABLE IF NOT EXISTS versions (
|
|
id TEXT PRIMARY KEY, created_at TEXT NOT NULL, metadata TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS additions (
|
|
text_key TEXT PRIMARY KEY, text TEXT NOT NULL, stem TEXT NOT NULL,
|
|
label INTEGER NOT NULL, confidence REAL NOT NULL,
|
|
teacher TEXT NOT NULL, version TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS jobs (
|
|
id TEXT PRIMARY KEY, created_at TEXT NOT NULL, status TEXT NOT NULL,
|
|
message TEXT NOT NULL, version TEXT, file_name TEXT NOT NULL);
|
|
CREATE UNIQUE INDEX IF NOT EXISTS one_training ON jobs ((1))
|
|
WHERE status IN ('queued', 'running');
|
|
''')
|
|
columns = {r['name'] for r in db.execute('PRAGMA table_info(jobs)')}
|
|
if 'audit' not in columns:
|
|
db.execute('ALTER TABLE jobs ADD COLUMN audit TEXT')
|
|
meta = json.loads((ARTIFACTS / 'metadata_model.json').read_text(encoding='utf-8'))
|
|
meta.update(version='v1', label_source='Dataset penelitian berlabel', added_rows=0)
|
|
db.execute('INSERT OR IGNORE INTO versions VALUES (?, ?, ?)',
|
|
('v1', meta.get('dibuat_pada_utc', now()), json.dumps(meta)))
|
|
db.execute("INSERT OR IGNORE INTO settings VALUES ('active', 'v1')")
|
|
# A stopped process cannot finish its queued job. No partial data are committed.
|
|
db.execute("UPDATE jobs SET status='failed', message=? WHERE status IN ('queued','running')",
|
|
('Training terhenti saat server dimatikan. Silakan unggah kembali.',))
|
|
|
|
|
|
initialize()
|
|
|
|
|
|
def bundle(version=None):
|
|
with connect() as db:
|
|
version = version or db.execute("SELECT value FROM settings WHERE key='active'").fetchone()[0]
|
|
row = db.execute('SELECT metadata FROM versions WHERE id=?', (version,)).fetchone()
|
|
if row is None:
|
|
raise ValueError('Versi model tidak ditemukan.')
|
|
if json.loads(row[0]).get('invalidated'):
|
|
raise ValueError('Model ini tidak valid karena input training keliru dan tidak boleh digunakan.')
|
|
with _cache_lock:
|
|
if version not in _cache:
|
|
folder = ARTIFACTS if version == 'v1' else STORE / version
|
|
model = joblib.load(folder / 'model_logistic_regression.joblib')
|
|
vectorizer = joblib.load(folder / 'tfidf_vectorizer.joblib')
|
|
meta = json.loads(row[0])
|
|
if model.n_features_in_ != len(vectorizer.get_feature_names_out()) or list(model.classes_) != [0, 1]:
|
|
raise ValueError('Pasangan model dan TF-IDF tidak valid.')
|
|
_cache[version] = (model, vectorizer, meta)
|
|
return _cache[version]
|
|
|
|
|
|
def seed():
|
|
global _seed_cache
|
|
if _seed_cache is not None:
|
|
return _seed_cache
|
|
frame = pd.read_csv(ARTIFACTS / 'training_seed.csv', keep_default_na=False)
|
|
model, vec, meta = bundle('v1')
|
|
if len(frame) != meta['total_data'] or not set(frame.label).issubset({0, 1}):
|
|
raise ValueError('Dataset awal tidak sesuai metadata penelitian.')
|
|
train, test = train_test_split(frame, test_size=meta['split']['test_size'],
|
|
random_state=meta['split']['random_state'], stratify=frame.label)
|
|
predictions = model.predict(vec.transform(test.stemmed_text))
|
|
if confusion_matrix(test.label, predictions, labels=[0, 1]).tolist() != meta['confusion_matrix']:
|
|
raise ValueError('Evaluasi model awal tidak cocok. Training dibatalkan agar data uji tidak berubah.')
|
|
_seed_cache = (frame, train, test)
|
|
return _seed_cache
|
|
|
|
|
|
def history():
|
|
with connect() as db:
|
|
active = db.execute("SELECT value FROM settings WHERE key='active'").fetchone()[0]
|
|
versions = [dict(json.loads(r['metadata']), created_at=r['created_at'], active=r['id'] == active)
|
|
for r in db.execute('SELECT * FROM versions ORDER BY rowid DESC')]
|
|
jobs = [dict(r) for r in db.execute('SELECT * FROM jobs ORDER BY rowid DESC LIMIT 20')]
|
|
for job in jobs:
|
|
job['audit'] = json.loads(job['audit']) if job.get('audit') else None
|
|
return dict(active_version=active, versions=versions, jobs=jobs)
|
|
|
|
|
|
def activate(version):
|
|
bundle(version) # Load and validate before changing the active pointer.
|
|
with connect() as db:
|
|
db.execute("UPDATE settings SET value=? WHERE key='active'", (version,))
|
|
return history()
|
|
|
|
|
|
def queue(file_name):
|
|
job_id = uuid4().hex
|
|
try:
|
|
with connect() as db:
|
|
db.execute('INSERT INTO jobs (id, created_at, status, message, version, file_name) VALUES (?, ?, ?, ?, ?, ?)',
|
|
(job_id, now(), 'queued', 'Menunggu proses training.', None, file_name[:200]))
|
|
except sqlite3.IntegrityError:
|
|
raise ValueError('Training lain masih berjalan. Tunggu hingga selesai.') from None
|
|
return job_id
|
|
|
|
|
|
def train_job(job_id, reviews, selected_column=None):
|
|
try:
|
|
with connect() as db:
|
|
db.execute("UPDATE jobs SET status='running', message=? WHERE id=?",
|
|
('Memeriksa data dan membuat label otomatis...', job_id))
|
|
frame, initial_train, test = seed()
|
|
teacher_model, teacher_vec, teacher_meta = bundle()
|
|
from model_service import preprocessing_detail
|
|
with connect() as db:
|
|
old = [dict(r) for r in db.execute('SELECT * FROM additions ORDER BY rowid')]
|
|
count = db.execute('SELECT COUNT(*) FROM versions').fetchone()[0]
|
|
test_text = {key(x) for x in test.komentar}
|
|
test_stem = {key(x) for x in test.stemmed_text}
|
|
initial_text = {key(x) for x in initial_train.komentar}
|
|
old_text = {r['text_key'] for r in old}
|
|
seen_stem = {key(x) for x in frame.stemmed_text} | {key(r['stem']) for r in old}
|
|
uploaded_text = set()
|
|
additions, skipped = [], dict(duplicate=0, empty=0, unknown=0)
|
|
skipped_rows = []
|
|
def skip(row_number, text, category, reason):
|
|
skipped[category] += 1
|
|
skipped_rows.append(dict(source_row_number=row_number, text=text, reason=reason))
|
|
for index, raw in enumerate(reviews):
|
|
text = raw.strip()
|
|
row_number = index + 2 # CSV record number, including the header.
|
|
if not text:
|
|
skip(row_number, text, 'empty', 'Ulasan kosong')
|
|
continue
|
|
text_key = key(text)
|
|
if text_key in test_text:
|
|
skip(row_number, text, 'duplicate', 'Cocok dengan data uji yang dilindungi')
|
|
continue
|
|
if text_key in initial_text:
|
|
skip(row_number, text, 'duplicate', 'Sudah ada dalam data latih awal')
|
|
continue
|
|
if text_key in old_text:
|
|
skip(row_number, text, 'duplicate', 'Sudah ada dalam data tambahan sebelumnya')
|
|
continue
|
|
if text_key in uploaded_text:
|
|
skip(row_number, text, 'duplicate', 'Duplikat ulasan dalam file unggahan')
|
|
continue
|
|
uploaded_text.add(text_key)
|
|
stem = preprocessing_detail(text)['stemming']
|
|
if not stem.strip():
|
|
skip(row_number, text, 'empty', 'Tidak ada kata setelah preprocessing')
|
|
continue
|
|
if key(stem) in test_stem:
|
|
skip(row_number, text, 'duplicate', 'Hasil preprocessing cocok dengan data uji yang dilindungi')
|
|
continue
|
|
if key(stem) in seen_stem:
|
|
skip(row_number, text, 'duplicate', 'Hasil preprocessing sama dengan data yang sudah ada')
|
|
continue
|
|
features = teacher_vec.transform([stem])
|
|
if not features.nnz:
|
|
skip(row_number, text, 'unknown', 'Tidak ada fitur yang dikenali model pemberi label')
|
|
continue
|
|
label = int(teacher_model.predict(features)[0])
|
|
confidence = float(teacher_model.predict_proba(features)[0].max())
|
|
additions.append(dict(text_key=text_key, text=text, stem=stem, label=label,
|
|
confidence=confidence, teacher=teacher_meta['version']))
|
|
seen_stem.add(key(stem))
|
|
audit = dict(uploaded_rows=len(reviews), accepted_rows=len(additions),
|
|
skipped_rows=skipped_rows, training_before=len(initial_train)+len(old), selected_column=selected_column)
|
|
with connect() as db:
|
|
db.execute('UPDATE jobs SET audit=? WHERE id=?', (json.dumps(audit), job_id))
|
|
if not additions:
|
|
with connect() as db:
|
|
db.execute("UPDATE jobs SET status='skipped', message=? WHERE id=?",
|
|
('Tidak ada data baru yang dapat dilatih. Data latih tetap.', job_id))
|
|
return
|
|
texts = initial_train.stemmed_text.tolist() + [r['stem'] for r in old + additions]
|
|
labels = initial_train.label.tolist() + [r['label'] for r in old + additions]
|
|
with connect() as db:
|
|
db.execute('UPDATE jobs SET message=? WHERE id=?',
|
|
('Melatih TF-IDF dan Logistic Regression, lalu mengevaluasi model...', job_id))
|
|
base_model, base_vec, base_meta = bundle('v1')
|
|
vec, model = clone(base_vec), clone(base_model)
|
|
model.fit(vec.fit_transform(texts), labels)
|
|
predicted = model.predict(vec.transform(test.stemmed_text))
|
|
report = classification_report(test.label, predicted, labels=[0, 1], output_dict=True, zero_division=0)
|
|
version = f'v{count + 1}'
|
|
meta = {k: base_meta[k] for k in ('nama_model', 'label_mapping', 'split', 'preprocessing_order')}
|
|
meta.update(version=version, total_data=len(labels) + len(test),
|
|
jumlah_data_training=len(labels), jumlah_data_testing=len(test),
|
|
jumlah_sentimen_negatif=int(labels.count(0) + (test.label == 0).sum()),
|
|
jumlah_sentimen_positif=int(labels.count(1) + (test.label == 1).sum()),
|
|
jumlah_fitur_tfidf=len(vec.get_feature_names_out()),
|
|
akurasi=float(accuracy_score(test.label, predicted)),
|
|
confusion_matrix=confusion_matrix(test.label, predicted, labels=[0, 1]).tolist(),
|
|
classification_report=report, teacher_version=teacher_meta['version'],
|
|
label_source='Pseudo-labeling (label otomatis dari model aktif)',
|
|
added_rows=len(additions), cumulative_added_rows=len(old) + len(additions),
|
|
audit=audit, skipped=skipped, test_seed_sha256=hashlib.sha256(
|
|
(ARTIFACTS / 'training_seed.csv').read_bytes()).hexdigest(),
|
|
training_parameters=model.get_params())
|
|
# Write artifacts first; publish the version and corpus together in one transaction.
|
|
folder = STORE / version
|
|
folder.mkdir(exist_ok=True)
|
|
joblib.dump(model, folder / 'model_logistic_regression.joblib')
|
|
joblib.dump(vec, folder / 'tfidf_vectorizer.joblib')
|
|
(folder / 'metadata_model.json').write_text(json.dumps(meta, indent=2), encoding='utf-8')
|
|
with connect() as db:
|
|
db.execute('INSERT INTO versions VALUES (?, ?, ?)', (version, now(), json.dumps(meta)))
|
|
db.executemany('INSERT INTO additions VALUES (?, ?, ?, ?, ?, ?, ?)',
|
|
[(r['text_key'], r['text'], r['stem'], r['label'], r['confidence'],
|
|
r['teacher'], version) for r in additions])
|
|
db.execute("UPDATE jobs SET status='completed', message=?, version=? WHERE id=?",
|
|
(f'{len(additions)} ulasan ditambahkan. Model siap ditinjau; model aktif tetap.', version, job_id))
|
|
except Exception as error:
|
|
logging.exception('Training failed: %s', job_id)
|
|
message = str(error) if isinstance(error, ValueError) else 'Training gagal. Periksa log backend lalu coba kembali.'
|
|
with connect() as db:
|
|
db.execute("UPDATE jobs SET status='failed', message=? WHERE id=?", (message, job_id))
|