| import os |
| import shutil |
| import json |
| from datetime import datetime |
| from typing import Optional |
| from huggingface_hub import HfApi, login |
| import sqlite3 |
| import tempfile |
|
|
| class DatabaseManager: |
| def __init__(self, hf_token: Optional[str] = None, repo_id: Optional[str] = None): |
| """Initialize Database Manager for Hugging Face Hub storage""" |
| self.repo_id = repo_id or os.getenv('REPO_ID', 'your-username/student-feedback-db') |
| self.hf_token = hf_token or os.getenv('HF_TOKEN') |
| self.db_path = 'instance/feedback_analysis.db' |
| self.backup_dir = 'backups' |
| self.is_local = not self.hf_token |
| |
| os.makedirs(self.backup_dir, exist_ok=True) |
| |
| if self.hf_token: |
| try: |
| login(token=self.hf_token) |
| self.api = HfApi() |
| except Exception: |
| self.api = None |
| else: |
| self.api = None |
| |
| def sqlite_to_json(self, db_path: str) -> dict: |
| """Convert SQLite database to JSON format""" |
| try: |
| conn = sqlite3.connect(db_path) |
| cursor = conn.cursor() |
| |
| cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") |
| tables = cursor.fetchall() |
| |
| data = {} |
| |
| for table in tables: |
| table_name = table[0] |
| |
| if table_name in ['sqlite_sequence', 'sqlite_master']: |
| continue |
| |
| cursor.execute(f"PRAGMA table_info({table_name})") |
| columns = [col[1] for col in cursor.fetchall()] |
| |
| cursor.execute(f"SELECT * FROM {table_name}") |
| rows = cursor.fetchall() |
| |
| table_data = [] |
| for row in rows: |
| row_dict = {} |
| for i, value in enumerate(row): |
| if columns[i] == 'is_admin': |
| value = bool(value) if value is not None else False |
| row_dict[columns[i]] = value |
| table_data.append(row_dict) |
| |
| data[table_name] = { |
| 'columns': columns, |
| 'data': table_data |
| } |
| |
| try: |
| cursor.execute("SELECT * FROM sqlite_sequence") |
| sequence_rows = cursor.fetchall() |
| if sequence_rows: |
| sequence_data = [] |
| for row in sequence_rows: |
| sequence_data.append({ |
| 'name': row[0], |
| 'seq': row[1] |
| }) |
| data['sqlite_sequence'] = { |
| 'columns': ['name', 'seq'], |
| 'data': sequence_data |
| } |
| except Exception: |
| pass |
| |
| conn.close() |
| return data |
| except Exception: |
| return {} |
| |
| def json_to_sqlite(self, json_data: dict, db_path: str): |
| """Convert JSON data back to SQLite database""" |
| try: |
| if os.path.exists(db_path): |
| os.remove(db_path) |
| |
| os.makedirs(os.path.dirname(db_path), exist_ok=True) |
| |
| conn = sqlite3.connect(db_path) |
| cursor = conn.cursor() |
| |
| for table_name, table_info in json_data.items(): |
| if table_name in ['sqlite_sequence', 'sqlite_master']: |
| continue |
| |
| columns = table_info['columns'] |
| data = table_info['data'] |
| |
| if not columns: |
| continue |
| |
| column_defs = [] |
| for col in columns: |
| if col == 'id': |
| column_defs.append(f"{col} INTEGER PRIMARY KEY AUTOINCREMENT") |
| elif col == 'is_admin': |
| column_defs.append(f"{col} BOOLEAN DEFAULT 0") |
| else: |
| column_defs.append(f"{col} TEXT") |
| |
| create_sql = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join(column_defs)})" |
| cursor.execute(create_sql) |
| |
| if data: |
| placeholders = ', '.join(['?' for _ in columns]) |
| insert_sql = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})" |
| |
| for row in data: |
| values = [] |
| for col in columns: |
| value = row.get(col) |
| if col == 'is_admin': |
| if isinstance(value, bool): |
| values.append(int(value)) |
| elif isinstance(value, str): |
| values.append(1 if value.lower() in ['true', '1'] else 0) |
| else: |
| values.append(int(value) if value else 0) |
| else: |
| values.append(value) |
| cursor.execute(insert_sql, values) |
| |
| if 'sqlite_sequence' in json_data: |
| sequence_data = json_data['sqlite_sequence']['data'] |
| for seq_row in sequence_data: |
| table_name = seq_row.get('name') |
| seq_value = seq_row.get('seq') |
| if table_name and seq_value: |
| cursor.execute(f"INSERT INTO sqlite_sequence (name, seq) VALUES (?, ?)", |
| (table_name, seq_value)) |
| |
| conn.commit() |
| conn.close() |
| except Exception: |
| pass |
| |
| def backup_database(self, force: bool = False) -> bool: |
| """Backup database to Hugging Face Hub""" |
| if self.is_local: |
| return True |
| |
| if not self.api or not os.path.exists(self.db_path): |
| return False |
| |
| try: |
| json_data = self.sqlite_to_json(self.db_path) |
| |
| if not json_data: |
| return False |
| |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| temp_file = f"{self.backup_dir}/feedback_backup_{timestamp}.json" |
| |
| with open(temp_file, 'w', encoding='utf-8') as f: |
| json.dump(json_data, f, indent=2, ensure_ascii=False) |
| |
| try: |
| self.api.upload_file( |
| path_or_fileobj=temp_file, |
| path_in_repo='feedback_backup.json', |
| repo_id=self.repo_id, |
| repo_type="dataset", |
| commit_message=f"Backup database - {timestamp}" |
| ) |
| except Exception as upload_error: |
| if "No files have been modified" in str(upload_error): |
| return True |
| else: |
| raise upload_error |
| |
| os.remove(temp_file) |
| return True |
| except Exception: |
| return False |
| |
| def restore_database(self) -> bool: |
| """Restore database from Hugging Face Hub""" |
| if not self.api: |
| return False |
| |
| try: |
| temp_dir = tempfile.mkdtemp() |
| temp_file = os.path.join(temp_dir, 'feedback_backup.json') |
| |
| self.api.hf_hub_download( |
| repo_id=self.repo_id, |
| filename='feedback_backup.json', |
| local_dir=temp_dir, |
| repo_type="dataset" |
| ) |
| |
| with open(temp_file, 'r', encoding='utf-8') as f: |
| json_data = json.load(f) |
| |
| self.json_to_sqlite(json_data, self.db_path) |
| shutil.rmtree(temp_dir) |
| return True |
| except Exception: |
| return False |
| |
| def check_database_exists(self) -> bool: |
| """Check if database file exists locally""" |
| return os.path.exists(self.db_path) |
| |
| def initialize_database_if_needed(self): |
| """Initialize database from backup if local database doesn't exist""" |
| if not self.check_database_exists() and not self.is_local: |
| self.restore_database() |
|
|
| |
| db_manager = DatabaseManager() |
|
|
