Python Database v2.0

Programlama ve Script dilleri konusunda bilgi paylaşım alanıdır.
Cevapla
Kullanıcı avatarı
TRWE_2012
Zettabyte1
Zettabyte1
Mesajlar: 15657
Kayıt: 25 Eyl 2013, 13:38
cinsiyet: Erkek
Teşekkür etti: 2738 kez
Teşekkür edildi: 5662 kez

Python Database v2.0

Mesaj gönderen TRWE_2012 »


Merhaba

Aşağıdaki ekran görüntüsünde Proje 6 : .... başlığını değiştirmeyi unutmuşum...(kafa kazan gibi olunca böyle oluyor,dalgınlık %1500...!!!) onun için orayı dikkate almayın.

Resim
Ekran Görüntüsü :
Resim
KOD İÇERİĞİ : (personnel_db_v2.py)

Kod: Tümünü seç


#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# =============================================================================
# Project      : VB6 -> Python Conversion
# Module       : Personnel Database  (Section 12.4)  — v2.0
# Description  : Personnel record management application.
#                - SQLite database (stdlib sqlite3)
#                - Fields : ID, Full Name, Birth Date, Department,
#                           Monthly Salary, Address, Phone, Mobile
#                - Menu   : Record -> New / Save / Delete / Open DB / Quit
#                - Operations panel : Clear Fields / Find Record (by ID)
#                - Navigation panel : Previous (←) / Next (→)
#                - Open DB  : user selects any .db file via file dialog
#                - Auto-split : when active .db exceeds 1024 MB a new shard
#                              is created automatically (personnel-2.db …)
#                - Shard scan: on open, sibling shard files are detected and
#                              reported to the user
# Platform     : GNU/Linux Mint 22.2 MATE x64  |  Windows 11 24H2 x64
# Python       : 3.8+  (stdlib only: tkinter, sqlite3, os, sys, re)
# =============================================================================

import os
import sys
import re

# ---------------------------------------------------------------------------
# VENV BOOTSTRAPPER  (v1.1 - robust)
# ---------------------------------------------------------------------------
VENV_DIR       = "/home/linuxmaster/python-toolchain"
_SENTINEL_FLAG = "_VENV_BOOTSTRAP_DONE"
_candidates    = [
    os.path.join(VENV_DIR, "bin", "python3"),
    os.path.join(VENV_DIR, "bin", "python"),
    os.path.join(VENV_DIR, "Scripts", "python.exe"),
]

def _venv_is_healthy() -> bool:
    lib = os.path.join(VENV_DIR, "lib")
    if os.path.isdir(lib):
        for e in os.listdir(lib):
            if os.path.isdir(os.path.join(lib, e, "site-packages")):
                return True
    return os.path.isdir(os.path.join(VENV_DIR, "Lib", "site-packages"))

def _bootstrap_venv() -> None:
    if os.environ.get(_SENTINEL_FLAG) == "1":
        return
    inside = os.path.normcase(os.path.realpath(sys.executable)).startswith(
        os.path.normcase(os.path.realpath(VENV_DIR)))
    if inside or not _venv_is_healthy():
        return
    for c in _candidates:
        if os.path.isfile(c):
            os.environ[_SENTINEL_FLAG] = "1"
            os.execv(c, [c] + sys.argv)

_bootstrap_venv()

# ---------------------------------------------------------------------------
# IMPORTS
# ---------------------------------------------------------------------------
import sqlite3
import tkinter as tk
from tkinter import messagebox, simpledialog, filedialog

# ---------------------------------------------------------------------------
# SHARD CONSTANTS
# ---------------------------------------------------------------------------
DB_SPLIT_LIMIT_MB  = 1024          # split threshold in megabytes
DB_SPLIT_LIMIT_B   = DB_SPLIT_LIMIT_MB * 1024 * 1024

# ---------------------------------------------------------------------------
# ACTIVE DATABASE PATH  (module-level, updated at runtime)
# ---------------------------------------------------------------------------
_DB_PATH: str = os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "personnel.db"
)

def get_db_path() -> str:
    return _DB_PATH

def set_db_path(path: str) -> None:
    global _DB_PATH
    _DB_PATH = path

# ---------------------------------------------------------------------------
# SHARD HELPERS
# ---------------------------------------------------------------------------

