database.py

# database.py
# CV PDF Project

# Copyright 2025 pyrus-code
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import sqlite3
import json
import os
import defaults
from datetime import datetime
from pathlib import Path

from app_paths import DB_FILE, FONTS_DIR


def get_connection():
    """Establishes a connection to the SQLite database."""
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row
    return conn


def _clear_all_tables(cursor, exclude=None):
    """A helper function to delete all rows from all tables, with an exclusion option."""
    if exclude is None:
        exclude = []

    tables = ["personal_info", "photos", "education", "languages", "skills", "software_skills",
              "settings", "jobs", "snapshots", "signatures", "standard_cvs"]

    for table in tables:
        if table not in exclude:
            cursor.execute(f"DELETE FROM {table}")


def _find_font_path_for_db(relative_path_str):
    if not relative_path_str:
        return ""

    font_filename = os.path.basename(relative_path_str)

    local_path = FONTS_DIR / font_filename
    if local_path.exists():
        return str(local_path.resolve())

    if os.name == 'nt':
        win_path = Path(os.environ.get("SystemRoot", "C:\\Windows")) / "Fonts" / font_filename
        if win_path.exists():
            return str(win_path.resolve())

    return relative_path_str


def _populate_from_defaults(cursor):
    """A helper function to populate tables with default data."""
    default_data = defaults.get_default_data()
    for key, value in default_data["personal_info"].items():
        cursor.execute("INSERT INTO personal_info (key, value) VALUES (?, ?)", (key, value))

    cursor.execute("SELECT COUNT(*) FROM photos")
    if cursor.fetchone()[0] == 0:
        for photo in default_data["photos"]:
            cursor.execute("INSERT INTO photos (label, path, size_adjustment) VALUES (?, ?, ?)",
                           (photo['label'], photo['path'], 0))

    cursor.execute("SELECT COUNT(*) FROM signatures")
    if cursor.fetchone()[0] == 0:
        cursor.execute("INSERT INTO signatures (label, path, size_adjustment) VALUES (?, ?, ?)", ('Default', '', 0))

    for i, edu in enumerate(default_data["education"]):
        cursor.execute("INSERT INTO education (type, data, order_index) VALUES (?, ?, ?)",
                       (edu['type'], json.dumps(edu), i))
    for i, lang in enumerate(default_data["languages"]):
        cursor.execute("INSERT INTO languages (language, proficiency, details, order_index) VALUES (?, ?, ?, ?)",
                       (lang['language'], lang['proficiency'], lang['details'], i))
    for i, skill in enumerate(default_data["skills"]):
        cursor.execute("INSERT INTO skills (skill_name, order_index) VALUES (?, ?)", (skill, i))
    for i, skill in enumerate(default_data["software_skills"]):
        cursor.execute("INSERT INTO software_skills (skill, level, comments, order_index) VALUES (?, ?, ?, ?)",
                       (skill['skill'], skill['level'], skill.get('comments', ''), i))

    default_settings = default_data["settings"]
    default_settings['calibri_path'] = _find_font_path_for_db(default_settings.get('calibri_path', ''))
    default_settings['calibri_bold_path'] = _find_font_path_for_db(default_settings.get('calibri_bold_path', ''))
    default_settings['times_path'] = _find_font_path_for_db(default_settings.get('times_path', ''))
    default_settings['times_bold_path'] = _find_font_path_for_db(default_settings.get('times_bold_path', ''))

    for key, value in default_settings.items():
        cursor.execute("INSERT INTO settings (key, value) VALUES (?, ?)", (key, value))

    for i, job in enumerate(default_data["jobs"]):
        cursor.execute("INSERT INTO jobs (title, company, dates, order_index) VALUES (?, ?, ?, ?)",
                       (job['title'], job['company'], job['dates'], i))


