izu/OBE-Mapping
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
afb9e72ba2a101611ed5f3af592f7c69c1b896ae
OBE-Mapping/app/db.py
T
Power BI Dev afb9e72ba2 Tambah asesmen per komponen, batasi Dasbor/Rekap ke kaprodi, penugasan per kelas 
- Halaman nilai: tombol isi-cepat per komponen penilaian (bagian G RPS) yang
  menyalin nilai ke seluruh Sub-CPMK terkait; komponen berulang (Kuis, Tugas,
  Presentasi, Tes Lisan) mendukung banyak sesi dengan rata-rata otomatis,
  komponen sekali-jalan (UTS/UAS) mengedit sesi yang sama. Sub-CPMK yang
  diukur >1 komponen digulung rata tertimbang bobot (hitung.skor_dari_sesi).
- Hapus jalur unggah Excel & mode "ketik rata-rata kelas" (agregat) dari
  input nilai — proses diseragamkan: hanya grid manual + tombol komponen,
  keduanya bersumber dari daftar mahasiswa impor kaprodi.
- Dasbor Monitoring, Rekap CPL/CPMK, dan Riwayat kini khusus kaprodi (dosen
  tidak lagi melihat rollup lintas-MK yang bukan miliknya).
- Penugasan dosen ke MK bisa dibatasi per kelas (kolom kelas baru di tabel
  penugasan, migrasi otomatis & aman untuk DB yang sudah berjalan) — dosen
  MK dengan banyak kelas paralel hanya melihat & menilai kelasnya sendiri;
  hapus nilai MK penuh dikunci untuk dosen yang dibatasi kelas.
2026-07-21 12:44:57 +07:00

