# ============================================================ # 뉴스 키워드 자동 감지 + 카카오톡 전송 봇 v3 # ============================================================ import schedule, time, requests, json, os, base64 from datetime import datetime from playwright.sync_api import sync_playwright from config import 키워드_목록, IMGBB_API_KEY, 체크_주기_분 os.makedirs("screenshots", exist_ok=True) 알림_보낸_URL = set() # ── 사이트 설정 (각 사이트별 URL과 계정) ────────────────── 사이트_목록 = [ { "이름": "더벨", "로그인_URL": "https://www.thebell.co.kr/front/common/login.asp", "기사_URL": "https://www.thebell.co.kr/front/AllHeadLine.asp", "아이디": "dscinvest", "비밀번호": "the2221", "활성화": True, }, { "이름": "딜사이트플러스", "로그인_URL": "https://dealsiteplus.co.kr/member/login", "기사_URL": "https://dealsiteplus.co.kr/newsflash", "아이디": "dscinvest", "비밀번호": "dscinvest3190*", "활성화": True, }, { "이름": "넘버스", "로그인_URL": "https://www.numbers.co.kr/", "기사_URL": "https://www.numbers.co.kr/news/articleList.html?sc_multi_code=S1&view_type=sm", "아이디": "dscinvest", "비밀번호": "dsc3190*", "활성화": True, }, ] # ── 카카오 토큰 ──────────────────────────────────────────── def 토큰_불러오기(): if not os.path.exists("kakao_token.json"): print("❌ kakao_token.json 없음 → get_kakao_token.py 먼저 실행하세요") exit() with open("kakao_token.json", "r") as f: return json.load(f) def 토큰_갱신(info): resp = requests.post("https://kauth.kakao.com/oauth/token", data={ "grant_type": "refresh_token", "client_id": info["REST_API_KEY"], "refresh_token": info["refresh_token"], }) if resp.status_code == 200: data = resp.json() info["access_token"] = data["access_token"] if "refresh_token" in data: info["refresh_token"] = data["refresh_token"] with open("kakao_token.json", "w") as f: json.dump(info, f, ensure_ascii=False, indent=2) return info def 이미지_업로드(경로): try: with open(경로, "rb") as f: data = base64.b64encode(f.read()).decode() resp = requests.post("https://api.imgbb.com/1/upload", data={ "key": IMGBB_API_KEY, "image": data, "expiration": 86400, }, timeout=15) if resp.status_code == 200: return resp.json()["data"]["url"] except Exception as e: print(f" ⚠️ 이미지 업로드 실패: {e}") return None def 카카오_전송(사이트명, 키워드, 제목, 기사URL, 이미지URL=None): info = 토큰_불러오기() if 이미지URL: template = { "object_type": "feed", "content": { "title": f"[{사이트명}] {키워드} 감지", "description": 제목[:80], "image_url": 이미지URL, "link": {"web_url": 기사URL, "mobile_web_url": 기사URL} } } else: template = { "object_type": "text", "text": f"🔔 [{사이트명}] 키워드: {키워드}\n\n{제목}\n\n{기사URL}", "link": {"web_url": 기사URL, "mobile_web_url": 기사URL} } def _전송(token): return requests.post( "https://kapi.kakao.com/v2/api/talk/memo/default/send", headers={"Authorization": f"Bearer {token}"}, data={"template_object": json.dumps(template, ensure_ascii=False)}, timeout=10 ) resp = _전송(info["access_token"]) if resp.status_code == 401: info = 토큰_갱신(info) resp = _전송(info["access_token"]) return resp.status_code == 200 # ── 로그인 함수 ──────────────────────────────────────────── def 로그인_더벨(page, 아이디, 비밀번호): try: page.goto("https://www.thebell.co.kr/front/common/login.asp", wait_until="domcontentloaded", timeout=20000) page.wait_for_timeout(2000) # 더벨 로그인 폼 page.fill('input[name="userId"]', 아이디) page.fill('input[name="userPwd"]', 비밀번호) page.keyboard.press("Enter") page.wait_for_timeout(3000) print(" ✅ 더벨 로그인 완료") return True except Exception as e: print(f" ⚠️ 더벨 로그인 오류: {e}") return False def 로그인_딜사이트(page, 아이디, 비밀번호): try: page.goto("https://dealsiteplus.co.kr/member/login", wait_until="domcontentloaded", timeout=20000) page.wait_for_timeout(2000) # 딜사이트플러스 로그인 폼 (여러 selector 시도) selectors_id = [ 'input[name="loginId"]', 'input[name="user_id"]', 'input[name="userid"]', 'input[name="id"]', 'input[type="text"]', ] selectors_pw = [ 'input[name="loginPwd"]', 'input[name="user_pw"]', 'input[name="password"]', 'input[name="passwd"]', 'input[type="password"]', ] for sel in selectors_id: try: page.fill(sel, 아이디, timeout=2000) break except: continue for sel in selectors_pw: try: page.fill(sel, 비밀번호, timeout=2000) break except: continue page.keyboard.press("Enter") page.wait_for_timeout(3000) print(" ✅ 딜사이트플러스 로그인 완료") return True except Exception as e: print(f" ⚠️ 딜사이트플러스 로그인 오류: {e}") return False def 로그인_넘버스(page, 아이디, 비밀번호): try: page.goto("https://www.numbers.co.kr/", wait_until="domcontentloaded", timeout=20000) page.wait_for_timeout(2000) # 로그인 버튼 찾아서 클릭 try: page.click('a:has-text("로그인")', timeout=3000) page.wait_for_timeout(1500) except: pass selectors_id = [ 'input[name="member_id"]', 'input[name="userId"]', 'input[name="id"]', 'input[type="text"]', ] selectors_pw = [ 'input[name="member_pw"]', 'input[name="userPwd"]', 'input[name="password"]', 'input[type="password"]', ] for sel in selectors_id: try: page.fill(sel, 아이디, timeout=2000) break except: continue for sel in selectors_pw: try: page.fill(sel, 비밀번호, timeout=2000) break except: continue page.keyboard.press("Enter") page.wait_for_timeout(3000) print(" ✅ 넘버스 로그인 완료") return True except Exception as e: print(f" ⚠️ 넘버스 로그인 오류: {e}") return False 로그인_함수 = { "더벨": 로그인_더벨, "딜사이트플러스": 로그인_딜사이트, "넘버스": 로그인_넘버스, } # ── 기사 스캔 ───────────────────────────────────────────── def 기사_스캔(page, 사이트): 발견_목록 = [] try: page.goto(사이트["기사_URL"], wait_until="domcontentloaded", timeout=20000) page.wait_for_timeout(2000) base = "/".join(사이트["기사_URL"].split("/")[:3]) for link in page.query_selector_all("a[href]"): try: 제목 = (link.get_attribute("title") or link.inner_text() or "").strip() href = (link.get_attribute("href") or "").strip() if len(제목) < 6 or not href: continue if href.startswith("/"): href = base + href if not href.startswith("http") or href in 알림_보낸_URL: continue for 키워드 in 키워드_목록: if 키워드 in 제목: 발견_목록.append({"키워드": 키워드, "제목": 제목, "URL": href}) 알림_보낸_URL.add(href) break except: continue except Exception as e: print(f" ⚠️ 스캔 오류: {e}") return 발견_목록 def 스크린샷_찍기(context, url, 파일명): 페이지 = context.new_page() try: 페이지.goto(url, wait_until="domcontentloaded", timeout=15000) 페이지.wait_for_timeout(2000) 경로 = f"screenshots/{파일명}.png" 페이지.screenshot(path=경로, clip={"x": 0, "y": 0, "width": 1280, "height": 900}) return 경로 except Exception as e: print(f" ⚠️ 스크린샷 오류: {e}") return None finally: 페이지.close() # ── 메인 체크 ───────────────────────────────────────────── def 뉴스_체크(): 지금 = datetime.now().strftime("%H:%M:%S") print(f"\n{'='*45}") print(f"[{지금}] 뉴스 체크 시작") print(f"{'='*45}") with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context( viewport={"width": 1280, "height": 900}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" ) for 사이트 in 사이트_목록: if not 사이트.get("활성화", True): continue 이름 = 사이트["이름"] print(f"\n▶ {이름} 확인 중...") page = context.new_page() try: login_fn = 로그인_함수.get(이름) if login_fn: login_fn(page, 사이트["아이디"], 사이트["비밀번호"]) 발견_목록 = 기사_스캔(page, 사이트) if not 발견_목록: print(f" — 키워드 기사 없음") else: for 항목 in 발견_목록: print(f" ✅ [{항목['키워드']}] {항목['제목'][:45]}") ts = datetime.now().strftime("%Y%m%d_%H%M%S") 스크린샷 = 스크린샷_찍기(context, 항목["URL"], f"{이름}_{항목['키워드']}_{ts}") 이미지URL = 이미지_업로드(스크린샷) if 스크린샷 else None ok = 카카오_전송(이름, 항목["키워드"], 항목["제목"], 항목["URL"], 이미지URL) print(f" 카카오: {'✅' if ok else '❌'} " f"{'(이미지)' if 이미지URL else '(텍스트)'}") except Exception as e: print(f" ❌ 오류: {e}") finally: page.close() browser.close() print(f"\n⏰ 완료. {체크_주기_분}분 후 재확인.") # ── 실행 ────────────────────────────────────────────────── if __name__ == "__main__": print("=" * 45) print(" 뉴스 키워드 모니터링 봇 v3") print(f" 사이트: 더벨 / 딜사이트플러스 / 넘버스") print(f" 키워드: {', '.join(키워드_목록[:5])} 외 {max(0,len(키워드_목록)-5)}개") print(f" 주기: {체크_주기_분}분마다") print("=" * 45) 뉴스_체크() schedule.every(체크_주기_분).minutes.do(뉴스_체크) print(f"\n⏰ 스케줄 실행 중. 종료: Ctrl+C\n") while True: schedule.run_pending() time.sleep(30)