root
717 lines
29 KiB
Python
717 lines
29 KiB
Python
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="./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'<img src="{local_img_path}" class="choice-image" style="max-height: 120px; display: block; margin-top: 8px; border-radius: 4px;" />'
|
|
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)
|