FaizanMirZa77/FormatX
0
1import os2import psycopg23import psycopg2.extras4import models5from pwd_hasher import verify_password6from jwt_generator import generate_jwt_token7from dotenv import load_dotenv8 9load_dotenv()10 11DATABASE_URL = os.getenv("DATABASE_URL")12 13def get_connection():14 return psycopg2.connect(DATABASE_URL, cursor_factory=psycopg2.extras.RealDictCursor)15 16def init_db():17 """Create all tables if they don't exist."""18 with get_connection() as conn:19 with conn.cursor() as cur:20 cur.execute("""21 CREATE TABLE IF NOT EXISTS users (22 id SERIAL PRIMARY KEY,23 username VARCHAR(100) NOT NULL,24 email VARCHAR(255) UNIQUE NOT NULL,25 password TEXT NOT NULL,26 is_active BOOLEAN DEFAULT FALSE,27 profile_picture_url TEXT DEFAULT NULL28 );29 """)30 # Migrate existing tables that may not have the column yet31 cur.execute("""32 ALTER TABLE users33 ADD COLUMN IF NOT EXISTS profile_picture_url TEXT DEFAULT NULL;34 """)35 cur.execute("""36 CREATE TABLE IF NOT EXISTS format_jobs (37 id SERIAL PRIMARY KEY,38 job_id VARCHAR(255) UNIQUE NOT NULL,39 user_email VARCHAR(255) NOT NULL,40 download_url TEXT NOT NULL,41 output_format VARCHAR(10) NOT NULL,42 job_type VARCHAR(50) NOT NULL,43 raw_document_name VARCHAR(255),44 created_at TIMESTAMPTZ DEFAULT NOW()45 );46 """)47 conn.commit()48 49init_db()50 51 52def login_user(loginDto: models.UserDto):53 try:54 with get_connection() as conn:55 with conn.cursor() as cur:56 cur.execute("SELECT * FROM users WHERE email = %s", (loginDto.email,))57 user = cur.fetchone()58 if user and user["is_active"] and verify_password(loginDto.password, user["password"]):59 userJwtDto = models.UserJwtPayloadDto(60 email=user["email"],61 username=user["username"]62 )63 return generate_jwt_token(userJwtDto)64 return None65 except Exception as e:66 print(f"Login error: {e}")67 return None68 69 70def register_user(userDto: models.RegisterUserDto):71 try:72 with get_connection() as conn:73 with conn.cursor() as cur:74 cur.execute("""75 INSERT INTO users (username, email, password, is_active)76 VALUES (%s, %s, %s, FALSE)77 """, (userDto.username, userDto.email, userDto.password))78 conn.commit()79 return True80 except Exception as e:81 print(f"Failed to insert data: {e}")82 return False83 84 85def is_email_registered(email: str) -> bool:86 try:87 with get_connection() as conn:88 with conn.cursor() as cur:89 cur.execute("SELECT * FROM users WHERE email = %s", (email,))90 user = cur.fetchone()91 if user and user["is_active"]:92 return True93 elif user:94 # Inactive user — delete and treat as unregistered95 cur.execute("DELETE FROM users WHERE email = %s", (email,))96 conn.commit()97 return False98 return False99 except Exception as e:100 print(f"Error checking email existence: {e}")101 return False102 103 104def activate_user(email: str):105 try:106 with get_connection() as conn:107 with conn.cursor() as cur:108 cur.execute("""109 UPDATE users SET is_active = TRUE110 WHERE email = %s111 RETURNING *112 """, (email,))113 result = cur.fetchone()114 conn.commit()115 if result:116 print(f"User {email} has been activated.")117 return result118 else:119 print(f"No user found with email: {email}")120 return None121 except Exception as e:122 print(f"Error activating user: {e}")123 return None124 125 126def get_user_by_email(email: str):127 try:128 with get_connection() as conn:129 with conn.cursor() as cur:130 cur.execute("SELECT * FROM users WHERE email = %s", (email,))131 result = cur.fetchone()132 if result:133 return result134 print(f"No user found with email: {email}")135 return None136 except Exception as e:137 print(f"Error fetching user: {e}")138 return None139 140 141def update_profile_picture(email: str, picture_url: str) -> bool:142 try:143 with get_connection() as conn:144 with conn.cursor() as cur:145 cur.execute("""146 UPDATE users SET profile_picture_url = %s WHERE email = %s147 """, (picture_url, email))148 conn.commit()149 return True150 except Exception as e:151 print(f"Error updating profile picture: {e}")152 return False153 154 155def get_user_profile_stats(email: str) -> dict:156 """Aggregate all activity stats for the profile screen."""157 stats = {158 "total_documents_formatted": 0,159 "total_downloads": 0,160 "plagiarism_checks": 0,161 "grammar_checks": 0,162 "citation_docs": 0,163 "tests_taken": 0,164 "last_activity": None,165 "profile_picture_url": None,166 }167 try:168 with get_connection() as conn:169 with conn.cursor() as cur:170 # Profile picture171 cur.execute("SELECT profile_picture_url FROM users WHERE email = %s", (email,))172 user_row = cur.fetchone()173 if user_row:174 stats["profile_picture_url"] = user_row.get("profile_picture_url")175 176 # Total formatted documents177 cur.execute(178 "SELECT COUNT(*) FROM format_jobs WHERE user_email = %s", (email,)179 )180 stats["total_documents_formatted"] = cur.fetchone()["count"]181 182 # Downloads = formatted docs (each job is downloadable)183 stats["total_downloads"] = stats["total_documents_formatted"]184 185 # Plagiarism checks186 cur.execute(187 "SELECT COUNT(*) FROM plagiarism_history WHERE user_email = %s", (email,)188 )189 stats["plagiarism_checks"] = cur.fetchone()["count"]190 191 # Grammar checks192 cur.execute(193 "SELECT COUNT(*) FROM grammar_history WHERE user_email = %s", (email,)194 )195 stats["grammar_checks"] = cur.fetchone()["count"]196 197 # Citation documents198 cur.execute(199 "SELECT COUNT(*) FROM citation_history WHERE user_email = %s", (email,)200 )201 stats["citation_docs"] = cur.fetchone()["count"]202 203 # Tests taken204 cur.execute(205 "SELECT COUNT(*) FROM test_results WHERE user_email = %s", (email,)206 )207 stats["tests_taken"] = cur.fetchone()["count"]208 209 # Last activity (most recent across all tables)210 cur.execute("""211 SELECT MAX(ts) AS last_activity FROM (212 SELECT MAX(created_at) AS ts FROM format_jobs WHERE user_email = %s213 UNION ALL214 SELECT MAX(created_at) AS ts FROM plagiarism_history WHERE user_email = %s215 UNION ALL216 SELECT MAX(checked_at) AS ts FROM grammar_history WHERE user_email = %s217 UNION ALL218 SELECT MAX(created_at) AS ts FROM citation_history WHERE user_email = %s219 UNION ALL220 SELECT MAX(taken_at) AS ts FROM test_results WHERE user_email = %s221 ) sub222 """, (email, email, email, email, email))223 row = cur.fetchone()224 if row and row["last_activity"]:225 stats["last_activity"] = row["last_activity"].isoformat()226 except Exception as e:227 print(f"Error fetching profile stats: {e}")228 return stats229 230 231def reset_user_password(inputDto: models.ResetPasswordDto):232 try:233 with get_connection() as conn:234 with conn.cursor() as cur:235 cur.execute("""236 UPDATE users SET password = %s WHERE email = %s237 RETURNING *238 """, (inputDto.password, inputDto.email))239 result = cur.fetchone()240 conn.commit()241 if result:242 return result243 else:244 print(f"No user found with email: {inputDto.email}")245 return None246 except Exception as e:247 print(f"Error resetting password: {e}")248 return None249 250def init_format_jobs_table():251 """Create format_jobs table if it doesn't exist."""252 with get_connection() as conn:253 with conn.cursor() as cur:254 cur.execute("""255 CREATE TABLE IF NOT EXISTS format_jobs (256 id SERIAL PRIMARY KEY,257 job_id VARCHAR(255) UNIQUE NOT NULL,258 user_email VARCHAR(255) NOT NULL,259 download_url TEXT NOT NULL,260 output_format VARCHAR(10) NOT NULL,261 job_type VARCHAR(50) NOT NULL,262 raw_document_name VARCHAR(255),263 created_at TIMESTAMPTZ DEFAULT NOW()264 );265 """)266 cur.execute("""267 CREATE TABLE IF NOT EXISTS grammar_history (268 id SERIAL PRIMARY KEY,269 document_id VARCHAR(255) UNIQUE NOT NULL,270 user_email VARCHAR(255) NOT NULL,271 original_name VARCHAR(255) NOT NULL,272 size_kb NUMERIC(10,2) NOT NULL,273 corrected_url TEXT,274 grammar_count INTEGER,275 spelling_count INTEGER,276 punctuation_count INTEGER,277 flesch_score NUMERIC(5,1),278 grade_level INTEGER,279 checked_at TIMESTAMPTZ DEFAULT NOW()280 );281 """)282 cur.execute("""283 CREATE TABLE IF NOT EXISTS questions (284 id SERIAL PRIMARY KEY,285 subject VARCHAR(50) NOT NULL,286 difficulty VARCHAR(10) NOT NULL,287 question_type VARCHAR(10) NOT NULL,288 question_text TEXT NOT NULL,289 option_a TEXT,290 option_b TEXT,291 option_c TEXT,292 option_d TEXT,293 correct_answer TEXT NOT NULL,294 keywords TEXT,295 marks INTEGER NOT NULL DEFAULT 1296 );297 """)298 cur.execute("""299 CREATE TABLE IF NOT EXISTS test_results (300 id SERIAL PRIMARY KEY,301 result_id VARCHAR(255) UNIQUE NOT NULL,302 user_email VARCHAR(255) NOT NULL,303 subject VARCHAR(50) NOT NULL,304 difficulty VARCHAR(10) NOT NULL,305 total_marks INTEGER NOT NULL,306 obtained_marks NUMERIC(5,2) NOT NULL,307 percentage NUMERIC(5,2) NOT NULL,308 grade VARCHAR(2) NOT NULL,309 total_time_seconds INTEGER NOT NULL DEFAULT 0,310 taken_at TIMESTAMPTZ DEFAULT NOW()311 );312 """)313 # Migrate existing tables that may not have the column yet314 # Migrate existing grammar_history tables that may not have the new columns315 for col, definition in [316 ("corrected_url", "TEXT"),317 ("grammar_count", "INTEGER"),318 ("spelling_count", "INTEGER"),319 ("punctuation_count", "INTEGER"),320 ("flesch_score", "NUMERIC(5,1)"),321 ("grade_level", "INTEGER"),322 ]:323 cur.execute(f"""324 ALTER TABLE grammar_history325 ADD COLUMN IF NOT EXISTS {col} {definition};326 """)327 cur.execute("""328 ALTER TABLE test_results329 ADD COLUMN IF NOT EXISTS total_time_seconds INTEGER NOT NULL DEFAULT 0;330 """)331 cur.execute("""332 CREATE TABLE IF NOT EXISTS citation_history (333 id SERIAL PRIMARY KEY,334 citation_id VARCHAR(255) UNIQUE NOT NULL,335 user_email VARCHAR(255) NOT NULL,336 original_name VARCHAR(255) NOT NULL,337 citation_style VARCHAR(20) NOT NULL,338 citation_count INTEGER NOT NULL DEFAULT 0,339 download_url TEXT NOT NULL,340 created_at TIMESTAMPTZ DEFAULT NOW()341 );342 """)343 cur.execute("""344 CREATE TABLE IF NOT EXISTS plagiarism_history (345 id SERIAL PRIMARY KEY,346 report_id VARCHAR(255) UNIQUE NOT NULL,347 user_email VARCHAR(255) NOT NULL,348 original_name VARCHAR(255) NOT NULL,349 ai_percentage NUMERIC(5,2) NOT NULL DEFAULT 0,350 plagiarism_percent NUMERIC(5,2) NOT NULL DEFAULT 0,351 unique_percentage NUMERIC(5,2) NOT NULL DEFAULT 0,352 matched_sources INTEGER NOT NULL DEFAULT 0,353 download_url TEXT NOT NULL,354 created_at TIMESTAMPTZ DEFAULT NOW()355 );356 """)357 conn.commit()358 359def save_format_job(job_id: str, user_email: str, download_url: str, output_format: str, job_type: str, raw_document_name: str):360 try:361 with get_connection() as conn:362 with conn.cursor() as cur:363 cur.execute("""364 INSERT INTO format_jobs (job_id, user_email, download_url, output_format, job_type, raw_document_name)365 VALUES (%s, %s, %s, %s, %s, %s)366 """, (job_id, user_email, download_url, output_format, job_type, raw_document_name))367 conn.commit()368 return True369 except Exception as e:370 print(f"Failed to save job record: {e}")371 return False372 373 374def get_formatted_docs_history(user_email: str) -> list[dict]:375 try:376 with get_connection() as conn:377 with conn.cursor() as cur:378 cur.execute("""379 SELECT job_id, download_url, output_format, raw_document_name, created_at380 FROM format_jobs381 WHERE user_email = %s382 ORDER BY created_at DESC383 """, (user_email,))384 rows = cur.fetchall()385 return [dict(row) for row in rows]386 except Exception as e:387 print(f"Error fetching formatted docs history: {e}")388 return []389 390 391def get_last_formatted_time(user_email: str) -> str | None:392 """393 Returns the created_at timestamp of the user's most recent format job,394 or None if they have never formatted a document.395 """396 try:397 with get_connection() as conn:398 with conn.cursor() as cur:399 cur.execute("""400 SELECT created_at FROM format_jobs401 WHERE user_email = %s402 ORDER BY created_at DESC403 LIMIT 1404 """, (user_email,))405 row = cur.fetchone()406 return row["created_at"] if row else None407 except Exception as e:408 print(f"Error fetching last formatted time: {e}")409 return None410 411init_format_jobs_table()412 413 414def save_grammar_history(document_id: str, user_email: str, original_name: str, size_kb: float):415 try:416 with get_connection() as conn:417 with conn.cursor() as cur:418 cur.execute("""419 INSERT INTO grammar_history (document_id, user_email, original_name, size_kb)420 VALUES (%s, %s, %s, %s)421 ON CONFLICT (document_id) DO NOTHING422 """, (document_id, user_email, original_name, round(size_kb, 2)))423 conn.commit()424 return True425 except Exception as e:426 print(f"Failed to save grammar history: {e}")427 return False428 429 430def get_grammar_history(user_email: str, limit: int = 10) -> list[dict]:431 try:432 with get_connection() as conn:433 with conn.cursor() as cur:434 cur.execute("""435 SELECT document_id, original_name, size_kb, checked_at,436 corrected_url, grammar_count, spelling_count,437 punctuation_count, flesch_score, grade_level438 FROM grammar_history439 WHERE user_email = %s440 ORDER BY checked_at DESC441 LIMIT %s442 """, (user_email, limit))443 rows = cur.fetchall()444 return [dict(row) for row in rows]445 except Exception as e:446 print(f"Error fetching grammar history: {e}")447 return []448 449 450def update_grammar_summary(451 document_id: str,452 corrected_url: str,453 grammar_count: int,454 spelling_count: int,455 punctuation_count: int,456 flesch_score: float,457 grade_level: int,458) -> bool:459 try:460 with get_connection() as conn:461 with conn.cursor() as cur:462 cur.execute("""463 UPDATE grammar_history464 SET corrected_url = %s,465 grammar_count = %s,466 spelling_count = %s,467 punctuation_count = %s,468 flesch_score = %s,469 grade_level = %s470 WHERE document_id = %s471 """, (corrected_url, grammar_count, spelling_count,472 punctuation_count, round(flesch_score, 1),473 grade_level, document_id))474 conn.commit()475 return True476 except Exception as e:477 print(f"Failed to update grammar summary: {e}")478 return False479 480 481# ── Test Evaluation ───────────────────────────────────────────────────────────482 483def get_questions(subject: str, difficulty: str, limit: int = 10) -> list[dict]:484 try:485 with get_connection() as conn:486 with conn.cursor() as cur:487 cur.execute("""488 SELECT id, subject, difficulty, question_type,489 question_text, option_a, option_b, option_c, option_d,490 correct_answer, keywords, marks491 FROM questions492 WHERE subject = %s AND difficulty = %s493 ORDER BY RANDOM()494 LIMIT %s495 """, (subject, difficulty, limit))496 return [dict(row) for row in cur.fetchall()]497 except Exception as e:498 print(f"Error fetching questions: {e}")499 return []500 501 502def get_questions_by_ids(ids: list[int]) -> list[dict]:503 if not ids:504 return []505 try:506 with get_connection() as conn:507 with conn.cursor() as cur:508 cur.execute("""509 SELECT id, subject, difficulty, question_type,510 question_text, option_a, option_b, option_c, option_d,511 correct_answer, keywords, marks512 FROM questions513 WHERE id = ANY(%s)514 """, (ids,))515 rows = {row["id"]: dict(row) for row in cur.fetchall()}516 return [rows[i] for i in ids if i in rows]517 except Exception as e:518 print(f"Error fetching questions by ids: {e}")519 return []520 521 522def insert_question(subject: str, difficulty: str, question_type: str,523 question_text: str, correct_answer: str, keywords: str,524 marks: int, option_a: str = None, option_b: str = None,525 option_c: str = None, option_d: str = None) -> bool:526 try:527 with get_connection() as conn:528 with conn.cursor() as cur:529 cur.execute("""530 INSERT INTO questions531 (subject, difficulty, question_type, question_text,532 option_a, option_b, option_c, option_d,533 correct_answer, keywords, marks)534 VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)535 """, (subject, difficulty, question_type, question_text,536 option_a, option_b, option_c, option_d,537 correct_answer, keywords, marks))538 conn.commit()539 return True540 except Exception as e:541 print(f"Error inserting question: {e}")542 return False543 544 545def question_bank_count(subject: str, difficulty: str) -> int:546 try:547 with get_connection() as conn:548 with conn.cursor() as cur:549 cur.execute(550 "SELECT COUNT(*) FROM questions WHERE subject=%s AND difficulty=%s",551 (subject, difficulty)552 )553 return cur.fetchone()["count"]554 except Exception:555 return 0556 557 558def save_test_result(result_id: str, user_email: str, subject: str,559 difficulty: str, total_marks: int, obtained_marks: float,560 percentage: float, grade: str,561 total_time_seconds: int = 0) -> bool:562 try:563 with get_connection() as conn:564 with conn.cursor() as cur:565 cur.execute("""566 INSERT INTO test_results567 (result_id, user_email, subject, difficulty,568 total_marks, obtained_marks, percentage, grade,569 total_time_seconds)570 VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)571 """, (result_id, user_email, subject, difficulty,572 total_marks, obtained_marks, percentage, grade,573 max(0, total_time_seconds)))574 conn.commit()575 return True576 except Exception as e:577 print(f"Error saving test result: {e}")578 return False579 580 581def get_test_history(user_email: str, limit: int = 20) -> list[dict]:582 try:583 with get_connection() as conn:584 with conn.cursor() as cur:585 cur.execute("""586 SELECT result_id, subject, difficulty, total_marks,587 obtained_marks, percentage, grade,588 total_time_seconds, taken_at589 FROM test_results590 WHERE user_email = %s591 ORDER BY taken_at DESC592 LIMIT %s593 """, (user_email, limit))594 return [dict(row) for row in cur.fetchall()]595 except Exception as e:596 print(f"Error fetching test history: {e}")597 return []598 599 600# ── Citation ──────────────────────────────────────────────────────────────────601 602def save_citation_history(603 citation_id: str,604 user_email: str,605 original_name: str,606 citation_style: str,607 citation_count: int,608 download_url: str,609) -> bool:610 try:611 with get_connection() as conn:612 with conn.cursor() as cur:613 cur.execute("""614 INSERT INTO citation_history615 (citation_id, user_email, original_name,616 citation_style, citation_count, download_url)617 VALUES (%s, %s, %s, %s, %s, %s)618 ON CONFLICT (citation_id) DO NOTHING619 """, (citation_id, user_email, original_name,620 citation_style, citation_count, download_url))621 conn.commit()622 return True623 except Exception as e:624 print(f"Failed to save citation history: {e}")625 return False626 627 628def get_citation_history(user_email: str, limit: int = 10) -> list[dict]:629 try:630 with get_connection() as conn:631 with conn.cursor() as cur:632 cur.execute("""633 SELECT citation_id, original_name, citation_style,634 citation_count, download_url, created_at635 FROM citation_history636 WHERE user_email = %s637 ORDER BY created_at DESC638 LIMIT %s639 """, (user_email, limit))640 return [dict(row) for row in cur.fetchall()]641 except Exception as e:642 print(f"Error fetching citation history: {e}")643 return []644 645 646# ── Plagiarism History ────────────────────────────────────────────────────────647 648def save_plagiarism_history(649 report_id: str,650 user_email: str,651 original_name: str,652 ai_percentage: float,653 plagiarism_percent: float,654 unique_percentage: float,655 matched_sources: int,656 download_url: str = "",657) -> bool:658 try:659 with get_connection() as conn:660 with conn.cursor() as cur:661 cur.execute("""662 INSERT INTO plagiarism_history663 (report_id, user_email, original_name,664 ai_percentage, plagiarism_percent, unique_percentage,665 matched_sources, download_url)666 VALUES (%s, %s, %s, %s, %s, %s, %s, %s)667 ON CONFLICT (report_id) DO NOTHING668 """, (report_id, user_email, original_name,669 round(ai_percentage, 2), round(plagiarism_percent, 2),670 round(unique_percentage, 2), matched_sources, download_url))671 conn.commit()672 return True673 except Exception as e:674 print(f"Failed to save plagiarism history: {e}")675 return False676 677 678def update_plagiarism_report_url(report_id: str, download_url: str) -> bool:679 try:680 with get_connection() as conn:681 with conn.cursor() as cur:682 cur.execute("""683 UPDATE plagiarism_history684 SET download_url = %s685 WHERE report_id = %s686 """, (download_url, report_id))687 conn.commit()688 return True689 except Exception as e:690 print(f"Failed to update plagiarism report URL: {e}")691 return False692 693 694def get_plagiarism_history(user_email: str, limit: int = 10) -> list[dict]:695 try:696 with get_connection() as conn:697 with conn.cursor() as cur:698 cur.execute("""699 SELECT report_id, original_name,700 ai_percentage, plagiarism_percent, unique_percentage,701 matched_sources, download_url, created_at702 FROM plagiarism_history703 WHERE user_email = %s704 ORDER BY created_at DESC705 LIMIT %s706 """, (user_email, limit))707 return [dict(row) for row in cur.fetchall()]708 except Exception as e:709 print(f"Error fetching plagiarism history: {e}")710 return []711 712 713# ── Clear History ─────────────────────────────────────────────────────────────714 715def clear_formatted_docs_history(user_email: str) -> bool:716 try:717 with get_connection() as conn:718 with conn.cursor() as cur:719 cur.execute(720 "DELETE FROM format_jobs WHERE user_email = %s",721 (user_email,)722 )723 conn.commit()724 return True725 except Exception as e:726 print(f"Error clearing formatted docs history: {e}")727 return False728 729 730def clear_grammar_history(user_email: str) -> bool:731 try:732 with get_connection() as conn:733 with conn.cursor() as cur:734 cur.execute(735 "DELETE FROM grammar_history WHERE user_email = %s",736 (user_email,)737 )738 conn.commit()739 return True740 except Exception as e:741 print(f"Error clearing grammar history: {e}")742 return False743 744 745def clear_plagiarism_history(user_email: str) -> bool:746 try:747 with get_connection() as conn:748 with conn.cursor() as cur:749 cur.execute(750 "DELETE FROM plagiarism_history WHERE user_email = %s",751 (user_email,)752 )753 conn.commit()754 return True755 except Exception as e:756 print(f"Error clearing plagiarism history: {e}")757 return False758 759 760def clear_citation_history(user_email: str) -> bool:761 try:762 with get_connection() as conn:763 with conn.cursor() as cur:764 cur.execute(765 "DELETE FROM citation_history WHERE user_email = %s",766 (user_email,)767 )768 conn.commit()769 return True770 except Exception as e:771 print(f"Error clearing citation history: {e}")772 return False773 774 775def clear_test_history(user_email: str) -> bool:776 try:777 with get_connection() as conn:778 with conn.cursor() as cur:779 cur.execute(780 "DELETE FROM test_results WHERE user_email = %s",781 (user_email,)782 )783 conn.commit()784 return True785 except Exception as e:786 print(f"Error clearing test history: {e}")787 return False788