Spaces:
Sleeping
Sleeping
| import os | |
| from typing import Optional | |
| import mysql.connector | |
| from mysql.connector import Error | |
| class DatabaseConfig: | |
| def __init__(self): | |
| self.host = os.getenv('DB_HOST', 'localhost') | |
| self.port = os.getenv('DB_PORT', 3306) | |
| self.database = os.getenv('DB_NAME', 'DOC_OC') | |
| self.user = os.getenv('DB_USER', 'root') | |
| self.password = os.getenv('DB_PASSWORD', 'aditya') | |
| def get_connection(self): | |
| try: | |
| connection = mysql.connector.connect( | |
| host=self.host, | |
| port=self.port, | |
| database=self.database, | |
| user=self.user, | |
| password=self.password | |
| ) | |
| return connection | |
| except Error as e: | |
| print(f"Database connection error: {e}") | |
| return None | |
| class MarksheetDB: | |
| def __init__(self): | |
| self.config = DatabaseConfig() | |
| def create_user(self, username: str, email: str) -> Optional[int]: | |
| connection = self.config.get_connection() | |
| if not connection: | |
| return None | |
| try: | |
| cursor = connection.cursor() | |
| query = "INSERT INTO users (username, email) VALUES (%s, %s)" | |
| cursor.execute(query, (username, email)) | |
| connection.commit() | |
| user_id = cursor.lastrowid | |
| return user_id | |
| except Error as e: | |
| print(f"Error creating user: {e}") | |
| return None | |
| finally: | |
| if connection.is_connected(): | |
| cursor.close() | |
| connection.close() | |
| def get_user_by_email(self, email: str) -> Optional[dict]: | |
| connection = self.config.get_connection() | |
| if not connection: | |
| return None | |
| try: | |
| cursor = connection.cursor(dictionary=True) | |
| query = "SELECT * FROM users WHERE email = %s" | |
| cursor.execute(query, (email,)) | |
| user = cursor.fetchone() | |
| return user | |
| except Error as e: | |
| print(f"Error getting user: {e}") | |
| return None | |
| finally: | |
| if connection.is_connected(): | |
| cursor.close() | |
| connection.close() | |
| def save_10th_marksheet(self, user_id: int, data: dict) -> Optional[int]: | |
| connection = self.config.get_connection() | |
| if not connection: | |
| return None | |
| try: | |
| cursor = connection.cursor() | |
| subjects = data.get('subjects', []) | |
| # Prepare subject data (up to 6 subjects) | |
| subject_data = {} | |
| for i in range(6): | |
| if i < len(subjects): | |
| subject = subjects[i] | |
| subject_data[f'subject{i+1}_code'] = subject.get('code') | |
| subject_data[f'subject{i+1}_name'] = subject.get('name') | |
| subject_data[f'subject{i+1}_theory_marks'] = subject.get('theory_marks') | |
| subject_data[f'subject{i+1}_practical_marks'] = subject.get('practical_marks') | |
| subject_data[f'subject{i+1}_internal_marks'] = subject.get('internal_marks') | |
| subject_data[f'subject{i+1}_total_marks'] = subject.get('total_marks') | |
| subject_data[f'subject{i+1}_marks'] = subject.get('marks') | |
| subject_data[f'subject{i+1}_grade'] = subject.get('grade') | |
| subject_data[f'subject{i+1}_percentage'] = subject.get('percentage') | |
| else: | |
| subject_data[f'subject{i+1}_code'] = None | |
| subject_data[f'subject{i+1}_name'] = None | |
| subject_data[f'subject{i+1}_theory_marks'] = None | |
| subject_data[f'subject{i+1}_practical_marks'] = None | |
| subject_data[f'subject{i+1}_internal_marks'] = None | |
| subject_data[f'subject{i+1}_total_marks'] = None | |
| subject_data[f'subject{i+1}_marks'] = None | |
| subject_data[f'subject{i+1}_grade'] = None | |
| subject_data[f'subject{i+1}_percentage'] = None | |
| query = """ | |
| INSERT INTO marksheet_10th | |
| (user_id, board, student_name, roll_number, unique_id, school_name, school_code, | |
| mother_name, father_name, subject1_code, subject1_name, subject1_theory_marks, | |
| subject1_practical_marks, subject1_internal_marks, subject1_total_marks, | |
| subject1_marks, subject1_grade, subject1_percentage, subject2_code, subject2_name, | |
| subject2_theory_marks, subject2_practical_marks, subject2_internal_marks, | |
| subject2_total_marks, subject2_marks, subject2_grade, subject2_percentage, | |
| subject3_code, subject3_name, subject3_theory_marks, subject3_practical_marks, | |
| subject3_internal_marks, subject3_total_marks, subject3_marks, subject3_grade, | |
| subject3_percentage, subject4_code, subject4_name, subject4_theory_marks, | |
| subject4_practical_marks, subject4_internal_marks, subject4_total_marks, | |
| subject4_marks, subject4_grade, subject4_percentage, subject5_code, subject5_name, | |
| subject5_theory_marks, subject5_practical_marks, subject5_internal_marks, | |
| subject5_total_marks, subject5_marks, subject5_grade, subject5_percentage, | |
| subject6_code, subject6_name, subject6_theory_marks, subject6_practical_marks, | |
| subject6_internal_marks, subject6_total_marks, subject6_marks, subject6_grade, | |
| subject6_percentage) | |
| VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, | |
| %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, | |
| %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, | |
| %s, %s, %s, %s, %s, %s, %s, %s, %s) | |
| ON DUPLICATE KEY UPDATE | |
| board = VALUES(board), student_name = VALUES(student_name), uploaded_at = CURRENT_TIMESTAMP | |
| """ | |
| cursor.execute(query, ( | |
| user_id, data.get('board', ''), data.get('student_name'), data.get('roll_number'), | |
| data.get('unique_id'), data.get('school_name'), data.get('school_code'), | |
| data.get('mother_name'), data.get('father_name'), | |
| *[subject_data[f'subject{i+1}_{field}'] for i in range(6) | |
| for field in ['code', 'name', 'theory_marks', 'practical_marks', 'internal_marks', | |
| 'total_marks', 'marks', 'grade', 'percentage']] | |
| )) | |
| connection.commit() | |
| return cursor.lastrowid or 1 | |
| except Error as e: | |
| print(f"Error saving 10th marksheet: {e}") | |
| return None | |
| finally: | |
| if connection.is_connected(): | |
| cursor.close() | |
| connection.close() | |
| def save_12th_marksheet(self, user_id: int, data: dict) -> Optional[int]: | |
| connection = self.config.get_connection() | |
| if not connection: | |
| return None | |
| try: | |
| cursor = connection.cursor() | |
| subjects = data.get('subjects', []) | |
| # Prepare subject data (up to 6 subjects) | |
| subject_data = {} | |
| for i in range(6): | |
| if i < len(subjects): | |
| subject = subjects[i] | |
| subject_data[f'subject{i+1}_code'] = subject.get('code') | |
| subject_data[f'subject{i+1}_name'] = subject.get('name') | |
| subject_data[f'subject{i+1}_theory_marks'] = subject.get('theory_marks') | |
| subject_data[f'subject{i+1}_practical_marks'] = subject.get('practical_marks') | |
| subject_data[f'subject{i+1}_internal_marks'] = subject.get('internal_marks') | |
| subject_data[f'subject{i+1}_total_marks'] = subject.get('total_marks') | |
| subject_data[f'subject{i+1}_marks'] = subject.get('marks') | |
| subject_data[f'subject{i+1}_grade'] = subject.get('grade') | |
| subject_data[f'subject{i+1}_percentage'] = subject.get('percentage') | |
| else: | |
| subject_data[f'subject{i+1}_code'] = None | |
| subject_data[f'subject{i+1}_name'] = None | |
| subject_data[f'subject{i+1}_theory_marks'] = None | |
| subject_data[f'subject{i+1}_practical_marks'] = None | |
| subject_data[f'subject{i+1}_internal_marks'] = None | |
| subject_data[f'subject{i+1}_total_marks'] = None | |
| subject_data[f'subject{i+1}_marks'] = None | |
| subject_data[f'subject{i+1}_grade'] = None | |
| subject_data[f'subject{i+1}_percentage'] = None | |
| query = """ | |
| INSERT INTO marksheet_12th | |
| (user_id, board, student_name, roll_number, unique_id, school_name, school_code, | |
| mother_name, father_name, subject1_code, subject1_name, subject1_theory_marks, | |
| subject1_practical_marks, subject1_internal_marks, subject1_total_marks, | |
| subject1_marks, subject1_grade, subject1_percentage, subject2_code, subject2_name, | |
| subject2_theory_marks, subject2_practical_marks, subject2_internal_marks, | |
| subject2_total_marks, subject2_marks, subject2_grade, subject2_percentage, | |
| subject3_code, subject3_name, subject3_theory_marks, subject3_practical_marks, | |
| subject3_internal_marks, subject3_total_marks, subject3_marks, subject3_grade, | |
| subject3_percentage, subject4_code, subject4_name, subject4_theory_marks, | |
| subject4_practical_marks, subject4_internal_marks, subject4_total_marks, | |
| subject4_marks, subject4_grade, subject4_percentage, subject5_code, subject5_name, | |
| subject5_theory_marks, subject5_practical_marks, subject5_internal_marks, | |
| subject5_total_marks, subject5_marks, subject5_grade, subject5_percentage, | |
| subject6_code, subject6_name, subject6_theory_marks, subject6_practical_marks, | |
| subject6_internal_marks, subject6_total_marks, subject6_marks, subject6_grade, | |
| subject6_percentage) | |
| VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, | |
| %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, | |
| %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, | |
| %s, %s, %s, %s, %s, %s, %s, %s, %s) | |
| ON DUPLICATE KEY UPDATE | |
| board = VALUES(board), student_name = VALUES(student_name), uploaded_at = CURRENT_TIMESTAMP | |
| """ | |
| cursor.execute(query, ( | |
| user_id, data.get('board', ''), data.get('student_name'), data.get('roll_number'), | |
| data.get('unique_id'), data.get('school_name'), data.get('school_code'), | |
| data.get('mother_name'), data.get('father_name'), | |
| *[subject_data[f'subject{i+1}_{field}'] for i in range(6) | |
| for field in ['code', 'name', 'theory_marks', 'practical_marks', 'internal_marks', | |
| 'total_marks', 'marks', 'grade', 'percentage']] | |
| )) | |
| connection.commit() | |
| return cursor.lastrowid or 1 | |
| except Error as e: | |
| print(f"Error saving 12th marksheet: {e}") | |
| return None | |
| finally: | |
| if connection.is_connected(): | |
| cursor.close() | |
| connection.close() | |
| def save_college_marksheet(self, user_id: int, semester: int, data: dict) -> Optional[int]: | |
| connection = self.config.get_connection() | |
| if not connection: | |
| return None | |
| try: | |
| cursor = connection.cursor() | |
| college_data = data.get('college', {}) | |
| student_data = data.get('student', {}) | |
| result_data = data.get('result', {}) | |
| subjects = data.get('subjects', []) | |
| # Prepare subject data (up to 10 subjects) | |
| subject_data = {} | |
| for i in range(10): | |
| if i < len(subjects): | |
| subject = subjects[i] | |
| subject_data[f'subject{i+1}_code'] = subject.get('code') | |
| subject_data[f'subject{i+1}_name'] = subject.get('name') | |
| subject_data[f'subject{i+1}_credits'] = subject.get('credits') | |
| subject_data[f'subject{i+1}_internal_marks'] = subject.get('internal_marks') | |
| subject_data[f'subject{i+1}_external_marks'] = subject.get('external_marks') | |
| subject_data[f'subject{i+1}_total_marks'] = subject.get('total') | |
| subject_data[f'subject{i+1}_grade'] = subject.get('grade') | |
| subject_data[f'subject{i+1}_grade_point'] = subject.get('grade_point') | |
| else: | |
| subject_data[f'subject{i+1}_code'] = None | |
| subject_data[f'subject{i+1}_name'] = None | |
| subject_data[f'subject{i+1}_credits'] = None | |
| subject_data[f'subject{i+1}_internal_marks'] = None | |
| subject_data[f'subject{i+1}_external_marks'] = None | |
| subject_data[f'subject{i+1}_total_marks'] = None | |
| subject_data[f'subject{i+1}_grade'] = None | |
| subject_data[f'subject{i+1}_grade_point'] = None | |
| # Build dynamic query | |
| subject_fields = [] | |
| subject_values = [] | |
| for i in range(10): | |
| for field in ['code', 'name', 'credits', 'internal_marks', 'external_marks', 'total_marks', 'grade', 'grade_point']: | |
| subject_fields.append(f'subject{i+1}_{field}') | |
| subject_values.append(subject_data[f'subject{i+1}_{field}']) | |
| query = f""" | |
| INSERT INTO college_marksheets | |
| (user_id, semester, course, session, student_name, roll_no, enrollment_no, | |
| father_name, sgpa, cgpa, total_credits_registered, total_credits_earned, status, | |
| {', '.join(subject_fields)}) | |
| VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, {', '.join(['%s'] * len(subject_values))}) | |
| ON DUPLICATE KEY UPDATE | |
| course = VALUES(course), student_name = VALUES(student_name), uploaded_at = CURRENT_TIMESTAMP | |
| """ | |
| cursor.execute(query, ( | |
| user_id, semester, college_data.get('course'), college_data.get('session'), | |
| student_data.get('name'), student_data.get('roll_no'), student_data.get('enrollment_no'), | |
| student_data.get('father_name'), result_data.get('sgpa'), result_data.get('cgpa'), | |
| result_data.get('total_credits_registered'), result_data.get('total_credits_earned'), | |
| result_data.get('status'), *subject_values | |
| )) | |
| connection.commit() | |
| return cursor.lastrowid or 1 | |
| except Error as e: | |
| print(f"Error saving college marksheet: {e}") | |
| return None | |
| finally: | |
| if connection.is_connected(): | |
| cursor.close() | |
| connection.close() | |
| def get_user_marksheets(self, user_id: int) -> dict: | |
| connection = self.config.get_connection() | |
| if not connection: | |
| return {} | |
| try: | |
| cursor = connection.cursor(dictionary=True) | |
| # Get 10th marksheet | |
| cursor.execute("SELECT * FROM marksheet_10th WHERE user_id = %s", (user_id,)) | |
| marksheet_10th = cursor.fetchone() | |
| # Get 12th marksheet | |
| cursor.execute("SELECT * FROM marksheet_12th WHERE user_id = %s", (user_id,)) | |
| marksheet_12th = cursor.fetchone() | |
| # Get college marksheets | |
| cursor.execute("SELECT * FROM college_marksheets WHERE user_id = %s ORDER BY semester", (user_id,)) | |
| college_marksheets = cursor.fetchall() | |
| return { | |
| 'marksheet_10th': marksheet_10th, | |
| 'marksheet_12th': marksheet_12th, | |
| 'college_marksheets': college_marksheets | |
| } | |
| except Error as e: | |
| print(f"Error getting marksheets: {e}") | |
| return {} | |
| finally: | |
| if connection.is_connected(): | |
| cursor.close() | |
| connection.close() |