def reset_current_data_to_defaults():
    conn = get_connection()
    cursor = conn.cursor()
    _clear_all_tables(cursor, exclude=['snapshots', 'photos', 'signatures', 'standard_cvs', 'user_defaults'])
    _populate_from_defaults(cursor)
    conn.commit()
    conn.close()


def load_data_from_snapshot(db_data):
    conn = get_connection()
    cursor = conn.cursor()
    _clear_all_tables(cursor, exclude=['snapshots', 'photos', 'signatures', 'standard_cvs', 'user_defaults'])
    conn.commit()
    conn.close()

    if 'personal_info' in db_data: update_personal_info(db_data['personal_info'])
    if 'settings' in db_data: save_settings(db_data['settings'])

    conn = get_connection()
    cursor = conn.cursor()
    if 'education' in db_data:
        for i, edu in enumerate(db_data["education"]):
            cursor.execute("INSERT INTO education (type, data, order_index) VALUES (?, ?, ?)",
                           (edu['type'], json.dumps(edu), i))
    if 'languages' in db_data:
        for i, lang in enumerate(db_data["languages"]):
            cursor.execute("INSERT INTO languages (language, proficiency, details, order_index) VALUES (?, ?, ?, ?)",
                           (lang['language'], lang['proficiency'], lang['details'], i))
    if 'skills' in db_data:
        for i, skill in enumerate(db_data["skills"]):
            cursor.execute("INSERT INTO skills (skill_name, order_index) VALUES (?, ?)", (skill['skill_name'], i))
    if 'software_skills' in db_data:
        for i, skill in enumerate(db_data["software_skills"]):
            cursor.execute("INSERT INTO software_skills (skill, level, comments, order_index) VALUES (?, ?, ?, ?)",
                           (skill['skill'], skill['level'], skill.get('comments', ''), i))
    if 'jobs' in db_data:
        for i, job in enumerate(db_data["jobs"]):
            cursor.execute("INSERT INTO jobs (title, company, dates, order_index) VALUES (?, ?, ?, ?)",
                           (job['title'], job['company'], job['dates'], i))
    if 'signatures' in db_data:
        for sig in db_data['signatures']:
            cursor.execute("INSERT INTO signatures (label, path, size_adjustment) VALUES (?, ?, ?)",
                           (sig['label'], sig['path'], sig.get('size_adjustment', 0)))

    conn.commit()
    conn.close()


def upgrade_schema(conn):
    """Upgrades the database schema if new tables or fields are needed."""
    cursor = conn.cursor()

    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='skills'")
    if cursor.fetchone() is None:
        print("Upgrading database: creating 'skills' table.")
        cursor.execute('''
            CREATE TABLE skills (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                skill_name TEXT,
                order_index INTEGER
            )
        ''')
        for i, skill in enumerate(defaults.get_default_data()["skills"]):
            cursor.execute("INSERT INTO skills (skill_name, order_index) VALUES (?, ?)", (skill, i))

    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='user_defaults'")
    if cursor.fetchone() is None:
        print("Upgrading database: creating 'user_defaults' table.")
        cursor.execute('CREATE TABLE user_defaults (key TEXT PRIMARY KEY, data TEXT)')

    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='standard_cvs'")
    if cursor.fetchone() is None:
        print("Upgrading database: creating 'standard_cvs' table.")
        cursor.execute('''
            CREATE TABLE standard_cvs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL UNIQUE,
                content TEXT
            )
        ''')

    cursor.execute("PRAGMA table_info(software_skills)")
    columns = [col['name'] for col in cursor.fetchall()]
    if 'comments' not in columns:
        cursor.execute("ALTER TABLE software_skills ADD COLUMN comments TEXT")

    cursor.execute("PRAGMA table_info(photos)")
    columns = [col['name'] for col in cursor.fetchall()]
    if 'size_adjustment' not in columns:
        print("Upgrading database: adding 'size_adjustment' to 'photos' table.")
        cursor.execute("ALTER TABLE photos ADD COLUMN size_adjustment INTEGER DEFAULT 0")

    cursor.execute("PRAGMA table_info(signatures)")
    columns = [col['name'] for col in cursor.fetchall()]
    if 'size_adjustment' not in columns:
        print("Upgrading database: adding 'size_adjustment' to 'signatures' table.")
        cursor.execute("ALTER TABLE signatures ADD COLUMN size_adjustment INTEGER DEFAULT 0")

    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='snapshots'")
    if cursor.fetchone() is None:
        cursor.execute('''
            CREATE TABLE snapshots (
                id INTEGER PRIMARY KEY AUTOINCREMENT, job_title TEXT, company TEXT,
                timestamp TEXT, full_data TEXT
            )''')

    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='signatures'")
    if cursor.fetchone() is None:
        cursor.execute('CREATE TABLE signatures (id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT, path TEXT, size_adjustment INTEGER DEFAULT 0)')
        cursor.execute("INSERT INTO signatures (label, path, size_adjustment) VALUES (?, ?, ?)", ('Default', '', 0))

    conn.commit()


