Power BI Dev
af369883f3
Dokumen RTM resmi sesuai template baku prodi (FORM RPS & RTM UNTAN), diisi
di halaman /rps/{kode}/rtm sendiri setelah RPS ditandai selesai — blok Tugas
bisa ditambah/dihapus dinamis, dan bisa diunduh sebagai .docx. Murni
dokumentasi, tak menyimpan atau memengaruhi nilai/skor mahasiswa sama sekali.
561 lines
20 KiB
Python
561 lines
20 KiB
Python
# -*- 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_nim_mahasiswa(conn: sqlite3.Connection) -> None:
|
|
"""Kolom lama bernama 'npm' — istilah baku adalah NIM, bukan NPM."""
|
|
kolom = {r["name"] for r in conn.execute("PRAGMA table_info(mahasiswa)")}
|
|
if "npm" in kolom and "nim" not in kolom:
|
|
conn.execute("ALTER TABLE mahasiswa RENAME COLUMN npm TO nim")
|
|
|
|
|
|
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", "rtm"):
|
|
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 (
|
|
nim TEXT NOT NULL,
|
|
nama TEXT NOT NULL,
|
|
mk_kode TEXT NOT NULL,
|
|
kelas TEXT NOT NULL DEFAULT '',
|
|
tahun TEXT NOT NULL,
|
|
PRIMARY KEY (nim, mk_kode, tahun)
|
|
)"""
|
|
)
|
|
_migrasi_nim_mahasiswa(conn)
|
|
|
|
|
|
# ── 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", "rtm")
|
|
|
|
|
|
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 (nim, nama, mk_kode, kelas, tahun) "
|
|
"VALUES (?,?,?,?,?)",
|
|
[(b["nim"], 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 nim, nama, kelas FROM mahasiswa WHERE mk_kode=? AND tahun=? "
|
|
"ORDER BY kelas, nim",
|
|
(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)
|