311 lines
14 KiB
Python
311 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
네이버 지도 + 블로그 통합 크롤러 (리팩토링 버전)
|
|
메인 실행 파일
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import platform
|
|
import json
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Dict
|
|
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.options import Options
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
|
|
from .logger_utils import setup_logger, config
|
|
from .naver_map_crawler import NaverMapCrawler
|
|
from .naver_blog_crawler import NaverBlogCrawler, ImageFilterConfig
|
|
|
|
|
|
class NaverIntegratedCrawler:
|
|
def __init__(self, project_dir:str,
|
|
image_filter: ImageFilterConfig = None):
|
|
"""
|
|
초기화
|
|
Args:
|
|
project_dir: 프로젝트 디렉토리
|
|
image_filter: 이미지 필터 설정
|
|
"""
|
|
self.project_dir = project_dir
|
|
self.image_filter = image_filter or ImageFilterConfig()
|
|
|
|
# 디렉토리 생성
|
|
os.makedirs(os.path.join(self.project_dir, 'debug'), exist_ok=True)
|
|
os.makedirs(os.path.join(self.project_dir, 'data'), exist_ok=True)
|
|
os.makedirs(os.path.join(self.project_dir, 'logs'), exist_ok=True)
|
|
os.makedirs(os.path.join(self.project_dir, 'images'), exist_ok=True)
|
|
|
|
# 로거 설정
|
|
self.logger = setup_logger(
|
|
'NaverIntegratedCrawler',
|
|
os.path.join(self.project_dir, 'logs'),
|
|
'integrated_crawler'
|
|
)
|
|
|
|
self.driver = None
|
|
self.wait = None
|
|
self.map_crawler = None
|
|
self.blog_crawler = None
|
|
|
|
def setup_driver(self):
|
|
"""웹드라이버 설정"""
|
|
self.logger.info("웹드라이버 설정 시작...")
|
|
|
|
# 설정 파일에서 웹드라이버 설정 읽기
|
|
webdriver_config = config.get('webdriver', {})
|
|
|
|
chrome_options = Options()
|
|
chrome_options.add_argument('--no-sandbox')
|
|
chrome_options.add_argument('--disable-dev-shm-usage')
|
|
chrome_options.add_argument('--disable-blink-features=AutomationControlled')
|
|
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
|
chrome_options.add_experimental_option('useAutomationExtension', False)
|
|
|
|
# 추가 안정성 옵션들
|
|
chrome_options.add_argument('--disable-gpu')
|
|
chrome_options.add_argument(f'--window-size={webdriver_config.get("window_size", "1920,1080")}')
|
|
chrome_options.add_argument('--disable-web-security')
|
|
chrome_options.add_argument('--allow-running-insecure-content')
|
|
|
|
# User-Agent 설정
|
|
user_agent = webdriver_config.get('user_agent',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')
|
|
chrome_options.add_argument(f'--user-agent={user_agent}')
|
|
|
|
# 디버깅을 위해 headless 모드 설정
|
|
headless = webdriver_config.get('headless', False)
|
|
if headless or (platform.system() == 'Linux' and not os.environ.get('DISPLAY')):
|
|
chrome_options.add_argument('--headless')
|
|
self.logger.info(" - Headless 모드로 실행")
|
|
else:
|
|
self.logger.info(" - GUI 모드로 실행")
|
|
|
|
self.driver = webdriver.Chrome(options=chrome_options)
|
|
|
|
# 페이지 로드 타임아웃 설정
|
|
page_load_timeout = webdriver_config.get('page_load_timeout', 30)
|
|
implicit_wait = webdriver_config.get('implicit_wait', 10)
|
|
|
|
self.driver.set_page_load_timeout(page_load_timeout)
|
|
self.driver.implicitly_wait(implicit_wait)
|
|
|
|
self.wait = WebDriverWait(self.driver, 30) # 20 -> 30초로 증가
|
|
#self.driver.maximize_window()
|
|
|
|
# 자동화 감지 방지
|
|
self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
|
|
|
|
# 크롤러 인스턴스 생성
|
|
self.map_crawler = NaverMapCrawler(self.driver, self.wait, self.project_dir)
|
|
self.blog_crawler = NaverBlogCrawler(self.driver, self.wait, self.project_dir, self.image_filter)
|
|
|
|
self.logger.info(" ✓ 웹드라이버 초기화 완료\n")
|
|
|
|
def collect_all_info(self, query: str = "선돌막국수", max_blogs: int = 5) -> Dict:
|
|
"""
|
|
지도 정보와 블로그 정보 통합 수집
|
|
|
|
Args:
|
|
query: 검색어
|
|
max_blogs: 최대 블로그 수집 개수
|
|
|
|
Returns:
|
|
수집 결과 딕셔너리
|
|
"""
|
|
self.logger.info(f"=== {query} 통합 정보 수집 시작 ===\n")
|
|
|
|
results = {
|
|
'query': query,
|
|
'collection_time': datetime.now().isoformat(),
|
|
'place_info': {},
|
|
'map_blog_reviews': [],
|
|
'detailed_blogs': [],
|
|
'photo_urls': [],
|
|
'total_blogs_visited': 0
|
|
}
|
|
|
|
try:
|
|
# 1. 네이버 지도 검색
|
|
if not self.map_crawler.search_place(query):
|
|
self.logger.error("장소 검색 실패")
|
|
return results
|
|
|
|
# 2. 지도에서 기본 정보 수집
|
|
place_info = self.map_crawler.collect_place_info()
|
|
if place_info:
|
|
results['place_info'] = place_info
|
|
self.logger.info(f"✓ 장소 기본 정보 수집 완료: {place_info.get('name', 'N/A')}")
|
|
|
|
# 3. 사진 수집
|
|
self.logger.info("\n=== 사진 수집 시작 ===")
|
|
photo_urls = self.map_crawler.collect_photo_urls()
|
|
results['photo_urls'] = photo_urls
|
|
self.logger.info(f"✓ 사진 {len(photo_urls)}개 수집 완료\n")
|
|
|
|
# 4. 지도의 블로그 리뷰 수집
|
|
map_reviews = self.map_crawler.collect_map_blog_reviews()
|
|
results['map_blog_reviews'] = map_reviews
|
|
self.logger.info(f"✓ 지도에서 블로그 리뷰 {len(map_reviews)}개 수집\n")
|
|
|
|
# 5. 블로그 상세 정보 수집
|
|
blog_links = self.map_crawler.collect_blog_links(max_blogs)
|
|
|
|
if blog_links:
|
|
self.logger.info(f"✓ {len(blog_links)}개의 블로그 링크 수집 완료\n")
|
|
|
|
# 각 블로그 방문
|
|
for idx, (title, url) in enumerate(blog_links):
|
|
self.logger.info(f"\n[{idx+1}/{len(blog_links)}] 블로그 방문 중...")
|
|
self.logger.info(f"제목: {title}")
|
|
self.logger.info(f"URL: {url[:80]}...")
|
|
|
|
blog_content = self.blog_crawler.visit_and_extract_blog(url, title, idx+1)
|
|
if blog_content:
|
|
results['detailed_blogs'].append(blog_content)
|
|
results['total_blogs_visited'] += 1
|
|
self.logger.info(f"✓ 내용 수집 완료 (사진 {len(blog_content['images'])}개 포함)")
|
|
|
|
# 다음 블로그를 위해 목록으로 돌아가기
|
|
if idx < len(blog_links) - 1:
|
|
self.map_crawler.return_to_blog_list()
|
|
time.sleep(1)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"\n오류 발생: {e}", exc_info=True)
|
|
|
|
# 결과 저장
|
|
self._save_integrated_results(results)
|
|
|
|
return results
|
|
|
|
def _save_integrated_results(self, results: Dict):
|
|
"""통합 결과 저장"""
|
|
# JSON 저장
|
|
json_file = os.path.join(self.project_dir, 'data',
|
|
f'integrated_result_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json')
|
|
|
|
with open(json_file, 'w', encoding='utf-8') as f:
|
|
json.dump(results, f, ensure_ascii=False, indent=2)
|
|
|
|
# 이미지 다운로드
|
|
download_info = self.blog_crawler.download_all_images(results)
|
|
results['image_download_info'] = download_info
|
|
|
|
# 텍스트 보고서 작성
|
|
self._generate_report(results)
|
|
|
|
self.logger.info(f"\n=== 수집 완료 ===")
|
|
self.logger.info(f"JSON 파일: {json_file}")
|
|
|
|
def _generate_report(self, results: Dict):
|
|
"""텍스트 보고서 생성"""
|
|
report_file = os.path.join(self.project_dir, 'data',
|
|
f'integrated_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.txt')
|
|
|
|
with open(report_file, 'w', encoding='utf-8') as f:
|
|
f.write(f"=== {results['query']} 통합 수집 결과 ===\n\n")
|
|
f.write(f"수집 일시: {results['collection_time']}\n\n")
|
|
|
|
# 장소 정보
|
|
f.write("=== 장소 기본 정보 ===\n")
|
|
place = results['place_info']
|
|
if place:
|
|
f.write(f"상호명: {place.get('name', 'N/A')}\n")
|
|
f.write(f"주소: {place.get('address', 'N/A')}\n")
|
|
f.write(f"전화번호: {place.get('tel', 'N/A')}\n")
|
|
f.write(f"카테고리: {place.get('category', 'N/A')}\n")
|
|
f.write(f"영업시간: {place.get('business_hours', 'N/A')}\n")
|
|
f.write(f"홈페이지: {place.get('homepage', 'N/A')}\n")
|
|
f.write(f"설명: {place.get('description', 'N/A')}\n")
|
|
|
|
# 사진 정보
|
|
f.write(f"\n=== 사진 정보 ===\n")
|
|
f.write(f"수집한 사진 URL 수: {len(results['photo_urls'])}개\n")
|
|
for i, url in enumerate(results['photo_urls'][:5], 1):
|
|
f.write(f"{i}. {url}\n")
|
|
if len(results['photo_urls']) > 5:
|
|
f.write(f"... 외 {len(results['photo_urls'])-5}개\n")
|
|
|
|
# 지도 블로그 리뷰
|
|
f.write(f"\n=== 지도 블로그 리뷰 ===\n")
|
|
f.write(f"리뷰 수: {len(results['map_blog_reviews'])}개\n")
|
|
|
|
ad_count = sum(1 for r in results['map_blog_reviews'] if r['is_ad'])
|
|
f.write(f"광고성 리뷰: {ad_count}개\n\n")
|
|
|
|
for i, review in enumerate(results['map_blog_reviews'][:5], 1):
|
|
f.write(f"{i}. {review['title']}\n")
|
|
if review['is_ad']:
|
|
f.write(f" [광고] {', '.join(review['ad_phrases'])}\n")
|
|
f.write(f" 날짜: {review.get('date', 'N/A')}\n")
|
|
|
|
# 상세 블로그 정보
|
|
f.write(f"\n=== 상세 블로그 정보 ===\n")
|
|
f.write(f"방문한 블로그 수: {results['total_blogs_visited']}개\n\n")
|
|
|
|
total_blog_images = 0
|
|
for blog in results['detailed_blogs']:
|
|
f.write(f"\n[{blog['index']}번째 블로그]\n")
|
|
f.write(f"제목: {blog['title']}\n")
|
|
f.write(f"블로그 타이틀: {blog['blog_title']}\n")
|
|
f.write(f"URL: {blog['url']}\n")
|
|
if blog['is_ad']:
|
|
f.write(f"광고성 포스트: YES\n")
|
|
f.write(f"광고 문구: {', '.join(blog['ad_phrases'][:2])}\n")
|
|
f.write(f"블로그 내 이미지: {len(blog['images'])}개\n")
|
|
|
|
if blog['images']:
|
|
f.write("이미지 URL:\n")
|
|
for i, img_url in enumerate(blog['images'][:3], 1):
|
|
f.write(f" {i}. {img_url}\n")
|
|
if len(blog['images']) > 3:
|
|
f.write(f" ... 외 {len(blog['images'])-3}개\n")
|
|
|
|
f.write(f"내용 (일부):\n{blog['content'][:300]}...\n")
|
|
f.write("-" * 80 + "\n")
|
|
|
|
total_blog_images += len(blog['images'])
|
|
|
|
# 통계
|
|
f.write(f"\n=== 수집 통계 ===\n")
|
|
f.write(f"총 수집된 이미지 URL: {len(results['photo_urls']) + total_blog_images}개\n")
|
|
f.write(f" - 사진탭: {len(results['photo_urls'])}개\n")
|
|
f.write(f" - 블로그: {total_blog_images}개\n")
|
|
|
|
# 이미지 다운로드 정보
|
|
if 'image_download_info' in results:
|
|
dl_info = results['image_download_info']
|
|
f.write(f"\n=== 이미지 다운로드 ===\n")
|
|
f.write(f"다운로드 디렉토리: {dl_info.get('download_dir', 'N/A')}\n")
|
|
f.write(f"다운로드 성공: {dl_info.get('downloaded', 0)}개\n")
|
|
f.write(f"전체 처리: {dl_info.get('total_images', 0)}개\n")
|
|
|
|
if 'statistics' in dl_info:
|
|
stats = dl_info['statistics']
|
|
f.write(f"\n상세 통계:\n")
|
|
f.write(f" - 신규 저장: {stats.get('success', 0)}개\n")
|
|
f.write(f" - 기존 파일: {stats.get('file_exists', 0)}개\n")
|
|
f.write(f" - 중복 제거: {stats.get('duplicate', 0)}개\n")
|
|
f.write(f" - 크기 필터링: {stats.get('size_filtered', 0)}개\n")
|
|
f.write(f" - 파일크기 필터링: {stats.get('file_size_filtered', 0)}개\n")
|
|
f.write(f" - 포맷 필터링: {stats.get('format_filtered', 0)}개\n")
|
|
|
|
# 광고성 포스트 통계
|
|
ad_blogs = sum(1 for b in results['detailed_blogs'] if b['is_ad'])
|
|
f.write(f"\n광고성 포스트: {ad_blogs}/{len(results['detailed_blogs'])}개 ")
|
|
if results['detailed_blogs']:
|
|
f.write(f"({ad_blogs/len(results['detailed_blogs'])*100:.1f}%)\n")
|
|
|
|
self.logger.info(f"보고서 저장: {report_file}")
|
|
|
|
def close(self):
|
|
"""리소스 정리"""
|
|
if self.driver:
|
|
self.driver.quit()
|
|
self.logger.info("웹드라이버 종료")
|