def init_db(force_default=False):
    db_exists = os.path.exists(DB_FILE)
    if force_default and db_exists:
        try:
            os.remove(DB_FILE)
            db_exists = False
        except OSError as e:
            print(f"Error removing database file: {e}")
            return


    conn = get_connection()

    if db_exists and not force_default:
        upgrade_schema(conn)
        conn.close()
        return

    cursor = conn.cursor()
    tables = ["personal_info", "photos", "education", "languages", "skills", "software_skills",
              "settings", "jobs", "snapshots", "signatures", "standard_cvs", "user_defaults"]
    for table in tables:
        cursor.execute(f"DROP TABLE IF EXISTS {table}")

    cursor.execute('CREATE TABLE personal_info (key TEXT PRIMARY KEY, value TEXT)')
    cursor.execute('CREATE TABLE photos (id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT, path TEXT, size_adjustment INTEGER DEFAULT 0)')
    cursor.execute(
        'CREATE TABLE education (id INTEGER PRIMARY KEY AUTOINCREMENT, type TEXT, data TEXT, order_index INTEGER)')
    cursor.execute(
        'CREATE TABLE languages (id INTEGER PRIMARY KEY AUTOINCREMENT, language TEXT, proficiency TEXT, details TEXT, order_index INTEGER)')
    cursor.execute(
        'CREATE TABLE skills (id INTEGER PRIMARY KEY AUTOINCREMENT, skill_name TEXT, order_index INTEGER)')
    cursor.execute(
        'CREATE TABLE software_skills (id INTEGER PRIMARY KEY AUTOINCREMENT, skill TEXT, level TEXT, comments TEXT, order_index INTEGER)')
    cursor.execute('CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT)')
    cursor.execute(
        'CREATE TABLE jobs (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, company TEXT, dates TEXT, order_index INTEGER)')
    cursor.execute(
        'CREATE TABLE snapshots (id INTEGER PRIMARY KEY AUTOINCREMENT, job_title TEXT, company TEXT, timestamp TEXT, full_data TEXT)')
    cursor.execute('CREATE TABLE signatures (id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT, path TEXT, size_adjustment INTEGER DEFAULT 0)')
    cursor.execute('''
        CREATE TABLE standard_cvs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL UNIQUE,
            content TEXT
        )
    ''')
    cursor.execute('CREATE TABLE user_defaults (key TEXT PRIMARY KEY, data TEXT)')

    _populate_from_defaults(cursor)

    conn.commit()
    conn.close()


