diff --git a/.gitignore b/.gitignore
index ba8194a..83bfc80 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,4 +3,6 @@ bin
frontend/dist
frontend/node_modules
build/linux/appimage/build
-build/windows/nsis/MicrosoftEdgeWebview2Setup.exe
\ No newline at end of file
+build/windows/nsis/MicrosoftEdgeWebview2Setup.exe
+config.json
+
diff --git a/check_all_dbs.py b/check_all_dbs.py
new file mode 100644
index 0000000..eacf254
--- /dev/null
+++ b/check_all_dbs.py
@@ -0,0 +1,19 @@
+import sqlite3
+import os
+
+for root, dirs, files in os.walk(r"./"):
+ for f in files:
+ if f.endswith(".db") or ".db" in f:
+ path = os.path.join(root, f)
+ try:
+ conn = sqlite3.connect(path)
+ c = conn.cursor()
+ c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='QuizQuestions'")
+ if c.fetchone():
+ c.execute("SELECT COUNT(*) FROM QuizQuestions WHERE category = ?", ("파생상품투자권유자문인력",))
+ count = c.fetchone()[0]
+ if count > 0:
+ print(f"File: {os.path.relpath(path, r'./')} | Count: {count}")
+ conn.close()
+ except Exception as e:
+ pass
diff --git a/check_bak_db.py b/check_bak_db.py
new file mode 100644
index 0000000..f551c97
--- /dev/null
+++ b/check_bak_db.py
@@ -0,0 +1,17 @@
+import sqlite3
+import sys
+
+def check_db(path):
+ print(f"Checking: {path}")
+ try:
+ conn = sqlite3.connect(path)
+ c = conn.cursor()
+ c.execute("SELECT COUNT(*) FROM QuizQuestions WHERE category = ?", ("파생상품투자권유자문인력",))
+ count = c.fetchone()[0]
+ print(f" Count of '파생상품투자권유자문인력': {count}")
+ conn.close()
+ except Exception as e:
+ print(f" Error: {e}")
+
+check_db(r"./qple_quiz.db")
+check_db(r"./qple_quiz.db.bak_before_restore")
diff --git a/check_table_data.py b/check_table_data.py
new file mode 100644
index 0000000..302cfe2
--- /dev/null
+++ b/check_table_data.py
@@ -0,0 +1,75 @@
+import sqlite3
+import os
+
+db_files = [
+ r'./qple_quiz.db',
+ r'./qple_quiz_corrupted.db.bak',
+ r'./qple_quiz_0621.db'
+]
+
+for db in db_files:
+ if not os.path.exists(db):
+ print(f"File not found: {db}")
+ continue
+ print(f"\nDB: {db} (Size: {os.path.getsize(db)} bytes)")
+ try:
+ conn = sqlite3.connect(db)
+ cursor = conn.cursor()
+
+ # list tables
+ cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
+ tables = [row[0] for row in cursor.fetchall()]
+ print(" Tables:", tables)
+
+ for t in ['QuizQuestions', 'QuizExplain', 'Bookmark', 'DescriptiveGrading']:
+ if t in tables:
+ cursor.execute(f"SELECT COUNT(*) FROM {t};")
+ cnt = cursor.fetchone()[0]
+ print(f" {t}: {cnt} rows")
+
+ # if QuizQuestions, list source counts
+ if t == 'QuizQuestions':
+ cursor.execute("PRAGMA table_info(QuizQuestions);")
+ cols = [c[1] for c in cursor.fetchall()]
+ print(f" Columns: {cols}")
+
+ cursor.execute("SELECT source, COUNT(*), MIN(id), MAX(id) FROM QuizQuestions GROUP BY source;")
+ src_cnt = cursor.fetchall()
+ print(f" QuizQuestions breakdown: {src_cnt}")
+ elif t == 'QuizExplain':
+ # check if source column exists
+ cursor.execute("PRAGMA table_info(QuizExplain);")
+ cols = [c[1] for c in cursor.fetchall()]
+ print(f" Columns: {cols}")
+
+ if 'source' in cols:
+ cursor.execute("SELECT source, COUNT(*), MIN(id), MAX(id) FROM QuizExplain GROUP BY source;")
+ print(f" QuizExplain breakdown: {cursor.fetchall()}")
+ else:
+ cursor.execute("SELECT COUNT(*), MIN(id), MAX(id) FROM QuizExplain;")
+ print(f" QuizExplain (no source): {cursor.fetchall()}")
+ elif t == 'Bookmark':
+ cursor.execute("PRAGMA table_info(Bookmark);")
+ cols = [c[1] for c in cursor.fetchall()]
+ print(f" Columns: {cols}")
+
+ if 'source' in cols:
+ cursor.execute("SELECT source, COUNT(*), MIN(question_id), MAX(question_id) FROM Bookmark GROUP BY source;")
+ print(f" Bookmark breakdown: {cursor.fetchall()}")
+ else:
+ cursor.execute("SELECT COUNT(*), MIN(question_id), MAX(question_id) FROM Bookmark;")
+ print(f" Bookmark (no source): {cursor.fetchall()}")
+ elif t == 'DescriptiveGrading':
+ cursor.execute("PRAGMA table_info(DescriptiveGrading);")
+ cols = [c[1] for c in cursor.fetchall()]
+ print(f" Columns: {cols}")
+
+ if 'source' in cols:
+ cursor.execute("SELECT source, COUNT(*) FROM DescriptiveGrading GROUP BY source;")
+ print(f" DescriptiveGrading breakdown: {cursor.fetchall()}")
+ else:
+ cursor.execute("SELECT COUNT(*) FROM DescriptiveGrading;")
+ print(f" DescriptiveGrading (no source): {cursor.fetchall()}")
+ conn.close()
+ except Exception as e:
+ print(f" Error inspecting: {e}")
diff --git a/create_quiz_db.py b/create_quiz_db.py
new file mode 100644
index 0000000..9d7dde5
--- /dev/null
+++ b/create_quiz_db.py
@@ -0,0 +1,96 @@
+import sqlite3
+import json
+
+def create_db():
+ # Connect to the database (or create it if it doesn't exist)
+ conn = sqlite3.connect('qple_quiz.db')
+ cursor = conn.cursor()
+
+ # Create the QuizQuestions table
+ # Based on inferred fields: Category, Question, Choices (MultipleChoiceList), Answer, Explanation
+ cursor.execute('''
+ CREATE TABLE IF NOT EXISTS QuizQuestions (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ category TEXT,
+ question_text TEXT,
+ choice1 TEXT,
+ choice2 TEXT,
+ choice3 TEXT,
+ choice4 TEXT,
+ correct_answer INTEGER,
+ explanation TEXT,
+ difficulty TEXT,
+ subject TEXT DEFAULT '',
+ passage TEXT,
+ image_url TEXT,
+ correct_answer_str TEXT,
+ raw_json TEXT
+ )
+ ''')
+
+ # Sample data reflecting the app's themes (Financial, Job, General Certificate)
+ quiz_data = [
+ (
+ '금융(Financial)',
+ '다음 중 유동자산에 해당하지 않는 것은?',
+ '현금', '외상매출금', '단기대여금', '토지',
+ 4,
+ '토지는 비유동자산(유형자산)에 해당합니다.',
+ 'Normal'
+ ),
+ (
+ '직무(Job)',
+ '프로젝트 관리에서 SWOT 분석의 요소가 아닌 것은?',
+ '강점(Strength)', '기회(Opportunity)', '위협(Threat)', '수익(Profit)',
+ 4,
+ 'SWOT 분석은 강점(Strength), 약점(Weakness), 기회(Opportunity), 위협(Threat)의 약자입니다.',
+ 'Easy'
+ ),
+ (
+ '자격증(Certificate)',
+ '정보처리기사: OSI 7계층 중 물리적 전송 매체와 관련된 계층은?',
+ '물리 계층', '데이터 링크 계층', '네트워크 계층', '전송 계층',
+ 1,
+ '물리 계층(Physical Layer)은 실제 장치와 전송 매체 간의 물리적인 연결을 담당합니다.',
+ 'Hard'
+ ),
+ (
+ '금융(Financial)',
+ '주식 시장에서 KOSPI 지수의 기준 시점은?',
+ '1980년 1월 4일', '1990년 1월 4일', '2000년 1월 4일', '1970년 1월 4일',
+ 1,
+ 'KOSPI 지수는 1980년 1월 4일의 시가총액을 100으로 설정하여 계산합니다.',
+ 'Normal'
+ )
+ ]
+
+ # Insert sample data
+ cursor.executemany('''
+ INSERT INTO QuizQuestions (category, question_text, choice1, choice2, choice3, choice4, correct_answer, explanation, difficulty)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+ ''', quiz_data)
+
+ # Create a table for App Configuration as well
+ cursor.execute('''
+ CREATE TABLE IF NOT EXISTS AppConfig (
+ key TEXT PRIMARY KEY,
+ value TEXT
+ )
+ ''')
+
+ config_data = {
+ "serverUrl": "https://qple-api-prod.qpleapp.com/",
+ "googlePacketVersion": "1.5.1",
+ "applePacketVersion": "1.5.1"
+ }
+
+ for key, value in config_data.items():
+ cursor.execute('INSERT OR REPLACE INTO AppConfig (key, value) VALUES (?, ?)', (key, value))
+
+ # Save (commit) the changes and close the connection
+ conn.commit()
+ conn.close()
+ print("Database 'qple_quiz.db' created and populated successfully.")
+
+if __name__ == "__main__":
+ create_db()
diff --git a/inspect_db.py b/inspect_db.py
new file mode 100644
index 0000000..62911e2
--- /dev/null
+++ b/inspect_db.py
@@ -0,0 +1,11 @@
+import sqlite3
+import json
+
+db_file = r'./qple_quiz.db'
+conn = sqlite3.connect(db_file)
+conn.row_factory = sqlite3.Row
+cursor = conn.cursor()
+cursor.execute("SELECT question_text, choice1, choice2, choice3, choice4, correct_answer_str, raw_json FROM QuizQuestions WHERE correct_answer_str = '' OR length(correct_answer_str) <= 1 LIMIT 5;")
+rows = [dict(row) for row in cursor.fetchall()]
+print(json.dumps(rows, ensure_ascii=False, indent=2))
+conn.close()
diff --git a/scrape_cbtbank.py b/scrape_cbtbank.py
new file mode 100644
index 0000000..e2255ea
--- /dev/null
+++ b/scrape_cbtbank.py
@@ -0,0 +1,405 @@
+import os
+import sys
+import time
+import json
+import sqlite3
+import re
+import argparse
+import hashlib
+import requests
+from bs4 import BeautifulSoup
+from urllib.parse import quote, unquote, urlparse
+
+def html_to_markdown(html_content):
+ if not html_content:
+ return ""
+ soup = BeautifulSoup(html_content, 'html.parser')
+
+ # Process paragraph tags
+ for p in soup.find_all('p'):
+ p.insert_after('\n\n')
+
+ # Process line breaks
+ for br in soup.find_all('br'):
+ br.replace_with('\n')
+
+ # Process list items
+ for li in soup.find_all('li'):
+ li.insert_before('- ')
+ li.insert_after('\n')
+
+ # Process bold text
+ for strong in soup.find_all(['strong', 'b']):
+ text = strong.get_text()
+ if text:
+ strong.replace_with(f"**{text}**")
+
+ # Process links
+ for a in soup.find_all('a'):
+ text = a.get_text()
+ href = a.get('href', '')
+ if text and href:
+ a.replace_with(f"[{text}]({href})")
+
+ text = soup.get_text()
+
+ # Clean up excess newlines
+ lines = [line.rstrip() for line in text.split('\n')]
+ cleaned_text = '\n'.join(lines)
+ while '\n\n\n' in cleaned_text:
+ cleaned_text = cleaned_text.replace('\n\n\n', '\n\n')
+ return cleaned_text.strip()
+
+def get_image_filename(url):
+ ext = os.path.splitext(urlparse(url).path)[1]
+ if not ext or len(ext) > 5:
+ ext = ".png"
+ h = hashlib.md5(url.encode('utf-8')).hexdigest()
+ return f"cbtbank_{h}{ext}"
+
+def download_and_save_image(image_url, target_dir, delay=1.0):
+ if not image_url:
+ return None
+
+ # If it is a relative URL, make it absolute
+ if image_url.startswith('/'):
+ image_url = f"https://cbtbank.kr{image_url}"
+
+ if not image_url.startswith('http'):
+ return image_url
+
+ try:
+ os.makedirs(target_dir, exist_ok=True)
+ filename = get_image_filename(image_url)
+ dest_path = os.path.join(target_dir, filename)
+
+ if os.path.exists(dest_path):
+ return f"/assets/{filename}"
+
+ time.sleep(delay)
+ headers = {
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+ }
+ r = requests.get(image_url, headers=headers, timeout=15)
+ if r.status_code == 200:
+ with open(dest_path, "wb") as f:
+ f.write(r.content)
+ print(f" [Image] Downloaded image: {filename}")
+ return f"/assets/{filename}"
+ else:
+ print(f" [Image Warning] Failed to download: {image_url} (HTTP {r.status_code})")
+ except Exception as e:
+ print(f" [Image Error] Error downloading image: {e}")
+ return image_url
+
+def get_existing_question_ids(conn):
+ cursor = conn.cursor()
+ cursor.execute("SELECT id FROM QuizQuestions WHERE source = 'cbtbank'")
+ return {row[0] for row in cursor.fetchall()}
+
+def scrape_exam(exam_name, selected_rounds=None, overwrite=False, delay=1.0, db_file="qple_quiz.db", assets_dir="./frontend/public/assets"):
+ conn = sqlite3.connect(db_file)
+
+ # Ensure correct_answers column exists in QuizQuestions table
+ cursor = conn.cursor()
+ cursor.execute("PRAGMA table_info(QuizQuestions)")
+ columns = [col[1] for col in cursor.fetchall()]
+ if "correct_answers" not in columns:
+ cursor.execute("ALTER TABLE QuizQuestions ADD COLUMN correct_answers TEXT DEFAULT ''")
+ conn.commit()
+ print(">>> Database Migration: Added correct_answers column to QuizQuestions table.")
+
+ existing_ids = get_existing_question_ids(conn)
+
+ print(f"\n>>> Connecting to SQLite DB: {db_file}")
+ print(f">>> Target Exam: {exam_name}")
+
+ # 1. Fetch main exam list page to find rounds
+ url = f"https://cbtbank.kr/category/{quote(exam_name)}"
+ headers = {
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+ }
+
+ print(f">>> Fetching list of exam rounds from: {url}")
+ r = requests.get(url, headers=headers, timeout=15)
+ r.encoding = 'utf-8'
+ if r.status_code != 200:
+ print(f"[Error] Failed to fetch exam page: HTTP {r.status_code}")
+ conn.close()
+ return
+
+ soup = BeautifulSoup(r.text, 'html.parser')
+
+ # Find all links pointing to /exam/...
+ rounds = []
+ for a in soup.find_all('a'):
+ href = a.get('href', '')
+ if href.startswith('/exam/'):
+ round_id = href.split('/')[-1]
+ round_name = a.get_text().strip()
+ rounds.append({
+ 'id': round_id,
+ 'name': round_name,
+ 'href': href
+ })
+
+ # Filter duplicate rounds preserving order
+ seen = set()
+ unique_rounds = []
+ for rnd in rounds:
+ if rnd['id'] not in seen:
+ seen.add(rnd['id'])
+ unique_rounds.append(rnd)
+
+ print(f"Found {len(unique_rounds)} unique rounds on cbtbank.kr:")
+ for rnd in unique_rounds:
+ print(f" - {rnd['name']} (ID: {rnd['id']})")
+
+ if selected_rounds:
+ filtered_rounds = [rnd for rnd in unique_rounds if rnd['id'] in selected_rounds or rnd['name'] in selected_rounds]
+ print(f"\n>>> Filtered to {len(filtered_rounds)} selected rounds: {[rnd['name'] for rnd in filtered_rounds]}")
+ unique_rounds = filtered_rounds
+
+ # 2. Iterate and scrape each round page
+ for round_idx, rnd in enumerate(unique_rounds, 1):
+ round_id = rnd['id']
+ round_name = rnd['name']
+ round_url = f"https://cbtbank.kr/exam/{round_id}"
+
+ print("\n" + "="*50)
+ print(f"Scraping Round ({round_idx}/{len(unique_rounds)}): {round_name}")
+ print(f"URL: {round_url}")
+ print("="*50)
+
+ time.sleep(delay)
+ r = requests.get(round_url, headers=headers, timeout=15)
+ r.encoding = 'utf-8'
+ if r.status_code != 200:
+ print(f" [Error] Failed to fetch round page: HTTP {r.status_code}")
+ continue
+
+ round_soup = BeautifulSoup(r.text, 'html.parser')
+ exams_div = round_soup.find(class_='exams')
+ if not exams_div:
+ print(" [Error] Could not find the questions container (class='exams') on this page.")
+ continue
+
+ current_subject = "일반"
+ parsed_questions_count = 0
+
+ # Iterate over the direct children of div.exams
+ for child in exams_div.children:
+ if not child.name:
+ continue
+
+ cl = child.get('class', [])
+ cl_str = " ".join(cl)
+
+ # 1. Update current subject if we encounter exam-class-title
+ if 'exam-class-title' in cl:
+ subject_text = child.get_text().strip()
+ if ':' in subject_text:
+ current_subject = subject_text.split(':', 1)[1].strip()
+ else:
+ current_subject = subject_text.strip()
+ print(f"\n >>> Subject switched to: {current_subject}")
+ continue
+
+ # 2. Process question containers (exam-box) nested inside row text-dark
+ if 'text-dark' in cl:
+ boxes = child.find_all(class_='exam-box')
+ for box in boxes:
+ q_id = box.get('question-id')
+ if not q_id:
+ continue
+
+ import zlib
+ q_id_int = zlib.crc32(q_id.encode('utf-8')) & 0x7fffffff
+ print(f" Question ID: {q_id} (Int: {q_id_int}) ({current_subject})")
+
+ if q_id_int in existing_ids and not overwrite:
+ print(f" -> {q_id} (Int: {q_id_int}) already exists in DB. Skipping.")
+ continue
+
+ # Extract title tag
+ title_tag = box.find(class_='exam-title')
+ if not title_tag:
+ print(" [Warning] Could not find exam-title. Skipping.")
+ continue
+
+ # Get question image (if any) in box (excluding choices and replies)
+ q_img_url = None
+ all_imgs = box.find_all('img')
+ for img in all_imgs:
+ # Check that this image is not inside reply or choices
+ if img.find_parent(class_='reply') or img.find_parent(class_='question-choice'):
+ continue
+ src = img.get('src')
+ if src and not any(x in src for x in ['kakao.png', 'line.png', 'facebook.png', 'twitter.png', 'band.png']):
+ q_img_url = src
+ break
+
+ # Clean question text by extracting the span.exam-number (question index)
+ import copy
+ title_tag_copy = copy.copy(title_tag)
+ num_span = title_tag_copy.find(class_='exam-number')
+ if num_span:
+ num_span.extract()
+ # Also extract any images from the copy so they don't leak into text
+ for img in title_tag_copy.find_all('img'):
+ img.extract()
+
+ q_text = title_tag_copy.get_text().strip()
+ q_text = re.sub(r'^[\s\.]+', '', q_text).strip()
+
+ # Extract passage (if any)
+ passage = ""
+ passage_div = box.find(class_='passage')
+ if passage_div:
+ passage = html_to_markdown(str(passage_div))
+
+ # Extract choices
+ choices = ["", "", "", ""]
+ correct_idx = -1
+ correct_str = ""
+ correct_answers_serialized = ""
+
+ choice_list = box.find(class_='question-choice')
+ if choice_list:
+ ol = choice_list.find('ol')
+ correct_attr = ol.get('correct') if ol else None
+
+ li_items = choice_list.find_all('li')
+ for li_idx, li in enumerate(li_items):
+ # Check if there is an image in choice
+ choice_img = li.find('img')
+ if choice_img:
+ src = choice_img.get('src')
+ if src:
+ local_src = download_and_save_image(src, assets_dir, delay)
+ choice_text = f'
'
+ else:
+ choice_text = ""
+ else:
+ choice_text = li.get_text().strip()
+
+ if li_idx < 4:
+ choices[li_idx] = choice_text
+
+ # Check correct choice by class
+ if 'correct' in li.get('class', []):
+ correct_idx = li_idx
+ correct_str = choice_text
+
+ # Fallback to correct attribute on ol
+ if correct_idx == -1 and correct_attr:
+ try:
+ correct_idx = int(correct_attr) - 1
+ if 0 <= correct_idx < len(choices):
+ correct_str = choices[correct_idx]
+ except ValueError:
+ pass
+
+ # If it's a descriptive question or no choices were found
+ if not choice_list or not any(choices):
+ choices = ["", "", "", ""]
+ correct_idx = -1
+ correct_str = "주관식"
+
+ correct_answers_serialized = str(correct_idx) if correct_idx != -1 else ""
+
+ # Download question image if present
+ final_image_url = ""
+ if q_img_url:
+ final_image_url = download_and_save_image(q_img_url, assets_dir, delay)
+
+ # Extract explanations
+ human_explanations = []
+ ai_explanations = []
+
+ reply_list = box.find(class_='view-reply')
+ if reply_list:
+ items = reply_list.find_all(class_='reply-item')
+ for item in items:
+ nick_tag = item.find(class_='nick')
+ nick = nick_tag.get_text().strip() if nick_tag else ""
+
+ comment_div = item.find(class_='reply-comment')
+ if comment_div:
+ comment_text = html_to_markdown(str(comment_div))
+ if "AI" in nick or "ai" in nick.lower():
+ ai_explanations.append(f"### 🤖 {nick} AI 해설\n\n{comment_text}")
+ else:
+ human_explanations.append(f"**{nick}**: {comment_text}")
+
+ human_explanation = "\n\n---\n\n".join(human_explanations)
+ ai_explanation = "\n\n---\n\n".join(ai_explanations)
+
+ # Construct metadata raw json
+ meta_json = json.dumps({
+ "source": "cbtbank.kr",
+ "q_id": q_id,
+ "round": round_name,
+ "subject": current_subject
+ }, ensure_ascii=False)
+
+ # Save to database
+ cursor = conn.cursor()
+ try:
+ cursor.execute('''
+ INSERT OR REPLACE INTO QuizQuestions (
+ id, source, category, question_text, choice1, choice2, choice3, choice4,
+ correct_answer, correct_answer_str, explanation, difficulty, subject,
+ passage, image_url, raw_json, correct_answers, oid
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ ''', (
+ q_id_int, 'cbtbank', exam_name, q_text,
+ choices[0], choices[1], choices[2], choices[3],
+ correct_idx, correct_str, human_explanation, round_name, current_subject,
+ passage, final_image_url, meta_json, correct_answers_serialized, q_id_int
+ ))
+
+ if ai_explanation:
+ cursor.execute('''
+ INSERT OR REPLACE INTO QuizExplain (id, source, explain)
+ VALUES (?, ?, ?)
+ ''', (q_id_int, 'cbtbank', ai_explanation))
+
+ conn.commit()
+ existing_ids.add(q_id_int)
+ parsed_questions_count += 1
+ print(f" -> Successfully saved to DB!")
+ except Exception as db_err:
+ conn.rollback()
+ print(f" [DB Error] Failed to save {q_id} (Int: {q_id_int}): {db_err}")
+
+ print(f"Completed round {round_name}. Saved {parsed_questions_count} new questions.")
+
+ conn.close()
+ print("\n>>> Scrape operation completed successfully!")
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description="Cbtbank Scraper for Quiz Database")
+ parser.add_argument("--exam", type=str, default="정보통신기사", help="Name of the exam/category on cbtbank.kr (e.g. 정보통신기사)")
+ parser.add_argument("--rounds", type=str, default="", help="Comma-separated round names/IDs to scrape (e.g. 'cey20230617'). Default is all.")
+ parser.add_argument("--overwrite", action="store_true", help="Overwrite existing questions in the database")
+ parser.add_argument("--delay", type=float, default=1.0, help="Delay in seconds between requests (default: 1.0)")
+ parser.add_argument("--db", type=str, default="qple_quiz.db", help="Path to sqlite database file (default: qple_quiz.db)")
+ parser.add_argument("--assets", type=str, default="./frontend/public/assets", help="Path to assets directory to store downloaded images")
+
+ args = parser.parse_args()
+
+ selected = [r.strip() for r in args.rounds.split(',') if r.strip()] if args.rounds else None
+
+ try:
+ scrape_exam(
+ exam_name=args.exam,
+ selected_rounds=selected,
+ overwrite=args.overwrite,
+ delay=args.delay,
+ db_file=args.db,
+ assets_dir=args.assets
+ )
+ except KeyboardInterrupt:
+ print("\n[Interrupted] Scraper stopped by user.")
+ sys.exit(0)
diff --git a/scrape_everything.py b/scrape_everything.py
new file mode 100644
index 0000000..be80a50
--- /dev/null
+++ b/scrape_everything.py
@@ -0,0 +1,90 @@
+import sys
+import os
+import time
+import shutil
+
+# Add current dir to path to import scraping scripts
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+try:
+ import scrape_newbt
+ import scrape_cbtbank
+except ImportError as e:
+ print(f"Failed to import scraping modules: {e}")
+ sys.exit(1)
+
+# Configuration
+db_file = r"./qple_quiz.db"
+assets_dir = r"./frontend/public/assets"
+delay = 0.1 # Moderate delay to prevent server blocking but process fast
+
+# 1. Newbt Exams List
+newbt_exams = [
+ "ISMS-P 인증심사원",
+ "AWS Solutions Architect Associate",
+ "Artificial Intelligence",
+ "DevOps and Agile",
+ "정보처리기사(구)",
+ "정보처리기사",
+ "PSAT 헌법",
+ "PSAT 자료해석",
+ "PSAT 상황판단",
+ "PSAT 언어논리",
+ "데이터분석 준전문가",
+ "리눅스마스터 1급",
+ "리눅스마스터 1급 실기",
+ "리눅스마스터 2급",
+ "정보시스템감리사"
+]
+
+# 2. CBTBank Exams List
+cbtbank_exams = [
+ "정보통신기사(구)",
+ "컴퓨터시스템기사(A형)",
+ "컴퓨터시스템기사(B형)"
+]
+
+print("======================================================================")
+print(">>> STARTING UNIFIED BATCH SCRAPING PIPELINE")
+print("======================================================================")
+
+# Step A: Scrape from Newbt
+print(f"\n>>> [PART 1/2] Scraping {len(newbt_exams)} categories from newbt.kr...")
+for idx, exam in enumerate(newbt_exams, 1):
+ print(f"\n----------------------------------------------------------------------")
+ print(f"[Newbt {idx}/{len(newbt_exams)}] SCRAPING: {exam}")
+ print(f"----------------------------------------------------------------------")
+ try:
+ scrape_newbt.scrape_exam(
+ exam_name=exam,
+ selected_rounds=None,
+ overwrite=False,
+ delay=delay,
+ db_file=db_file,
+ assets_dir=assets_dir
+ )
+ except Exception as e:
+ print(f"[Error] Failed scraping {exam} from newbt: {e}")
+ time.sleep(1.0)
+
+# Step B: Scrape from CBTBank
+print(f"\n>>> [PART 2/2] Scraping {len(cbtbank_exams)} categories from cbtbank.kr...")
+for idx, exam in enumerate(cbtbank_exams, 1):
+ print(f"\n----------------------------------------------------------------------")
+ print(f"[CBTBank {idx}/{len(cbtbank_exams)}] SCRAPING: {exam}")
+ print(f"----------------------------------------------------------------------")
+ try:
+ scrape_cbtbank.scrape_exam(
+ exam_name=exam,
+ selected_rounds=None,
+ overwrite=False,
+ delay=delay,
+ db_file=db_file,
+ assets_dir=assets_dir
+ )
+ except Exception as e:
+ print(f"[Error] Failed scraping {exam} from cbtbank: {e}")
+ time.sleep(1.0)
+
+print("\n======================================================================")
+print(">>> ALL SCRAPING TASKS COMPLETE!")
+print("======================================================================")
diff --git a/scrape_newbt.py b/scrape_newbt.py
new file mode 100644
index 0000000..59e554a
--- /dev/null
+++ b/scrape_newbt.py
@@ -0,0 +1,716 @@
+import os
+import sys
+import time
+import json
+import sqlite3
+import re
+import argparse
+import hashlib
+import requests
+from bs4 import BeautifulSoup
+from urllib.parse import quote, unquote, urlparse
+
+def table_to_markdown(table_tag):
+ # Find only direct tr elements of the current table to avoid recursion into nested tables
+ all_trs = []
+ for tr in table_tag.find_all('tr'):
+ if tr.find_parent('table') == table_tag:
+ all_trs.append(tr)
+
+ # 1. 테이블의 최대 열 개수(max_cols) 계산
+ max_cols = 0
+ for tr in all_trs:
+ cols_count = 0
+ cells = tr.find_all(['td', 'th'])
+ for td in cells:
+ colspan = 1
+ if td.has_attr('colspan'):
+ try:
+ colspan = int(td['colspan'])
+ except ValueError:
+ colspan = 1
+ cols_count += colspan
+ if cols_count > max_cols:
+ max_cols = cols_count
+
+ if max_cols == 0:
+ return ""
+
+ # 2. 각 행을 분석하여 데이터 행(table row) 또는 평문 행(text row)으로 분류
+ classified_rows = []
+ for tr in all_trs:
+ cells = tr.find_all(['td', 'th'])
+ if not cells:
+ continue
+
+ # colspan을 고려한 셀 확장 리스트 생성
+ expanded_cells = []
+ for td in cells:
+ colspan = 1
+ if td.has_attr('colspan'):
+ try:
+ colspan = int(td['colspan'])
+ except ValueError:
+ colspan = 1
+ text = td.get_text().strip().replace('\n', ' ')
+ expanded_cells.append(text)
+ for _ in range(colspan - 1):
+ expanded_cells.append("")
+
+ # 평문 행 여부 판단: 실제 td/th가 1개이거나, 확장 후 비어있지 않은 셀이 1개뿐인 경우
+ is_text = False
+ if len(cells) == 1:
+ is_text = True
+ elif len([c for c in expanded_cells if c]) == 1:
+ is_text = True
+
+ if is_text:
+ text_content = cells[0].get_text().strip()
+ classified_rows.append({
+ "type": "text",
+ "content": text_content
+ })
+ else:
+ classified_rows.append({
+ "type": "table",
+ "cells": expanded_cells
+ })
+
+ # 3. 조립 진행 (연속된 table 행은 마크다운 표로 결합하고, text 행은 개별 평문 단락으로 출력)
+ output_parts = []
+ current_table_rows = []
+
+ def flush_table():
+ if not current_table_rows:
+ return
+
+ # 1. Filter out completely empty rows
+ non_empty_rows = []
+ for r in current_table_rows:
+ if any(cell.strip() for cell in r):
+ non_empty_rows.append(r)
+
+ if not non_empty_rows:
+ current_table_rows.clear()
+ return
+
+ # 2. Find max columns in these non-empty rows
+ t_max_cols = max(len(r) for r in non_empty_rows)
+ if t_max_cols == 0:
+ current_table_rows.clear()
+ return
+
+ # 3. Pad all rows to t_max_cols
+ padded_rows = []
+ for r in non_empty_rows:
+ row_copy = list(r)
+ if len(row_copy) < t_max_cols:
+ row_copy += [""] * (t_max_cols - len(row_copy))
+ else:
+ row_copy = row_copy[:t_max_cols]
+ padded_rows.append(row_copy)
+
+ # 4. Find which columns are completely empty (have no content in any row)
+ non_empty_col_indices = []
+ for col_idx in range(t_max_cols):
+ has_content = False
+ for r in padded_rows:
+ if r[col_idx].strip():
+ has_content = True
+ break
+ if has_content:
+ non_empty_col_indices.append(col_idx)
+
+ if not non_empty_col_indices:
+ current_table_rows.clear()
+ return
+
+ # 5. Re-build rows with only non-empty columns
+ final_rows = []
+ for r in padded_rows:
+ final_rows.append([r[idx] for idx in non_empty_col_indices])
+
+ final_max_cols = len(non_empty_col_indices)
+
+ # 6. Generate markdown table
+ table_md = []
+ headers = final_rows[0]
+ table_md.append("| " + " | ".join(headers) + " |")
+ table_md.append("| " + " | ".join(["---"] * final_max_cols) + " |")
+ for row in final_rows[1:]:
+ table_md.append("| " + " | ".join(row) + " |")
+
+ output_parts.append("\n\n" + "\n".join(table_md) + "\n\n")
+ current_table_rows.clear()
+
+ for row in classified_rows:
+ if row["type"] == "text":
+ flush_table()
+ text = row["content"]
+ # 타이틀 스타일 표제어 굵게 강조
+ if text.startswith('<') and text.endswith('>'):
+ output_parts.append(f"\n\n**{text}**\n\n")
+ else:
+ output_parts.append(f"\n\n{text}\n\n")
+ else:
+ current_table_rows.append(row["cells"])
+
+ flush_table()
+ return "".join(output_parts)
+
+def fix_corrupted_symbols(text):
+ if not text:
+ return text
+ lines = text.split('\n')
+ symbols = ['㉠', '㉡', '㉢', '㉣', '㉤', '㉥']
+ symbol_idx = 0
+
+ new_lines = []
+ for line in lines:
+ stripped = line.strip()
+ if stripped.startswith('?') and symbol_idx < len(symbols):
+ import re
+ replaced = re.sub(r'^\?\s*', symbols[symbol_idx] + ' ', line)
+ new_lines.append(replaced)
+ symbol_idx += 1
+ else:
+ if '?' in line:
+ parts = line.split('?')
+ if len(parts) > 1 and all(p.strip() for p in parts[1:]) and line.startswith('?'):
+ new_line = parts[0]
+ for part in parts[1:]:
+ if symbol_idx < len(symbols):
+ new_line += symbols[symbol_idx] + part
+ symbol_idx += 1
+ else:
+ new_line += '?' + part
+ new_lines.append(new_line)
+ continue
+ new_lines.append(line)
+
+ return '\n'.join(new_lines)
+
+def deduplicate_consecutive_lines(text):
+ if not text:
+ return text
+ lines = text.split('\n')
+ cleaned_lines = []
+
+ last_non_empty = None
+ for line in lines:
+ stripped = line.strip()
+ if not stripped:
+ cleaned_lines.append(line)
+ continue
+
+ import re
+ c_line = re.sub(r'[\s\W_]', '', stripped)
+ if last_non_empty is not None:
+ c_last = re.sub(r'[\s\W_]', '', last_non_empty)
+ if c_line == c_last and len(c_line) > 10:
+ continue
+
+ cleaned_lines.append(line)
+ last_non_empty = line
+
+ return '\n'.join(cleaned_lines)
+
+def html_to_markdown(html_content, assets_dir=None, delay=1.0):
+ if not html_content:
+ return ""
+ soup = BeautifulSoup(html_content, 'html.parser')
+
+ # Process paragraph tags
+ for p in soup.find_all('p'):
+ p.insert_after('\n\n')
+
+ # Process line breaks
+ for br in soup.find_all('br'):
+ br.replace_with('\n')
+
+ # Process list items
+ for li in soup.find_all('li'):
+ li.insert_before('- ')
+ li.insert_after('\n')
+
+ # Process bold text
+ for strong in soup.find_all(['strong', 'b']):
+ # If it is already replaced or empty, skip
+ text = strong.get_text()
+ if text:
+ strong.replace_with(f"**{text}**")
+
+ # Process links
+ for a in soup.find_all('a'):
+ text = a.get_text()
+ href = a.get('href', '')
+ if text and href:
+ a.replace_with(f"[{text}]({href})")
+
+ # Process images inside HTML to download them locally and replace with markdown image syntax
+ for img in soup.find_all('img'):
+ src = img.get('src', '')
+ if src and not any(x in src for x in ['o.png', 'x.png', 't.png', 'check-mark.png']):
+ if assets_dir:
+ local_path = download_and_save_image(src, assets_dir, delay)
+ if local_path:
+ img.replace_with(f"")
+ else:
+ img.replace_with(f"")
+
+ text = soup.get_text()
+
+ # Clean up excess newlines
+ lines = [line.rstrip() for line in text.split('\n')]
+ cleaned_text = '\n'.join(lines)
+ while '\n\n\n' in cleaned_text:
+ cleaned_text = cleaned_text.replace('\n\n\n', '\n\n')
+ return cleaned_text.strip()
+
+def get_image_filename(url):
+ ext = os.path.splitext(urlparse(url).path)[1]
+ if not ext or len(ext) > 5:
+ ext = ".png"
+ h = hashlib.md5(url.encode('utf-8')).hexdigest()
+ return f"newbt_{h}{ext}"
+
+def download_and_save_image(image_url, target_dir, delay=1.0):
+ if not image_url:
+ return None
+
+ # If it is a relative URL, make it absolute
+ if image_url.startswith('/'):
+ image_url = f"https://newbt.kr{image_url}"
+
+ if not image_url.startswith('http'):
+ return image_url
+
+ try:
+ os.makedirs(target_dir, exist_ok=True)
+ filename = get_image_filename(image_url)
+ dest_path = os.path.join(target_dir, filename)
+
+ if os.path.exists(dest_path):
+ return f"/assets/{filename}"
+
+ time.sleep(delay)
+ headers = {
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+ }
+ r = requests.get(image_url, headers=headers, timeout=15)
+ if r.status_code == 200:
+ with open(dest_path, "wb") as f:
+ f.write(r.content)
+ print(f" [Image] Downloaded image: {filename}")
+ return f"/assets/{filename}"
+ else:
+ print(f" [Image Warning] Failed to download: {image_url} (HTTP {r.status_code})")
+ except Exception as e:
+ print(f" [Image Error] Error downloading image: {e}")
+ return image_url
+
+def fetch_choices_and_answer(q_id, delay=1.0):
+ url = f"https://newbt.kr/question/examples/{q_id}"
+ headers = {
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+ }
+ time.sleep(delay)
+ try:
+ r = requests.get(url, headers=headers, timeout=10)
+ if r.status_code == 200:
+ res = r.json()
+ if res.get("success") and "data" in res:
+ data = res["data"]
+ # Sort choices by number
+ data = sorted(data, key=lambda x: int(x.get("number", 0)))
+ choices = [x.get("contents", "") for x in data]
+
+ # Find all correct answers
+ correct_indices = []
+ correct_strs = []
+ for idx, x in enumerate(data):
+ if x.get("is_answer") == "1":
+ correct_indices.append(idx)
+ correct_strs.append(x.get("contents", ""))
+
+ correct_idx = correct_indices[0] if correct_indices else -1
+ correct_str = " | ".join(correct_strs) if correct_strs else ""
+ correct_answers_serialized = ",".join(map(str, correct_indices)) if correct_indices else ""
+
+ return choices, correct_idx, correct_str, correct_answers_serialized
+ except Exception as e:
+ print(f" [API Error] Choices fetch error for Q{q_id}: {e}")
+ return [], -1, "", ""
+
+def fetch_explanations(q_id, delay=1.0, assets_dir=None):
+ headers = {
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+ }
+
+ # 1. Fetch human explanation
+ human_explanation = ""
+ time.sleep(delay)
+ try:
+ url_human = f"https://newbt.kr/resolve/get/{q_id}"
+ r = requests.get(url_human, headers=headers, timeout=10)
+ if r.status_code == 200:
+ res = r.json()
+ if res.get("success") and "data" in res:
+ data = res["data"]
+ contents = []
+ for x in data:
+ c = x.get("content", "").strip()
+ if c:
+ contents.append(html_to_markdown(c, assets_dir, delay))
+ if contents:
+ human_explanation = "\n\n---\n\n".join(contents)
+ except Exception as e:
+ print(f" [API Error] Human explanation fetch error for Q{q_id}: {e}")
+
+ # 2. Fetch AI explanation
+ ai_explanations = []
+ time.sleep(delay)
+ try:
+ url_ai = f"https://newbt.kr/resolve/getResolvedResults/{q_id}"
+ r = requests.get(url_ai, headers=headers, timeout=10)
+ if r.status_code == 200:
+ res = r.json()
+ if res.get("success") and "data" in res:
+ data = res["data"]
+ for x in data:
+ c = x.get("content", "")
+ if c and c.strip():
+ model_name = x.get("name", "AI")
+ cleaned_c = html_to_markdown(c, assets_dir, delay)
+ ai_explanations.append(f"### 🤖 {model_name} AI 해설\n\n{cleaned_c}")
+ except Exception as e:
+ print(f" [API Error] AI explanation fetch error for Q{q_id}: {e}")
+
+ ai_explanation_combined = "\n\n---\n\n".join(ai_explanations)
+ return human_explanation, ai_explanation_combined
+
+def get_existing_question_ids(conn):
+ cursor = conn.cursor()
+ cursor.execute("SELECT id FROM QuizQuestions WHERE source = 'newbt'")
+ return {row[0] for row in cursor.fetchall()}
+
+
+def scrape_exam(exam_name, selected_rounds=None, overwrite=False, delay=1.0, db_file="qple_quiz.db", assets_dir="./frontend/public/assets"):
+ conn = sqlite3.connect(db_file)
+
+ # Ensure correct_answers column exists in QuizQuestions table
+ cursor = conn.cursor()
+ cursor.execute("PRAGMA table_info(QuizQuestions)")
+ columns = [col[1] for col in cursor.fetchall()]
+ if "correct_answers" not in columns:
+ cursor.execute("ALTER TABLE QuizQuestions ADD COLUMN correct_answers TEXT DEFAULT ''")
+ conn.commit()
+ print(">>> Database Migration: Added correct_answers column to QuizQuestions table.")
+
+ existing_ids = get_existing_question_ids(conn)
+
+ print(f"\n>>> Connecting to SQLite DB: {db_file}")
+ print(f">>> Target Exam: {exam_name}")
+
+ # 1. Fetch main exam list page to find rounds
+ url = f"https://newbt.kr/시험/{quote(exam_name)}"
+ headers = {
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+ }
+
+ print(f">>> Fetching list of exam rounds from: {url}")
+ r = requests.get(url, headers=headers, timeout=15)
+ r.encoding = 'utf-8'
+ if r.status_code != 200:
+ print(f"[Error] Failed to fetch exam page: HTTP {r.status_code}")
+ conn.close()
+ return
+
+ soup = BeautifulSoup(r.text, 'html.parser')
+
+ # Find all links pointing to /시험/{exam_name}/...
+ rounds = []
+ for a in soup.find_all('a'):
+ href = a.get('href', '')
+ if href.startswith(f"/시험/{exam_name}/") or href.startswith(f"/시험/{quote(exam_name)}/"):
+ parts = [unquote(p) for p in href.split('/') if p]
+ if len(parts) >= 3:
+ round_name = parts[2].replace('+', ' ')
+ rounds.append({
+ 'name': round_name,
+ 'href': href
+ })
+
+ # Filter duplicate rounds preserving order
+ seen = set()
+ unique_rounds = []
+ for rnd in rounds:
+ if rnd['name'] not in seen:
+ seen.add(rnd['name'])
+ unique_rounds.append(rnd)
+
+ print(f"Found {len(unique_rounds)} unique rounds on newbt.kr:")
+ for rnd in unique_rounds:
+ print(f" - {rnd['name']} ({rnd['href']})")
+
+ if selected_rounds:
+ filtered_rounds = [rnd for rnd in unique_rounds if rnd['name'] in selected_rounds]
+ print(f"\n>>> Filtered to {len(filtered_rounds)} selected rounds: {[rnd['name'] for rnd in filtered_rounds]}")
+ unique_rounds = filtered_rounds
+
+ # 2. Iterate and scrape each round page
+ for r_idx, rnd in enumerate(unique_rounds, 1):
+ round_name = rnd['name']
+ round_href = rnd['href']
+ round_url = f"https://newbt.kr{round_href}"
+
+ print(f"\n==================================================")
+ print(f"Scraping Round ({r_idx}/{len(unique_rounds)}): {round_name}")
+ print(f"URL: {round_url}")
+ print(f"==================================================")
+
+ time.sleep(delay)
+ res_round = requests.get(round_url, headers=headers, timeout=15)
+ res_round.encoding = 'utf-8'
+ if res_round.status_code != 200:
+ print(f" [Error] Failed to fetch round page: HTTP {res_round.status_code}")
+ continue
+
+ round_soup = BeautifulSoup(res_round.text, 'html.parser')
+
+ current_subject = "Unknown"
+ parsed_questions = []
+
+ # Traverse elements to find headers and questions
+ for el in round_soup.find_all(['h3', 'div']):
+ if el.name == 'h3':
+ parent = el.parent
+ classes = parent.get('class') or []
+ if 'col-sm-12' in classes:
+ if not el.find_parent(id='resultModal'):
+ current_subject = el.get_text().strip()
+ elif el.name == 'div' and 'blog-post' in el.get('class', []):
+ q_id_attr = el.get('id')
+ if q_id_attr and q_id_attr.startswith('q'):
+ q_id = int(q_id_attr[1:])
+
+ subject_tag = el.find('h5', class_='subject')
+ if not subject_tag:
+ continue
+
+ q_text = subject_tag.get_text().strip()
+ # Remove leading question number
+ q_text = re.sub(r'^\s*\d+\s*\.\s*', '', q_text)
+ # Restore stripped <보기> tags
+ q_text = q_text.replace("다음 는", "다음 <보기>는")
+ q_text = q_text.replace(" 의 내용", " <보기>의 내용")
+
+ passage_tag = el.find('pre', class_='contents')
+ passage = ""
+ if passage_tag:
+ import copy
+ p_tag_copy = copy.deepcopy(passage_tag)
+ tables = p_tag_copy.find_all('table')
+ if tables:
+ # 1. Extract all text inside tables
+ table_texts = []
+ for t in p_tag_copy.find_all('table'):
+ table_texts.append(t.get_text())
+ combined_table_text = "".join(table_texts)
+
+ # 2. Extract all text outside tables
+ non_table_text_parts = []
+ for child in p_tag_copy.children:
+ if child.name != 'table':
+ non_table_text_parts.append(child.get_text() if child.name else str(child))
+ combined_non_table_text = "".join(non_table_text_parts)
+
+ # Helper to clean text for comparison
+ def clean_comparison_text(text):
+ return re.sub(r'[\s\W_]', '', text)
+
+ c_non = clean_comparison_text(combined_non_table_text)
+ c_tab = clean_comparison_text(combined_table_text)
+
+ is_duplicate = False
+ if c_non and c_tab:
+ if c_non in c_tab or c_tab in c_non:
+ is_duplicate = True
+ else:
+ common_chars = set(c_non) & set(c_tab)
+ if len(common_chars) / len(set(c_non)) > 0.80:
+ is_duplicate = True
+
+ if is_duplicate:
+ # Remove direct non-table children
+ for child in list(p_tag_copy.children):
+ if child.name != 'table':
+ child.extract()
+
+ # 3. Replace tables with markdown (innermost tables first, i.e., in reverse)
+ all_tables = p_tag_copy.find_all('table')
+ for t in reversed(all_tables):
+ if t.parent:
+ md_table = table_to_markdown(t)
+ t.replace_with(md_table)
+
+ passage = p_tag_copy.get_text().strip()
+
+ # Extract raw image if any
+ img_url = None
+ for img in el.find_all('img'):
+ src = img.get('src', '')
+ if src and not any(x in src for x in ['o.png', 'x.png', 't.png', 'check-mark.png']):
+ img_url = src
+ break
+
+ # Extract choice images mapping
+ choice_images = {}
+ ul_example = el.find('ul', class_='example')
+ if ul_example:
+ for img in ul_example.find_all('img'):
+ src = img.get('src', '')
+ alt = img.get('alt', '')
+ if src and not any(x in src for x in ['o.png', 'x.png', 't.png', 'check-mark.png']):
+ digits = re.findall(r'\d+', alt)
+ if digits:
+ num = digits[0]
+ choice_images[num] = src
+
+ parsed_questions.append({
+ 'id': q_id,
+ 'subject': current_subject,
+ 'question_text': q_text,
+ 'passage': deduplicate_consecutive_lines(passage),
+ 'image_url': img_url,
+ 'choice_images': choice_images
+ })
+
+ print(f"Found {len(parsed_questions)} questions in round {round_name}.")
+
+ # Process each question
+ for q_idx, q in enumerate(parsed_questions, 1):
+ q_id = q['id']
+ subject_name = q['subject']
+
+ # Form target/difficulty category structure
+ # Form target/difficulty category structure
+ # category: "정보보안기사", difficulty: round_name, subject: subject_name
+ q_category = exam_name
+ q_difficulty = round_name
+ q_subject = subject_name
+
+ print(f"\n [{q_idx}/{len(parsed_questions)}] Question ID: {q_id} ({subject_name})")
+
+ # Check skip
+ if q_id in existing_ids and not overwrite:
+ print(f" -> Q{q_id} already exists in DB. Skipping (use --overwrite to update).")
+ continue
+
+ # Fetch choices, correct answer and descriptions from API
+ choices, correct_idx, correct_str, correct_answers = fetch_choices_and_answer(q_id, delay)
+ if not choices:
+ # Descriptive/Essay question
+ print(f" -> [Info] Q{q_id} has no choices (Descriptive/Essay question). Importing anyway.")
+ choices = ["", "", "", "", ""]
+ correct_idx = -1
+ correct_str = "주관식"
+ correct_answers = ""
+ elif correct_idx == -1:
+ print(f" -> [Warning] Failed to find correct answer index for Q{q_id}. Skipping.")
+ continue
+
+ # Make sure we have 5 choices
+ choices = (choices + ["", "", "", "", ""])[:5]
+
+ # Replace choice image placeholders with local img tags
+ choice_img_map = q.get('choice_images', {})
+ for idx in range(len(choices)):
+ placeholders = re.findall(r'\[_이미지#(\d+)_\]', choices[idx])
+ for num in placeholders:
+ if num in choice_img_map:
+ img_src = choice_img_map[num]
+ local_img_path = download_and_save_image(img_src, assets_dir, delay)
+ if local_img_path:
+ img_tag = f'
'
+ choices[idx] = choices[idx].replace(f'[_이미지#{num}_]', img_tag)
+
+ choices = [fix_corrupted_symbols(c) for c in choices]
+ correct_str = fix_corrupted_symbols(correct_str)
+
+ human_explanation, ai_explanation = fetch_explanations(q_id, delay, assets_dir)
+
+ # Handle image download if present
+ final_image_url = ""
+ if q['image_url']:
+ final_image_url = download_and_save_image(q['image_url'], assets_dir, delay)
+
+ # Construct metadata raw json
+ meta_json = json.dumps({
+ "source": "newbt.kr",
+ "q_id": q_id,
+ "round": round_name,
+ "subject": subject_name
+ }, ensure_ascii=False)
+
+ # Save to Database
+ cursor = conn.cursor()
+ try:
+ # 1. Insert/Replace into QuizQuestions
+ cursor.execute('''
+ INSERT OR REPLACE INTO QuizQuestions (
+ id, source, category, question_text, choice1, choice2, choice3, choice4, choice5,
+ correct_answer, correct_answer_str, explanation, difficulty, subject,
+ passage, image_url, raw_json, correct_answers, oid
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ ''', (
+ q_id, 'newbt', q_category, q['question_text'],
+ choices[0], choices[1], choices[2], choices[3], choices[4],
+ correct_idx, correct_str, human_explanation, q_difficulty, q_subject,
+ q['passage'], final_image_url, meta_json, correct_answers, q_id
+ ))
+
+
+ # 2. Insert/Replace into QuizExplain
+ if ai_explanation:
+ cursor.execute('''
+ INSERT OR REPLACE INTO QuizExplain (id, source, explain)
+ VALUES (?, ?, ?)
+ ''', (q_id, 'newbt', ai_explanation))
+
+ conn.commit()
+ existing_ids.add(q_id)
+ print(f" -> Q{q_id} successfully saved to DB!")
+ except Exception as e:
+ conn.rollback()
+ print(f" -> [DB Error] Failed to save Q{q_id}: {e}")
+
+
+ conn.close()
+ print("\n>>> Scrape operation completed successfully!")
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description="Newbt Scraper for Quiz Database")
+ parser.add_argument("--exam", type=str, default="정보보안기사", help="Name of the exam/category on newbt.kr (e.g. 정보보안기사)")
+ parser.add_argument("--rounds", type=str, default="", help="Comma-separated round names to scrape (e.g. '제9회,제4회'). Default is all.")
+ parser.add_argument("--overwrite", action="store_true", help="Overwrite existing questions in the database")
+ parser.add_argument("--delay", type=float, default=1.0, help="Delay in seconds between requests (default: 1.0)")
+ parser.add_argument("--db", type=str, default="qple_quiz.db", help="Path to sqlite database file (default: qple_quiz.db)")
+ parser.add_argument("--assets", type=str, default="./frontend/public/assets", help="Path to assets directory to store downloaded images")
+
+ args = parser.parse_args()
+
+ selected = [r.strip() for r in args.rounds.split(',') if r.strip()] if args.rounds else None
+
+ try:
+ scrape_exam(
+ exam_name=args.exam,
+ selected_rounds=selected,
+ overwrite=args.overwrite,
+ delay=args.delay,
+ db_file=args.db,
+ assets_dir=args.assets
+ )
+ except KeyboardInterrupt:
+ print("\n[Interrupted] Scraper stopped by user.")
+ sys.exit(0)
diff --git a/test_captured_headers.py b/test_captured_headers.py
new file mode 100644
index 0000000..35d9966
--- /dev/null
+++ b/test_captured_headers.py
@@ -0,0 +1,30 @@
+import urllib.request
+import urllib.error
+import urllib.parse
+
+def test_user_info_req():
+ url = "https://qple-api-prod.qpleapp.com/Login/UserInfoReq"
+
+ # Empty Thrift payload or simple struct (this might still 500 if the struct isn't perfect, but we'll try)
+ # The user noted Content-Length was 169.
+ # We will just try an empty body or simple byte sequence for now to see if we get a Thrift response or 500
+ data = b'\x00' * 169
+
+ req = urllib.request.Request(url, data=data, method="POST")
+ req.add_header('Accept', '*/*')
+ req.add_header('AppVersion', '1.5.1')
+ req.add_header('Authorization', 'Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJyYW5kTnVtYmVyIjoxNDA1MzU0OTgzLCJ1c2VyR1VJRCI6IjY1NTU1OTdkYWU2Mjk0ZGU3NTMwNTRjMiIsImFjY291bnRJZCI6IjEwMTMwMjUiLCJkZXZpY2VJRCI6IjU2Y2E1ZTcxZTVkOTlhZmRkMTdkYzQyZGFjNjNjMDU1IiwiZXhwaXJlc19kYXRlIjotODU4NDE5MzcyNzYzOTEyMTA3OX0.mD2vhhSYn-M4Oth_EoVhVjC4Fdn8fIIY9zhnAtkXgZ4')
+ req.add_header('Content-Type', 'application/x-thrift')
+ req.add_header('Cookie', 'CHARCTER_INFO=nickname=ëì¬ì¹´&branchName=ITìì¤íë¶')
+ req.add_header('User-Agent', 'UnityPlayer/2022.3.62f2 (UnityWebRequest/1.0, libcurl/8.10.1-DEV)')
+ req.add_header('X-Unity-Version', '2022.3.62f2')
+
+ try:
+ with urllib.request.urlopen(req, timeout=10) as response:
+ print(f"Status: {response.status}")
+ print(response.read())
+ except urllib.error.HTTPError as e:
+ print(f"HTTP {e.code}")
+ print(e.read()[:500])
+
+test_user_info_req()