630 lines
20 KiB
Python
630 lines
20 KiB
Python
from csv import DictReader
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
from statistics import median
|
|
|
|
from flask import Flask, jsonify, request
|
|
|
|
try:
|
|
import pymysql
|
|
except ImportError:
|
|
pymysql = None
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parents[1]
|
|
LARAVEL_ENV_PATH = BASE_DIR / ".env"
|
|
|
|
|
|
def dataset_candidates():
|
|
env_path = os.getenv("CAR_DATASET_PATH")
|
|
paths = []
|
|
|
|
if env_path:
|
|
paths.append(Path(env_path))
|
|
|
|
paths.extend(
|
|
[
|
|
BASE_DIR / "storage" / "app" / "dataset_final.csv",
|
|
BASE_DIR / "database" / "dataset_final.csv",
|
|
BASE_DIR / "database" / "dataset_final.xls",
|
|
BASE_DIR / "storage" / "app" / "dataset_mobil.csv",
|
|
]
|
|
)
|
|
|
|
return paths
|
|
|
|
app = Flask(__name__)
|
|
|
|
|
|
def parse_env_file(path):
|
|
values = {}
|
|
|
|
if not path.exists():
|
|
return values
|
|
|
|
with path.open("r", encoding="utf-8") as env_file:
|
|
for line in env_file:
|
|
line = line.strip()
|
|
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
|
|
key, value = line.split("=", 1)
|
|
values[key.strip()] = value.strip().strip('"').strip("'")
|
|
|
|
return values
|
|
|
|
|
|
def env_value(key, default=None):
|
|
return os.getenv(key) or parse_env_file(LARAVEL_ENV_PATH).get(key, default)
|
|
|
|
|
|
def database_config():
|
|
return {
|
|
"host": env_value("DB_HOST", "127.0.0.1"),
|
|
"port": to_int(env_value("DB_PORT", 3306), default=3306),
|
|
"user": env_value("DB_USERNAME", "root"),
|
|
"password": env_value("DB_PASSWORD", ""),
|
|
"database": env_value("DB_DATABASE", ""),
|
|
"charset": "utf8mb4",
|
|
"cursorclass": pymysql.cursors.DictCursor if pymysql else None,
|
|
}
|
|
|
|
|
|
def to_int(value, default=0):
|
|
try:
|
|
return int(float(value))
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def normalize_text(value):
|
|
return str(value or "").strip().lower()
|
|
|
|
|
|
def as_list(value):
|
|
if value in (None, ""):
|
|
return []
|
|
|
|
if isinstance(value, (list, tuple, set)):
|
|
return [normalize_text(item) for item in value if normalize_text(item)]
|
|
|
|
return [normalize_text(value)]
|
|
|
|
|
|
def clamp(value, minimum=0.0, maximum=1.0):
|
|
return max(minimum, min(maximum, value))
|
|
|
|
|
|
def closeness_score(value, target, tolerance):
|
|
if target in (None, ""):
|
|
return 0.5
|
|
|
|
target = to_int(target, default=None)
|
|
if target in (None, 0):
|
|
return 0.5
|
|
|
|
return clamp(1 - abs(value - target) / max(tolerance, 1))
|
|
|
|
|
|
def parse_request_options(payload):
|
|
preferences = payload.get("preferences") or payload.get("preferensi") or {}
|
|
filters = payload.get("filters") or payload.get("filter") or {}
|
|
|
|
# Backward compatibility with the first simple API.
|
|
if payload.get("cluster") is not None and preferences.get("cluster") is None:
|
|
preferences["cluster"] = payload.get("cluster")
|
|
|
|
return preferences, filters
|
|
|
|
|
|
def category_row_to_car(row, index):
|
|
merk = str(row.get("merk") or "").strip()
|
|
model = str(row.get("model") or "").strip()
|
|
name = str(row.get("name") or "").strip()
|
|
composed_name = f"{merk} {model}".strip() or name
|
|
|
|
return {
|
|
"id": row.get("id") or index,
|
|
"merk": merk,
|
|
"model": model,
|
|
"nama": composed_name,
|
|
"kategori": str(row.get("kategori") or "").strip(),
|
|
"tahun": to_int(row.get("year")),
|
|
"mesin_cc": to_int(row.get("mesin_cc")),
|
|
"transmisi": str(row.get("transmisi") or "").strip(),
|
|
"bahan_bakar": str(row.get("bahan_bakar") or "").strip(),
|
|
"harga": to_int(row.get("price")),
|
|
"cluster": None,
|
|
"source": "database",
|
|
}
|
|
|
|
|
|
def load_cars_from_database():
|
|
if pymysql is None:
|
|
raise RuntimeError("PyMySQL belum terpasang. Jalankan pip install -r flask_api/requirements.txt")
|
|
|
|
config = database_config()
|
|
|
|
if not config["database"]:
|
|
raise RuntimeError("DB_DATABASE belum diatur.")
|
|
|
|
connection = pymysql.connect(**config)
|
|
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, name, merk, model, kategori, price, year, mesin_cc, transmisi, bahan_bakar
|
|
FROM categories
|
|
WHERE status = 'available'
|
|
AND price > 0
|
|
AND year IS NOT NULL
|
|
AND mesin_cc IS NOT NULL
|
|
ORDER BY created_at DESC, id DESC
|
|
"""
|
|
)
|
|
rows = cursor.fetchall()
|
|
finally:
|
|
connection.close()
|
|
|
|
cars = []
|
|
for index, row in enumerate(rows, start=1):
|
|
car = category_row_to_car(row, index)
|
|
if car["harga"] > 0:
|
|
cars.append(car)
|
|
|
|
return cars
|
|
|
|
|
|
def load_cars_from_dataset():
|
|
dataset_path = get_dataset_path()
|
|
|
|
if not dataset_path:
|
|
searched_paths = [str(path) for path in dataset_candidates()]
|
|
raise FileNotFoundError(f"Dataset tidak ditemukan. Lokasi yang dicek: {searched_paths}")
|
|
|
|
cars = []
|
|
with dataset_path.open("r", encoding="utf-8-sig", newline="") as dataset:
|
|
for index, row in enumerate(DictReader(dataset), start=1):
|
|
price = to_int(row.get("Harga"))
|
|
if price <= 0:
|
|
continue
|
|
|
|
cars.append(
|
|
{
|
|
"id": index,
|
|
"merk": row.get("merk", "").strip(),
|
|
"model": row.get("model", "").strip(),
|
|
"nama": f"{row.get('merk', '').strip()} {row.get('model', '').strip()}".strip(),
|
|
"kategori": row.get("kategori", "").strip(),
|
|
"tahun": to_int(row.get("tahun")),
|
|
"mesin_cc": to_int(row.get("mesin_cc")),
|
|
"transmisi": row.get("transmisi", "").strip(),
|
|
"bahan_bakar": row.get("bahan_bakar", "").strip(),
|
|
"harga": price,
|
|
"cluster": to_int(row.get("cluster"), default=None),
|
|
"source": "dataset",
|
|
}
|
|
)
|
|
|
|
return cars
|
|
|
|
|
|
def load_cars():
|
|
try:
|
|
cars = load_cars_from_database()
|
|
if cars:
|
|
return cars
|
|
except Exception:
|
|
pass
|
|
|
|
return load_cars_from_dataset()
|
|
|
|
|
|
def get_dataset_path():
|
|
for path in dataset_candidates():
|
|
if path.exists() and path.is_file():
|
|
return path
|
|
|
|
return None
|
|
|
|
|
|
def matches_filters(car, filters):
|
|
allowed_categories = as_list(filters.get("kategori") or filters.get("category"))
|
|
allowed_transmissions = as_list(filters.get("transmisi") or filters.get("transmission"))
|
|
allowed_fuels = as_list(filters.get("bahan_bakar") or filters.get("fuel"))
|
|
allowed_brands = as_list(filters.get("merk") or filters.get("brand"))
|
|
allowed_clusters = [to_int(cluster, default=None) for cluster in as_list(filters.get("cluster"))]
|
|
allowed_clusters = [cluster for cluster in allowed_clusters if cluster is not None]
|
|
|
|
if allowed_categories and normalize_text(car["kategori"]) not in allowed_categories:
|
|
return False
|
|
if allowed_transmissions and normalize_text(car["transmisi"]) not in allowed_transmissions:
|
|
return False
|
|
if allowed_fuels and normalize_text(car["bahan_bakar"]) not in allowed_fuels:
|
|
return False
|
|
if allowed_brands and normalize_text(car["merk"]) not in allowed_brands:
|
|
return False
|
|
if allowed_clusters and car["cluster"] not in allowed_clusters:
|
|
return False
|
|
|
|
min_year = to_int(filters.get("tahun_min") or filters.get("min_tahun"), default=None)
|
|
max_year = to_int(filters.get("tahun_max") or filters.get("max_tahun"), default=None)
|
|
min_engine = to_int(filters.get("mesin_min") or filters.get("min_mesin_cc"), default=None)
|
|
max_engine = to_int(filters.get("mesin_max") or filters.get("max_mesin_cc"), default=None)
|
|
|
|
if min_year is not None and car["tahun"] < min_year:
|
|
return False
|
|
if max_year is not None and car["tahun"] > max_year:
|
|
return False
|
|
if min_engine is not None and car["mesin_cc"] < min_engine:
|
|
return False
|
|
if max_engine is not None and car["mesin_cc"] > max_engine:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def preference_match_score(car, preferences):
|
|
scores = []
|
|
reasons = []
|
|
|
|
preference_fields = [
|
|
("merk", "merk"),
|
|
("kategori", "kategori"),
|
|
("transmisi", "transmisi"),
|
|
("bahan_bakar", "bahan bakar"),
|
|
]
|
|
|
|
for field, label in preference_fields:
|
|
desired_values = as_list(preferences.get(field))
|
|
if not desired_values:
|
|
continue
|
|
|
|
if normalize_text(car[field]) in desired_values:
|
|
scores.append(1)
|
|
reasons.append(f"Cocok dengan preferensi {label}: {car[field]}")
|
|
else:
|
|
scores.append(0)
|
|
|
|
preferred_cluster = preferences.get("cluster")
|
|
if preferred_cluster is not None:
|
|
preferred_cluster = to_int(preferred_cluster, default=None)
|
|
scores.append(1 if car["cluster"] == preferred_cluster else 0)
|
|
if car["cluster"] == preferred_cluster:
|
|
reasons.append(f"Sesuai cluster preferensi: {car['cluster']}")
|
|
|
|
return (sum(scores) / len(scores) if scores else 0.5), reasons
|
|
|
|
|
|
def score_car(car, budget, preferences, stats, strategy, weights):
|
|
budget_ratio = car["harga"] / max(budget, 1)
|
|
budget_gap = budget - car["harga"]
|
|
|
|
if strategy == "hemat":
|
|
target_ratio = 0.80
|
|
elif strategy == "premium":
|
|
target_ratio = 0.97
|
|
elif strategy in ["harga_pas", "harga pas", "spesifik", "specific"]:
|
|
target_ratio = 1.00
|
|
else:
|
|
target_ratio = 0.90
|
|
|
|
budget_score = clamp(1 - abs(budget_ratio - target_ratio) / 0.35)
|
|
year_score = clamp((car["tahun"] - stats["min_year"]) / max(stats["max_year"] - stats["min_year"], 1))
|
|
engine_score = closeness_score(
|
|
car["mesin_cc"],
|
|
preferences.get("mesin_cc") or preferences.get("target_mesin_cc"),
|
|
tolerance=700,
|
|
)
|
|
preference_score, preference_reasons = preference_match_score(car, preferences)
|
|
|
|
preferred_cluster = preferences.get("cluster")
|
|
cluster_score = 0.5
|
|
if preferred_cluster is not None:
|
|
cluster_score = 1 if car["cluster"] == to_int(preferred_cluster, default=None) else 0
|
|
elif stats["median_cluster"] is not None:
|
|
cluster_score = 1 / (1 + abs(car["cluster"] - stats["median_cluster"]))
|
|
|
|
weighted_score = (
|
|
budget_score * weights["budget"]
|
|
+ preference_score * weights["preference"]
|
|
+ year_score * weights["year"]
|
|
+ engine_score * weights["engine"]
|
|
+ cluster_score * weights["cluster"]
|
|
)
|
|
|
|
reasons = []
|
|
if budget_gap >= 0:
|
|
reasons.append(f"Masih di bawah budget dengan sisa Rp{budget_gap:,.0f}")
|
|
else:
|
|
reasons.append(f"Melebihi budget Rp{abs(budget_gap):,.0f}")
|
|
|
|
if year_score >= 0.75:
|
|
reasons.append(f"Tahun relatif muda: {car['tahun']}")
|
|
if engine_score >= 0.8 and (preferences.get("mesin_cc") or preferences.get("target_mesin_cc")):
|
|
reasons.append(f"Kapasitas mesin dekat preferensi: {car['mesin_cc']} cc")
|
|
|
|
reasons.extend(preference_reasons)
|
|
|
|
return {
|
|
"total": round(weighted_score * 100, 2),
|
|
"detail": {
|
|
"budget": round(budget_score * 100, 2),
|
|
"preferensi": round(preference_score * 100, 2),
|
|
"tahun": round(year_score * 100, 2),
|
|
"mesin": round(engine_score * 100, 2),
|
|
"cluster": round(cluster_score * 100, 2),
|
|
},
|
|
"alasan": reasons[:5],
|
|
}
|
|
|
|
|
|
def default_weights(payload_weights):
|
|
weights = {
|
|
"budget": 0.35,
|
|
"preference": 0.30,
|
|
"year": 0.20,
|
|
"engine": 0.10,
|
|
"cluster": 0.05,
|
|
}
|
|
|
|
for key, value in (payload_weights or {}).items():
|
|
if key in weights:
|
|
weights[key] = max(float(value), 0)
|
|
|
|
total = sum(weights.values()) or 1
|
|
return {key: value / total for key, value in weights.items()}
|
|
|
|
|
|
def recommendation_weights(payload_weights, strategy):
|
|
if payload_weights:
|
|
return default_weights(payload_weights)
|
|
|
|
if normalize_text(strategy) in ["tahun_muda", "tahun muda", "kondisi_baru"]:
|
|
return {
|
|
"budget": 0.30,
|
|
"preference": 0.10,
|
|
"year": 0.40,
|
|
"engine": 0.20,
|
|
"cluster": 0.00,
|
|
}
|
|
|
|
return default_weights(None)
|
|
|
|
|
|
def dataset_stats(cars):
|
|
years = [car["tahun"] for car in cars if car["tahun"] > 0]
|
|
clusters = [car["cluster"] for car in cars if car["cluster"] is not None]
|
|
|
|
return {
|
|
"min_year": min(years) if years else 0,
|
|
"max_year": max(years) if years else 1,
|
|
"median_cluster": median(clusters) if clusters else None,
|
|
}
|
|
|
|
|
|
def scaled_value(value, minimum, maximum):
|
|
if maximum == minimum:
|
|
return 0.0
|
|
|
|
return (value - minimum) / (maximum - minimum)
|
|
|
|
|
|
def distance(point, centroid):
|
|
return math.sqrt(sum((point[index] - centroid[index]) ** 2 for index in range(len(point))))
|
|
|
|
|
|
def car_features(car):
|
|
return [car["harga"], car["tahun"], car["mesin_cc"]]
|
|
|
|
|
|
def build_kmeans_model(cars, cluster_count=6, iterations=30):
|
|
usable_cars = [car for car in cars if car["harga"] > 0 and car["tahun"] > 0 and car["mesin_cc"] > 0]
|
|
|
|
if not usable_cars:
|
|
return None
|
|
|
|
feature_rows = [car_features(car) for car in usable_cars]
|
|
columns = list(zip(*feature_rows))
|
|
ranges = [(min(column), max(column)) for column in columns]
|
|
vectors = [
|
|
[scaled_value(value, ranges[index][0], ranges[index][1]) for index, value in enumerate(row)]
|
|
for row in feature_rows
|
|
]
|
|
|
|
cluster_count = max(1, min(cluster_count, len(vectors)))
|
|
sorted_indexes = sorted(range(len(usable_cars)), key=lambda index: usable_cars[index]["harga"])
|
|
centroids = []
|
|
|
|
for cluster_index in range(cluster_count):
|
|
sample_position = round(cluster_index * (len(sorted_indexes) - 1) / max(cluster_count - 1, 1))
|
|
centroids.append(vectors[sorted_indexes[sample_position]][:])
|
|
|
|
labels = [0 for _ in vectors]
|
|
|
|
for _ in range(iterations):
|
|
changed = False
|
|
|
|
for index, vector in enumerate(vectors):
|
|
nearest_cluster = min(range(cluster_count), key=lambda cluster: distance(vector, centroids[cluster]))
|
|
if labels[index] != nearest_cluster:
|
|
changed = True
|
|
labels[index] = nearest_cluster
|
|
|
|
new_centroids = []
|
|
for cluster in range(cluster_count):
|
|
members = [vectors[index] for index, label in enumerate(labels) if label == cluster]
|
|
if not members:
|
|
new_centroids.append(centroids[cluster])
|
|
continue
|
|
|
|
new_centroids.append(
|
|
[sum(member[column] for member in members) / len(members) for column in range(len(vectors[0]))]
|
|
)
|
|
|
|
centroids = new_centroids
|
|
|
|
if not changed:
|
|
break
|
|
|
|
car_labels = {id(car): labels[index] for index, car in enumerate(usable_cars)}
|
|
|
|
return {
|
|
"cars": usable_cars,
|
|
"labels": car_labels,
|
|
"centroids": centroids,
|
|
"ranges": ranges,
|
|
"cluster_count": cluster_count,
|
|
}
|
|
|
|
|
|
def predict_budget_cluster(model, budget):
|
|
model_cars = model["cars"]
|
|
target_year = median([car["tahun"] for car in model_cars if car["harga"] <= budget] or [car["tahun"] for car in model_cars])
|
|
target_engine = median([car["mesin_cc"] for car in model_cars if car["harga"] <= budget] or [car["mesin_cc"] for car in model_cars])
|
|
target_features = [budget, target_year, target_engine]
|
|
target_vector = [
|
|
scaled_value(value, model["ranges"][index][0], model["ranges"][index][1])
|
|
for index, value in enumerate(target_features)
|
|
]
|
|
|
|
return min(
|
|
range(model["cluster_count"]),
|
|
key=lambda cluster: distance(target_vector, model["centroids"][cluster]),
|
|
)
|
|
|
|
|
|
def apply_kmeans_budget_filter(cars, budget, filters):
|
|
model = build_kmeans_model(cars, cluster_count=to_int(filters.get("jumlah_cluster"), default=4))
|
|
|
|
if not model:
|
|
return cars, None
|
|
|
|
target_cluster = predict_budget_cluster(model, budget)
|
|
same_cluster = [
|
|
car for car in cars
|
|
if model["labels"].get(id(car)) == target_cluster and car["harga"] <= budget
|
|
]
|
|
|
|
exact_matches = [car for car in same_cluster if car["harga"] == budget]
|
|
selected = exact_matches or same_cluster or [car for car in cars if car["harga"] <= budget]
|
|
|
|
return selected, {
|
|
"algorithm": "k_means",
|
|
"features": ["harga", "tahun", "mesin_cc"],
|
|
"cluster_count": model["cluster_count"],
|
|
"target_cluster": target_cluster,
|
|
"exact_budget_match": bool(exact_matches),
|
|
}
|
|
|
|
|
|
def recommend_by_budget(
|
|
budget,
|
|
limit=10,
|
|
preferences=None,
|
|
filters=None,
|
|
strategy="seimbang",
|
|
include_above_budget=False,
|
|
max_over_budget_percent=0,
|
|
weights=None,
|
|
):
|
|
cars = load_cars()
|
|
preferences = preferences or {}
|
|
filters = filters or {}
|
|
normalized_strategy = normalize_text(strategy)
|
|
weights = recommendation_weights(weights, normalized_strategy)
|
|
max_price = budget * (1 + max(max_over_budget_percent, 0) / 100) if include_above_budget else budget
|
|
candidates = [car for car in cars if car["harga"] <= max_price and matches_filters(car, filters)]
|
|
candidates, model_info = apply_kmeans_budget_filter(candidates, budget, filters)
|
|
|
|
if normalized_strategy in ["harga_pas", "harga pas", "spesifik", "specific"]:
|
|
tolerance_percent = to_int(filters.get("toleransi_harga_percent"), default=10)
|
|
minimum_price = budget * (1 - max(tolerance_percent, 0) / 100)
|
|
tight_candidates = [car for car in candidates if car["harga"] >= minimum_price]
|
|
candidates = tight_candidates
|
|
|
|
stats = dataset_stats(cars)
|
|
|
|
recommendations = []
|
|
for car in candidates:
|
|
scoring = score_car(car, budget, preferences, stats, normalized_strategy, weights)
|
|
recommendations.append(
|
|
{
|
|
**car,
|
|
"model_cluster": model_info["target_cluster"] if model_info else None,
|
|
"selisih_budget": budget - car["harga"],
|
|
"score": scoring["total"],
|
|
"score_detail": scoring["detail"],
|
|
"alasan": scoring["alasan"],
|
|
}
|
|
)
|
|
|
|
recommendations = sorted(
|
|
recommendations,
|
|
key=lambda car: (car["score"], car["tahun"], car["harga"]),
|
|
reverse=True,
|
|
)[:limit]
|
|
|
|
return recommendations, model_info
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
dataset_path = get_dataset_path()
|
|
cars = load_cars()
|
|
data_source = cars[0].get("source", "unknown") if cars else "empty"
|
|
|
|
return jsonify(
|
|
{
|
|
"status": "ok",
|
|
"source": data_source,
|
|
"dataset": str(dataset_path) if dataset_path else None,
|
|
"database": database_config()["database"],
|
|
"total_data": len(cars),
|
|
}
|
|
)
|
|
|
|
|
|
@app.post("/recommend")
|
|
def recommend():
|
|
payload = request.get_json(silent=True) or {}
|
|
budget = to_int(payload.get("budget"))
|
|
limit = to_int(payload.get("limit"), default=10)
|
|
preferences, filters = parse_request_options(payload)
|
|
|
|
if budget <= 0:
|
|
return jsonify({"message": "Budget harus lebih dari 0."}), 422
|
|
|
|
if limit <= 0:
|
|
limit = 10
|
|
|
|
recommendations = recommend_by_budget(
|
|
budget=budget,
|
|
limit=min(limit, 50),
|
|
preferences=preferences,
|
|
filters=filters,
|
|
strategy=payload.get("strategy") or payload.get("strategi") or "seimbang",
|
|
include_above_budget=bool(payload.get("include_above_budget", False)),
|
|
max_over_budget_percent=to_int(payload.get("max_over_budget_percent"), default=0),
|
|
weights=payload.get("weights") or payload.get("bobot"),
|
|
)
|
|
recommendations, model_info = recommendations
|
|
|
|
return jsonify(
|
|
{
|
|
"budget": budget,
|
|
"total_rekomendasi": len(recommendations),
|
|
"model": model_info,
|
|
"preferences": preferences,
|
|
"filters": filters,
|
|
"recommendations": recommendations,
|
|
}
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=5000, debug=False, use_reloader=False)
|