def _shard_base(path: str) -> str:
    """
    Return the stem of the shard family.
    personnel.db        -> personnel
    personnel-2.db      -> personnel
    /some/dir/data-5.db -> data
    """
    name = os.path.basename(path)
    stem = name[:-3] if name.lower().endswith(".db") else name
    # strip trailing -N suffix
    stem = re.sub(r"-\d+$", "", stem)
    return stem


def _shard_index(path: str) -> int:
    """
    Return the shard index of a file.
    personnel.db   -> 1
    personnel-2.db -> 2
    """
    name = os.path.basename(path)
    stem = name[:-3] if name.lower().endswith(".db") else name
    m    = re.search(r"-(\d+)$", stem)
    return int(m.group(1)) if m else 1


def _shard_path(base_path: str, index: int) -> str:
    """
    Build a shard path from the first-shard path and an index.
    base_path=/a/personnel.db  index=2  ->  /a/personnel-2.db
    """
    directory = os.path.dirname(base_path)
    base      = _shard_base(base_path)
    filename  = f"{base}.db" if index == 1 else f"{base}-{index}.db"
    return os.path.join(directory, filename)


def find_all_shards(path: str) -> list:
    """
    Return sorted list of all shard paths that belong to the same family
    as `path`, found in the same directory.
    """
    directory = os.path.dirname(os.path.abspath(path))
    base      = _shard_base(path)
    shards    = []
    try:
        for name in os.listdir(directory):
            if not name.lower().endswith(".db"):
                continue
            stem = name[:-3]
            # match  base.db  or  base-N.db
            if stem == base or re.match(rf"^{re.escape(base)}-\d+$", stem):
                shards.append(os.path.join(directory, name))
    except OSError:
        pass
    shards.sort(key=lambda p: _shard_index(p))
    return shards


def db_file_size_bytes(path: str | None = None) -> int:
    """Return file size in bytes; 0 if file does not exist."""
    target = path or get_db_path()
    try:
        return os.path.getsize(target)
    except OSError:
        return 0


def next_shard_path() -> str:
    """Compute the path of the next shard after the current active one."""
    current = get_db_path()
    idx     = _shard_index(current)
    return _shard_path(current, idx + 1)


# ---------------------------------------------------------------------------
# DATABASE LAYER
# ---------------------------------------------------------------------------

def db_connect(path: str | None = None) -> sqlite3.Connection:
    conn = sqlite3.connect(path or get_db_path())
    conn.row_factory = sqlite3.Row
    return conn