def load_all_data():
    if not os.path.exists(DB_FILE):
        init_db()

    conn = get_connection()
    cursor = conn.cursor()
    data = {}

    default_personal_info = defaults.get_default_data()['personal_info']
    db_personal_info = {row['key']: row['value'] for row in cursor.execute("SELECT key, value FROM personal_info")}
    data['personal_info'] = {**default_personal_info, **db_personal_info}

    default_settings = defaults.get_default_data()['settings']
    db_settings = {row['key']: row['value'] for row in cursor.execute("SELECT key, value FROM settings")}
    data['settings'] = {**default_settings, **db_settings}

    data['photos'] = [dict(row) for row in cursor.execute("SELECT id, label, path, size_adjustment FROM photos")]
    data['education'] = [dict(json.loads(row['data']), id=row['id']) for row in
                         cursor.execute("SELECT id, data FROM education ORDER BY order_index")]
    data['languages'] = [dict(row) for row in cursor.execute("SELECT * FROM languages ORDER BY order_index")]
    data['skills'] = [dict(row) for row in cursor.execute("SELECT * FROM skills ORDER BY order_index")]
    data['software_skills'] = [dict(row) for row in
                               cursor.execute("SELECT * FROM software_skills ORDER BY order_index")]
    data['jobs'] = [dict(row) for row in cursor.execute("SELECT * FROM jobs ORDER BY order_index")]
    data['signatures'] = [dict(row) for row in cursor.execute("SELECT id, label, path, size_adjustment FROM signatures")]
    data['standard_cvs'] = [dict(row) for row in cursor.execute("SELECT * FROM standard_cvs ORDER BY name")]

    conn.close()
    return data


def save_as_user_default():
    conn = get_connection()
    cursor = conn.cursor()

    current_data = load_all_data()
    current_data.pop('standard_cvs', None)

    cursor.execute("DELETE FROM user_defaults")

    for key, value in current_data.items():
        cursor.execute("INSERT INTO user_defaults (key, data) VALUES (?, ?)", (key, json.dumps(value)))

    conn.commit()
    conn.close()


def restore_from_user_default():
    conn = get_connection()
    cursor = conn.cursor()

    cursor.execute("SELECT key, data FROM user_defaults")
    user_defaults = {row['key']: json.loads(row['data']) for row in cursor.fetchall()}

    if not user_defaults:
        conn.close()
        return False

    _clear_all_tables(cursor, exclude=['snapshots', 'standard_cvs', 'user_defaults'])

    for key, value in user_defaults.get('personal_info', {}).items():
        cursor.execute("INSERT INTO personal_info (key, value) VALUES (?, ?)", (key, value))

    for item in user_defaults.get('photos', []):
        cursor.execute("INSERT INTO photos (label, path, size_adjustment) VALUES (?, ?, ?)",
                       (item['label'], item['path'], item.get('size_adjustment', 0)))

    for item in user_defaults.get('signatures', []):
        cursor.execute("INSERT INTO signatures (label, path, size_adjustment) VALUES (?, ?, ?)",
                       (item['label'], item['path'], item.get('size_adjustment', 0)))

    for i, item in enumerate(user_defaults.get('education', [])):
        cursor.execute("INSERT INTO education (type, data, order_index) VALUES (?, ?, ?)",
                       (item['type'], json.dumps(item), i))

    for i, item in enumerate(user_defaults.get('languages', [])):
        cursor.execute("INSERT INTO languages (language, proficiency, details, order_index) VALUES (?, ?, ?, ?)",
                       (item['language'], item['proficiency'], item['details'], i))

    for i, item in enumerate(user_defaults.get('skills', [])):
        cursor.execute("INSERT INTO skills (skill_name, order_index) VALUES (?, ?)",
                       (item['skill_name'], i))

    for i, item in enumerate(user_defaults.get('software_skills', [])):
        cursor.execute("INSERT INTO software_skills (skill, level, comments, order_index) VALUES (?, ?, ?, ?)",
                       (item['skill'], item['level'], item.get('comments', ''), i))

    for key, value in user_defaults.get('settings', {}).items():
        cursor.execute("INSERT INTO settings (key, value) VALUES (?, ?)", (key, value))

    for i, item in enumerate(user_defaults.get('jobs', [])):
        cursor.execute("INSERT INTO jobs (title, company, dates, order_index) VALUES (?, ?, ?, ?)",
                       (item['title'], item['company'], item['dates'], i))

    conn.commit()
    conn.close()
    return True


