"""PulsaPay — server HTTP + REST API (Python standard library). Jalankan: python server.py Buka: http://localhost:8090 """ import csv import datetime import http.cookies import http.server import io import json import mimetypes import os import re import socket import sqlite3 import sys import tempfile import traceback import urllib.parse import db BASE_DIR = os.path.dirname(os.path.abspath(__file__)) STATIC_DIR = os.path.join(BASE_DIR, "static") PORT = int(os.environ.get("PORT", "8090")) HOST = os.environ.get("HOST", "0.0.0.0") COOKIE_NAME = "pp_session" SESSION_DAYS = 30 MAX_BODY = 6 * 1024 * 1024 TYPE_LABEL = {"topup": "Top Up", "transfer": "Transfer", "tarik": "Tarik Tunai"} TYPE_PREFIX = {"topup": "TP", "transfer": "TR", "tarik": "TT"} SERVICE_LABEL = {"hp": "HP", "laptop": "Laptop/Komputer", "printer": "Printer"} SERVICE_STATUS = { "antri": "Antri", "proses": "Dikerjakan", "selesai": "Selesai", "diambil": "Diambil", "batal": "Batal", } SERVICE_DONE = ("selesai", "diambil") class ApiError(Exception): def __init__(self, message, status=400): super().__init__(message) self.message = message self.status = status # -------------------------------------------------------------------------- # Utilitas # -------------------------------------------------------------------------- def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def to_int(value, field, minimum=0, maximum=None, default=None): if value is None or value == "": if default is not None: return default raise ApiError("{} wajib diisi.".format(field)) try: number = int(str(value).replace(".", "").replace(",", "").strip()) except (TypeError, ValueError): raise ApiError("{} harus berupa angka.".format(field)) if minimum is not None and number < minimum: raise ApiError("{} minimal {}.".format(field, minimum)) if maximum is not None and number > maximum: raise ApiError("{} maksimal {}.".format(field, maximum)) return number def clean_text(value, max_len=300): if value is None: return "" text = str(value).strip() if len(text) > max_len: raise ApiError("Teks terlalu panjang (maksimal {} karakter).".format(max_len)) return text def parse_date(value, default=None): if not value: if default is not None: return default raise ApiError("Tanggal wajib diisi.") try: return datetime.date.fromisoformat(str(value)[:10]).isoformat() except ValueError: raise ApiError("Format tanggal tidak valid.") def row_to_dict(row): return {key: row[key] for key in row.keys()} if row is not None else None def tx_select(conn, where="", params=(), order="t.created_at DESC, t.id DESC", limit=None, offset=0): sql = ( "SELECT t.*, u.name AS user_name FROM transactions t " "LEFT JOIN users u ON u.id = t.created_by " + ("WHERE " + where + " " if where else "") + "ORDER BY " + order ) args = list(params) if limit is not None: sql += " LIMIT ? OFFSET ?" args.extend([limit, offset]) return [row_to_dict(row) for row in conn.execute(sql, args).fetchall()] def aggregate_transactions(rows): total = { "count": len(rows), "volume": 0, "pendapatan": 0, "modal": 0, "fee": 0, "topup_count": 0, "topup_margin": 0, "transfer_count": 0, "transfer_fee": 0, "tarik_count": 0, "tarik_fee": 0, } for row in rows: total["volume"] += row["nominal"] total["modal"] += row["cost"] total["fee"] += row["fee"] if row["type"] == "topup": total["topup_count"] += 1 total["topup_margin"] += row["profit"] elif row["type"] == "transfer": total["transfer_count"] += 1 total["transfer_fee"] += row["fee"] elif row["type"] == "tarik": total["tarik_count"] += 1 total["tarik_fee"] += row["fee"] total["pendapatan"] = total["topup_margin"] + total["fee"] return total def expenses_between(conn, start, end): rows = conn.execute( "SELECT e.*, u.name AS user_name FROM expenses e " "LEFT JOIN users u ON u.id = e.created_by " "WHERE e.date BETWEEN ? AND ? ORDER BY e.date DESC, e.id DESC", (start, end), ).fetchall() return [row_to_dict(row) for row in rows] def service_select(conn, where="", params=(), order="s.created_at DESC, s.id DESC", limit=None, offset=0): sql = ( "SELECT s.*, u.name AS user_name FROM services s " "LEFT JOIN users u ON u.id = s.created_by " + ("WHERE " + where + " " if where else "") + "ORDER BY " + order ) args = list(params) if limit is not None: sql += " LIMIT ? OFFSET ?" args.extend([limit, offset]) return [row_to_dict(row) for row in conn.execute(sql, args).fetchall()] def attach_service_parts(conn, services): if not services: return services ids = [service["id"] for service in services] placeholders = ",".join("?" for _ in ids) rows = conn.execute( "SELECT * FROM service_parts WHERE service_id IN ({}) ORDER BY id".format(placeholders), ids, ).fetchall() grouped = {} for row in rows: grouped.setdefault(row["service_id"], []).append(row_to_dict(row)) for service in services: service["parts"] = grouped.get(service["id"], []) service["device_label"] = SERVICE_LABEL.get(service["device_type"], service["device_type"]) service["status_label"] = SERVICE_STATUS.get(service["status"], service["status"]) return services def completed_services_between(conn, start, end): rows = conn.execute( "SELECT * FROM services WHERE status IN ('selesai', 'diambil') " "AND date(COALESCE(completed_at, created_at)) BETWEEN ? AND ?", (start, end), ).fetchall() return [row_to_dict(row) for row in rows] def aggregate_inprogress_services(conn, start=None, end=None): """Aggregate services that are not yet taken (antri, dikerjaka). Returns total biaya jasa and count for in-progress services.""" query = "SELECT * FROM services WHERE status IN ('antri', 'proses')" params = [] if start and end: query += " AND date(created_at) BETWEEN ? AND ?" params = [start, end] elif start: query += " AND created_at >= ?" params = [start] rows = conn.execute(query, params).fetchall() total = { "count": len(rows), "volume": 0, "service_fee": 0, "parts_cost": 0, "parts_price": 0, } for row in rows: total["volume"] += row["total"] total["service_fee"] += row["service_fee"] total["parts_cost"] += row["parts_cost"] total["parts_price"] += row["parts_price"] return total def aggregate_all_services_from_rows(rows): """Aggregate services from already-fetched rows. Returns both completed and in-progress data.""" total = { "count": len(rows), "volume": 0, "pendapatan": 0, "service_fee": 0, "parts_cost": 0, "parts_price": 0, } for row in rows: total["volume"] += row["total"] total["service_fee"] += row["service_fee"] total["parts_cost"] += row["parts_cost"] total["parts_price"] += row["parts_price"] if row["status"] in ("selesai", "diambil"): total["pendapatan"] += row["service_fee"] - row["parts_cost"] return total def parse_service_parts(conn, data): raw_parts = data.get("parts") or [] if not isinstance(raw_parts, list): raise ApiError("Daftar sparepart tidak valid.") parts = [] for raw in raw_parts: if not isinstance(raw, dict): continue try: sparepart_id = int(raw.get("sparepart_id") or 0) except (TypeError, ValueError): raise ApiError("Sparepart tidak valid.") if not sparepart_id: continue row = conn.execute("SELECT * FROM spareparts WHERE id = ?", (sparepart_id,)).fetchone() if not row: raise ApiError("Sparepart tidak ditemukan.", 404) if not row["active"]: raise ApiError("Sparepart {} sudah nonaktif.".format(row["name"])) qty = to_int(raw.get("qty"), "Jumlah sparepart", minimum=1) price_value = raw.get("price") if price_value in (None, ""): price = row["price"] else: price = to_int(price_value, "Harga jual sparepart", minimum=0) parts.append( { "sparepart_id": sparepart_id, "name": row["name"], "qty": qty, "cost": row["cost"], "price": price, "subtotal": qty * price, } ) return parts def validate_parts_stock(conn, parts): needed = {} for part in parts: needed[part["sparepart_id"]] = needed.get(part["sparepart_id"], 0) + part["qty"] for sparepart_id, qty in needed.items(): row = conn.execute( "SELECT name, stock, unit FROM spareparts WHERE id = ?", (sparepart_id,) ).fetchone() if not row: raise ApiError("Sparepart tidak ditemukan.", 404) if qty > row["stock"]: raise ApiError( "Stok {} tidak cukup. Tersedia {} {}.".format(row["name"], row["stock"], row["unit"]) ) def consume_parts(conn, service_id, code, parts, user_id): for part in parts: conn.execute( "UPDATE spareparts SET stock = stock - ? WHERE id = ?", (part["qty"], part["sparepart_id"]), ) conn.execute( "INSERT INTO sparepart_moves(sparepart_id, qty, type, cost, note, service_id, created_at, created_by) " "VALUES(?, ?, 'service', ?, ?, ?, ?, ?)", ( part["sparepart_id"], -part["qty"], part["cost"], "Servis " + code, service_id, now_str(), user_id, ), ) def restore_parts(conn, service_id, code, parts, user_id, note="Batal servis"): for part in parts: if not part.get("sparepart_id"): continue conn.execute( "UPDATE spareparts SET stock = stock + ? WHERE id = ?", (part["qty"], part["sparepart_id"]), ) conn.execute( "INSERT INTO sparepart_moves(sparepart_id, qty, type, cost, note, service_id, created_at, created_by) " "VALUES(?, ?, 'return', ?, ?, ?, ?, ?)", ( part["sparepart_id"], part["qty"], part["cost"], "{} {}".format(note, code), service_id, now_str(), user_id, ), ) # -------------------------------------------------------------------------- # Handler # -------------------------------------------------------------------------- class AppHandler(http.server.BaseHTTPRequestHandler): server_version = "PulsaPay/1.0" # ---------------------------------------------------------------- infra def log_message(self, fmt, *args): sys.stdout.write("[%s] %s\n" % (datetime.datetime.now().strftime("%H:%M:%S"), fmt % args)) sys.stdout.flush() def db(self): if not hasattr(self, "_conn"): self._conn = db.connect() return self._conn def close_db(self): if hasattr(self, "_conn"): try: self._conn.close() except Exception: pass del self._conn def read_body(self): try: length = int(self.headers.get("Content-Length") or 0) except ValueError: length = 0 if length > MAX_BODY: raise ApiError("Data yang dikirim terlalu besar.", 413) raw = self.rfile.read(length) if length else b"" if not raw: return {} try: data = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): raise ApiError("Format data tidak valid.") if not isinstance(data, dict): raise ApiError("Format data tidak valid.") return data def send_json(self, payload, status=200, extra_headers=None): body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") for key, value in (extra_headers or {}).items(): self.send_header(key, value) self.end_headers() if self.command != "HEAD": self.wfile.write(body) def send_bytes(self, body, content_type, filename=None, status=200): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) if filename: self.send_header( "Content-Disposition", 'attachment; filename="{}"'.format(filename) ) self.end_headers() if self.command != "HEAD": self.wfile.write(body) # --------------------------------------------------------------- auth def session_token(self): header = self.headers.get("Cookie") if not header: return None jar = http.cookies.SimpleCookie() try: jar.load(header) except http.cookies.CookieError: return None item = jar.get(COOKIE_NAME) return item.value if item else None def current_user(self): token = self.session_token() if not token: return None row = self.db().execute( "SELECT s.token, s.expires_at, u.id, u.username, u.name, u.role, u.active " "FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.token = ?", (token,), ).fetchone() if not row: return None if row["expires_at"] < now_str() or not row["active"]: self.db().execute("DELETE FROM sessions WHERE token = ?", (token,)) self.db().commit() return None return { "id": row["id"], "username": row["username"], "name": row["name"], "role": row["role"], } def require_user(self): user = self.current_user() if not user: raise ApiError("Sesi berakhir. Silakan masuk kembali.", 401) return user def require_admin(self): user = self.require_user() if user["role"] != "admin": raise ApiError("Hanya admin yang dapat melakukan tindakan ini.", 403) return user # -------------------------------------------------------------- router def do_GET(self): self.dispatch("GET") def do_POST(self): self.dispatch("POST") def do_PUT(self): self.dispatch("PUT") def do_DELETE(self): self.dispatch("DELETE") def do_HEAD(self): self.dispatch("GET") def dispatch(self, method): parsed = urllib.parse.urlparse(self.path) path = parsed.path query = {k: v[-1] for k, v in urllib.parse.parse_qs(parsed.query).items()} try: if path.startswith("/api"): if self.api(method, path, query) is False: return elif method in ("GET", "HEAD"): self.serve_static(path) return else: raise ApiError("Alamat tidak ditemukan.", 404) except ApiError as exc: self.send_json({"error": exc.message}, exc.status) except (BrokenPipeError, ConnectionResetError): pass except Exception: traceback.print_exc() self.send_json( {"error": "Terjadi kesalahan pada server. Coba lagi."}, 500 ) finally: self.close_db() # --------------------------------------------------------------- API def api(self, method, path, query): conn = self.db() # Publik --------------------------------------------------------- if path == "/api/health" and method == "GET": self.send_json({"ok": True, "time": now_str(), "app": "PulsaPay"}) return if path == "/api/public" and method == "GET": settings = db.get_settings(conn) self.send_json( { "business_name": settings["business_name"], "tagline": settings["tagline"], "logo": settings["logo"], "accent": settings["accent"], } ) return if path == "/api/session" and method == "GET": session_user = self.current_user() settings = db.get_settings(conn) if session_user: self.send_json({"user": session_user, "settings": settings}) else: self.send_json( { "user": None, "settings": { "business_name": settings["business_name"], "tagline": settings["tagline"], "logo": settings["logo"], "accent": settings["accent"], }, } ) return if path == "/api/login" and method == "POST": self.handle_login() return if path == "/api/logout" and method == "POST": self.handle_logout() return # Semua di bawah ini butuh login -------------------------------- user = self.require_user() if path == "/api/me" and method == "GET": self.send_json({"user": user, "settings": db.get_settings(conn)}) return if path == "/api/me/password" and method == "POST": data = self.read_body() self.change_password(user, data, own=True) return if path == "/api/settings" and method == "GET": self.send_json({"settings": db.get_settings(conn)}) return if path == "/api/settings" and method == "PUT": self.require_admin() data = self.read_body() self.update_settings(data) return if path == "/api/dashboard" and method == "GET": self.handle_dashboard() return # Produk --------------------------------------------------------- if path == "/api/products" and method == "GET": self.handle_products_list(query) return if path == "/api/products" and method == "POST": self.require_admin() self.handle_product_save(self.read_body(), None) return match = re.fullmatch(r"/api/products/(\d+)", path) if match: product_id = int(match.group(1)) if method == "PUT": self.require_admin() self.handle_product_save(self.read_body(), product_id) return if method == "DELETE": self.require_admin() self.handle_product_delete(product_id) return # Transaksi ------------------------------------------------------ if path == "/api/transactions" and method == "GET": self.handle_transactions_list(query) return if path == "/api/transactions" and method == "POST": self.handle_transaction_create(user) return match = re.fullmatch(r"/api/transactions/(\d+)", path) if match: tx_id = int(match.group(1)) if method == "GET": row = tx_select(conn, "t.id = ?", (tx_id,), limit=1) if not row: raise ApiError("Transaksi tidak ditemukan.", 404) self.send_json({"transaction": row[0]}) return if method == "DELETE": self.require_admin() self.handle_transaction_delete(tx_id) return # Biaya ---------------------------------------------------------- if path == "/api/expenses" and method == "GET": start = parse_date(query.get("from"), datetime.date.today().replace(day=1).isoformat()) end = parse_date(query.get("to"), datetime.date.today().isoformat()) rows = expenses_between(conn, start, end) self.send_json( { "expenses": rows, "total": sum(r["amount"] for r in rows), "from": start, "to": end, } ) return if path == "/api/expenses" and method == "POST": self.handle_expense_create(user) return match = re.fullmatch(r"/api/expenses/(\d+)", path) if match and method == "DELETE": self.require_admin() self.handle_expense_delete(int(match.group(1))) return # Saldo / mutasi ------------------------------------------------- if path == "/api/balance-moves" and method == "GET": rows = conn.execute( "SELECT b.*, u.name AS user_name FROM balance_moves b " "LEFT JOIN users u ON u.id = b.created_by ORDER BY b.id DESC LIMIT 100" ).fetchall() self.send_json( { "moves": [row_to_dict(r) for r in rows], "balances": db.balances(conn), } ) return if path == "/api/balance-moves" and method == "POST": self.require_admin() self.handle_balance_move(user) return # Sparepart ------------------------------------------------------ if path == "/api/spareparts" and method == "GET": self.handle_spareparts_list(query) return if path == "/api/spareparts" and method == "POST": self.require_admin() self.handle_sparepart_save(self.read_body(), None) return match = re.fullmatch(r"/api/spareparts/(\d+)/stock", path) if match and method == "POST": self.require_admin() self.handle_sparepart_stock(user, int(match.group(1))) return match = re.fullmatch(r"/api/spareparts/(\d+)", path) if match: sparepart_id = int(match.group(1)) if method == "PUT": self.require_admin() self.handle_sparepart_save(self.read_body(), sparepart_id) return if method == "DELETE": self.require_admin() self.handle_sparepart_delete(sparepart_id) return if path == "/api/sparepart-moves" and method == "GET": self.handle_sparepart_moves(query) return # Service -------------------------------------------------------- if path == "/api/services" and method == "GET": self.handle_services_list(query) return if path == "/api/services" and method == "POST": self.handle_service_create(user) return match = re.fullmatch(r"/api/services/(\d+)", path) if match: service_id = int(match.group(1)) if method == "GET": rows = service_select(conn, "s.id = ?", (service_id,), limit=1) if not rows: raise ApiError("Servis tidak ditemukan.", 404) attach_service_parts(conn, rows) self.send_json({"service": rows[0]}) return if method == "PUT": self.handle_service_update(user, service_id) return if method == "DELETE": self.require_admin() self.handle_service_delete(service_id) return # Laporan -------------------------------------------------------- if path == "/api/reports" and method == "GET": self.handle_report(query) return if path == "/api/reports/export.csv" and method == "GET": self.handle_report_csv(query) return # Pengguna ------------------------------------------------------- if path == "/api/users" and method == "GET": self.require_admin() rows = conn.execute( "SELECT id, username, name, role, active, created_at FROM users ORDER BY id" ).fetchall() self.send_json({"users": [row_to_dict(r) for r in rows]}) return if path == "/api/users" and method == "POST": self.require_admin() self.handle_user_create(self.read_body()) return match = re.fullmatch(r"/api/users/(\d+)", path) if match: target_id = int(match.group(1)) if method == "PUT": self.require_admin() self.handle_user_update(user, target_id, self.read_body()) return if method == "DELETE": self.require_admin() self.handle_user_delete(user, target_id) return match = re.fullmatch(r"/api/users/(\d+)/password", path) if match and method == "POST": target_id = int(match.group(1)) if user["role"] != "admin" and user["id"] != target_id: raise ApiError("Anda tidak dapat mengubah password pengguna lain.", 403) self.change_password( user, self.read_body(), target_id=target_id, own=(user["role"] != "admin" and target_id == user["id"]), ) return # Backup --------------------------------------------------------- if path == "/api/backup" and method == "GET": self.require_admin() self.handle_backup() return False raise ApiError("Alamat API tidak ditemukan.", 404) # ------------------------------------------------------------- login def handle_login(self): data = self.read_body() username = clean_text(data.get("username"), 60).lower() password = str(data.get("password") or "") if not username or not password: raise ApiError("Username dan password wajib diisi.") row = self.db().execute( "SELECT * FROM users WHERE lower(username) = ?", (username,) ).fetchone() if not row or not row["active"] or not db.verify_password(password, row["password_hash"]): raise ApiError("Username atau password salah.", 401) token = db.secrets.token_hex(32) created = now_str() expires = (datetime.datetime.now() + datetime.timedelta(days=SESSION_DAYS)).strftime( "%Y-%m-%d %H:%M:%S" ) self.db().execute("DELETE FROM sessions WHERE expires_at < ?", (created,)) self.db().execute( "INSERT INTO sessions(token, user_id, created_at, expires_at) VALUES(?, ?, ?, ?)", (token, row["id"], created, expires), ) self.db().commit() cookie = "{}={}; Path=/; HttpOnly; SameSite=Lax; Max-Age={}".format( COOKIE_NAME, token, SESSION_DAYS * 86400 ) self.send_json( { "user": { "id": row["id"], "username": row["username"], "name": row["name"], "role": row["role"], }, "settings": db.get_settings(self.db()), }, extra_headers={"Set-Cookie": cookie}, ) def handle_logout(self): token = self.session_token() if token: self.db().execute("DELETE FROM sessions WHERE token = ?", (token,)) self.db().commit() self.send_json( {"ok": True}, extra_headers={ "Set-Cookie": "{}=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0".format( COOKIE_NAME ) }, ) # ---------------------------------------------------------- settings def update_settings(self, data): allowed_text = { "business_name": 80, "tagline": 120, "address": 250, "phone": 60, "receipt_footer": 200, } allowed_money = { "saldo_prastika_awal", "saldo_seabank_awal", "kas_awal", "fee_transfer", "fee_tarik", } conn = self.db() for key, max_len in allowed_text.items(): if key in data: value = clean_text(data.get(key), max_len) if key == "business_name" and not value: raise ApiError("Nama usaha wajib diisi.") db.set_setting(conn, key, value) for key in allowed_money: if key in data: value = to_int(data.get(key), key.replace("_", " "), minimum=0) db.set_setting(conn, key, value) if "accent" in data: accent = clean_text(data.get("accent"), 20) if not re.fullmatch(r"#[0-9a-fA-F]{6}", accent): raise ApiError("Warna aksen tidak valid.") db.set_setting(conn, "accent", accent.lower()) if "logo" in data: logo = str(data.get("logo") or "") if logo and not logo.startswith("data:image/"): raise ApiError("Format logo tidak valid.") if len(logo) > 900000: raise ApiError("Ukuran logo terlalu besar. Gunakan gambar di bawah 500 KB.") db.set_setting(conn, "logo", logo) conn.commit() self.send_json({"settings": db.get_settings(conn)}) # ------------------------------------------------------------ produk def handle_products_list(self, query): conn = self.db() include_inactive = query.get("all") == "1" sql = "SELECT * FROM products" if not include_inactive: sql += " WHERE active = 1" sql += " ORDER BY category, price, name" rows = conn.execute(sql).fetchall() products = [] for row in rows: item = row_to_dict(row) item["margin"] = item["price"] - item["cost"] products.append(item) self.send_json({"products": products}) def handle_product_save(self, data, product_id): conn = self.db() name = clean_text(data.get("name"), 100) category = clean_text(data.get("category"), 50) provider = clean_text(data.get("provider"), 50) if not name: raise ApiError("Nama produk wajib diisi.") if not category: raise ApiError("Kategori wajib diisi.") nominal = to_int(data.get("nominal", 0), "Nominal", minimum=0) cost = to_int(data.get("cost", 0), "Harga modal", minimum=0) price = to_int(data.get("price", 0), "Harga jual", minimum=0) active = 1 if data.get("active", True) else 0 if price <= 0: raise ApiError("Harga jual harus lebih dari 0.") if product_id is None: cursor = conn.execute( "INSERT INTO products(name, category, provider, nominal, cost, price, active, created_at) " "VALUES(?, ?, ?, ?, ?, ?, ?, ?)", (name, category, provider, nominal, cost, price, active, now_str()), ) product_id = cursor.lastrowid else: if not conn.execute("SELECT 1 FROM products WHERE id = ?", (product_id,)).fetchone(): raise ApiError("Produk tidak ditemukan.", 404) conn.execute( "UPDATE products SET name = ?, category = ?, provider = ?, nominal = ?, " "cost = ?, price = ?, active = ? WHERE id = ?", (name, category, provider, nominal, cost, price, active, product_id), ) conn.commit() row = conn.execute("SELECT * FROM products WHERE id = ?", (product_id,)).fetchone() item = row_to_dict(row) item["margin"] = item["price"] - item["cost"] self.send_json({"product": item}) def handle_product_delete(self, product_id): conn = self.db() row = conn.execute("SELECT * FROM products WHERE id = ?", (product_id,)).fetchone() if not row: raise ApiError("Produk tidak ditemukan.", 404) conn.execute("DELETE FROM products WHERE id = ?", (product_id,)) conn.commit() self.send_json({"ok": True}) # --------------------------------------------------------- transaksi def handle_transactions_list(self, query): conn = self.db() clauses = [] params = [] tx_type = query.get("type") if tx_type in TYPE_LABEL: clauses.append("t.type = ?") params.append(tx_type) start = query.get("from") end = query.get("to") if start: clauses.append("date(t.created_at) >= ?") params.append(parse_date(start)) if end: clauses.append("date(t.created_at) <= ?") params.append(parse_date(end)) search = clean_text(query.get("q"), 80) if search: like = "%" + search + "%" clauses.append( "(t.code LIKE ? OR t.product_name LIKE ? OR t.customer_name LIKE ? " "OR t.destination LIKE ? OR t.destination_name LIKE ? OR t.note LIKE ?)" ) params.extend([like] * 6) where = " AND ".join(clauses) try: limit = min(max(int(query.get("limit", 100)), 1), 500) offset = max(int(query.get("offset", 0)), 0) except ValueError: limit, offset = 100, 0 rows = tx_select(conn, where, params, limit=limit, offset=offset) count = conn.execute( "SELECT COUNT(*) AS n FROM transactions t " + ("WHERE " + where if where else ""), params, ).fetchone()["n"] self.send_json( { "transactions": rows, "count": count, "limit": limit, "offset": offset, "summary": aggregate_transactions(rows), } ) def handle_transaction_create(self, user): data = self.read_body() conn = self.db() tx_type = str(data.get("type") or "") if tx_type not in TYPE_LABEL: raise ApiError("Jenis transaksi tidak valid.") product_id = None product_name = "" category = "" nominal = 0 cost = 0 price = 0 fee = 0 total = 0 profit = 0 destination = clean_text(data.get("destination"), 120) destination_name = clean_text(data.get("destination_name"), 120) customer_name = clean_text(data.get("customer_name"), 120) note = clean_text(data.get("note"), 300) if tx_type == "topup": try: product_id = int(data.get("product_id") or 0) except (TypeError, ValueError): product_id = 0 product = conn.execute( "SELECT * FROM products WHERE id = ? AND active = 1", (product_id,) ).fetchone() if not product: raise ApiError("Pilih produk top up terlebih dahulu.") if not destination: raise ApiError("Nomor / ID tujuan wajib diisi.") nominal = product["nominal"] or product["price"] cost = product["cost"] price = to_int( data.get("price") if data.get("price") not in (None, "") else product["price"], "Harga jual", minimum=1, ) total = price profit = price - cost product_name = product["name"] category = product["category"] else: nominal = to_int(data.get("nominal"), "Nominal", minimum=1) fee = to_int(data.get("fee", 0), "Biaya admin", minimum=0) total = nominal + fee profit = fee if not destination: if tx_type == "transfer": raise ApiError("Rekening / tujuan transfer wajib diisi.") if tx_type == "tarik" and not (customer_name or destination): raise ApiError("Nama pelanggan atau tujuan wajib diisi.") current = db.balances(conn) if tx_type == "topup" and cost > current["prastika"]: raise ApiError( "Saldo Prastika tidak cukup. Saldo saat ini {}.".format( format_rupiah(current["prastika"]) ) ) if tx_type == "transfer" and nominal > current["seabank"]: raise ApiError( "Saldo SeaBank tidak cukup. Saldo saat ini {}.".format( format_rupiah(current["seabank"]) ) ) if tx_type == "tarik" and nominal > current["kas"]: raise ApiError( "Kas tunai tidak cukup untuk tarik tunai. Kas saat ini {}.".format( format_rupiah(current["kas"]) ) ) code = db.make_code(conn, TYPE_PREFIX[tx_type]) cursor = conn.execute( "INSERT INTO transactions(code, type, product_id, product_name, category, " "customer_name, destination, destination_name, nominal, cost, price, fee, total, " "profit, note, created_at, created_by) " "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( code, tx_type, product_id, product_name, category, customer_name, destination, destination_name, nominal, cost, price, fee, total, profit, note, now_str(), user["id"], ), ) conn.commit() tx_id = cursor.lastrowid row = tx_select(conn, "t.id = ?", (tx_id,), limit=1)[0] self.send_json( {"transaction": row, "balances": db.balances(conn), "label": TYPE_LABEL[tx_type]}, status=201, ) def handle_transaction_delete(self, tx_id): conn = self.db() row = conn.execute("SELECT * FROM transactions WHERE id = ?", (tx_id,)).fetchone() if not row: raise ApiError("Transaksi tidak ditemukan.", 404) conn.execute("DELETE FROM transactions WHERE id = ?", (tx_id,)) conn.commit() self.send_json({"ok": True, "balances": db.balances(conn)}) # ------------------------------------------------------------- biaya def handle_expense_create(self, user): data = self.read_body() conn = self.db() date = parse_date(data.get("date"), datetime.date.today().isoformat()) category = clean_text(data.get("category"), 60) note = clean_text(data.get("note"), 300) amount = to_int(data.get("amount"), "Jumlah biaya", minimum=1) if not category: raise ApiError("Kategori biaya wajib diisi.") current = db.balances(conn) if amount > current["kas"]: raise ApiError( "Kas tidak cukup. Kas saat ini {}. Tambah saldo kas terlebih dahulu.".format( format_rupiah(current["kas"]) ) ) cursor = conn.execute( "INSERT INTO expenses(date, category, amount, note, created_at, created_by) " "VALUES(?, ?, ?, ?, ?, ?)", (date, category, amount, note, now_str(), user["id"]), ) conn.commit() row = conn.execute( "SELECT e.*, u.name AS user_name FROM expenses e LEFT JOIN users u ON u.id = e.created_by " "WHERE e.id = ?", (cursor.lastrowid,), ).fetchone() self.send_json( {"expense": row_to_dict(row), "balances": db.balances(conn)}, status=201 ) def handle_expense_delete(self, expense_id): conn = self.db() if not conn.execute("SELECT 1 FROM expenses WHERE id = ?", (expense_id,)).fetchone(): raise ApiError("Biaya tidak ditemukan.", 404) conn.execute("DELETE FROM expenses WHERE id = ?", (expense_id,)) conn.commit() self.send_json({"ok": True, "balances": db.balances(conn)}) # ------------------------------------------------------------- saldo def handle_balance_move(self, user): data = self.read_body() conn = self.db() account = str(data.get("account") or "") if account not in ("prastika", "seabank", "kas"): raise ApiError("Akun saldo tidak valid.") try: amount = int(str(data.get("amount")).replace(".", "").replace(",", "")) except (TypeError, ValueError): raise ApiError("Jumlah harus berupa angka.") if amount == 0: raise ApiError("Jumlah tidak boleh nol.") note = clean_text(data.get("note"), 200) current = db.balances(conn) new_value = current[account] + amount if new_value < 0: raise ApiError( "Saldo tidak cukup. Saldo {} saat ini {}.".format( account, format_rupiah(current[account]) ) ) conn.execute( "INSERT INTO balance_moves(account, amount, note, created_at, created_by) " "VALUES(?, ?, ?, ?, ?)", (account, amount, note or ("Tambah saldo" if amount > 0 else "Koreksi saldo"), now_str(), user["id"]), ) conn.commit() self.send_json({"ok": True, "balances": db.balances(conn)}) # --------------------------------------------------------- sparepart def handle_spareparts_list(self, query): conn = self.db() include_inactive = query.get("all") == "1" clauses = [] params = [] if not include_inactive: clauses.append("active = 1") search = clean_text(query.get("q"), 80) if search: like = "%" + search + "%" clauses.append("(name LIKE ? OR category LIKE ?)") params.extend([like, like]) where = " AND ".join(clauses) sql = "SELECT * FROM spareparts" + (" WHERE " + where if where else "") + " ORDER BY active DESC, name" rows = conn.execute(sql, params).fetchall() items = [] for row in rows: item = row_to_dict(row) item["margin"] = item["price"] - item["cost"] item["stock_value"] = item["stock"] * item["cost"] item["low"] = item["stock"] <= item["min_stock"] items.append(item) stats = conn.execute( "SELECT COUNT(*) AS total, " "COALESCE(SUM(stock * cost), 0) AS value, " "COALESCE(SUM(CASE WHEN active = 1 AND stock <= min_stock THEN 1 ELSE 0 END), 0) AS low " "FROM spareparts" ).fetchone() self.send_json( { "spareparts": items, "count": len(items), "stats": { "total": stats["total"], "value": stats["value"], "low": stats["low"], }, } ) def handle_sparepart_save(self, data, sparepart_id): conn = self.db() name = clean_text(data.get("name"), 120) if not name: raise ApiError("Nama sparepart wajib diisi.") category = clean_text(data.get("category"), 60) or "Umum" unit = clean_text(data.get("unit"), 20) or "pcs" cost = to_int(data.get("cost", 0), "Harga modal", minimum=0) price = to_int(data.get("price", 0), "Harga jual", minimum=0) stock = to_int(data.get("stock", 0), "Stok", minimum=0) min_stock = to_int(data.get("min_stock", 0), "Stok minimum", minimum=0) active = 1 if data.get("active", True) else 0 if price <= 0: raise ApiError("Harga jual sparepart harus lebih dari 0.") if sparepart_id is None: cursor = conn.execute( "INSERT INTO spareparts(name, category, unit, stock, min_stock, cost, price, active, created_at) " "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)", (name, category, unit, stock, min_stock, cost, price, active, now_str()), ) sparepart_id = cursor.lastrowid else: old = conn.execute("SELECT * FROM spareparts WHERE id = ?", (sparepart_id,)).fetchone() if not old: raise ApiError("Sparepart tidak ditemukan.", 404) conn.execute( "UPDATE spareparts SET name = ?, category = ?, unit = ?, stock = ?, min_stock = ?, " "cost = ?, price = ?, active = ? WHERE id = ?", (name, category, unit, stock, min_stock, cost, price, active, sparepart_id), ) diff = stock - old["stock"] if diff: conn.execute( "INSERT INTO sparepart_moves(sparepart_id, qty, type, cost, note, created_at, created_by) " "VALUES(?, ?, 'adjust', ?, ?, ?, ?)", (sparepart_id, diff, cost, "Koreksi stok dari edit barang", now_str(), None), ) conn.commit() row = conn.execute("SELECT * FROM spareparts WHERE id = ?", (sparepart_id,)).fetchone() item = row_to_dict(row) item["margin"] = item["price"] - item["cost"] item["stock_value"] = item["stock"] * item["cost"] item["low"] = item["stock"] <= item["min_stock"] self.send_json({"sparepart": item, "balances": db.balances(conn)}) def handle_sparepart_delete(self, sparepart_id): conn = self.db() if not conn.execute("SELECT 1 FROM spareparts WHERE id = ?", (sparepart_id,)).fetchone(): raise ApiError("Sparepart tidak ditemukan.", 404) used = conn.execute( "SELECT COUNT(*) AS n FROM service_parts WHERE sparepart_id = ?", (sparepart_id,) ).fetchone()["n"] moves = conn.execute( "SELECT COUNT(*) AS n FROM sparepart_moves WHERE sparepart_id = ?", (sparepart_id,) ).fetchone()["n"] if used or moves: raise ApiError( "Sparepart sudah punya riwayat pemakaian. Nonaktifkan saja agar riwayat tetap utuh." ) conn.execute("DELETE FROM spareparts WHERE id = ?", (sparepart_id,)) conn.commit() self.send_json({"ok": True}) def handle_sparepart_stock(self, user, sparepart_id): data = self.read_body() conn = self.db() row = conn.execute("SELECT * FROM spareparts WHERE id = ?", (sparepart_id,)).fetchone() if not row: raise ApiError("Sparepart tidak ditemukan.", 404) qty = to_int(data.get("qty"), "Jumlah stok masuk", minimum=1) cost_value = data.get("cost") if cost_value in (None, ""): cost = row["cost"] else: cost = to_int(cost_value, "Harga modal", minimum=0) if cost == 0: cost = row["cost"] note = clean_text(data.get("note"), 200) or "Tambah stok" pay_from_kas = bool(data.get("pay_from_kas")) total = qty * cost if pay_from_kas: current = db.balances(conn) if total > current["kas"]: raise ApiError( "Kas tidak cukup. Kas saat ini {}. Matikan potong kas atau tambah saldo kas.".format( format_rupiah(current["kas"]) ) ) conn.execute( "INSERT INTO balance_moves(account, amount, note, created_at, created_by) " "VALUES('kas', ?, ?, ?, ?)", (-total, "Beli stok {} x{}".format(row["name"], qty), now_str(), user["id"]), ) conn.execute( "UPDATE spareparts SET stock = stock + ?, cost = ? WHERE id = ?", (qty, cost, sparepart_id), ) conn.execute( "INSERT INTO sparepart_moves(sparepart_id, qty, type, cost, note, created_at, created_by) " "VALUES(?, ?, 'in', ?, ?, ?, ?)", (sparepart_id, qty, cost, note, now_str(), user["id"]), ) conn.commit() updated = conn.execute("SELECT * FROM spareparts WHERE id = ?", (sparepart_id,)).fetchone() item = row_to_dict(updated) item["margin"] = item["price"] - item["cost"] item["stock_value"] = item["stock"] * item["cost"] item["low"] = item["stock"] <= item["min_stock"] self.send_json({"ok": True, "sparepart": item, "balances": db.balances(conn)}) def handle_sparepart_moves(self, query): conn = self.db() try: limit = min(max(int(query.get("limit", 50)), 1), 200) except ValueError: limit = 50 rows = conn.execute( "SELECT m.*, s.name AS sparepart_name, s.unit, u.name AS user_name " "FROM sparepart_moves m LEFT JOIN spareparts s ON s.id = m.sparepart_id " "LEFT JOIN users u ON u.id = m.created_by ORDER BY m.id DESC LIMIT ?", (limit,), ).fetchall() self.send_json({"moves": [row_to_dict(row) for row in rows]}) # ------------------------------------------------------------ service def handle_services_list(self, query): conn = self.db() clauses = [] params = [] device_type = query.get("type") if device_type in SERVICE_LABEL: clauses.append("s.device_type = ?") params.append(device_type) status = query.get("status") if status in SERVICE_STATUS: clauses.append("s.status = ?") params.append(status) start = query.get("from") end = query.get("to") if start: clauses.append("date(s.created_at) >= ?") params.append(parse_date(start)) if end: clauses.append("date(s.created_at) <= ?") params.append(parse_date(end)) search = clean_text(query.get("q"), 80) if search: like = "%" + search + "%" clauses.append( "(s.code LIKE ? OR s.device_name LIKE ? OR s.customer_name LIKE ? " "OR s.customer_phone LIKE ? OR s.serial LIKE ? OR s.complaint LIKE ?)" ) params.extend([like] * 6) where = " AND ".join(clauses) try: limit = min(max(int(query.get("limit", 100)), 1), 500) offset = max(int(query.get("offset", 0)), 0) except ValueError: limit, offset = 100, 0 rows = service_select(conn, where, params, limit=limit, offset=offset) attach_service_parts(conn, rows) count = conn.execute( "SELECT COUNT(*) AS n FROM services s " + ("WHERE " + where if where else ""), params, ).fetchone()["n"] today = datetime.date.today().isoformat() month_start = datetime.date.today().replace(day=1).isoformat() today_count = conn.execute( "SELECT COUNT(*) AS n FROM services WHERE date(created_at) = ?", (today,) ).fetchone()["n"] open_count = conn.execute( "SELECT COUNT(*) AS n FROM services WHERE status IN ('antri', 'proses')" ).fetchone()["n"] month_services = completed_services_between(conn, month_start, today) month_summary = aggregate_all_services_from_rows(month_services) low_stock = conn.execute( "SELECT COUNT(*) AS n FROM spareparts WHERE active = 1 AND stock <= min_stock" ).fetchone()["n"] self.send_json( { "services": rows, "count": count, "limit": limit, "offset": offset, "stats": { "today": today_count, "open": open_count, "month_revenue": month_summary["pendapatan"], "month_count": month_summary["count"], "low_stock": low_stock, }, } ) def handle_service_create(self, user): data = self.read_body() conn = self.db() device_type = str(data.get("device_type") or "") if device_type not in SERVICE_LABEL: raise ApiError("Jenis perangkat tidak valid.") status = str(data.get("status") or "selesai") if status not in SERVICE_STATUS: raise ApiError("Status servis tidak valid.") device_name = clean_text(data.get("device_name"), 120) customer_name = clean_text(data.get("customer_name"), 120) if not device_name: raise ApiError("Merk / tipe perangkat wajib diisi.") if not customer_name: raise ApiError("Nama pelanggan wajib diisi.") service_fee = to_int(data.get("service_fee", 0), "Biaya jasa", minimum=0) parts = parse_service_parts(conn, data) if not parts and service_fee <= 0: raise ApiError("Isi biaya jasa atau tambahkan sparepart yang dipakai.") if status != "batal": validate_parts_stock(conn, parts) parts_cost = sum(part["cost"] * part["qty"] for part in parts) parts_price = sum(part["subtotal"] for part in parts) total = service_fee profit = service_fee - parts_cost code = db.make_service_code(conn) completed_at = now_str() if status in SERVICE_DONE else None cursor = conn.execute( "INSERT INTO services(code, device_type, device_name, serial, customer_name, customer_phone, " "complaint, diagnosis, technician, status, service_fee, parts_cost, parts_price, total, " "profit, note, created_at, completed_at, created_by) " "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( code, device_type, device_name, clean_text(data.get("serial"), 120), customer_name, clean_text(data.get("customer_phone"), 40), clean_text(data.get("complaint"), 500), clean_text(data.get("diagnosis"), 500), clean_text(data.get("technician"), 120), status, service_fee, parts_cost, parts_price, total, profit, clean_text(data.get("note"), 300), now_str(), completed_at, user["id"], ), ) service_id = cursor.lastrowid for part in parts: conn.execute( "INSERT INTO service_parts(service_id, sparepart_id, name, qty, cost, price, subtotal) " "VALUES(?, ?, ?, ?, ?, ?, ?)", ( service_id, part["sparepart_id"], part["name"], part["qty"], part["cost"], part["price"], part["subtotal"], ), ) if status != "batal": consume_parts(conn, service_id, code, parts, user["id"]) conn.commit() rows = service_select(conn, "s.id = ?", (service_id,), limit=1) attach_service_parts(conn, rows) self.send_json({"service": rows[0], "balances": db.balances(conn)}, status=201) def handle_service_update(self, user, service_id): data = self.read_body() conn = self.db() old_row = conn.execute("SELECT * FROM services WHERE id = ?", (service_id,)).fetchone() if not old_row: raise ApiError("Servis tidak ditemukan.", 404) old = row_to_dict(old_row) old_parts = [ row_to_dict(row) for row in conn.execute( "SELECT * FROM service_parts WHERE service_id = ? ORDER BY id", (service_id,) ).fetchall() ] code = old["code"] # Ubah status saja (dari halaman detail). if data and set(data.keys()) <= {"status"}: new_status = str(data.get("status") or "") if new_status not in SERVICE_STATUS: raise ApiError("Status servis tidak valid.") old_consumed = old["status"] != "batal" new_consumed = new_status != "batal" if old_consumed and not new_consumed: restore_parts(conn, service_id, code, old_parts, user["id"], note="Batal servis") elif not old_consumed and new_consumed: validate_parts_stock(conn, old_parts) consume_parts(conn, service_id, code, old_parts, user["id"]) completed_at = old["completed_at"] if new_status in SERVICE_DONE: completed_at = completed_at or now_str() else: completed_at = None conn.execute( "UPDATE services SET status = ?, completed_at = ? WHERE id = ?", (new_status, completed_at, service_id), ) conn.commit() rows = service_select(conn, "s.id = ?", (service_id,), limit=1) attach_service_parts(conn, rows) self.send_json({"service": rows[0], "balances": db.balances(conn)}) return device_type = str(data.get("device_type") or "") if device_type not in SERVICE_LABEL: raise ApiError("Jenis perangkat tidak valid.") status = str(data.get("status") or old["status"]) if status not in SERVICE_STATUS: raise ApiError("Status servis tidak valid.") device_name = clean_text(data.get("device_name"), 120) customer_name = clean_text(data.get("customer_name"), 120) if not device_name: raise ApiError("Merk / tipe perangkat wajib diisi.") if not customer_name: raise ApiError("Nama pelanggan wajib diisi.") service_fee = to_int(data.get("service_fee", 0), "Biaya jasa", minimum=0) parts = parse_service_parts(conn, data) if not parts and service_fee <= 0: raise ApiError("Isi biaya jasa atau tambahkan sparepart yang dipakai.") if old["status"] != "batal": restore_parts(conn, service_id, code, old_parts, user["id"], note="Edit servis") if status != "batal": validate_parts_stock(conn, parts) parts_cost = sum(part["cost"] * part["qty"] for part in parts) parts_price = sum(part["subtotal"] for part in parts) total = service_fee profit = service_fee - parts_cost if status in SERVICE_DONE: completed_at = old["completed_at"] if old["status"] in SERVICE_DONE and old["completed_at"] else now_str() else: completed_at = None conn.execute( "UPDATE services SET device_type = ?, device_name = ?, serial = ?, customer_name = ?, " "customer_phone = ?, complaint = ?, diagnosis = ?, technician = ?, status = ?, " "service_fee = ?, parts_cost = ?, parts_price = ?, total = ?, profit = ?, note = ?, " "completed_at = ? WHERE id = ?", ( device_type, device_name, clean_text(data.get("serial"), 120), customer_name, clean_text(data.get("customer_phone"), 40), clean_text(data.get("complaint"), 500), clean_text(data.get("diagnosis"), 500), clean_text(data.get("technician"), 120), status, service_fee, parts_cost, parts_price, total, profit, clean_text(data.get("note"), 300), completed_at, service_id, ), ) conn.execute("DELETE FROM service_parts WHERE service_id = ?", (service_id,)) for part in parts: conn.execute( "INSERT INTO service_parts(service_id, sparepart_id, name, qty, cost, price, subtotal) " "VALUES(?, ?, ?, ?, ?, ?, ?)", ( service_id, part["sparepart_id"], part["name"], part["qty"], part["cost"], part["price"], part["subtotal"], ), ) if status != "batal": consume_parts(conn, service_id, code, parts, user["id"]) conn.commit() rows = service_select(conn, "s.id = ?", (service_id,), limit=1) attach_service_parts(conn, rows) self.send_json({"service": rows[0], "balances": db.balances(conn)}) def handle_service_delete(self, service_id): conn = self.db() row = conn.execute("SELECT * FROM services WHERE id = ?", (service_id,)).fetchone() if not row: raise ApiError("Servis tidak ditemukan.", 404) parts = [ row_to_dict(item) for item in conn.execute( "SELECT * FROM service_parts WHERE service_id = ? ORDER BY id", (service_id,) ).fetchall() ] if row["status"] != "batal": restore_parts(conn, service_id, row["code"], parts, None, note="Hapus servis") conn.execute("DELETE FROM service_parts WHERE service_id = ?", (service_id,)) conn.execute("DELETE FROM services WHERE id = ?", (service_id,)) conn.commit() self.send_json({"ok": True, "balances": db.balances(conn)}) # --------------------------------------------------------- dashboard def handle_dashboard(self): conn = self.db() today = datetime.date.today() today_iso = today.isoformat() month_start = today.replace(day=1).isoformat() week_start = (today - datetime.timedelta(days=6)).isoformat() week_rows = tx_select( conn, "date(t.created_at) BETWEEN ? AND ?", (week_start, today_iso), order="t.created_at ASC, t.id ASC", ) today_rows = [r for r in week_rows if r["created_at"][:10] == today_iso] month_rows = tx_select( conn, "date(t.created_at) BETWEEN ? AND ?", (month_start, today_iso), order="t.created_at ASC, t.id ASC", ) week_services = completed_services_between(conn, week_start, today_iso) today_services = [ r for r in week_services if (r["completed_at"] or r["created_at"])[:10] == today_iso ] month_services = completed_services_between(conn, month_start, today_iso) inprogress_week = aggregate_inprogress_services(conn, week_start, today_iso) inprogress_month = aggregate_inprogress_services(conn, month_start, today_iso) expenses_today = expenses_between(conn, today_iso, today_iso) expenses_month = expenses_between(conn, month_start, today_iso) today_summary = aggregate_transactions(today_rows) month_summary = aggregate_transactions(month_rows) today_service = aggregate_all_services_from_rows(today_services) month_service = aggregate_all_services_from_rows(month_services) for summary, service in ((today_summary, today_service), (month_summary, month_service)): summary["service_count"] = service["count"] summary["service_pendapatan"] = service["pendapatan"] summary["service_volume"] = service["volume"] summary["inprogress_count"] = service["inprogress_count"] if "inprogress_count" in service else (inprogress_week["count"] if summary == today_summary else inprogress_month["count"]) summary["service_fee"] = service["service_fee"] summary["inprogress_service_fee"] = inprogress_week["service_fee"] if summary == today_summary else inprogress_month["service_fee"] summary["inprogress_parts_cost"] = inprogress_week["parts_cost"] if summary == today_summary else inprogress_month["parts_cost"] summary["pendapatan"] += service["pendapatan"] summary["volume"] += service["volume"] summary["count"] += service["count"] today_summary["biaya"] = sum(e["amount"] for e in expenses_today) today_summary["laba"] = today_summary["pendapatan"] - today_summary["biaya"] month_summary["biaya"] = sum(e["amount"] for e in expenses_month) month_summary["laba"] = month_summary["pendapatan"] - month_summary["biaya"] series = [] day_names = ["Sen", "Sel", "Rab", "Kam", "Jum", "Sab", "Min"] for offset in range(6, -1, -1): day = today - datetime.timedelta(days=offset) iso = day.isoformat() rows = [r for r in week_rows if r["created_at"][:10] == iso] summary = aggregate_transactions(rows) service_rows = [ r for r in week_services if (r["completed_at"] or r["created_at"])[:10] == iso ] service = aggregate_all_services_from_rows(service_rows) expense_total = conn.execute( "SELECT COALESCE(SUM(amount), 0) AS v FROM expenses WHERE date = ?", (iso,) ).fetchone()["v"] revenue = summary["pendapatan"] + service["pendapatan"] series.append( { "date": iso, "label": day_names[day.weekday()], "pendapatan": revenue, "laba": revenue - expense_total, "count": summary["count"] + service["count"], "volume": summary["volume"] + service["volume"], } ) recent = tx_select(conn, limit=8) recent_services = service_select(conn, limit=5) attach_service_parts(conn, recent_services) settings = db.get_settings(conn) needs_setup = ( int(settings.get("saldo_prastika_awal") or 0) == 0 and int(settings.get("saldo_seabank_awal") or 0) == 0 and int(settings.get("kas_awal") or 0) == 0 and conn.execute("SELECT COUNT(*) AS n FROM transactions").fetchone()["n"] == 0 and conn.execute("SELECT COUNT(*) AS n FROM services").fetchone()["n"] == 0 and conn.execute("SELECT COUNT(*) AS n FROM balance_moves").fetchone()["n"] == 0 ) self.send_json( { "today": today_summary, "month": month_summary, "series": series, "recent": recent, "recent_services": recent_services, "expenses_today": expenses_today, "balances": db.balances(conn), "needs_setup": needs_setup, "server_time": now_str(), } ) # ------------------------------------------------------------ laporan def handle_report(self, query): conn = self.db() today = datetime.date.today() start = parse_date(query.get("from"), today.replace(day=1).isoformat()) end = parse_date(query.get("to"), today.isoformat()) if start > end: start, end = end, start rows = tx_select( conn, "date(t.created_at) BETWEEN ? AND ?", (start, end), order="t.created_at ASC, t.id ASC", ) service_rows = completed_services_between(conn, start, end) inprogress_summary = aggregate_inprogress_services(conn, start, end) expenses = expenses_between(conn, start, end) summary = aggregate_transactions(rows) service_summary = aggregate_all_services_from_rows(service_rows) summary["service_count"] = service_summary["count"] summary["service_pendapatan"] = service_summary["pendapatan"] summary["service_volume"] = service_summary["volume"] summary["service_fee"] = service_summary["service_fee"] summary["parts_cost"] = service_summary["parts_cost"] summary["parts_price"] = service_summary["parts_price"] summary["inprogress_count"] = inprogress_summary["count"] summary["inprogress_service_fee"] = inprogress_summary["service_fee"] summary["inprogress_parts_cost"] = inprogress_summary["parts_cost"] summary["count"] += service_summary["count"] summary["volume"] += service_summary["volume"] summary["pendapatan"] += service_summary["pendapatan"] summary["biaya"] = sum(e["amount"] for e in expenses) summary["laba"] = summary["pendapatan"] - summary["biaya"] summary["from"] = start summary["to"] = end by_day_map = {} for row in rows: key = row["created_at"][:10] item = by_day_map.setdefault( key, {"date": key, "count": 0, "volume": 0, "pendapatan": 0, "biaya": 0, "laba": 0} ) item["count"] += 1 item["volume"] += row["nominal"] item["pendapatan"] += row["profit"] if row["type"] == "topup" else row["fee"] for service in service_rows: key = (service["completed_at"] or service["created_at"])[:10] item = by_day_map.setdefault( key, {"date": key, "count": 0, "volume": 0, "pendapatan": 0, "biaya": 0, "laba": 0} ) item["count"] += 1 item["volume"] += service["total"] item["pendapatan"] += service["service_fee"] - service["parts_cost"] for expense in expenses: item = by_day_map.setdefault( expense["date"], {"date": expense["date"], "count": 0, "volume": 0, "pendapatan": 0, "biaya": 0, "laba": 0}, ) item["biaya"] += expense["amount"] by_day = [] for key in sorted(by_day_map.keys(), reverse=True): item = by_day_map[key] item["laba"] = item["pendapatan"] - item["biaya"] by_day.append(item) by_type = [] for tx_type in ("topup", "transfer", "tarik"): subset = [r for r in rows if r["type"] == tx_type] agg = aggregate_transactions(subset) by_type.append( { "type": tx_type, "label": TYPE_LABEL[tx_type], "count": agg["count"], "volume": agg["volume"], "pendapatan": agg["pendapatan"], } ) by_type.append( { "type": "service", "label": "Service", "count": service_summary["count"], "volume": service_summary["volume"], "pendapatan": service_summary["pendapatan"], } ) product_map = {} for row in rows: if row["type"] != "topup": continue item = product_map.setdefault( row["product_name"], {"name": row["product_name"], "category": row["category"], "count": 0, "volume": 0, "margin": 0}, ) item["count"] += 1 item["volume"] += row["nominal"] item["margin"] += row["profit"] top_products = sorted(product_map.values(), key=lambda x: x["count"], reverse=True)[:10] device_map = {} for service in service_rows: item = device_map.setdefault( service["device_type"], { "type": service["device_type"], "label": SERVICE_LABEL.get(service["device_type"], service["device_type"]), "count": 0, "volume": 0, "pendapatan": 0, }, ) item["count"] += 1 item["volume"] += service["total"] item["pendapatan"] += service["service_fee"] - service["parts_cost"] service_by_device = [ device_map[key] for key in ("hp", "laptop", "printer") if key in device_map ] top_spareparts = [ { "name": row["name"], "qty": row["qty"], "cost": row["cost"], } for row in conn.execute( "SELECT sp.name, SUM(sp.qty) AS qty, SUM(sp.cost * sp.qty) AS cost " "FROM service_parts sp JOIN services s ON s.id = sp.service_id " "WHERE s.status IN ('selesai', 'diambil') " "AND date(COALESCE(s.completed_at, s.created_at)) BETWEEN ? AND ? " "GROUP BY sp.name ORDER BY qty DESC LIMIT 10", (start, end), ).fetchall() ] self.send_json( { "summary": summary, "by_day": by_day, "by_type": by_type, "top_products": top_products, "service_by_device": service_by_device, "top_spareparts": top_spareparts, "expenses": expenses, "transactions": rows, } ) def handle_report_csv(self, query): conn = self.db() today = datetime.date.today() start = parse_date(query.get("from"), today.replace(day=1).isoformat()) end = parse_date(query.get("to"), today.isoformat()) if start > end: start, end = end, start rows = tx_select( conn, "date(t.created_at) BETWEEN ? AND ?", (start, end), order="t.created_at ASC, t.id ASC", ) output = io.StringIO() writer = csv.writer(output) writer.writerow( [ "Tanggal", "Kode", "Jenis", "Produk", "Kategori", "Pelanggan", "Tujuan", "Nama Tujuan", "Nominal", "Modal", "Harga Jual", "Fee", "Total", "Laba", "Catatan", "Kasir", ] ) for row in rows: writer.writerow( [ row["created_at"], row["code"], TYPE_LABEL.get(row["type"], row["type"]), row["product_name"], row["category"], row["customer_name"], row["destination"], row["destination_name"], row["nominal"], row["cost"], row["price"], row["fee"], row["total"], row["profit"], row["note"], row["user_name"], ] ) service_rows = service_select( conn, "s.status IN ('selesai', 'diambil') " "AND date(COALESCE(s.completed_at, s.created_at)) BETWEEN ? AND ?", (start, end), order="COALESCE(s.completed_at, s.created_at) ASC, s.id ASC", ) attach_service_parts(conn, service_rows) for service in service_rows: parts_note = ", ".join( "{} x{}".format(part["name"], part["qty"]) for part in service["parts"] ) note_parts = [] if parts_note: note_parts.append("Sparepart: " + parts_note) if service["complaint"]: note_parts.append("Keluhan: " + service["complaint"]) if service["note"]: note_parts.append(service["note"]) writer.writerow( [ service["completed_at"] or service["created_at"], service["code"], "Service", service["device_name"], SERVICE_LABEL.get(service["device_type"], service["device_type"]), service["customer_name"], service["serial"], service["technician"], service["service_fee"], service["parts_cost"], service["service_fee"], 0, service["service_fee"], service["service_fee"] - service["parts_cost"], " | ".join(note_parts), service["user_name"], ] ) data = "\ufeff" + output.getvalue() filename = "laporan-pulsapay-{}-{}.csv".format(start, end) self.send_bytes( data.encode("utf-8"), "text/csv; charset=utf-8", filename ) return False # ----------------------------------------------------------- pengguna def handle_user_create(self, data): conn = self.db() username = clean_text(data.get("username"), 40).lower() name = clean_text(data.get("name"), 80) password = str(data.get("password") or "") role = str(data.get("role") or "kasir") if not re.fullmatch(r"[a-z0-9._-]{3,30}", username): raise ApiError("Username 3-30 karakter, hanya huruf, angka, titik, garis bawah, atau strip.") if not name: raise ApiError("Nama pengguna wajib diisi.") if len(password) < 6: raise ApiError("Password minimal 6 karakter.") if role not in ("admin", "kasir"): raise ApiError("Peran pengguna tidak valid.") if conn.execute("SELECT 1 FROM users WHERE lower(username) = ?", (username,)).fetchone(): raise ApiError("Username sudah dipakai.") cursor = conn.execute( "INSERT INTO users(username, name, password_hash, role, active, created_at) " "VALUES(?, ?, ?, ?, 1, ?)", (username, name, db.hash_password(password), role, now_str()), ) conn.commit() row = conn.execute( "SELECT id, username, name, role, active, created_at FROM users WHERE id = ?", (cursor.lastrowid,), ).fetchone() self.send_json({"user": row_to_dict(row)}, status=201) def handle_user_update(self, current_user, target_id, data): conn = self.db() row = conn.execute("SELECT * FROM users WHERE id = ?", (target_id,)).fetchone() if not row: raise ApiError("Pengguna tidak ditemukan.", 404) name = clean_text(data.get("name", row["name"]), 80) role = str(data.get("role") or row["role"]) active = 1 if data.get("active", bool(row["active"])) else 0 if not name: raise ApiError("Nama pengguna wajib diisi.") if role not in ("admin", "kasir"): raise ApiError("Peran pengguna tidak valid.") if target_id == current_user["id"] and (role != "admin" or not active): raise ApiError("Anda tidak dapat menonaktifkan atau menurunkan peran akun sendiri.") if row["role"] == "admin" and (role != "admin" or not active): admin_count = conn.execute( "SELECT COUNT(*) AS n FROM users WHERE role = 'admin' AND active = 1" ).fetchone()["n"] if admin_count <= 1: raise ApiError("Minimal harus ada satu admin aktif.") conn.execute( "UPDATE users SET name = ?, role = ?, active = ? WHERE id = ?", (name, role, active, target_id), ) if not active: conn.execute("DELETE FROM sessions WHERE user_id = ?", (target_id,)) conn.commit() updated = conn.execute( "SELECT id, username, name, role, active, created_at FROM users WHERE id = ?", (target_id,), ).fetchone() self.send_json({"user": row_to_dict(updated)}) def handle_user_delete(self, current_user, target_id): if target_id == current_user["id"]: raise ApiError("Anda tidak dapat menghapus akun sendiri.") conn = self.db() row = conn.execute("SELECT * FROM users WHERE id = ?", (target_id,)).fetchone() if not row: raise ApiError("Pengguna tidak ditemukan.", 404) if row["role"] == "admin" and row["active"]: admin_count = conn.execute( "SELECT COUNT(*) AS n FROM users WHERE role = 'admin' AND active = 1" ).fetchone()["n"] if admin_count <= 1: raise ApiError("Minimal harus ada satu admin aktif.") conn.execute("DELETE FROM sessions WHERE user_id = ?", (target_id,)) conn.execute("DELETE FROM users WHERE id = ?", (target_id,)) conn.commit() self.send_json({"ok": True}) def change_password(self, current_user, data, target_id=None, own=False): conn = self.db() if own: target_id = current_user["id"] row = conn.execute("SELECT * FROM users WHERE id = ?", (target_id,)).fetchone() if not row: raise ApiError("Pengguna tidak ditemukan.", 404) old_password = str(data.get("old_password") or "") new_password = str(data.get("new_password") or data.get("password") or "") if own: if not db.verify_password(old_password, row["password_hash"]): raise ApiError("Password lama salah.") if len(new_password) < 6: raise ApiError("Password baru minimal 6 karakter.") conn.execute( "UPDATE users SET password_hash = ? WHERE id = ?", (db.hash_password(new_password), target_id), ) if target_id != current_user["id"]: conn.execute("DELETE FROM sessions WHERE user_id = ?", (target_id,)) conn.commit() self.send_json({"ok": True}) # ------------------------------------------------------------- backup def handle_backup(self): conn = self.db() fd, temp_path = tempfile.mkstemp(suffix=".db") os.close(fd) try: dest = sqlite3.connect(temp_path) conn.backup(dest) dest.close() with open(temp_path, "rb") as handle: data = handle.read() filename = "backup-pulsapay-{}.db".format(datetime.date.today().isoformat()) self.send_bytes(data, "application/octet-stream", filename) finally: try: os.remove(temp_path) except OSError: pass return False # ------------------------------------------------------------- static def serve_static(self, path): if path in ("", "/"): path = "/index.html" relative = path.lstrip("/") candidate = os.path.realpath(os.path.join(STATIC_DIR, relative)) static_root = os.path.realpath(STATIC_DIR) + os.sep if not (candidate + os.sep).startswith(static_root) and candidate != os.path.realpath(STATIC_DIR): self.send_error(403) return if not os.path.isfile(candidate): self.send_error(404) return mime, _ = mimetypes.guess_type(candidate) if relative.endswith(".webmanifest"): mime = "application/manifest+json" if not mime: mime = "application/octet-stream" if mime.startswith("text/") or mime in ( "application/javascript", "application/json", "image/svg+xml" ): mime += "; charset=utf-8" with open(candidate, "rb") as handle: body = handle.read() self.send_response(200) self.send_header("Content-Type", mime) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-cache") self.end_headers() if self.command != "HEAD": self.wfile.write(body) def format_rupiah(value): return "Rp " + format(int(value or 0), ",d").replace(",", ".") def local_ip(): try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.connect(("8.8.8.8", 80)) address = sock.getsockname()[0] sock.close() return address except OSError: return "127.0.0.1" def main(): db.init_db() try: server = http.server.ThreadingHTTPServer((HOST, PORT), AppHandler) except OSError as exc: print("Gagal menjalankan server:", exc) print("Port {} mungkin sedang dipakai. Coba: set PORT=8091".format(PORT)) return print("=" * 58) print(" PulsaPay — Pembukuan Top Up & SeaBank") print("=" * 58) print(" Lokal : http://localhost:{}".format(PORT)) if HOST == "0.0.0.0": print(" Jaringan: http://{}:{}".format(local_ip(), PORT)) print(" Login : admin / admin123") print(" Hentikan: Ctrl+C") print("=" * 58) try: server.serve_forever() except KeyboardInterrupt: print("\nServer dihentikan.") finally: server.server_close() if __name__ == "__main__": main()