494 lines
22 KiB
Python
494 lines
22 KiB
Python
from DrissionPage import ChromiumPage, ChromiumOptions
|
|
from selectolax.parser import HTMLParser
|
|
import time
|
|
import re
|
|
from utils.logger import get_logger
|
|
from utils.path_utils import get_data_dir
|
|
|
|
class VOCScraper:
|
|
def __init__(self, headless_mode=True, settings=None):
|
|
"""
|
|
VOC 스크래퍼 초기화
|
|
|
|
Args:
|
|
headless_mode: 헤드리스 모드 여부
|
|
settings: 설정 딕셔너리 (None일 경우 settings.json에서 로드)
|
|
"""
|
|
self.logger = get_logger("Scraper")
|
|
self.page = None
|
|
self.headless_mode = headless_mode
|
|
|
|
# 설정 로드 (전달받은 settings 또는 파일에서 로드)
|
|
if settings is not None:
|
|
self.settings = settings
|
|
else:
|
|
self.settings = self._load_settings()
|
|
|
|
# 크롤링 설정 추출 (하드코딩 제거)
|
|
crawl_settings = self.settings.get('crawling', {})
|
|
self.target_depts = crawl_settings.get('target_depts', ['차량'])
|
|
self.keywords = crawl_settings.get('keywords', [])
|
|
self.filter_mode = crawl_settings.get('filter_mode', 'OR') # 'AND' 또는 'OR'
|
|
self.max_retries = crawl_settings.get('max_retries', 3) # 재시도 횟수
|
|
self.retry_delay = crawl_settings.get('retry_delay', 2) # 재시도 간격(초)
|
|
|
|
def _load_settings(self):
|
|
"""settings.json에서 설정 로드"""
|
|
try:
|
|
import json
|
|
settings_file = get_data_dir() / "settings.json"
|
|
if settings_file.exists():
|
|
with open(settings_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
else:
|
|
self.logger.warning("settings.json을 찾을 수 없습니다. 기본값 사용")
|
|
return {}
|
|
except Exception as e:
|
|
self.logger.error(f"설정 파일 로드 실패: {e}")
|
|
return {}
|
|
|
|
def start_browser(self):
|
|
"""브라우저(Chromium) 실행 및 초기 설정"""
|
|
if self.page: return True
|
|
co = ChromiumOptions()
|
|
|
|
# 1. 헤드리스 모드 설정 (User Setting)
|
|
co.headless(self.headless_mode)
|
|
co.auto_port()
|
|
|
|
# 2. 브라우저 보안/자동화 감지 우회 설정
|
|
# co.set_argument('--ignore-certificate-errors') # 인증서 에러 무시
|
|
# co.set_argument('--no-sandbox')
|
|
co.set_argument('--disable-gpu')
|
|
|
|
# 3. 브라우저 불필요한 팝업 및 기능 비활성화
|
|
co.set_pref('credentials_enable_service', False) # 비밀번호 저장 제안 끄기
|
|
co.set_pref('profile.password_manager_enabled', False)
|
|
|
|
# '자동제어 중' 알림바 제거
|
|
co.set_argument('--disable-infobars')
|
|
|
|
# 콘솔 로그 레벨 조정 (에러만 출력)
|
|
co.set_argument('--log-level=3')
|
|
|
|
co.mute(True) # 오디오 음소거
|
|
try:
|
|
self.page = ChromiumPage(co)
|
|
return True
|
|
except Exception as e:
|
|
self.logger.error(f"브라우저 실행 실패: {e}")
|
|
return False
|
|
|
|
def is_logged_in(self):
|
|
"""현재 로그인 상태인지 확인 (특정 요소 존재 여부로 판단)"""
|
|
if not self.page: return False
|
|
try:
|
|
return bool(self.page.ele("css:td#m_td2", timeout=2))
|
|
except:
|
|
return False
|
|
|
|
def login(self, uid, upw):
|
|
"""관리자 페이지 로그인 시도"""
|
|
if not self.page: self.start_browser()
|
|
try:
|
|
self.logger.info(f"{uid} 계정으로 로그인 시도 중...")
|
|
self.page.get("https://www.humetro.busan.kr/voc/admin/login_admin.jsp")
|
|
|
|
# 입력 폼 대기 및 입력
|
|
uid_field = self.page.ele("css:input[name='userID']", timeout=10)
|
|
if uid_field:
|
|
uid_field.input(uid)
|
|
self.page.ele("css:input[name='password']").input(upw)
|
|
self.page.ele("css:div.btn").click()
|
|
|
|
# 로그인 성공 여부 확인
|
|
if self.page.ele("css:td#m_td2", timeout=20):
|
|
self.logger.info("로그인 성공.")
|
|
time.sleep(1)
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
self.logger.error(f"로그인 에러: {e}")
|
|
return False
|
|
|
|
def _check_data_exists(self):
|
|
"""현재 페이지에 데이터 행(리스트)이 실제 존재하는지 확인"""
|
|
try:
|
|
html = self.page.html
|
|
tree = HTMLParser(html)
|
|
rows = tree.css("tr")
|
|
for row in rows:
|
|
bg = row.attributes.get('bgcolor', '').lower()
|
|
# 흰색 배경(#ffffff) 행이 실제 데이터 행임
|
|
if bg == '#ffffff' and len(row.css('td')) > 5:
|
|
return True
|
|
return False
|
|
except:
|
|
return False
|
|
|
|
def _navigate_to_list_via_physical_click(self):
|
|
"""메뉴 호버 및 클릭을 통해 전체 VOC 목록으로 이동"""
|
|
try:
|
|
menu_td = self.page.ele("css:td#m_td2")
|
|
if not menu_td: return False
|
|
|
|
# 1. 메뉴 호버링으로 서브메뉴 띄우기
|
|
menu_td.hover()
|
|
time.sleep(0.5)
|
|
|
|
# 2. '전체VOC보기' 링크 클릭
|
|
target_link = self.page.ele("text:전체VOC보기")
|
|
if target_link:
|
|
target_link.click()
|
|
time.sleep(2)
|
|
|
|
# 혹시 데이터가 안 보이면 검색 버튼 명시적으로 클릭
|
|
if not self._check_data_exists():
|
|
search_btn = self.page.ele("css:img[alt='검색']")
|
|
if search_btn:
|
|
search_btn.click()
|
|
time.sleep(2)
|
|
return self._check_data_exists()
|
|
return False
|
|
return False
|
|
except Exception as e:
|
|
self.logger.warning(f"목록 페이지 이동 중 에러: {e}")
|
|
return False
|
|
|
|
def fetch_list_pages(self, max_pages_limit=0, keywords=None, target_depts=None):
|
|
"""VOC 목록 페이지를 순회하며 데이터 수집"""
|
|
self.stop_requested = False
|
|
all_results = []
|
|
|
|
# 필터링 조건 설정
|
|
keywords = keywords if keywords else []
|
|
target_depts = target_depts if target_depts else self.target_depts
|
|
|
|
try:
|
|
# 0. 항상 목록 페이지로 이동하여 최신 상태 확보
|
|
# 캐시된 페이지 사용 시 신규 게시글 누락 가능하므로 조건부 체크 제거
|
|
if not self._navigate_to_list_via_physical_click():
|
|
self.logger.error("목록 페이지 접근 실패. 수집을 중단합니다.")
|
|
return {"status": "error", "msg": "목록 페이지 접근 실패"}
|
|
|
|
# 1. 페이지네이션 정보 확인
|
|
total_cnt_ele = self.page.ele("text:page )", timeout=2)
|
|
max_pages = 1
|
|
if total_cnt_ele:
|
|
# 예: ( 1 page / 7 page )
|
|
txt = total_cnt_ele.text
|
|
if "/" in txt:
|
|
try:
|
|
max_pages = int(txt.split("/")[1].replace("page", "").replace(")", "").strip())
|
|
except: pass
|
|
|
|
# 수집 한도 설정 (인자 우선, 없으면 self.settings 등 참조해야 하나,
|
|
# 여기선 인자가 0이면 settings가 있을 경우 참조하는 식으로 구성)
|
|
# 그러나 self.settings 직접 참조가 없으므로 인자에 의존하거나 동적 계산
|
|
if max_pages_limit > 0:
|
|
if max_pages > max_pages_limit:
|
|
max_pages = max_pages_limit
|
|
|
|
self.logger.info(f"수집 대상 페이지 수: {max_pages}")
|
|
|
|
# 2. 페이지 순회
|
|
for page_num in range(1, max_pages + 1):
|
|
if self.stop_requested: break
|
|
|
|
self.logger.debug(f"페이지 {page_num} 수집 중...")
|
|
if page_num > 1:
|
|
# 페이지 번호 클릭 ([2], [3]...)
|
|
# 자바스크립트 함수 호출이므로 text로 찾아서 클릭
|
|
page_btn = self.page.ele(f"text:[{page_num}]")
|
|
if page_btn:
|
|
page_btn.click()
|
|
time.sleep(1.5)
|
|
else: break
|
|
|
|
# 현재 페이지 HTML 파싱
|
|
tree = HTMLParser(self.page.html)
|
|
|
|
# 배경색이 흰색인 행만 데이터 행으로 간주 (헤더 제외)
|
|
rows = tree.css("tr[bgcolor='#ffffff']")
|
|
|
|
for row in rows:
|
|
if self.stop_requested: break
|
|
cols = row.css("td")
|
|
if len(cols) < 7: continue
|
|
|
|
# [0]접수번호, [1]채널, [2]제목, [3]고객명, [4]부서, [5]공개, [6]상태
|
|
voc_id = cols[0].text(strip=True)
|
|
channel = cols[1].text(strip=True)
|
|
|
|
# 제목: a 태그의 title 속성(전체 제목) 우선, 없으면 텍스트
|
|
title_a = cols[2].css_first("a")
|
|
title = title_a.attributes.get("title", "") if title_a else ""
|
|
if not title and title_a: title = title_a.text(strip=True)
|
|
|
|
writer = cols[3].text(strip=True)
|
|
dept = cols[4].text(strip=True)
|
|
|
|
is_public_txt = cols[5].text(strip=True)
|
|
is_public = 1 if is_public_txt == '○' or is_public_txt == 'O' else 0
|
|
|
|
# 비공개 게시글도 메타데이터에 포함 (is_public=0으로 마킹)
|
|
# 상세 페이지 접근 시 권한 문제 발생 가능하므로 로깅만 수행
|
|
if is_public == 0:
|
|
self.logger.debug(f"비공개 게시글 수집: {voc_id} (상세 조회 스킵 예정)")
|
|
|
|
status = cols[6].text(strip=True)
|
|
|
|
post_info = {
|
|
"id": voc_id,
|
|
"title": title,
|
|
"writer": writer,
|
|
"department": dept,
|
|
"is_public": is_public,
|
|
"status": status,
|
|
"channel": channel,
|
|
"is_related": 0
|
|
}
|
|
|
|
# 타겟 부서/키워드 필터링 (고도화된 로직)
|
|
is_target = self._check_filter_match(title, dept)
|
|
|
|
post_info["is_related"] = 1 if is_target else 0
|
|
|
|
# 상세 수집 전략: 관련성 있는 게시글만 상세 조회 시도
|
|
# 비공개 게시글은 권한 오류 가능하지만, try-catch로 처리됨
|
|
# 차량 등 관련 부서의 경우 is_public 상관없이 시도 (update_detail은 이미 오류처리)
|
|
|
|
if is_target:
|
|
# 관심 게시글: 공개/비공개 구분 없이 상세 조회 시도
|
|
# 비공개면 fetch_detail_content에서 타임아웃으로 처리됨
|
|
detail = self.fetch_detail_content(voc_id)
|
|
if detail:
|
|
# 상세에서 가져온 더 정확한 정보로 덮어쓰기
|
|
post_info.update(detail)
|
|
|
|
all_results.append(post_info)
|
|
|
|
return {"status": "success", "data": all_results}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"목록 수집 중 에러: {e}")
|
|
return {"status": "error", "msg": str(e)}
|
|
|
|
def fetch_detail_content(self, voc_id):
|
|
"""특정 VOC의 모든 상세 데이터를 추출 (제목, 역명, 질의내용, 답변, 첨부파일, 작성자 등)"""
|
|
try:
|
|
# 1. 목록에서 링크 찾기 시도
|
|
target_link = self.page.ele(f"css:a[href*='vocCode={voc_id}']", timeout=1)
|
|
|
|
navigated_directly = False
|
|
|
|
if target_link:
|
|
# 링크가 있으면 클릭
|
|
target_link.click()
|
|
else:
|
|
# 링크가 없으면 (페이지가 넘어갔거나 등) 직접 URL 이동 시도
|
|
# URL 패턴 추정: /voc/admin/view.jsp?vocCode={id}
|
|
# 주의: 세션이 유지되어야 함
|
|
direct_url = f"https://www.humetro.busan.kr/voc/admin/view.jsp?vocCode={voc_id}"
|
|
self.logger.info(f" -> 목록에서 링크 없음, 직접 이동 시도: {voc_id}")
|
|
self.page.get(direct_url)
|
|
navigated_directly = True
|
|
|
|
# 2. 상세 페이지 로딩 대기
|
|
# 성공 시 '질의내용' 텍스트 존재
|
|
if self.page.ele("text:질의내용", timeout=10):
|
|
tree = HTMLParser(self.page.html)
|
|
data = {
|
|
"id": voc_id,
|
|
"title": "", "station": "", "content": "", "attachment": "",
|
|
"answer": "", "channel": "", "date": "", "voc_type": "",
|
|
"response_type": "", "summary": "",
|
|
"writer": "" # 작성자 추가
|
|
}
|
|
|
|
# 제목 추출
|
|
title_a = tree.css_first("td.title, td[style*='font-weight:bold']")
|
|
if title_a: data["title"] = title_a.text(strip=True)
|
|
|
|
# 라벨 기반 데이터 추출
|
|
labels = tree.css("td[bgcolor='#E0EDEF']")
|
|
for label in labels:
|
|
key = label.text(strip=True)
|
|
|
|
parent = label.parent
|
|
if not parent: continue
|
|
tds = parent.css("td")
|
|
|
|
idx = -1
|
|
for i, td in enumerate(tds):
|
|
if td == label:
|
|
idx = i
|
|
break
|
|
|
|
if idx != -1 and idx + 1 < len(tds):
|
|
val_node = tds[idx+1]
|
|
val_txt = val_node.text(separator="\n", strip=True)
|
|
|
|
if "제목" in key and not data["title"]: data["title"] = val_txt
|
|
elif "역명" in key: data["station"] = val_txt
|
|
elif "질의내용" in key: data["content"] = val_txt
|
|
elif "첨부파일" in key: data["attachment"] = val_txt
|
|
elif "답변내용" in key: data["answer"] = val_txt
|
|
elif "접수채널" in key: data["channel"] = val_txt
|
|
elif "등록일자" in key: data["date"] = val_txt
|
|
elif "VOC유형" in key: data["voc_type"] = val_txt
|
|
elif "요약" in key: data["summary"] = val_txt
|
|
elif "고객명" in key: data["writer"] = val_txt # 작성자 추출
|
|
|
|
elif "응답구분" in key:
|
|
checked_input = val_node.css_first("input[checked]")
|
|
if checked_input:
|
|
val = checked_input.attributes.get('value', '6')
|
|
mapping = {'1':'게시판', '2':'우편', '3':'팩스', '4':'전화', '5':'E-mail', '6':'기타'}
|
|
data["response_type"] = mapping.get(val, "기타")
|
|
else:
|
|
data["response_type"] = ""
|
|
|
|
# 복귀 (직접 이동했으면 뒤로가기 혹은 목록으로 이동)
|
|
if navigated_directly:
|
|
# 목록으로 명시적 이동보다는 뒤로가기가 안전할 수 있음
|
|
# 하지만 직접 이동했으므로 history.back()이 먹힐지 확인 필요.
|
|
# 일단 back() 시도
|
|
self.page.back()
|
|
|
|
# 만약 back() 후에도 여전히 view 페이지면 강제 목록 이동
|
|
if "view.jsp" in self.page.url:
|
|
# 목록 URL로 이동 (검색 조건 유지 안될 수 있음 주의)
|
|
# self.page.get("https://www.humetro.busan.kr/voc/admin/list.jsp?act=searchList")
|
|
# 일단 back 한번 더 시도해보고 안되면 둠 (다음 루프에서 처리)
|
|
pass
|
|
else:
|
|
self.page.back()
|
|
|
|
time.sleep(0.5)
|
|
return data
|
|
else:
|
|
self.logger.error(f" -> [Error] 상세 페이지 {voc_id} 로딩 실패/타임아웃. URL: {self.page.url}")
|
|
if "vocCode=" in self.page.url:
|
|
self.page.back()
|
|
return None
|
|
|
|
except Exception as e:
|
|
self.logger.error(f" -> [Exception] {voc_id} Detail: {e}")
|
|
if "vocCode=" in self.page.url:
|
|
try: self.page.back()
|
|
except: pass
|
|
return None
|
|
|
|
def download_attachment(self, voc_id, save_dir):
|
|
"""특정 VOC의 첨부파일을 다운로드합니다."""
|
|
if not self.page: self.start_browser()
|
|
|
|
try:
|
|
# 1. 상세 페이지 이동 확인 (현재 아니면 이동)
|
|
if f"vocCode={voc_id}" not in self.page.url:
|
|
# 목록에서 찾아서 클릭 (이미 fetch_detail_content에서 쓰던 로직 재사용)
|
|
target_link = self.page.ele(f"css:a[href*='vocCode={voc_id}']", timeout=5)
|
|
if not target_link:
|
|
return None, "목록에서 게시글을 찾을 수 없습니다."
|
|
target_link.click()
|
|
self.page.wait.ele("text:질의내용")
|
|
|
|
# 2. 첨부파일 링크 찾기
|
|
# '첨부파일' 라벨 옆의 td 안의 a 태그 탐색
|
|
label = self.page.ele("text:첨부파일", timeout=5)
|
|
if not label:
|
|
return None, "첨부파일 항목을 찾을 수 없습니다."
|
|
|
|
# 레이아웃: td(라벨) -> td(값) -> a(링크)
|
|
val_td = label.parent().next()
|
|
link = val_td.ele("css:a")
|
|
|
|
if not link:
|
|
return None, "첨부파일 링크가 없습니다."
|
|
|
|
# 3. 다운로드 시작
|
|
filename = link.text
|
|
self.logger.info(f"다운로드 중: {filename}...")
|
|
|
|
# DrissionPage의 download 기능 사용
|
|
# save_path가 디렉토리이면 그 안에 저장됨
|
|
res = self.page.download(link, save_path=save_dir)
|
|
|
|
if res:
|
|
# res는 성공 시 (파일명, 경로) 튜플 혹은 성공여부 반환 (버전에 따라 다름)
|
|
# 보통 (True, 최종완료경로) 형태
|
|
final_path = res[1] if isinstance(res, tuple) else res
|
|
return final_path, None
|
|
else:
|
|
return None, "다운로드에 실패했습니다."
|
|
|
|
except Exception as e:
|
|
return None, f"다운로드 중 예외 발생: {e}"
|
|
finally:
|
|
# 상세 페이지에서 다시 목록으로 (필요시)
|
|
if "vocCode=" in self.page.url:
|
|
self.page.back()
|
|
time.sleep(0.5)
|
|
|
|
def _check_filter_match(self, title: str, dept: str) -> bool:
|
|
"""
|
|
고도화된 필터링 로직
|
|
|
|
설정에 따라 AND/OR 모드로 동작하며, 단어 경계 매칭을 지원합니다.
|
|
|
|
Args:
|
|
title: 게시글 제목
|
|
dept: 담당 부서
|
|
|
|
Returns:
|
|
bool: 필터링 조건 충족 여부
|
|
"""
|
|
if not title:
|
|
title = ""
|
|
if not dept:
|
|
dept = ""
|
|
|
|
# 키워드 매칭 (단어 경계 고려)
|
|
match_keyword = False
|
|
if self.keywords:
|
|
for kw in self.keywords:
|
|
if not kw:
|
|
continue
|
|
# 단순 substring 매칭 (기존 방식 유지하면서 개선)
|
|
if kw in title:
|
|
match_keyword = True
|
|
self.logger.debug(f"키워드 매칭: '{kw}' in '{title[:30]}...'")
|
|
break
|
|
|
|
# 부서 매칭 (정확한 매칭 또는 시작swith)
|
|
match_dept = False
|
|
if self.target_depts:
|
|
for target in self.target_depts:
|
|
if not target:
|
|
continue
|
|
# 정확한 일치 또는 부서명이 타겟으로 시작/포함
|
|
if target == dept or dept.startswith(target) or target in dept:
|
|
match_dept = True
|
|
self.logger.debug(f"부서 매칭: '{target}' in '{dept}'")
|
|
break
|
|
|
|
# 필터 모드에 따른 결과 반환
|
|
if self.filter_mode == "AND":
|
|
# 키워드가 없으면 부서만, 부서가 없으면 키워드만 체크
|
|
if self.keywords and self.target_depts:
|
|
return match_keyword and match_dept
|
|
elif self.keywords:
|
|
return match_keyword
|
|
elif self.target_depts:
|
|
return match_dept
|
|
else:
|
|
return False
|
|
else: # OR 모드 (기본값)
|
|
return match_keyword or match_dept
|
|
|
|
def close(self):
|
|
"""브라우저 종료"""
|
|
if self.page:
|
|
self.page.quit()
|