553 lines
20 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
"""Lapisan data: kurikulum (read-only, dari seed JSON) + isian RPS (SQLite)."""
import json
import os
import pathlib
import sqlite3
from datetime import datetime, timezone, timedelta
BASE = pathlib.Path(__file__).resolve().parent.parent
SEED = BASE / "seed" / "kurikulum.json"
DB_PATH = os.environ.get("DB_PATH", str(BASE / "data" / "app.db"))
WIB = timezone(timedelta(hours=7))
TAHUN_AJARAN_AKTIF = os.environ.get("TAHUN_AJARAN", "2026/2027 Ganjil")
def tahun_aktif() -> str:
"""Tahun ajaran aktif — dari pengaturan prodi, fallback env TAHUN_AJARAN."""
return ambil_pengaturan().get("tahun_aktif") or TAHUN_AJARAN_AKTIF
# ── kurikulum (dimuat sekali saat start) ─────────────────────────────────────
_kur = None
_sks = None
def kurikulum() -> dict:
global _kur
if _kur is None:
_kur = json.loads(SEED.read_text(encoding="utf-8"))
return _kur
def sks_mk() -> dict:
"""{kode: {'sks': x, 'semester': n}} dari seed/sks.json (boleh kosong)."""
global _sks
if _sks is None:
p = SEED.parent / "sks.json"
_sks = json.loads(p.read_text(encoding="utf-8")) if p.exists() else {}
return _sks
def daftar_mk() -> list[dict]:
"""Semua MK terurut nomor kurikulum, dengan ringkasan pemetaan."""
k = kurikulum()
out = []
for kode, mk in k["mk"].items():
sub_mk = k["sub"].get(kode, {"cpmks": {}})
n_sub = sum(len(c["subs"]) for c in sub_mk["cpmks"].values())
out.append(
{
"kode": kode,
"nama": mk["nama"],
"no": mk.get("no", 999),
"cpls": sorted(mk.get("cpls", [])),
"n_cpmk": len(sub_mk["cpmks"]),
"n_sub": n_sub,
}
)
return sorted(out, key=lambda m: m["no"])
def peta_mk(kode: str) -> dict | None:
"""Pemetaan lengkap satu MK: CPL, CPMK, dan Sub-CPMK yang dipikulnya."""
k = kurikulum()
mk = k["mk"].get(kode)
if not mk:
return None
sub_mk = k["sub"].get(kode, {"cpmks": {}})
cpmks = []
for ckey, cval in sub_mk["cpmks"].items():
info = k["cpmk"].get(ckey, {})
cpmks.append(
{
"kode": info.get("kode", ckey),
"key": ckey,
"desk": cval["desk"],
"cpl": info.get("cpl", []),
"subs": cval["subs"],
}
)
cpmks.sort(key=lambda c: c["kode"])
cpl_kode = sorted({cp for c in cpmks for cp in c["cpl"]} | set(mk.get("cpls", [])))
cpls = [{"kode": cp, "desk": k["cpl"][cp]["desk"]} for cp in cpl_kode if cp in k["cpl"]]
subs = [
{"kode": s["kode"], "uraian": s["uraian"], "cpmk": c["kode"]}
for c in cpmks
for s in c["subs"]
]
return {"kode": kode, "nama": mk["nama"], "cpls": cpls, "cpmks": cpmks, "subs": subs}
# ── SQLite: isian RPS ────────────────────────────────────────────────────────
def get_conn() -> sqlite3.Connection:
pathlib.Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
def _migrasi_role_pengguna(conn: sqlite3.Connection) -> None:
"""Skema lama hanya kenal role 'dosen'; pecah jadi dosen_mk/dosen_rps_mk.
Akun 'dosen' lama dipetakan ke 'dosen_rps_mk' agar hak akses (termasuk
edit RPS) yang sudah mereka miliki tidak berkurang oleh migrasi ini.
"""
row = conn.execute(
"SELECT sql FROM sqlite_master WHERE type='table' AND name='pengguna'"
).fetchone()
if not row or "dosen_rps_mk" in row["sql"]:
return
conn.execute("ALTER TABLE pengguna RENAME TO pengguna_lama")
conn.execute(
"""CREATE TABLE pengguna (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
nama TEXT NOT NULL,
role TEXT NOT NULL
CHECK (role IN ('kaprodi','dosen_mk','dosen_rps_mk')),
pw_hash TEXT NOT NULL,
dibuat TEXT NOT NULL
)"""
)
conn.execute(
"""INSERT INTO pengguna (id, username, nama, role, pw_hash, dibuat)
SELECT id, username, nama,
CASE WHEN role='dosen' THEN 'dosen_rps_mk' ELSE role END,
pw_hash, dibuat
FROM pengguna_lama"""
)
conn.execute("DROP TABLE pengguna_lama")
def _migrasi_kelas_penugasan(conn: sqlite3.Connection) -> None:
"""Skema lama penugasan hanya (user_id, mk_kode) — tambah kolom kelas
(kosong = seluruh kelas MK ini, dipakai dosen_rps_mk; terisi = dosen_mk
yang hanya mengampu kelas tertentu)."""
kolom = {r["name"] for r in conn.execute("PRAGMA table_info(penugasan)")}
if "kelas" not in kolom:
conn.execute("ALTER TABLE penugasan ADD COLUMN kelas TEXT NOT NULL DEFAULT ''")
def init_db() -> None:
with get_conn() as conn:
for tabel in ("rps", "portofolio", "nilai"):
conn.execute(
f"""CREATE TABLE IF NOT EXISTS {tabel} (
mk_kode TEXT NOT NULL,
tahun TEXT NOT NULL,
data TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'draft',
updated_at TEXT NOT NULL,
updated_by TEXT NOT NULL DEFAULT '',
PRIMARY KEY (mk_kode, tahun)
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS rps_arsip (
id INTEGER PRIMARY KEY AUTOINCREMENT,
mk_kode TEXT NOT NULL,
tahun TEXT NOT NULL,
data TEXT NOT NULL,
status TEXT NOT NULL,
updated_at TEXT NOT NULL,
updated_by TEXT NOT NULL
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS pengaturan (
kunci TEXT PRIMARY KEY,
nilai TEXT NOT NULL
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS pengguna (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
nama TEXT NOT NULL,
role TEXT NOT NULL
CHECK (role IN ('kaprodi','dosen_mk','dosen_rps_mk')),
pw_hash TEXT NOT NULL,
dibuat TEXT NOT NULL
)"""
)
_migrasi_role_pengguna(conn)
conn.execute(
"""CREATE TABLE IF NOT EXISTS sesi (
token TEXT PRIMARY KEY,
user_id INTEGER NOT NULL,
dibuat TEXT NOT NULL
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS penugasan (
user_id INTEGER NOT NULL,
mk_kode TEXT NOT NULL,
PRIMARY KEY (user_id, mk_kode)
)"""
)
_migrasi_kelas_penugasan(conn)
conn.execute(
"""CREATE TABLE IF NOT EXISTS mahasiswa (
npm TEXT NOT NULL,
nama TEXT NOT NULL,
mk_kode TEXT NOT NULL,
kelas TEXT NOT NULL DEFAULT '',
tahun TEXT NOT NULL,
PRIMARY KEY (npm, mk_kode, tahun)
)"""
)
# ── pengaturan prodi ─────────────────────────────────────────────────────────
PENGATURAN_DEFAULT = {
"tahun_aktif": TAHUN_AJARAN_AKTIF,
"ambang": "70",
"metode_antar_mk": "rata", # 'rata' (bagi rata) | 'sks' (tertimbang SKS)
# skala huruf mutu — dari contoh portofolio; format: HURUF|MIN|MAX|ANGKA
"skala_huruf": (
"A+|91|100|4.00\nA|86|90|3.75\nA-|81|85|3.50\nB+|76|80|3.25\n"
"B|71|75|3.00\nB-|66|70|2.75\nC+|61|65|2.50\nC|56|60|2.00\n"
"D|41|55|1.00\nE|0|40|0.00"
),
}
def ambil_pengaturan() -> dict:
with get_conn() as conn:
rows = conn.execute("SELECT kunci, nilai FROM pengaturan").fetchall()
out = dict(PENGATURAN_DEFAULT)
out.update({r["kunci"]: r["nilai"] for r in rows})
return out
def simpan_pengaturan(d: dict) -> None:
with get_conn() as conn:
for k, v in d.items():
if k in PENGATURAN_DEFAULT:
conn.execute(
"INSERT INTO pengaturan (kunci, nilai) VALUES (?,?) "
"ON CONFLICT(kunci) DO UPDATE SET nilai=excluded.nilai",
(k, str(v)),
)
_TABEL_DOK = ("rps", "portofolio", "nilai")
def simpan_dok(tabel: str, mk_kode: str, data: dict, status: str, updated_by: str,
tahun: str | None = None) -> None:
assert tabel in _TABEL_DOK
tahun = tahun or tahun_aktif()
now = datetime.now(WIB).strftime("%Y-%m-%d %H:%M")
payload = json.dumps(data, ensure_ascii=False)
with get_conn() as conn:
conn.execute(
f"""INSERT INTO {tabel} (mk_kode, tahun, data, status, updated_at, updated_by)
VALUES (?,?,?,?,?,?)
ON CONFLICT(mk_kode, tahun) DO UPDATE SET
data=excluded.data, status=excluded.status,
updated_at=excluded.updated_at, updated_by=excluded.updated_by""",
(mk_kode, tahun, payload, status, now, updated_by),
)
# snapshot arsip RPS setiap kali status ditetapkan 'selesai'
if tabel == "rps" and status == "selesai":
conn.execute(
"""INSERT INTO rps_arsip (mk_kode, tahun, data, status, updated_at, updated_by)
VALUES (?,?,?,?,?,?)""",
(mk_kode, tahun, payload, status, now, updated_by),
)
def hapus_dok(tabel: str, mk_kode: str, tahun: str | None = None) -> None:
assert tabel in _TABEL_DOK
tahun = tahun or tahun_aktif()
with get_conn() as conn:
conn.execute(f"DELETE FROM {tabel} WHERE mk_kode=? AND tahun=?",
(mk_kode, tahun))
def ambil_dok(tabel: str, mk_kode: str, tahun: str | None = None) -> dict | None:
assert tabel in _TABEL_DOK
tahun = tahun or tahun_aktif()
with get_conn() as conn:
row = conn.execute(
f"SELECT * FROM {tabel} WHERE mk_kode=? AND tahun=?", (mk_kode, tahun)
).fetchone()
if not row:
return None
return {
"data": json.loads(row["data"]),
"status": row["status"],
"updated_at": row["updated_at"],
"updated_by": row["updated_by"],
}
def status_semua(tabel: str = "rps", tahun: str | None = None) -> dict[str, dict]:
assert tabel in _TABEL_DOK
tahun = tahun or tahun_aktif()
with get_conn() as conn:
rows = conn.execute(
f"SELECT mk_kode, status, updated_at, updated_by FROM {tabel} WHERE tahun=?",
(tahun,),
).fetchall()
return {r["mk_kode"]: dict(r) for r in rows}
def semua_nilai(tahun: str | None = None) -> dict[str, dict]:
tahun = tahun or tahun_aktif()
with get_conn() as conn:
rows = conn.execute(
"SELECT mk_kode, data FROM nilai WHERE tahun=?", (tahun,)
).fetchall()
return {r["mk_kode"]: json.loads(r["data"]) for r in rows}
def daftar_tahun() -> list[str]:
"""Semua tahun ajaran yang pernah punya data (RPS/nilai/portofolio)."""
tahun = {tahun_aktif()}
with get_conn() as conn:
for tabel in _TABEL_DOK:
for r in conn.execute(f"SELECT DISTINCT tahun FROM {tabel}"):
tahun.add(r["tahun"])
return sorted(tahun)
def tahun_rps_lain(mk_kode: str) -> list[str]:
"""Tahun lain (selain aktif) yang punya RPS untuk MK ini — bahan duplikasi."""
aktif = tahun_aktif()
with get_conn() as conn:
rows = conn.execute(
"SELECT tahun FROM rps WHERE mk_kode=? AND tahun<>? ORDER BY tahun DESC",
(mk_kode, aktif),
).fetchall()
return [r["tahun"] for r in rows]
def dump_semua() -> dict:
"""Seluruh isi database sebagai dict (untuk cadangan JSON)."""
out = {}
with get_conn() as conn:
for tabel in (*_TABEL_DOK, "rps_arsip", "pengaturan",
"pengguna", "penugasan", "mahasiswa"):
rows = conn.execute(f"SELECT * FROM {tabel}").fetchall()
out[tabel] = [dict(r) for r in rows]
for tabel in _TABEL_DOK + ("rps_arsip",):
for r in out.get(tabel, []):
r["data"] = json.loads(r["data"])
return out
# ── pengguna, sesi, penugasan ────────────────────────────────────────────────
def ada_pengguna() -> bool:
with get_conn() as conn:
return conn.execute("SELECT 1 FROM pengguna LIMIT 1").fetchone() is not None
def buat_pengguna(username: str, nama: str, role: str, pw_hash: str) -> int | None:
now = datetime.now(WIB).strftime("%Y-%m-%d %H:%M")
try:
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO pengguna (username, nama, role, pw_hash, dibuat) "
"VALUES (?,?,?,?,?)",
(username, nama, role, pw_hash, now),
)
return cur.lastrowid
except sqlite3.IntegrityError:
return None
def pengguna_by_username(username: str) -> dict | None:
with get_conn() as conn:
row = conn.execute(
"SELECT * FROM pengguna WHERE username=?", (username,)
).fetchone()
return dict(row) if row else None
def pengguna_by_id(user_id: int) -> dict | None:
with get_conn() as conn:
row = conn.execute("SELECT * FROM pengguna WHERE id=?", (user_id,)).fetchone()
return dict(row) if row else None
def daftar_pengguna() -> list[dict]:
with get_conn() as conn:
rows = conn.execute(
"SELECT id, username, nama, role, dibuat FROM pengguna "
"ORDER BY role, nama"
).fetchall()
tugas = conn.execute(
"SELECT user_id, mk_kode, kelas FROM penugasan ORDER BY mk_kode"
).fetchall()
per_user_kode: dict[int, list[str]] = {}
per_user_kelas: dict[int, dict[str, str]] = {}
for t in tugas:
per_user_kode.setdefault(t["user_id"], []).append(t["mk_kode"])
per_user_kelas.setdefault(t["user_id"], {})[t["mk_kode"]] = t["kelas"] or ""
return [{**dict(r), "mk": per_user_kode.get(r["id"], []),
"kelas_map": per_user_kelas.get(r["id"], {})} for r in rows]
def hapus_pengguna(user_id: int) -> None:
with get_conn() as conn:
conn.execute("DELETE FROM pengguna WHERE id=?", (user_id,))
conn.execute("DELETE FROM sesi WHERE user_id=?", (user_id,))
conn.execute("DELETE FROM penugasan WHERE user_id=?", (user_id,))
def ubah_password(user_id: int, pw_hash: str) -> None:
with get_conn() as conn:
conn.execute("UPDATE pengguna SET pw_hash=? WHERE id=?", (pw_hash, user_id))
# sesi lama gugur agar pemegang password lama langsung keluar
conn.execute("DELETE FROM sesi WHERE user_id=?", (user_id,))
def set_penugasan(user_id: int, penugasan: list[tuple[str, str]]) -> None:
"""penugasan = [(mk_kode, kelas), ...] — kelas '' berarti seluruh kelas MK
ini; boleh CSV ('A,B') bila dosen mengampu beberapa kelas MK yang sama."""
with get_conn() as conn:
conn.execute("DELETE FROM penugasan WHERE user_id=?", (user_id,))
conn.executemany(
"INSERT OR IGNORE INTO penugasan (user_id, mk_kode, kelas) VALUES (?,?,?)",
[(user_id, k, kelas) for k, kelas in penugasan],
)
def mk_diampu(user_id: int) -> set[str]:
with get_conn() as conn:
rows = conn.execute(
"SELECT mk_kode FROM penugasan WHERE user_id=?", (user_id,)
).fetchall()
return {r["mk_kode"] for r in rows}
def kelas_diampu(user_id: int, mk_kode: str) -> set[str] | None:
"""Kelas yang diizinkan dosen ini untuk MK tsb. None = seluruh kelas
(tak dibatasi); set kosong = tidak ditugaskan sama sekali ke MK ini."""
with get_conn() as conn:
row = conn.execute(
"SELECT kelas FROM penugasan WHERE user_id=? AND mk_kode=?",
(user_id, mk_kode),
).fetchone()
if not row:
return set()
kelas = (row["kelas"] or "").strip()
if not kelas:
return None
return {k.strip() for k in kelas.split(",") if k.strip()}
def pengampu_per_mk() -> dict[str, list[str]]:
"""{mk_kode: [nama dosen, ...]} — untuk kolom pengampu di beranda."""
with get_conn() as conn:
rows = conn.execute(
"SELECT p.mk_kode, u.nama FROM penugasan p "
"JOIN pengguna u ON u.id = p.user_id ORDER BY u.nama"
).fetchall()
out: dict[str, list[str]] = {}
for r in rows:
out.setdefault(r["mk_kode"], []).append(r["nama"])
return out
def buat_sesi(user_id: int, token: str) -> None:
now = datetime.now(WIB).strftime("%Y-%m-%d %H:%M")
with get_conn() as conn:
conn.execute(
"INSERT INTO sesi (token, user_id, dibuat) VALUES (?,?,?)",
(token, user_id, now),
)
def user_dari_sesi(token: str) -> dict | None:
with get_conn() as conn:
row = conn.execute(
"SELECT u.* FROM sesi s JOIN pengguna u ON u.id = s.user_id "
"WHERE s.token=?",
(token,),
).fetchone()
return dict(row) if row else None
def hapus_sesi(token: str) -> None:
with get_conn() as conn:
conn.execute("DELETE FROM sesi WHERE token=?", (token,))
# ── mahasiswa (kontrak MK per kelas, diimpor kaprodi) ────────────────────────
def ganti_mahasiswa(baris: list[dict], tahun: str | None = None) -> None:
"""Ganti seluruh daftar mahasiswa tahun tsb dengan hasil impor terbaru."""
tahun = tahun or tahun_aktif()
with get_conn() as conn:
conn.execute("DELETE FROM mahasiswa WHERE tahun=?", (tahun,))
conn.executemany(
"INSERT OR REPLACE INTO mahasiswa (npm, nama, mk_kode, kelas, tahun) "
"VALUES (?,?,?,?,?)",
[(b["npm"], b["nama"], b["mk_kode"], b.get("kelas", ""), tahun)
for b in baris],
)
def mahasiswa_mk(mk_kode: str, tahun: str | None = None) -> list[dict]:
tahun = tahun or tahun_aktif()
with get_conn() as conn:
rows = conn.execute(
"SELECT npm, nama, kelas FROM mahasiswa WHERE mk_kode=? AND tahun=? "
"ORDER BY kelas, npm",
(mk_kode, tahun),
).fetchall()
return [dict(r) for r in rows]
def ringkasan_mahasiswa(tahun: str | None = None) -> dict[str, dict]:
"""{mk_kode: {'n': jumlah, 'kelas': [daftar kelas]}} tahun aktif."""
tahun = tahun or tahun_aktif()
with get_conn() as conn:
rows = conn.execute(
"SELECT mk_kode, kelas, COUNT(*) AS n FROM mahasiswa WHERE tahun=? "
"GROUP BY mk_kode, kelas",
(tahun,),
).fetchall()
out: dict[str, dict] = {}
for r in rows:
d = out.setdefault(r["mk_kode"], {"n": 0, "kelas": []})
d["n"] += r["n"]
if r["kelas"]:
d["kelas"].append(r["kelas"])
return out
# kompatibilitas rute lama
def simpan_rps(mk_kode, data, status, updated_by, tahun=None):
simpan_dok("rps", mk_kode, data, status, updated_by, tahun)
def ambil_rps(mk_kode, tahun=None):
return ambil_dok("rps", mk_kode, tahun)
Reference in New Issue View Git Blame Copy Permalink