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"![image]({local_path})") else: img.replace_with(f"![image]({src})") 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="./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="./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)