1277 lines
48 KiB
Python
1277 lines
48 KiB
Python
import os
|
|
import re
|
|
import time
|
|
import math
|
|
import warnings
|
|
import json
|
|
import urllib.request
|
|
import urllib.parse
|
|
from datetime import datetime
|
|
from flask import Flask, send_from_directory, request, jsonify, session
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
from sklearn.metrics import confusion_matrix, classification_report, accuracy_score
|
|
|
|
# Modul NLP dan Model Inferensi
|
|
try:
|
|
from nlp import preprocessing
|
|
except ImportError:
|
|
# pyrefly: ignore [missing-import]
|
|
from Nlp import preprocessing
|
|
|
|
try:
|
|
from models import model, tfidf
|
|
except ImportError:
|
|
from Models import model, tfidf
|
|
|
|
warnings.filterwarnings("ignore")
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
FRONTEND_DIR = os.path.join(BASE_DIR, "frontend")
|
|
SQLITE_DB_PATH = os.path.join(BASE_DIR, "sentiment_tiktok.db")
|
|
|
|
app = Flask(
|
|
__name__,
|
|
static_folder=FRONTEND_DIR,
|
|
static_url_path="/static"
|
|
)
|
|
app.secret_key = os.environ.get("SECRET_KEY", "sentimen-tiktok-secret-key-super-secure-2026")
|
|
|
|
# ==========================================
|
|
# 1. KONFIGURASI DATABASE (MYSQL LARAGON & SQLITE FALLBACK)
|
|
# ==========================================
|
|
|
|
MYSQL_CONFIG = {
|
|
"host": os.environ.get("MYSQL_HOST", "127.0.0.1"),
|
|
"port": int(os.environ.get("MYSQL_PORT", 3306)),
|
|
"user": os.environ.get("MYSQL_USER", "root"),
|
|
"password": os.environ.get("MYSQL_PASSWORD", ""),
|
|
"database": os.environ.get("MYSQL_DB", "sentimen_tiktok"),
|
|
"charset": "utf8mb4"
|
|
}
|
|
|
|
ACTIVE_DB_ENGINE = "mysql" # 'mysql' atau 'sqlite'
|
|
|
|
def try_mysql_connect():
|
|
"""Mencoba koneksi ke MySQL Laragon menggunakan PyMySQL."""
|
|
try:
|
|
import pymysql
|
|
from pymysql.cursors import DictCursor
|
|
conn = pymysql.connect(
|
|
host=MYSQL_CONFIG["host"],
|
|
port=MYSQL_CONFIG["port"],
|
|
user=MYSQL_CONFIG["user"],
|
|
password=MYSQL_CONFIG["password"],
|
|
database=MYSQL_CONFIG["database"],
|
|
charset=MYSQL_CONFIG["charset"],
|
|
cursorclass=DictCursor,
|
|
autocommit=True
|
|
)
|
|
return conn
|
|
except Exception as e:
|
|
return None
|
|
|
|
def get_db():
|
|
"""Mengembalikan koneksi database aktif (prioritas MySQL Laragon, fallback ke SQLite)."""
|
|
global ACTIVE_DB_ENGINE
|
|
mysql_conn = try_mysql_connect()
|
|
if mysql_conn is not None:
|
|
ACTIVE_DB_ENGINE = "mysql"
|
|
return mysql_conn
|
|
|
|
# Fallback ke SQLite jika MySQL tidak aktif
|
|
import sqlite3
|
|
ACTIVE_DB_ENGINE = "sqlite"
|
|
conn = sqlite3.connect(SQLITE_DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def db_query(sql, params=None):
|
|
"""Menjalankan SELECT query dan mengembalikan list of dict."""
|
|
conn = get_db()
|
|
try:
|
|
if ACTIVE_DB_ENGINE == "mysql":
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, params or ())
|
|
return cur.fetchall() or []
|
|
else:
|
|
# SQLite mode: ubah %s menjadi ?
|
|
sql_sqlite = sql.replace("%s", "?")
|
|
cur = conn.cursor()
|
|
cur.execute(sql_sqlite, params or ())
|
|
rows = cur.fetchall() or []
|
|
return [dict(r) for r in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
def db_query_one(sql, params=None):
|
|
"""Menjalankan SELECT query dan mengembalikan single dict atau None."""
|
|
conn = get_db()
|
|
try:
|
|
if ACTIVE_DB_ENGINE == "mysql":
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, params or ())
|
|
return cur.fetchone()
|
|
else:
|
|
sql_sqlite = sql.replace("%s", "?")
|
|
cur = conn.cursor()
|
|
cur.execute(sql_sqlite, params or ())
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
def db_execute(sql, params=None):
|
|
"""Menjalankan query DML (UPDATE/DELETE) dan mengembalikan affected rows."""
|
|
conn = get_db()
|
|
try:
|
|
if ACTIVE_DB_ENGINE == "mysql":
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, params or ())
|
|
conn.commit()
|
|
return cur.rowcount
|
|
else:
|
|
sql_sqlite = sql.replace("%s", "?")
|
|
cur = conn.cursor()
|
|
cur.execute(sql_sqlite, params or ())
|
|
conn.commit()
|
|
return cur.rowcount
|
|
finally:
|
|
conn.close()
|
|
|
|
def db_insert(sql, params=None):
|
|
"""Menjalankan query INSERT dan mengembalikan lastrowid."""
|
|
conn = get_db()
|
|
try:
|
|
if ACTIVE_DB_ENGINE == "mysql":
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, params or ())
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
else:
|
|
sql_sqlite = sql.replace("%s", "?")
|
|
cur = conn.cursor()
|
|
cur.execute(sql_sqlite, params or ())
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
finally:
|
|
conn.close()
|
|
|
|
def db_executemany(sql, params_list):
|
|
"""Menjalankan query INSERT batch dan mengembalikan affected rows."""
|
|
if not params_list:
|
|
return 0
|
|
conn = get_db()
|
|
try:
|
|
if ACTIVE_DB_ENGINE == "mysql":
|
|
with conn.cursor() as cur:
|
|
cur.executemany(sql, params_list)
|
|
conn.commit()
|
|
return cur.rowcount
|
|
else:
|
|
sql_sqlite = sql.replace("%s", "?")
|
|
cur = conn.cursor()
|
|
cur.executemany(sql_sqlite, params_list)
|
|
conn.commit()
|
|
return cur.rowcount
|
|
finally:
|
|
conn.close()
|
|
|
|
def init_db():
|
|
"""Memastikan struktur tabel dan akun admin awal tersedia."""
|
|
conn = get_db()
|
|
try:
|
|
if ACTIVE_DB_ENGINE == "mysql":
|
|
with conn.cursor() as cursor:
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
name VARCHAR(255) NOT NULL,
|
|
email VARCHAR(255) NOT NULL UNIQUE,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
''')
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS analyses (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
user_id INT NOT NULL,
|
|
tiktok_url TEXT NOT NULL,
|
|
video_id VARCHAR(255) DEFAULT NULL,
|
|
status VARCHAR(50) DEFAULT 'done',
|
|
total_comments INT DEFAULT 0,
|
|
positive_count INT DEFAULT 0,
|
|
positive_percentage DOUBLE DEFAULT 0.0,
|
|
negative_count INT DEFAULT 0,
|
|
negative_percentage DOUBLE DEFAULT 0.0,
|
|
error TEXT DEFAULT NULL,
|
|
created_at DATETIME(6) DEFAULT NULL,
|
|
KEY idx_analyses_user (user_id)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
''')
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS comments (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
analysis_id INT NOT NULL,
|
|
original_comment TEXT NOT NULL,
|
|
cleaned_comment TEXT DEFAULT NULL,
|
|
sentiment VARCHAR(50) NOT NULL,
|
|
probability DOUBLE NOT NULL,
|
|
KEY idx_comments_analysis (analysis_id)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
''')
|
|
# Periksa pengguna admin default
|
|
cursor.execute("SELECT id FROM users WHERE email = %s", ("admin@tiktok.com",))
|
|
if not cursor.fetchone():
|
|
cursor.execute(
|
|
"INSERT INTO users (name, email, password_hash) VALUES (%s, %s, %s)",
|
|
("Admin Peneliti", "admin@tiktok.com", generate_password_hash("admin123"))
|
|
)
|
|
conn.commit()
|
|
else:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
email TEXT UNIQUE NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
''')
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS analyses (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
tiktok_url TEXT NOT NULL,
|
|
video_id TEXT,
|
|
status TEXT DEFAULT 'done',
|
|
total_comments INTEGER DEFAULT 0,
|
|
positive_count INTEGER DEFAULT 0,
|
|
positive_percentage REAL DEFAULT 0.0,
|
|
negative_count INTEGER DEFAULT 0,
|
|
negative_percentage REAL DEFAULT 0.0,
|
|
error TEXT,
|
|
created_at TEXT,
|
|
FOREIGN KEY (user_id) REFERENCES users (id)
|
|
)
|
|
''')
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS comments (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
analysis_id INTEGER NOT NULL,
|
|
original_comment TEXT NOT NULL,
|
|
cleaned_comment TEXT,
|
|
sentiment TEXT NOT NULL,
|
|
probability REAL NOT NULL,
|
|
FOREIGN KEY (analysis_id) REFERENCES analyses (id) ON DELETE CASCADE
|
|
)
|
|
''')
|
|
cursor.execute("SELECT id FROM users WHERE email = ?", ("admin@tiktok.com",))
|
|
if not cursor.fetchone():
|
|
cursor.execute(
|
|
"INSERT INTO users (name, email, password_hash) VALUES (?, ?, ?)",
|
|
("Admin Peneliti", "admin@tiktok.com", generate_password_hash("admin123"))
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
# Inisialisasi DB saat start
|
|
init_db()
|
|
|
|
# ==========================================
|
|
# 2. MODUL WEB SCRAPING KOMENTAR TIKTOK OTOMATIS
|
|
# ==========================================
|
|
|
|
def scrape_tiktok_comments(tiktok_url: str, max_comments: int = 100):
|
|
"""
|
|
Modul Web Scraping Otomatis untuk mengekstrak komentar nyata dari tautan video TikTok.
|
|
Mengambil metadata video dan daftar komentar asli via endpoint scraping.
|
|
"""
|
|
comments = []
|
|
meta = {}
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
|
|
'Accept': 'application/json, text/plain, */*'
|
|
}
|
|
|
|
# Normalisasi tautan TikTok (bila user memasukkan ID angka atau judul dengan format (#ID))
|
|
clean_url = str(tiktok_url).strip()
|
|
id_match = re.search(r'#?(\d{15,22})', clean_url)
|
|
if not clean_url.startswith("http") and id_match:
|
|
clean_url = f"https://www.tiktok.com/@video/video/{id_match.group(1)}"
|
|
|
|
# 1. Ambil Metadata Video (Judul, Author, Total Komentar)
|
|
try:
|
|
info_api = f'https://www.tikwm.com/api/?url={urllib.parse.quote(clean_url)}'
|
|
req = urllib.request.Request(info_api, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=10) as res:
|
|
info_json = json.loads(res.read().decode('utf-8'))
|
|
if info_json.get('code') == 0 and info_json.get('data'):
|
|
d = info_json['data']
|
|
meta = {
|
|
"id": str(d.get("id", "")),
|
|
"title": d.get("title", ""),
|
|
"author": d.get("author", {}).get("nickname", ""),
|
|
"author_id": d.get("author", {}).get("unique_id", ""),
|
|
"comment_count": d.get("comment_count", 0),
|
|
"likes": d.get("digg_count", 0)
|
|
}
|
|
except Exception as e:
|
|
print("[!] Gagal mengambil metadata video TikTok:", e)
|
|
|
|
# 2. Ambil Komentar Asli (Dengan Paginasi Cursor hingga max_comments)
|
|
cursor = 0
|
|
# Berikan buffer halaman jika ada komentar duplikat atau emote murni
|
|
max_pages = max(1, math.ceil(max_comments / 40))
|
|
max_pages = min(max_pages, 25) # Batas maksimal aman dinamis hingga ~1000 komentar
|
|
|
|
# Berikan jeda sejenak setelah info request pertama
|
|
time.sleep(1.2)
|
|
|
|
for page in range(max_pages):
|
|
api_url = f'https://www.tikwm.com/api/comment/list?url={urllib.parse.quote(clean_url)}&count=50&cursor={cursor}'
|
|
req = urllib.request.Request(api_url, headers=headers)
|
|
success_page = False
|
|
|
|
# Coba hingga 2 kali jika terkena rate-limit TikWM
|
|
for attempt in range(2):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=12) as res:
|
|
cdata = json.loads(res.read().decode('utf-8'))
|
|
|
|
# Jika terkena limit TikWM "Free Api Limit: 1 request/second"
|
|
if cdata.get('code') != 0:
|
|
msg = str(cdata.get('msg', ''))
|
|
if 'limit' in msg.lower() or cdata.get('code') == -1:
|
|
time.sleep(1.8)
|
|
continue
|
|
break
|
|
|
|
c_data_obj = cdata.get('data', {})
|
|
items = c_data_obj.get('comments', [])
|
|
cursor = c_data_obj.get('cursor', 0)
|
|
has_more = c_data_obj.get('hasMore', False)
|
|
|
|
for it in items:
|
|
txt = it.get('text', '').strip()
|
|
if txt and txt not in comments:
|
|
comments.append(txt)
|
|
|
|
success_page = True
|
|
if not has_more or not items or len(comments) >= max_comments:
|
|
break
|
|
break
|
|
except Exception as e:
|
|
print(f"[!] Error saat scraping halaman komentar {page+1} (attempt {attempt+1}):", e)
|
|
time.sleep(1.5)
|
|
|
|
if len(comments) >= max_comments or not success_page:
|
|
break
|
|
|
|
# Jeda 1.4 detik antar halaman agar mematuhi aturan TikWM 1 req/detik
|
|
time.sleep(1.4)
|
|
|
|
return {
|
|
"success": len(comments) > 0,
|
|
"comments": comments[:max_comments],
|
|
"meta": meta
|
|
}
|
|
|
|
# ==========================================
|
|
# 4. HELPER PREDIKSI SENTIMEN
|
|
# ==========================================
|
|
|
|
def analyze_single_text(teks: str):
|
|
teks_bersih = preprocessing(teks)
|
|
|
|
# Bila hasil preprocessing kosong, tandai gagal dianalisis
|
|
if not teks_bersih or not teks_bersih.strip():
|
|
return {
|
|
"cleaned": teks_bersih if teks_bersih else "",
|
|
"sentiment": "GAGAL DIANALISIS",
|
|
"probability": 0.0,
|
|
"prob_negatif": 0.0,
|
|
"prob_positif": 0.0
|
|
}
|
|
|
|
# Evaluasi teks bersih ke TF-IDF
|
|
X = tfidf.transform([teks_bersih])
|
|
has_features = bool(X.nnz > 0)
|
|
|
|
# Bila fitur kata tidak ditemukan di TF-IDF, jangan gunakan bias model
|
|
if not has_features:
|
|
return {
|
|
"cleaned": teks_bersih,
|
|
"sentiment": "FITUR TIDAK DITEMUKAN",
|
|
"probability": 0.0,
|
|
"prob_negatif": 0.0,
|
|
"prob_positif": 0.0,
|
|
"fitur_ditemukan": False,
|
|
"keterangan": "Fitur tidak ditemukan"
|
|
}
|
|
|
|
pred = model.predict(X)[0]
|
|
proba = model.predict_proba(X)[0]
|
|
prob_neg = float(proba[0])
|
|
prob_pos = float(proba[1])
|
|
score = prob_pos if pred == 'POSITIF' else prob_neg
|
|
|
|
return {
|
|
"cleaned": teks_bersih,
|
|
"sentiment": str(pred),
|
|
"probability": round(score, 4),
|
|
"prob_negatif": round(prob_neg, 4),
|
|
"prob_positif": round(prob_pos, 4),
|
|
"fitur_ditemukan": True,
|
|
"keterangan": "Fitur berhasil diekstraksi"
|
|
}
|
|
|
|
# ==========================================
|
|
# 4. STATUS SISTEM & DATABASE
|
|
# ==========================================
|
|
|
|
@app.route("/api/db-status", methods=["GET"])
|
|
def api_db_status():
|
|
"""Mengembalikan status engine database yang sedang aktif."""
|
|
return jsonify({
|
|
"status": "connected",
|
|
"engine": ACTIVE_DB_ENGINE,
|
|
"database": MYSQL_CONFIG["database"] if ACTIVE_DB_ENGINE == "mysql" else "sentiment_tiktok.db",
|
|
"host": f"{MYSQL_CONFIG['host']}:{MYSQL_CONFIG['port']}" if ACTIVE_DB_ENGINE == "mysql" else "local-sqlite",
|
|
"info": "MySQL Laragon Aktif" if ACTIVE_DB_ENGINE == "mysql" else "SQLite Lokal Aktif"
|
|
})
|
|
|
|
# ==========================================
|
|
# 5. ENDPOINT OTENTIKASI & USER
|
|
# ==========================================
|
|
|
|
@app.route("/api/me", methods=["GET"])
|
|
def api_me():
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
user = db_query_one("SELECT id, name, email FROM users WHERE id = %s", (user_id,))
|
|
if not user:
|
|
session.clear()
|
|
return jsonify({"message": "User not found"}), 401
|
|
|
|
is_admin = (user["email"].lower() == "admin@tiktok.com" or "admin" in user["name"].lower())
|
|
return jsonify({"user": {"id": user["id"], "name": user["name"], "email": user["email"], "is_admin": is_admin}})
|
|
|
|
|
|
@app.route("/api/register", methods=["POST"])
|
|
def api_register():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
name = data.get("name", "").strip()
|
|
email = data.get("email", "").strip().lower()
|
|
pw = data.get("password", "")
|
|
pw2 = data.get("password_confirm", "")
|
|
|
|
if not name or not email or not pw:
|
|
return jsonify({"message": "Nama, email, dan password wajib diisi."}), 400
|
|
if pw != pw2:
|
|
return jsonify({"message": "Konfirmasi password tidak cocok."}), 400
|
|
if len(pw) < 6:
|
|
return jsonify({"message": "Password minimal 6 karakter."}), 400
|
|
|
|
existing = db_query_one("SELECT id FROM users WHERE email = %s", (email,))
|
|
if existing:
|
|
return jsonify({"message": "Email sudah terdaftar."}), 400
|
|
|
|
user_id = db_insert(
|
|
"INSERT INTO users (name, email, password_hash) VALUES (%s, %s, %s)",
|
|
(name, email, generate_password_hash(pw))
|
|
)
|
|
|
|
session["user_id"] = user_id
|
|
is_admin = (email == "admin@tiktok.com" or "admin" in name.lower())
|
|
return jsonify({"user": {"id": user_id, "name": name, "email": email, "is_admin": is_admin}})
|
|
|
|
|
|
@app.route("/api/login", methods=["POST"])
|
|
def api_login():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
email = data.get("email", "").strip().lower()
|
|
pw = data.get("password", "")
|
|
|
|
if not email or not pw:
|
|
return jsonify({"message": "Email dan password wajib diisi."}), 400
|
|
|
|
user = db_query_one("SELECT id, name, email, password_hash FROM users WHERE email = %s", (email,))
|
|
if not user:
|
|
return jsonify({"message": "Email atau password salah."}), 400
|
|
|
|
# Verifikasi password dengan hash atau fallback untuk akun admin default
|
|
is_valid = check_password_hash(user["password_hash"], pw)
|
|
if not is_valid and email == "admin@tiktok.com" and pw == "admin123":
|
|
# Perbarui hash admin ke format terkini
|
|
new_hash = generate_password_hash("admin123")
|
|
db_execute("UPDATE users SET password_hash = %s WHERE id = %s", (new_hash, user["id"]))
|
|
is_valid = True
|
|
|
|
if not is_valid:
|
|
return jsonify({"message": "Email atau password salah."}), 400
|
|
|
|
session["user_id"] = user["id"]
|
|
is_admin = (user["email"].lower() == "admin@tiktok.com" or "admin" in user["name"].lower())
|
|
return jsonify({"user": {"id": user["id"], "name": user["name"], "email": user["email"], "is_admin": is_admin}})
|
|
|
|
|
|
@app.route("/api/logout", methods=["POST"])
|
|
def api_logout():
|
|
session.clear()
|
|
session.modified = True
|
|
resp = jsonify({"success": True})
|
|
resp.delete_cookie(app.config.get("SESSION_COOKIE_NAME", "session"))
|
|
return resp
|
|
|
|
# ==========================================
|
|
# 6. ENDPOINT STATISTIK & ANALISIS
|
|
# ==========================================
|
|
|
|
@app.route("/api/stats", methods=["GET"])
|
|
def api_stats():
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
row = db_query_one('''
|
|
SELECT COUNT(*) as total_analyses,
|
|
COALESCE(SUM(total_comments), 0) as total_comments,
|
|
COALESCE(SUM(positive_count), 0) as positive,
|
|
COALESCE(SUM(negative_count), 0) as negative
|
|
FROM analyses WHERE user_id = %s
|
|
''', (user_id,))
|
|
|
|
failed_row = db_query_one('''
|
|
SELECT COUNT(*) as failed
|
|
FROM comments c
|
|
JOIN analyses a ON c.analysis_id = a.id
|
|
WHERE a.user_id = %s AND c.sentiment = 'GAGAL DIANALISIS'
|
|
''', (user_id,))
|
|
|
|
return jsonify({
|
|
"total_analyses": int(row["total_analyses"] or 0) if row else 0,
|
|
"total_comments": int(row["total_comments"] or 0) if row else 0,
|
|
"positive": int(row["positive"] or 0) if row else 0,
|
|
"negative": int(row["negative"] or 0) if row else 0,
|
|
"failed": int(failed_row["failed"] or 0) if failed_row else 0
|
|
})
|
|
|
|
|
|
@app.route("/api/analyses", methods=["GET", "POST"])
|
|
def api_analyses():
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
if request.method == "GET":
|
|
rows = db_query('''
|
|
SELECT id, tiktok_url, video_id, status, total_comments,
|
|
positive_count, positive_percentage, negative_count,
|
|
negative_percentage, error, created_at
|
|
FROM analyses WHERE user_id = %s
|
|
ORDER BY id DESC
|
|
''', (user_id,))
|
|
|
|
# Format created_at ke ISO string jika berupa datetime
|
|
for r in rows:
|
|
if hasattr(r.get("created_at"), "isoformat"):
|
|
r["created_at"] = r["created_at"].isoformat()
|
|
|
|
return jsonify({"data": rows})
|
|
|
|
# POST - Mulai Analisis
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
tiktok_url = data.get("tiktok_url", "").strip()
|
|
manual = data.get("manual_comments", "").strip()
|
|
max_comments = int(data.get("max_comments", 100))
|
|
max_comments = max(1, min(1000, max_comments))
|
|
|
|
# Validasi: Minimal salah satu harus diisi (URL atau Komentar Langsung)
|
|
if not tiktok_url and not manual:
|
|
return jsonify({"message": "Harap masukkan link video TikTok atau ketik komentar secara langsung."}), 400
|
|
|
|
|
|
now_dt = datetime.now()
|
|
raw_list = []
|
|
display_title = tiktok_url
|
|
video_id = "video"
|
|
status_source = "done"
|
|
scrape_notice = ""
|
|
|
|
# Penanganan Komentar: Manual (Kata Langsung / File) vs Scraping Link Otomatis
|
|
if manual:
|
|
status_source = "manual"
|
|
# Dukung multi-baris serta pemisah koma / titik koma jika 1 baris
|
|
lines = [line.strip() for line in manual.splitlines() if line.strip()]
|
|
for line in lines:
|
|
if ("," in line or ";" in line) and len(line) > 30 and len(lines) == 1:
|
|
sep = "," if "," in line else ";"
|
|
parts = [p.strip() for p in line.split(sep) if p.strip()]
|
|
raw_list.extend(parts)
|
|
else:
|
|
raw_list.append(line)
|
|
|
|
if not raw_list:
|
|
return jsonify({"message": "Komentar yang dimasukkan kosong atau tidak terbaca."}), 400
|
|
|
|
# Jika URL kosong, berikan judul otomatis yang informatif
|
|
if not tiktok_url:
|
|
if len(raw_list) == 1:
|
|
snippet = raw_list[0][:30] + ("..." if len(raw_list[0]) > 30 else "")
|
|
display_title = f'Uji Komentar: "{snippet}"'
|
|
else:
|
|
display_title = f"Input Komentar Mandiri ({len(raw_list)} Komentar - {now_dt.strftime('%d/%m/%Y %H:%M')})"
|
|
video_id = "manual"
|
|
else:
|
|
vid_match = re.search(r'/video/(\d+)', tiktok_url)
|
|
video_id = vid_match.group(1) if vid_match else re.sub(r'\D', '', tiktok_url)[:15] or "kustom"
|
|
display_title = tiktok_url
|
|
else:
|
|
# Mode Link Video Otomatis -> JALANKAN WEB SCRAPING OTOMATIS KE TIKTOK!
|
|
vid_match = re.search(r'/video/(\d+)', tiktok_url)
|
|
video_id = vid_match.group(1) if vid_match else re.sub(r'\D', '', tiktok_url)[:15] or "video"
|
|
|
|
print(f"[*] Menjalankan scraping komentar langsung untuk link: {tiktok_url} (target max: {max_comments})")
|
|
scrape_res = scrape_tiktok_comments(tiktok_url, max_comments=max_comments)
|
|
|
|
if scrape_res["success"] and len(scrape_res["comments"]) > 0:
|
|
raw_list = scrape_res["comments"]
|
|
status_source = "scraped"
|
|
author = scrape_res.get("meta", {}).get("author")
|
|
title = scrape_res.get("meta", {}).get("title")
|
|
real_vid_id = scrape_res.get("meta", {}).get("id")
|
|
if real_vid_id:
|
|
video_id = real_vid_id
|
|
if author:
|
|
display_title = f"TikTok @{author} (#{video_id})"
|
|
print(f"[OK] Berhasil men-scrape {len(raw_list)} komentar nyata dari TikTok!")
|
|
else:
|
|
print("[!] Scraping tidak menemukan komentar pada video ini.")
|
|
return jsonify({
|
|
"message": "Gagal mengambil komentar dari video TikTok tersebut. Pastikan video bersifat publik, tidak dikunci/privat, dan telah memiliki komentar aktif. Atau gunakan tab 'Input Kata Langsung' untuk menganalisis komentar secara mandiri."
|
|
}), 400
|
|
|
|
# Lakukan inferensi klasifikasi sentimen
|
|
processed_records = []
|
|
pos_count = 0
|
|
neg_count = 0
|
|
failed_count = 0
|
|
oov_count = 0
|
|
|
|
for comment in raw_list:
|
|
res = analyze_single_text(comment)
|
|
if res["sentiment"] == "POSITIF":
|
|
pos_count += 1
|
|
elif res["sentiment"] == "NEGATIF":
|
|
neg_count += 1
|
|
elif res["sentiment"] == "FITUR TIDAK DITEMUKAN":
|
|
oov_count += 1
|
|
else:
|
|
failed_count += 1
|
|
processed_records.append((comment, res["cleaned"], res["sentiment"], res["probability"]))
|
|
|
|
total = len(processed_records)
|
|
pos_pct = round((pos_count / total * 100), 1) if total > 0 else 0.0
|
|
neg_pct = round((neg_count / total * 100), 1) if total > 0 else 0.0
|
|
|
|
notices = []
|
|
if scrape_notice:
|
|
notices.append(scrape_notice)
|
|
if failed_count > 0:
|
|
notices.append(f"{failed_count} komentar ditandai 'GAGAL DIANALISIS' karena hasil preprocessing kosong.")
|
|
if oov_count > 0:
|
|
notices.append(f"{oov_count} komentar ditandai 'FITUR TIDAK DITEMUKAN' karena kata tidak ada di kosakata TF-IDF.")
|
|
final_notice = " | ".join(notices) if notices else None
|
|
|
|
# Simpan riwayat analisis ke database
|
|
analysis_id = db_insert('''
|
|
INSERT INTO analyses (
|
|
user_id, tiktok_url, video_id, status, total_comments,
|
|
positive_count, positive_percentage, negative_count,
|
|
negative_percentage, error, created_at
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
''', (user_id, display_title, video_id, status_source, total, pos_count, pos_pct, neg_count, neg_pct, final_notice, now_dt))
|
|
|
|
# Simpan komentar secara batch
|
|
db_executemany('''
|
|
INSERT INTO comments (analysis_id, original_comment, cleaned_comment, sentiment, probability)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
''', [(analysis_id, c[0], c[1], c[2], c[3]) for c in processed_records])
|
|
|
|
return jsonify({
|
|
"data": {
|
|
"id": analysis_id,
|
|
"status": status_source,
|
|
"is_scraped": (status_source == "scraped"),
|
|
"total_comments": total,
|
|
"notice": scrape_notice
|
|
}
|
|
})
|
|
|
|
|
|
@app.route("/api/analyses/<int:analysis_id>", methods=["GET", "DELETE"])
|
|
def api_analysis_detail(analysis_id):
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
if request.method == "DELETE":
|
|
db_execute("DELETE FROM comments WHERE analysis_id = %s", (analysis_id,))
|
|
db_execute("DELETE FROM analyses WHERE id = %s AND user_id = %s", (analysis_id, user_id))
|
|
return jsonify({"success": True})
|
|
|
|
row = db_query_one('''
|
|
SELECT id, tiktok_url, video_id, status, total_comments,
|
|
positive_count, positive_percentage, negative_count,
|
|
negative_percentage, error, created_at
|
|
FROM analyses WHERE id = %s AND user_id = %s
|
|
''', (analysis_id, user_id))
|
|
|
|
if not row:
|
|
return jsonify({"message": "Analisis tidak ditemukan."}), 404
|
|
|
|
if hasattr(row.get("created_at"), "isoformat"):
|
|
row["created_at"] = row["created_at"].isoformat()
|
|
|
|
return jsonify({"data": row})
|
|
|
|
|
|
@app.route("/api/analyses/<int:analysis_id>/comments", methods=["GET"])
|
|
def api_analysis_comments(analysis_id):
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
page = max(1, int(request.args.get("page", 1)))
|
|
limit = max(1, min(100, int(request.args.get("limit", 50))))
|
|
sentiment = request.args.get("sentiment", "").strip().upper()
|
|
offset = (page - 1) * limit
|
|
|
|
# Verifikasi kepemilikan analisis
|
|
valid = db_query_one("SELECT id FROM analyses WHERE id = %s AND user_id = %s", (analysis_id, user_id))
|
|
if not valid:
|
|
return jsonify({"message": "Analisis tidak ditemukan."}), 404
|
|
|
|
if sentiment in ["POSITIF", "NEGATIF", "GAGAL DIANALISIS", "GAGAL", "FITUR TIDAK DITEMUKAN"]:
|
|
query_sentiment = "GAGAL DIANALISIS" if sentiment in ["GAGAL DIANALISIS", "GAGAL"] else sentiment
|
|
tot_row = db_query_one(
|
|
"SELECT COUNT(*) as cnt FROM comments WHERE analysis_id = %s AND sentiment = %s",
|
|
(analysis_id, query_sentiment)
|
|
)
|
|
total = int(tot_row["cnt"] if tot_row else 0)
|
|
rows = db_query(
|
|
"SELECT id, original_comment, cleaned_comment, sentiment, probability FROM comments WHERE analysis_id = %s AND sentiment = %s ORDER BY id ASC LIMIT %s OFFSET %s",
|
|
(analysis_id, query_sentiment, limit, offset)
|
|
)
|
|
else:
|
|
tot_row = db_query_one(
|
|
"SELECT COUNT(*) as cnt FROM comments WHERE analysis_id = %s",
|
|
(analysis_id,)
|
|
)
|
|
total = int(tot_row["cnt"] if tot_row else 0)
|
|
rows = db_query(
|
|
"SELECT id, original_comment, cleaned_comment, sentiment, probability FROM comments WHERE analysis_id = %s ORDER BY id ASC LIMIT %s OFFSET %s",
|
|
(analysis_id, limit, offset)
|
|
)
|
|
|
|
total_pages = max(1, math.ceil(total / limit)) if total > 0 else 1
|
|
|
|
formatted_rows = []
|
|
for r in rows:
|
|
row_dict = dict(r)
|
|
c_clean = (row_dict.get("cleaned_comment") or "").strip()
|
|
sent = row_dict.get("sentiment")
|
|
if sent == "GAGAL DIANALISIS" or not c_clean:
|
|
row_dict["fitur_ditemukan"] = False
|
|
row_dict["keterangan"] = "Preprocessing kosong"
|
|
else:
|
|
has_feat = bool(tfidf.transform([c_clean]).nnz > 0)
|
|
row_dict["fitur_ditemukan"] = has_feat
|
|
row_dict["keterangan"] = "Fitur berhasil diekstraksi" if has_feat else "Fitur tidak ditemukan"
|
|
formatted_rows.append(row_dict)
|
|
|
|
return jsonify({
|
|
"page": page,
|
|
"limit": limit,
|
|
"total": total,
|
|
"total_pages": total_pages,
|
|
"data": formatted_rows
|
|
})
|
|
|
|
# ==========================================
|
|
# 7. STANDALONE PREDIKSI TUNGGAL (LIVE NLP PLAYGROUND / WORKBENCH)
|
|
# ==========================================
|
|
|
|
@app.route("/predict", methods=["POST"])
|
|
def predict():
|
|
"""Endpoint inferensi NLP langsung untuk kata/kalimat tunggal dengan detail fitur TF-IDF."""
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
komentar = data.get("komentar", "").strip()
|
|
|
|
if not komentar:
|
|
return jsonify({"success": False, "message": "Komentar tidak boleh kosong."}), 400
|
|
|
|
start_time = time.time()
|
|
hasil_preprocessing = preprocessing(komentar)
|
|
|
|
# Bila hasil preprocessing kosong, tandai gagal dianalisis
|
|
if not hasil_preprocessing or not hasil_preprocessing.strip():
|
|
elapsed_ms = round((time.time() - start_time) * 1000, 2)
|
|
return jsonify({
|
|
"success": True,
|
|
"teks_asli": komentar,
|
|
"sentimen": "GAGAL DIANALISIS",
|
|
"preprocessing": "",
|
|
"probabilitas": {
|
|
"negatif": 0.0,
|
|
"positif": 0.0
|
|
},
|
|
"keyakinan": 0.0,
|
|
"decision_value": 0.0,
|
|
"fitur_dominan": [],
|
|
"elapsed_ms": max(0.5, elapsed_ms),
|
|
"keterangan": "Hasil preprocessing kosong (teks hanya berisi emotikon, angka, simbol, atau kata henti sehingga tidak dapat dianalisis)."
|
|
})
|
|
|
|
X = tfidf.transform([hasil_preprocessing])
|
|
feature_names = tfidf.get_feature_names_out()
|
|
non_zero_indices = X.nonzero()[1]
|
|
has_features = len(non_zero_indices) > 0
|
|
elapsed_ms = round((time.time() - start_time) * 1000, 2)
|
|
|
|
# Bila fitur kata tidak ditemukan di TF-IDF, jangan gunakan bias model
|
|
if not has_features:
|
|
return jsonify({
|
|
"success": True,
|
|
"teks_asli": komentar,
|
|
"sentimen": "FITUR TIDAK DITEMUKAN",
|
|
"preprocessing": hasil_preprocessing,
|
|
"probabilitas": {
|
|
"negatif": 0.0,
|
|
"positif": 0.0
|
|
},
|
|
"keyakinan": 0.0,
|
|
"decision_value": 0.0,
|
|
"fitur_dominan": [],
|
|
"fitur_ditemukan": False,
|
|
"keterangan": "Fitur tidak ditemukan",
|
|
"elapsed_ms": max(0.5, elapsed_ms)
|
|
})
|
|
|
|
hasil_prediksi = model.predict(X)[0]
|
|
proba = model.predict_proba(X)[0]
|
|
prob_negatif = float(proba[0])
|
|
prob_positif = float(proba[1])
|
|
decision_val = float(model.decision_function(X)[0])
|
|
|
|
features_info = []
|
|
for idx in non_zero_indices:
|
|
token = feature_names[idx]
|
|
val = float(X[0, idx])
|
|
weight = float(model.coef_[0, idx])
|
|
features_info.append({
|
|
"token": token,
|
|
"tfidf": round(val, 4),
|
|
"weight": round(weight, 4)
|
|
})
|
|
|
|
features_info.sort(key=lambda item: abs(item['weight'] * item['tfidf']), reverse=True)
|
|
top_features = features_info[:8]
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"teks_asli": komentar,
|
|
"sentimen": str(hasil_prediksi),
|
|
"preprocessing": hasil_preprocessing,
|
|
"probabilitas": {
|
|
"negatif": round(prob_negatif, 4),
|
|
"positif": round(prob_positif, 4)
|
|
},
|
|
"keyakinan": round((prob_negatif if hasil_prediksi == 'NEGATIF' else prob_positif) * 100, 1),
|
|
"decision_value": round(decision_val, 4),
|
|
"fitur_dominan": top_features,
|
|
"fitur_ditemukan": True,
|
|
"keterangan": "Fitur berhasil diekstraksi",
|
|
"elapsed_ms": max(0.5, elapsed_ms)
|
|
})
|
|
|
|
# ==========================================
|
|
# 8. MODUL PENGUJIAN & EVALUASI MODEL (CONFUSION MATRIX)
|
|
# ==========================================
|
|
|
|
@app.route("/api/model-test/scrape-tiktok", methods=["POST"])
|
|
def api_model_test_scrape_tiktok():
|
|
"""Scrape komentar langsung dari tautan video TikTok untuk pengujian model."""
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
tiktok_url = data.get("tiktok_url", "").strip()
|
|
max_comments = int(data.get("max_comments", 50))
|
|
max_comments = max(1, min(1000, max_comments))
|
|
|
|
if not tiktok_url:
|
|
return jsonify({"message": "Harap masukkan link video TikTok."}), 400
|
|
|
|
scrape_res = scrape_tiktok_comments(tiktok_url, max_comments=max_comments)
|
|
if not scrape_res["success"] or not scrape_res["comments"]:
|
|
return jsonify({
|
|
"message": "Gagal mengambil komentar dari video TikTok tersebut. Pastikan tautan video bersifat publik dan telah memiliki komentar aktif."
|
|
}), 400
|
|
|
|
raw_comments = scrape_res["comments"]
|
|
clean_texts = [preprocessing(c) for c in raw_comments]
|
|
|
|
items = []
|
|
for i, comm in enumerate(raw_comments):
|
|
c_clean = clean_texts[i]
|
|
if not c_clean or not c_clean.strip():
|
|
pred = "GAGAL DIANALISIS"
|
|
conf = 0.0
|
|
prob_neg = 0.0
|
|
prob_pos = 0.0
|
|
has_feat = False
|
|
keterangan = "Hasil preprocessing kosong"
|
|
else:
|
|
X_single = tfidf.transform([c_clean])
|
|
has_feat = bool(X_single.nnz > 0)
|
|
if not has_feat:
|
|
pred = "FITUR TIDAK DITEMUKAN"
|
|
conf = 0.0
|
|
prob_neg = 0.0
|
|
prob_pos = 0.0
|
|
keterangan = "Fitur tidak ditemukan"
|
|
else:
|
|
pred = str(model.predict(X_single)[0])
|
|
proba = model.predict_proba(X_single)[0]
|
|
prob_neg = float(proba[0])
|
|
prob_pos = float(proba[1])
|
|
conf = prob_pos if pred == "POSITIF" else prob_neg
|
|
keterangan = "Fitur berhasil diekstraksi"
|
|
|
|
items.append({
|
|
"id": i + 1,
|
|
"original_comment": comm,
|
|
"cleaned_comment": c_clean,
|
|
"predicted_label": pred,
|
|
"actual_label": pred if pred in ["POSITIF", "NEGATIF"] else "NEGATIF", # Default ground truth
|
|
"confidence": round(conf * 100, 1),
|
|
"prob_positif": round(prob_pos, 4),
|
|
"prob_negatif": round(prob_neg, 4),
|
|
"fitur_ditemukan": has_feat,
|
|
"keterangan": keterangan
|
|
})
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"meta": scrape_res.get("meta", {}),
|
|
"total_comments": len(items),
|
|
"comments": items
|
|
})
|
|
|
|
|
|
@app.route("/api/model-test/analyses-list", methods=["GET"])
|
|
def api_model_test_analyses_list():
|
|
"""Daftar riwayat video TikTok yang pernah dianalisis sebelumnya di database."""
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
rows = db_query('''
|
|
SELECT id, tiktok_url, video_id, total_comments, created_at
|
|
FROM analyses WHERE user_id = %s AND total_comments > 0
|
|
ORDER BY id DESC LIMIT 20
|
|
''', (user_id,))
|
|
|
|
for r in rows:
|
|
if hasattr(r.get("created_at"), "isoformat"):
|
|
r["created_at"] = r["created_at"].isoformat()
|
|
|
|
return jsonify({"data": rows})
|
|
|
|
|
|
@app.route("/api/model-test/load-analysis/<int:analysis_id>", methods=["GET"])
|
|
def api_model_test_load_analysis(analysis_id):
|
|
"""Memuat komentar dari riwayat video TikTok database untuk diuji dengan Confusion Matrix."""
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
analysis = db_query_one("SELECT * FROM analyses WHERE id = %s AND user_id = %s", (analysis_id, user_id))
|
|
if not analysis:
|
|
return jsonify({"message": "Analisis tidak ditemukan."}), 404
|
|
|
|
comms = db_query(
|
|
"SELECT id, original_comment, cleaned_comment, sentiment, probability FROM comments WHERE analysis_id = %s ORDER BY id ASC LIMIT 200",
|
|
(analysis_id,)
|
|
)
|
|
|
|
items = []
|
|
for i, c in enumerate(comms):
|
|
items.append({
|
|
"id": i + 1,
|
|
"original_comment": c["original_comment"],
|
|
"cleaned_comment": c.get("cleaned_comment") or "",
|
|
"predicted_label": c["sentiment"],
|
|
"actual_label": c["sentiment"],
|
|
"confidence": round(float(c["probability"]) * 100, 1)
|
|
})
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"meta": {
|
|
"title": analysis["tiktok_url"],
|
|
"video_id": analysis["video_id"],
|
|
"total_comments": analysis["total_comments"]
|
|
},
|
|
"total_comments": len(items),
|
|
"comments": items
|
|
})
|
|
|
|
|
|
@app.route("/api/model-test/evaluate", methods=["POST"])
|
|
def api_model_test_evaluate():
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"message": "Unauthorized"}), 401
|
|
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
items = data.get("dataset", [])
|
|
|
|
if not items or not isinstance(items, list):
|
|
return jsonify({"message": "Dataset pengujian kosong atau format tidak sesuai."}), 400
|
|
|
|
# Normalisasi label dan validasi
|
|
valid_records = []
|
|
for idx, item in enumerate(items):
|
|
txt = str(item.get("comment", item.get("original_comment", ""))).strip()
|
|
lbl = str(item.get("actual_label", item.get("label", ""))).strip().upper()
|
|
if not txt:
|
|
continue
|
|
# Map variasi label
|
|
if lbl in ["1", "POS", "POSITIF", "POSITIVE"]:
|
|
clean_lbl = "POSITIF"
|
|
elif lbl in ["0", "-1", "NEG", "NEGATIF", "NEGATIVE"]:
|
|
clean_lbl = "NEGATIF"
|
|
else:
|
|
clean_lbl = "POSITIF" if "POS" in lbl else ("NEGATIF" if "NEG" in lbl else None)
|
|
|
|
if not clean_lbl:
|
|
return jsonify({"message": f"Baris #{idx+1}: Label '{lbl}' tidak valid. Gunakan 'POSITIF' atau 'NEGATIF'."}), 400
|
|
|
|
valid_records.append({"comment": txt, "label": clean_lbl})
|
|
|
|
if len(valid_records) < 2:
|
|
return jsonify({"message": "Minimal diperlukan 2 data berlabel untuk pengujian model."}), 400
|
|
|
|
start_time = time.time()
|
|
|
|
# Preprocessing NLP
|
|
comments = [it["comment"] for it in valid_records]
|
|
y_true = [it["label"] for it in valid_records]
|
|
clean_texts = [preprocessing(c) for c in comments]
|
|
|
|
# Transformasi TF-IDF dan Prediksi
|
|
y_pred = []
|
|
y_proba = []
|
|
for c in clean_texts:
|
|
if not c or not c.strip():
|
|
y_pred.append("GAGAL DIANALISIS")
|
|
y_proba.append([0.0, 0.0])
|
|
else:
|
|
X_single = tfidf.transform([c])
|
|
if X_single.nnz == 0:
|
|
y_pred.append("FITUR TIDAK DITEMUKAN")
|
|
y_proba.append([0.0, 0.0])
|
|
else:
|
|
pred_single = str(model.predict(X_single)[0])
|
|
proba_single = model.predict_proba(X_single)[0]
|
|
y_pred.append(pred_single)
|
|
y_proba.append([float(proba_single[0]), float(proba_single[1])])
|
|
|
|
# Hitung Confusion Matrix (Urutan: ['POSITIF', 'NEGATIF'])
|
|
labels = ["POSITIF", "NEGATIF"]
|
|
cm = confusion_matrix(y_true, y_pred, labels=labels)
|
|
# cm:
|
|
# Row 0: True Positif -> [TP, FN]
|
|
# Row 1: True Negatif -> [FP, TN]
|
|
tp = int(cm[0][0])
|
|
fn = int(cm[0][1])
|
|
fp = int(cm[1][0])
|
|
tn = int(cm[1][1])
|
|
total = len(valid_records)
|
|
|
|
# Perhitungan Metrik Evaluasi
|
|
acc = float(accuracy_score(y_true, y_pred))
|
|
|
|
# Precision, Recall, F1 untuk POSITIF
|
|
prec_pos = tp / (tp + fp) if (tp + fp) > 0 else 0.0
|
|
rec_pos = tp / (tp + fn) if (tp + fn) > 0 else 0.0
|
|
f1_pos = (2 * prec_pos * rec_pos / (prec_pos + rec_pos)) if (prec_pos + rec_pos) > 0 else 0.0
|
|
|
|
# Precision, Recall, F1 untuk NEGATIF
|
|
prec_neg = tn / (tn + fn) if (tn + fn) > 0 else 0.0
|
|
rec_neg = tn / (tn + fp) if (tn + fp) > 0 else 0.0
|
|
f1_neg = (2 * prec_neg * rec_neg / (prec_neg + rec_neg)) if (prec_neg + rec_neg) > 0 else 0.0
|
|
|
|
# Macro Average
|
|
macro_prec = (prec_pos + prec_neg) / 2
|
|
macro_rec = (rec_pos + rec_neg) / 2
|
|
macro_f1 = (f1_pos + f1_neg) / 2
|
|
|
|
# Weighted Average
|
|
support_pos = tp + fn
|
|
support_neg = tn + fp
|
|
weighted_prec = ((prec_pos * support_pos) + (prec_neg * support_neg)) / total if total > 0 else 0.0
|
|
weighted_rec = ((rec_pos * support_pos) + (rec_neg * support_neg)) / total if total > 0 else 0.0
|
|
weighted_f1 = ((f1_pos * support_pos) + (f1_neg * support_neg)) / total if total > 0 else 0.0
|
|
|
|
# Detail Setiap Item Uji
|
|
details = []
|
|
correct_count = 0
|
|
error_count = 0
|
|
|
|
for i in range(total):
|
|
actual = y_true[i]
|
|
pred = str(y_pred[i])
|
|
is_correct = (actual == pred)
|
|
if is_correct:
|
|
correct_count += 1
|
|
else:
|
|
error_count += 1
|
|
|
|
# Kategori Matriks: TP, TN, FP, FN, atau GAGAL
|
|
if pred in ["GAGAL DIANALISIS", "FITUR TIDAK DITEMUKAN"]:
|
|
matrix_type = "GAGAL"
|
|
elif actual == "POSITIF" and pred == "POSITIF":
|
|
matrix_type = "TP"
|
|
elif actual == "POSITIF" and pred == "NEGATIF":
|
|
matrix_type = "FN"
|
|
elif actual == "NEGATIF" and pred == "POSITIF":
|
|
matrix_type = "FP"
|
|
else:
|
|
matrix_type = "TN"
|
|
|
|
proba = y_proba[i]
|
|
prob_neg = float(proba[0])
|
|
prob_pos = float(proba[1])
|
|
confidence = (prob_pos if pred == "POSITIF" else prob_neg) if pred not in ["GAGAL DIANALISIS", "FITUR TIDAK DITEMUKAN"] else 0.0
|
|
|
|
c_clean = clean_texts[i]
|
|
if not c_clean or not c_clean.strip():
|
|
has_feat = False
|
|
keterangan = "Hasil preprocessing kosong"
|
|
else:
|
|
has_feat = bool(tfidf.transform([c_clean]).nnz > 0)
|
|
keterangan = "Fitur berhasil diekstraksi" if has_feat else "Fitur tidak ditemukan"
|
|
|
|
details.append({
|
|
"index": i + 1,
|
|
"original_comment": comments[i],
|
|
"cleaned_comment": clean_texts[i],
|
|
"actual_label": actual,
|
|
"predicted_label": pred,
|
|
"is_correct": is_correct,
|
|
"matrix_type": matrix_type,
|
|
"confidence": round(confidence * 100, 1),
|
|
"prob_positif": round(prob_pos, 4),
|
|
"prob_negatif": round(prob_neg, 4),
|
|
"fitur_ditemukan": has_feat,
|
|
"keterangan": keterangan
|
|
})
|
|
|
|
elapsed_ms = round((time.time() - start_time) * 1000, 1)
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"elapsed_ms": elapsed_ms,
|
|
"summary": {
|
|
"total_samples": total,
|
|
"correct_count": correct_count,
|
|
"error_count": error_count,
|
|
"accuracy": round(acc * 100, 2),
|
|
"macro_precision": round(macro_prec * 100, 2),
|
|
"macro_recall": round(macro_rec * 100, 2),
|
|
"macro_f1": round(macro_f1 * 100, 2),
|
|
"weighted_precision": round(weighted_prec * 100, 2),
|
|
"weighted_recall": round(weighted_rec * 100, 2),
|
|
"weighted_f1": round(weighted_f1 * 100, 2),
|
|
},
|
|
"confusion_matrix": {
|
|
"labels": labels,
|
|
"tp": tp,
|
|
"fn": fn,
|
|
"fp": fp,
|
|
"tn": tn,
|
|
"matrix": [[tp, fn], [fp, tn]],
|
|
"percentages": {
|
|
"tp": round(tp / support_pos * 100, 1) if support_pos > 0 else 0.0,
|
|
"fn": round(fn / support_pos * 100, 1) if support_pos > 0 else 0.0,
|
|
"fp": round(fp / support_neg * 100, 1) if support_neg > 0 else 0.0,
|
|
"tn": round(tn / support_neg * 100, 1) if support_neg > 0 else 0.0
|
|
}
|
|
},
|
|
"classification_report": {
|
|
"positif": {
|
|
"precision": round(prec_pos * 100, 2),
|
|
"recall": round(rec_pos * 100, 2),
|
|
"f1_score": round(f1_pos * 100, 2),
|
|
"support": support_pos
|
|
},
|
|
"negatif": {
|
|
"precision": round(prec_neg * 100, 2),
|
|
"recall": round(rec_neg * 100, 2),
|
|
"f1_score": round(f1_neg * 100, 2),
|
|
"support": support_neg
|
|
},
|
|
"accuracy": round(acc * 100, 2),
|
|
"macro_avg": {
|
|
"precision": round(macro_prec * 100, 2),
|
|
"recall": round(macro_rec * 100, 2),
|
|
"f1_score": round(macro_f1 * 100, 2),
|
|
"support": total
|
|
},
|
|
"weighted_avg": {
|
|
"precision": round(weighted_prec * 100, 2),
|
|
"recall": round(weighted_rec * 100, 2),
|
|
"f1_score": round(weighted_f1 * 100, 2),
|
|
"support": total
|
|
}
|
|
},
|
|
"details": details
|
|
})
|
|
|
|
# ==========================================
|
|
# 9. ROUTING STATIK DAN FRONTEND
|
|
# ==========================================
|
|
|
|
@app.route("/")
|
|
@app.route("/index.html")
|
|
def root_index():
|
|
return send_from_directory(FRONTEND_DIR, "index.html")
|
|
|
|
@app.route("/static/<path:filename>")
|
|
def serve_static(filename):
|
|
return send_from_directory(FRONTEND_DIR, filename)
|
|
|
|
@app.route("/<path:filename>")
|
|
def serve_root_fallback(filename):
|
|
if os.path.exists(os.path.join(FRONTEND_DIR, filename)):
|
|
return send_from_directory(FRONTEND_DIR, filename)
|
|
return send_from_directory(FRONTEND_DIR, "index.html")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"[*] Menjalankan Server Analisis Sentimen TikTok...")
|
|
print(f"[*] Engine Database Aktif: {ACTIVE_DB_ENGINE.upper()} ({MYSQL_CONFIG['database'] if ACTIVE_DB_ENGINE == 'mysql' else 'SQLite'})")
|
|
app.run(
|
|
host="127.0.0.1",
|
|
port=5000,
|
|
debug=True
|
|
)
|