def restore_factory_defaults():
    conn = get_connection()
    cursor = conn.cursor()
    _clear_all_tables(cursor)
    _populate_from_defaults(cursor)
    conn.commit()
    conn.close()


def delete_all_snapshots():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM snapshots")
    conn.commit()
    conn.close()


def save_settings(settings_dict):
    conn = get_connection()
    cursor = conn.cursor()
    for key, value in settings_dict.items():
        cursor.execute("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", (key, value))
    conn.commit()
    conn.close()


def update_item_order(table_name, id_order_list):
    conn = get_connection()
    cursor = conn.cursor()
    for index, item_id in enumerate(id_order_list):
        cursor.execute(f"UPDATE {table_name} SET order_index = ? WHERE id = ?", (index, item_id))
    conn.commit()
    conn.close()


def execute_query(query, params=()):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(query, params)
    conn.commit()
    conn.close()


def update_personal_info(p_info_dict):
    conn = get_connection()
    cursor = conn.cursor()
    for key, value in p_info_dict.items():
        cursor.execute("INSERT OR REPLACE INTO personal_info (key, value) VALUES (?, ?)", (key, value))
    conn.commit()
    conn.close()


def update_photos(photos_list):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM photos")
    for photo in photos_list:
        cursor.execute("INSERT INTO photos (label, path, size_adjustment) VALUES (?, ?, ?)",
                       (photo['label'], photo['path'], photo.get('size_adjustment', 0)))
    conn.commit()
    conn.close()


def update_signatures(signatures_list):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM signatures")
    for sig in signatures_list:
        cursor.execute("INSERT INTO signatures (label, path, size_adjustment) VALUES (?, ?, ?)",
                       (sig['label'], sig['path'], sig.get('size_adjustment', 0)))
    conn.commit()
    conn.close()


def clear_all_photo_paths():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("UPDATE photos SET path = ''")
    cursor.execute("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", ('default_photo_label', ''))
    conn.commit()
    conn.close()


def save_snapshot(job_title, company, full_data_json):
    conn = get_connection()
    cursor = conn.cursor()
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    cursor.execute(
        "INSERT INTO snapshots (job_title, company, timestamp, full_data) VALUES (?, ?, ?, ?)",
        (job_title, company, timestamp, full_data_json)
    )
    conn.commit()
    conn.close()


def load_snapshots():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT id, job_title, company, timestamp FROM snapshots ORDER BY timestamp DESC")
    return [dict(row) for row in cursor.fetchall()]


def get_snapshot_data(snapshot_id):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT full_data FROM snapshots WHERE id = ?", (snapshot_id,))
    row = cursor.fetchone()
    conn.close()
    return row['full_data'] if row else None


def delete_snapshot(snapshot_id):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM snapshots WHERE id = ?", (snapshot_id,))
    conn.commit()
    conn.close()


def load_standard_cvs():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT id, name, content FROM standard_cvs ORDER BY name")
    return [dict(row) for row in cursor.fetchall()]


def save_standard_cv(cv_data):
    conn = get_connection()
    cursor = conn.cursor()
    if cv_data.get('id'):
        cursor.execute(
            "UPDATE standard_cvs SET name = ?, content = ? WHERE id = ?",
            (cv_data['name'], cv_data['content'], cv_data['id'])
        )
    else:
        cursor.execute(
            "INSERT INTO standard_cvs (name, content) VALUES (?, ?)",
            (cv_data['name'], cv_data['content'])
        )
    conn.commit()
    conn.close()


def delete_standard_cv(cv_id):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM standard_cvs WHERE id = ?", (cv_id,))
    conn.commit()
    conn.close()