환경 변수에서 네이버 비즈니스 프로필 리뷰를 가지고 올지, 네이버 지도를 가져올지 선택하도록 함, 공통 코드 분리

lib/naver_review_crawler.py

@@ -0,0 +1,159 @@
+import os, sys
+from datetime import datetime
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+from conf.config import (
+    PLACE_IDS, START_DATE, END_DATE, DEBUG, 
+    MESSAGE_PLATFORMS, MATTERMOST_URL, MATTERMOST_BOT_TOKEN, MATTERMOST_CHANNEL_ID
+    )
+from lib.send_message import MessageSender
+from lib.lib import (
+    create_mobile_driver,
+    get_start_end_dates,
+    parse_korean_date,
+    click_more,
+    extract_shop_name,
+    clean_html_text,
+    send_failure_message
+)
+
+class NaverReviewMapCollector:
+    def __init__(self):
+        self.driver = None
+        self.total_reviews = 0
+        self.start_date, self.end_date = get_start_end_dates(DEBUG, START_DATE, END_DATE)
+        self.reviews_by_place = {}
+
+    def extract_reviews(self):
+        reviews = []
+        WebDriverWait(self.driver, 10).until(
+            EC.presence_of_element_located((By.ID, "_review_list"))
+        )
+        ul = self.driver.find_element(By.ID, "_review_list")
+        items = ul.find_elements(By.XPATH, './/li[contains(@class, "place_apply_pui")]')
+        for item in items:
+            try:
+                writer = "익명"
+                try:
+                    writer = item.find_element(By.XPATH, "./div[1]/a[2]/div/span/span").text.strip()
+                except:
+                    pass
+
+                date_obj = None
+                try:
+                    date_text = item.find_element(By.XPATH, "./div[7]/div[2]/div/span[1]/span[2]").text.strip()
+                    date_obj = parse_korean_date(date_text)
+                except:
+                    continue
+
+                text = ""
+                try:
+                    text = item.find_element(By.XPATH, "./div[5]/a").get_attribute("innerHTML")
+                except:
+                    continue
+
+                if date_obj and (self.start_date <= date_obj <= self.end_date):
+                    reviews.append({
+                        "작성자": writer,
+                        "작성일": date_obj,
+                        "내용": clean_html_text(text)
+                    })
+            except Exception as e:
+                print(f"[WARN] 리뷰 추출 실패: {e}")
+        return reviews
+
+    def send_to_message(self):
+        today_str = datetime.today().strftime("%Y년 %m월 %d일")
+        now_str = datetime.now().strftime("%H:%M:%S")
+        lines = [f"##### {today_str} 네이버 지도 리뷰 크롤링 결과", ""]
+        lines.append(f"**수집 시간 :** {now_str}")
+        lines.append(f"**총 리뷰 수 :** {self.total_reviews}")
+        lines.append("")
+
+        for place_name, reviews in self.reviews_by_place.items():
+            lines.append(f"- {place_name}: {len(reviews)}건  ")
+        lines.append("\n---\n")
+
+        for idx, (place_name, reviews) in enumerate(self.reviews_by_place.items(), start=1):
+            lines.append(f"**{idx}. {place_name}**  ")
+            lines.append("")
+            if not reviews:
+                lines.append("- 리뷰 없음\n")
+            else:
+                for r in reviews:
+                    lines.append(f"- **작성일** : {r['작성일']}  ")
+                    lines.append(f"  **작성자** : {r['작성자']}  ")
+                    lines.append(f"  **내용** : {r['내용']}\n")
+            lines.append("---")
+
+        message = "\n".join(lines)
+        if not MESSAGE_PLATFORMS:
+            print("[WARN] 메시지 전송 플랫폼 없음")
+            print(f"[DEBUG] {message}")
+            return
+
+        sender = MessageSender(
+            mattermost_url=MATTERMOST_URL,
+            mattermost_token=MATTERMOST_BOT_TOKEN,
+            mattermost_channel_id=MATTERMOST_CHANNEL_ID,
+        )
+
+        if DEBUG:
+            print("[DEBUG] 디버그 모드로 메시지 미전송")
+            print(message)
+        else:
+            sender.send(message, platforms=MESSAGE_PLATFORMS, use_webhook=False)
+
+    def run(self):
+        self.driver = create_mobile_driver()
+
+        for place_id in PLACE_IDS:
+            url = f"https://m.place.naver.com/place/{place_id}/review/visitor?reviewSort=recent"
+            print(f"[INFO] 접근: {url}")
+            self.driver.get(url)
+            shop_name = extract_shop_name(self.driver)
+            all_reviews = []
+            seen = set()
+
+            while True:
+                new_reviews = self.extract_reviews()
+                if not new_reviews:
+                    break
+
+                filtered = []
+                for r in new_reviews:
+                    key = (r["작성자"], r["작성일"], r["내용"])
+                    if key not in seen:
+                        seen.add(key)
+                        filtered.append(r)
+
+                if not filtered:
+                    break
+
+                all_reviews.extend(filtered)
+                if not click_more(self.driver):
+                    break
+
+            print(f"[DONE] {shop_name}: {len(all_reviews)}건 수집")
+            self.total_reviews += len(all_reviews)
+            self.reviews_by_place[shop_name] = all_reviews
+
+        self.driver.quit()
+
+        if not self.reviews_by_place:
+            sender = MessageSender(
+                mattermost_url=MATTERMOST_URL,
+                mattermost_token=MATTERMOST_BOT_TOKEN,
+                mattermost_channel_id=MATTERMOST_CHANNEL_ID,
+            )
+            send_failure_message(sender, MESSAGE_PLATFORMS)
+        else:
+            self.send_to_message()
+
+if __name__ == "__main__":
+    collector = NaverReviewMapCollector()
+    collector.run()