def db_init(path: str | None = None) -> None:
    """Create the personnel table if it does not exist."""
    with db_connect(path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS personnel (
                id          INTEGER PRIMARY KEY,
                full_name   TEXT    NOT NULL DEFAULT '',
                birth_date  TEXT    NOT NULL DEFAULT '',
                department  TEXT    NOT NULL DEFAULT '',
                salary      TEXT    NOT NULL DEFAULT '',
                address     TEXT    NOT NULL DEFAULT '',
                phone       TEXT    NOT NULL DEFAULT '',
                mobile      TEXT    NOT NULL DEFAULT ''
            )
        """)
        conn.commit()


def db_all_ids() -> list:
    with db_connect() as conn:
        rows = conn.execute(
            "SELECT id FROM personnel ORDER BY id"
        ).fetchall()
    return [r["id"] for r in rows]


def db_get(record_id: int) -> dict | None:
    with db_connect() as conn:
        row = conn.execute(
            "SELECT * FROM personnel WHERE id = ?", (record_id,)
        ).fetchone()
    return dict(row) if row else None


def db_save(data: dict) -> int:
    """Insert or update a record. Returns the record id."""
    with db_connect() as conn:
        existing = conn.execute(
            "SELECT id FROM personnel WHERE id = ?", (data["id"],)
        ).fetchone()
        if existing:
            conn.execute("""
                UPDATE personnel SET
                    full_name=?, birth_date=?, department=?,
                    salary=?, address=?, phone=?, mobile=?
                WHERE id=?
            """, (data["full_name"], data["birth_date"], data["department"],
                  data["salary"],    data["address"],    data["phone"],
                  data["mobile"],    data["id"]))
        else:
            conn.execute("""
                INSERT INTO personnel
                    (id, full_name, birth_date, department,
                     salary, address, phone, mobile)
                VALUES (?,?,?,?,?,?,?,?)
            """, (data["id"],        data["full_name"], data["birth_date"],
                  data["department"], data["salary"],   data["address"],
                  data["phone"],      data["mobile"]))
        conn.commit()
    return data["id"]


def db_delete(record_id: int) -> bool:
    with db_connect() as conn:
        cur = conn.execute(
            "DELETE FROM personnel WHERE id = ?", (record_id,)
        )
        conn.commit()
    return cur.rowcount > 0


# ---------------------------------------------------------------------------
# FIELD DEFINITIONS
# ---------------------------------------------------------------------------
FIELDS = [
    ("id",          "ID No"),
    ("full_name",   "Full Name"),
    ("birth_date",  "Birth Date"),
    ("department",  "Department"),
    ("salary",      "Monthly Salary"),
    ("address",     "Address"),
    ("phone",       "Phone"),
    ("mobile",      "Mobile"),
]

# ---------------------------------------------------------------------------
# MAIN APPLICATION
# ---------------------------------------------------------------------------

class PersonnelApp:

    def __init__(self, master: tk.Tk) -> None:
        self.master    = master
        self._ids: list = []
        self._idx: int  = -1
        self._entries: dict = {}

        master.resizable(False, False)
        self._build_menu()
        self._build_ui()
        self._center_window()
        master.protocol("WM_DELETE_WINDOW", self._on_quit)

        # Load default db
        self._load_db(get_db_path(), initial=True)

    # ------------------------------------------------------------------
    # DB LOAD / SWITCH
    # ------------------------------------------------------------------

    def _load_db(self, path: str, initial: bool = False) -> None:
        """
        Set active DB path, initialise schema, reload IDs,
        scan for sibling shards and inform the user.
        """
        set_db_path(path)
        db_init(path)
        self._reload_ids()

        if self._ids:
            self._show_record(0)
        else:
            self._clear_fields()

        self._update_title()

        # Shard scan (skip on very first load to avoid startup clutter)
        shards = find_all_shards(path)
        if len(shards) > 1:
            shard_names = "\n  ".join(os.path.basename(s) for s in shards)
            msg = (
                f"The following shard files were found in the same directory:\n\n"
                f"  {shard_names}\n\n"
                f"Currently active: {os.path.basename(path)}\n"
                f"Use Record → Open Database File to switch between shards."
            )
            messagebox.showinfo("Shard Files Detected", msg, parent=self.master)
        elif not initial:
            self._set_status(
                f"Loaded: {os.path.basename(path)}  "
                f"({db_file_size_bytes() / 1024:.1f} KB)"
            )

    def _update_title(self) -> None:
        path   = get_db_path()
        name   = os.path.basename(path)
        size_b = db_file_size_bytes()
        size_s = (f"{size_b / (1024*1024):.1f} MB"
                  if size_b >= 1024 * 1024 else f"{size_b / 1024:.1f} KB")
        self.master.title(f"Personnel Database  —  {name}  [{size_s}]")

    # ------------------------------------------------------------------
    # AUTO-SPLIT CHECK
    # ------------------------------------------------------------------

    def _check_split(self) -> bool:
        """
        Called before each save.  If the active db exceeds the limit,
        create and switch to the next shard.
        Returns True if a split was performed (caller must re-save).
        """
        if db_file_size_bytes() < DB_SPLIT_LIMIT_B:
            return False

        new_path = next_shard_path()
        new_name = os.path.basename(new_path)

        answer = messagebox.askyesno(
            "Database Size Limit Reached",
            f"The current database file has reached {DB_SPLIT_LIMIT_MB} MB.\n\n"
            f"A new shard will be created:\n  {new_name}\n\n"
            f"New records will be saved to this file.\n"
            f"Existing records remain in the previous shard.\n\n"
            f"Continue?",
            icon=messagebox.WARNING,
            parent=self.master,
        )
        if not answer:
            return False

        db_init(new_path)           # create schema in new shard
        set_db_path(new_path)       # switch active path
        self._reload_ids()
        self._update_title()
        self._set_status(
            f"Split performed — now writing to: {new_name}"
        )
        return True

    # ------------------------------------------------------------------
    # MENU
    # ------------------------------------------------------------------

    def _build_menu(self) -> None:
        menubar = tk.Menu(self.master)
        rec_m   = tk.Menu(menubar, tearoff=0)
        rec_m.add_command(label="New Record",        command=self._op_new)
        rec_m.add_command(label="Save Record",       command=self._op_save)
        rec_m.add_command(label="Delete Record",     command=self._op_delete)
        rec_m.add_separator()
        rec_m.add_command(label="Open Database File",command=self._op_open_db)
        rec_m.add_separator()
        rec_m.add_command(label="Quit",              command=self._on_quit)
        menubar.add_cascade(label="Record", menu=rec_m)
        self.master.config(menu=menubar)

    # ------------------------------------------------------------------
    # UI
    # ------------------------------------------------------------------

    def _build_ui(self) -> None:
        outer = tk.Frame(self.master, padx=10, pady=10)
        outer.pack()

        # LEFT — Personnel group
        lf = tk.LabelFrame(outer, text="Personnel",
                           font=("Helvetica", 9, "bold"), padx=8, pady=8)
        lf.grid(row=0, column=0, sticky="nsew", padx=(0, 10))

        for row, (key, label) in enumerate(FIELDS):
            tk.Label(lf, text=f"{label}:", anchor="e",
                     width=14).grid(row=row, column=0, sticky="e",
                                    pady=3, padx=(0, 6))
            entry = tk.Entry(lf, width=30, relief=tk.SUNKEN)
            entry.grid(row=row, column=1, sticky="w", pady=3)
            self._entries[key] = entry

        # RIGHT — Operations + Navigation
        rf = tk.Frame(outer)
        rf.grid(row=0, column=1, sticky="ns")

        ops = tk.LabelFrame(rf, text="Operations",
                            font=("Helvetica", 9, "bold"), padx=8, pady=8)
        ops.pack(fill=tk.X, pady=(0, 10))
        tk.Button(ops, text="Clear Fields", width=14,
                  command=self._op_clear).pack(pady=(0, 6))
        tk.Button(ops, text="Find Record",  width=14,
                  command=self._op_find).pack()

        nav = tk.LabelFrame(rf, text="Records",
                            font=("Helvetica", 9, "bold"), padx=8, pady=8)
        nav.pack(fill=tk.X)
        btn_f = tk.Frame(nav)
        btn_f.pack()
        tk.Button(btn_f, text="←", width=5, font=("Helvetica", 12),
                  command=self._op_prev).pack(side=tk.LEFT, padx=4)
        tk.Button(btn_f, text="→", width=5, font=("Helvetica", 12),
                  command=self._op_next).pack(side=tk.LEFT, padx=4)

        # Status
        self.lbl_status = tk.Label(outer, text="", anchor="w",
                                   font=("Courier", 9), fg="#444444")
        self.lbl_status.grid(row=1, column=0, columnspan=2,
                             sticky="w", pady=(8, 0))

    # ------------------------------------------------------------------
    # NAVIGATION HELPERS
    # ------------------------------------------------------------------

    def _reload_ids(self) -> None:
        self._ids = db_all_ids()

    def _show_record(self, idx: int) -> None:
        if not self._ids:
            self._clear_fields()
            self._set_status("No records in database.")
            return
        self._idx = max(0, min(idx, len(self._ids) - 1))
        rec = db_get(self._ids[self._idx])
        if rec:
            self._populate_fields(rec)
            size_b = db_file_size_bytes()
            size_s = (f"{size_b/(1024*1024):.1f} MB"
                      if size_b >= 1024*1024 else f"{size_b/1024:.1f} KB")
            self._set_status(
                f"Record {self._idx + 1} of {len(self._ids)}  "
                f"(ID: {self._ids[self._idx]})  |  "
                f"DB: {os.path.basename(get_db_path())}  [{size_s}]"
            )

    def _populate_fields(self, rec: dict) -> None:
        for key, entry in self._entries.items():
            entry.config(state=tk.NORMAL)
            entry.delete(0, tk.END)
            entry.insert(0, str(rec.get(key, "")))

    def _clear_fields(self) -> None:
        for entry in self._entries.values():
            entry.config(state=tk.NORMAL)
            entry.delete(0, tk.END)
        self._set_status("Fields cleared.")

    def _read_fields(self) -> dict:
        return {key: entry.get().strip()
                for key, entry in self._entries.items()}

    # ------------------------------------------------------------------
    # OPERATIONS
    # ------------------------------------------------------------------

    def _op_open_db(self) -> None:
        """Let the user select any .db file and load it as the active database."""
        path = filedialog.askopenfilename(
            title="Open Personnel Database",
            filetypes=[("SQLite Database", "*.db"), ("All files", "*.*")],
            initialdir=os.path.dirname(get_db_path()),
            parent=self.master,
        )
        if not path:
            return
        self._load_db(path)

    def _op_clear(self) -> None:
        self._clear_fields()
        self._idx = -1

    def _op_find(self) -> None:
        raw = simpledialog.askstring(
            "Find Record", "Enter ID number to search:",
            parent=self.master
        )
        if not raw:
            return
        try:
            search_id = int(raw.strip())
        except ValueError:
            messagebox.showwarning("Find", "ID must be a whole number.",
                                   parent=self.master)
            return
        rec = db_get(search_id)
        if rec:
            self._reload_ids()
            self._show_record(self._ids.index(search_id))
        else:
            messagebox.showinfo(
                "Find", f"No record found with ID: {search_id}",
                parent=self.master
            )

    def _op_new(self) -> None:
        self._clear_fields()
        next_id = (max(self._ids) + 1) if self._ids else 1
        self._entries["id"].insert(0, str(next_id))
        self._entries["full_name"].focus_set()
        self._set_status(
            "New record — fill in the fields and choose Record → Save."
        )

    def _op_save(self) -> None:
        data   = self._read_fields()
        raw_id = data.get("id", "").strip()
        if not raw_id:
            messagebox.showwarning("Save", "ID No cannot be empty.",
                                   parent=self.master)
            return
        try:
            data["id"] = int(raw_id)
        except ValueError:
            messagebox.showwarning("Save", "ID must be a whole number.",
                                   parent=self.master)
            return
        if not data.get("full_name"):
            messagebox.showwarning("Save", "Full Name cannot be empty.",
                                   parent=self.master)
            return

        # Auto-split check: may switch DB_PATH to next shard
        self._check_split()

        db_save(data)
        self._reload_ids()

        # Record might now live in a new shard — handle gracefully
        if data["id"] in self._ids:
            self._show_record(self._ids.index(data["id"]))
        else:
            self._show_record(len(self._ids) - 1)

        self._update_title()
        self._set_status(
            f"Record saved — ID: {data['id']}  |  "
            f"DB: {os.path.basename(get_db_path())}"
        )

    def _op_delete(self) -> None:
        raw = self._entries["id"].get().strip()
        if not raw:
            messagebox.showinfo("Delete", "No record loaded.",
                                parent=self.master)
            return
        try:
            rec_id = int(raw)
        except ValueError:
            return
        name = self._entries["full_name"].get().strip() or f"ID {rec_id}"
        if not messagebox.askyesno(
            "Delete Record",
            f'Permanently delete:\n"{name}" (ID: {rec_id})?\n\nThis cannot be undone.',
            icon=messagebox.WARNING, parent=self.master,
        ):
            return
        if db_delete(rec_id):
            self._reload_ids()
            if self._ids:
                self._show_record(min(self._idx, len(self._ids) - 1))
            else:
                self._clear_fields()
            self._update_title()
            self._set_status(f"Record deleted — ID: {rec_id}")
        else:
            messagebox.showwarning("Delete", "Record not found in database.",
                                   parent=self.master)

    def _op_prev(self) -> None:
        if not self._ids:
            return
        if self._idx <= 0:
            self._set_status("Already at the first record.")
            return
        self._show_record(self._idx - 1)

    def _op_next(self) -> None:
        if not self._ids:
            return
        if self._idx >= len(self._ids) - 1:
            self._set_status("Already at the last record.")
            return
        self._show_record(self._idx + 1)

    # ------------------------------------------------------------------
    # MISC
    # ------------------------------------------------------------------

    def _set_status(self, text: str) -> None:
        self.lbl_status.config(text=f"  {text}")

    def _on_quit(self) -> None:
        if messagebox.askyesno("Quit", "Are you sure you want to quit?",
                               icon=messagebox.QUESTION, parent=self.master):
            self.master.destroy()

    def _center_window(self) -> None:
        self.master.update_idletasks()
        w  = self.master.winfo_width()
        h  = self.master.winfo_height()
        sw = self.master.winfo_screenwidth()
        sh = self.master.winfo_screenheight()
        self.master.geometry(f"+{(sw - w) // 2}+{(sh - h) // 2}")


# ---------------------------------------------------------------------------
# ENTRY POINT
# ---------------------------------------------------------------------------

def main() -> None:
    root = tk.Tk()
    PersonnelApp(root)
    root.mainloop()
    input("\nPress ENTER to exit.")


if __name__ == "__main__":
    main()

Basit bir uygulama ama işlevsel, bunu basit adres defteri gibi kullanabilirsiniz bana göre..Güle güle kullanın.....
Cevapla

“Programlama ve Script dilleri” sayfasına dön