뉴스 크롤링

# main.py import os import csv import requests import openai from datetime import datetime, timedelta from difflib import SequenceMatcher from notion_client import Client # ------------------------------- # API 키 설정 (이미 채워져 있습니다) # ------------------------------- OPENAI_API_KEY = "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" NOTION_API_KEY = "ntn_t57603150027RZ5YV6OcyaqR8HmuaGO1yUzxm2599gDcVI" NOTION_DB_ID = "215157576e7a8087a6cdecac205628ae" SERP_API_KEY = "8be565108c16a94ace13d37bac8b05f8d24d15a800a8c8a9a3f343b3c1e5ba6f" openai.api_key = OPENAI_API_KEY notion = Client(auth=NOTION_API_KEY) # ------------------------------- # 로그 파일 및 중복 방지 설정 # ------------------------------- LOG_FILE = "news_log.csv" DUPLICATE_HOURS = 6 SIMILARITY_THRESHOLD = 0.8 # ------------------------------- # 유틸리티 함수 # ------------------------------- def load_keywords(): with open("keywords.txt", "r", encoding="utf-8") as f: return [line.strip() for line in f if line.strip()] def is_recent(ts_str, hours=DUPLICATE_HOURS): try: ts = datetime.fromisoformat(ts_str) except Exception: return False return datetime.now() - ts < timedelta(hours=hours) def is_similar(a: str, b: str, thresh=SIMILARITY_THRESHOLD): return SequenceMatcher(None, a, b).ratio() >= thresh def load_logged_entries(): entries = [] if not os.path.exists(LOG_FILE): return entries with open(LOG_FILE, "r", encoding="utf-8", newline="") as f: reader = csv.DictReader(f) for row in reader: if "timestamp" not in row or not row["timestamp"].strip(): continue if is_recent(row["timestamp"]): entries.append(row) return entries def log_entry(title: str, url: str, snippet: str): file_exists = os.path.exists(LOG_FILE) now = datetime.now().isoformat() with open(LOG_FILE, "a", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=["title", "url", "timestamp", "snippet"]) if not file_exists: writer.writeheader() writer.writerow({ "title": title, "url": url, "timestamp": now, "snippet": snippet }) # ------------------------------- # OpenAI 호출 함수 # ------------------------------- def expand_article(title: str, snippet: str, url: str) -> str: prompt = f""" 아래 뉴스 제목, 요약(스니펫), 링크를 참고하여 실제 뉴스 기사처럼 본문 전체를 복원해 주세요. 요약하지 말고 기자 스타일의 리드문과 본문(5~10문장)을 자연스럽게 작성해 주세요. 제목: {title} 요약: {snippet} 링크: {url} """ resp = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], temperature=0.7, ) return resp.choices[0].message.content.strip() def summarize_article(body_text: str) -> str: prompt = f""" 아래 기사의 본문을 2~3문장으로 한국어로 요약해 주세요. - 핵심 정보만 간결하게 전달 - 기자 스타일로 작성 본문: {body_text} """ resp = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], temperature=0.5, ) return resp.choices[0].message.content.strip() # ------------------------------- # Notion 저장 함수 # ------------------------------- def save_to_notion(title: str, url: str, body_text: str, summary_text: str, country: str, category: str): """ - '요약' 필드: summary_text - children 블록: body_text """ notion.pages.create( parent={"database_id": NOTION_DB_ID}, icon={"type": "emoji", "emoji": "📰"}, properties={ "제목": {"title": [{"text": {"content": title}}]}, "요약": {"rich_text": [{"text": {"content": summary_text}}]}, "국가": {"select": {"name": country}}, "분류": {"select": {"name": category}}, "링크": {"url": url}, "날짜": {"date": {"start": datetime.now().isoformat()}} }, children=[ { "object": "block", "type": "paragraph", "paragraph": { "rich_text": [ {"type": "text", "text": {"content": body_text}} ] } } ] ) # ------------------------------- # 메인 처리 흐름 # ------------------------------- def fetch_news(): keywords = load_keywords() logged = load_logged_entries() logged_keys = set((e["title"], e["url"]) for e in logged) logged_snippets = [e["snippet"] for e in logged] for kw in keywords: print(f"\n[검색] 키워드: {kw}") api_url = ( f"https://serpapi.com/search.json" f"?q={kw}&engine=google_news&hl=ko" f"&api_key={SERP_API_KEY}&tbs=qdr:d" ) try: res = requests.get(api_url).json() items = res.get("news_results", []) except Exception as e: print("❌ 뉴스 수집 오류:", e) continue for article in items: title = article.get("title", "").strip() url = article.get("link", "").strip() snippet = article.get("snippet", "").strip() if not title or not url: continue # 중복 검사: 제목+URL if (title, url) in logged_keys: print("중복, 스킵:", title) continue # 중복 검사: 스니펫 유사도 if any(is_similar(snippet, old) for old in logged_snippets): print("유사 기사, 스킵:", title) continue try: # 본문 복원 body = expand_article(title, snippet, url) # 요약 생성 summary = summarize_article(body) # 고정값 or GPT 추론 (원하는 로직으로 교체 가능) country, category = "한국", "경제" # Notion 저장 + 로그 기록 save_to_notion(title, url, body, summary, country, category) log_entry(title, url, snippet) print("[✅ 저장됨]", title) except Exception as e: print("❌ 처리 오류:", title, e) if __name__ == "__main__": fetch_news()

import os, requests, openai, datetime, json from urllib.parse import quote from notion_client import Client from numpy import dot from numpy.linalg import norm # ✅ 파현님 전용 API 키 OPENAI_API_KEY = "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" NOTION_API_KEY = "ntn_t57603150027RZ5YV6OcyaqR8HmuaGO1yUzxm2599gDcVI" NOTION_DB_ID = "215157576e7a8087a6cdecac205628ae" SERP_API_KEY = "8be565108c16a94ace13d37bac8b05f8d24d15a800a8c8a9a3f343b3c1e5ba6f" openai.api_key = OPENAI_API_KEY notion = Client(auth=NOTION_API_KEY) LOG_FILE = "news_log.csv" SUMMARY_LOG_FILE = "summary_log.json" def get_embedding(text, model="text-embedding-3-small"): res = openai.Embedding.create(input=[text], model=model) return res["data"][0]["embedding"] def cosine_similarity(vec1, vec2): return dot(vec1, vec2) / (norm(vec1) * norm(vec2)) def load_keywords(): with open("keywords.txt", "r", encoding="utf-8") as f: return [line.strip() for line in f if line.strip()] def load_logged_titles(): if not os.path.exists(LOG_FILE): return set() with open(LOG_FILE, "r", encoding="utf-8") as f: return set(line.strip() for line in f) def log_title(title): with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(title.strip() + "\n") def load_summary_embeddings(): if not os.path.exists(SUMMARY_LOG_FILE): return [] with open(SUMMARY_LOG_FILE, "r", encoding="utf-8") as f: return json.load(f) def log_summary_embedding(summary, embedding): logs = load_summary_embeddings() logs.append({"summary": summary, "embedding": embedding}) with open(SUMMARY_LOG_FILE, "w", encoding="utf-8") as f: json.dump(logs, f, ensure_ascii=False, indent=2) def expand_article(title, snippet, url): prompt = f""" 아래 뉴스 제목과 요약을 참고하여: ① 뉴스 요약을 3줄로 작성해주세요. 제목: {title} 요약: {snippet} 링크: {url} 반드시 아래 JSON 형식으로 출력: {{ "summary": "..." }} """ res = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) return json.loads(res['choices'][0]['message']['content']) def is_duplicate_by_embedding(new_summary, threshold=0.90): new_vec = get_embedding(new_summary) for log in load_summary_embeddings(): sim = cosine_similarity(new_vec, log["embedding"]) if sim > threshold: print(f"⚠️ 유사도 {sim:.3f} → 중복으로 간주") return True return False def save_to_notion(title, url, summary, body, country, category, keyword): print(f"[Notion 저장 시도] {title}") notion.pages.create( parent={"database_id": NOTION_DB_ID}, icon={"type": "emoji", "emoji": "📰"}, properties={ "제목": {"title": [{"text": {"content": title}}]}, "요약": {"rich_text": [{"text": {"content": summary}}]}, "국가": {"select": {"name": country}}, "분류": {"select": {"name": category}}, "링크": {"url": url}, "날짜": {"date": {"start": datetime.datetime.now().isoformat()}}, "키워드": {"multi_select": [{"name": keyword}]} }, children=[ { "object": "block", "type": "paragraph", "paragraph": { "rich_text": [ {"type": "text", "text": {"content": body}} ] } } ] ) log_title(title) log_summary_embedding(summary, get_embedding(summary)) print(f"[✅ 저장 완료] {title}") def fetch_news(): keywords = load_keywords() logged_titles = load_logged_titles() for kw in keywords: print(f"\n[검색 시작] 키워드: {kw}") encoded_kw = quote(kw) url = f"https://serpapi.com/search.json?q={encoded_kw}&engine=google_news&hl=ko&api_key={SERP_API_KEY}&tbs=qdr:d" try: res = requests.get(url).json() news_items = res.get("news_results", []) print(f"→ 수집된 뉴스 수: {len(news_items)}") for article in news_items: title = article.get("title") link = article.get("link") snippet = article.get("snippet", "") if not title or title.strip() in logged_titles: continue try: summary = expand_article(title, snippet, link)["summary"] if is_duplicate_by_embedding(summary): print(f"⛔ 중복 내용으로 저장 생략됨 [{title}]") continue save_to_notion(title, link, summary, snippet, "한국", "경제", kw) except Exception as e: print("❌ GPT 또는 Notion 오류:", title, e) except Exception as e: print("❌ 뉴스 수집 오류:", kw, e) if __name__ == "__main__": fetch_news()

import requests, json from bs4 import BeautifulSoup from openai import OpenAI from notion_client import Client # OpenAI 설정 openai.api_key = "YOUR_OPENAI_KEY" # Notion 설정 notion = Client(auth="YOUR_NOTION_SECRET") db_id = "YOUR_NOTION_DATABASE_ID" def summarize_article(title, url, content): prompt = f""" 다음 뉴스 기사 제목과 내용을 3줄로 요약하고, 국가와 분야를 추출해줘. 제목: {title} 내용: {content} 결과는 JSON 형식으로: {{"summary": "...", "country": "...", "category": "..."}} """ response = openai.ChatCompletion.create( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) return json.loads(response['choices'][0]['message']['content']) def crawl_and_save(site_url): res = requests.get(site_url) soup = BeautifulSoup(res.text, "html.parser") articles = soup.select("a")[:5] # 최신 기사 몇 개만 예시로 for article in articles: link = article['href'] title = article.text.strip() if not link.startswith("http"): continue content_res = requests.get(link) content_soup = BeautifulSoup(content_res.text, "html.parser") text = ' '.join([p.text for p in content_soup.select('p')]) result = summarize_article(title, link, text) # Notion에 저장 notion.pages.create( parent={"database_id": db_id}, properties={ "제목": {"title": [{"text": {"content": title}}]}, "국가": {"select": {"name": result["country"]}}, "분야": {"select": {"name": result["category"]}}, "요약": {"rich_text": [{"text": {"content": result["summary"]}}]}, "링크": {"url": link} } ) # 실행 crawl_and_save("https://www.wsj.com/")

news_collector/ ├── .env ├── main.py ├── notion_api.py ├── summarizer.py ├── sites_config.yaml └── logs/

# 📁 news_to_notion_bot/.env.example # → 복사하여 .env로 저장하고 다음 값 입력 OPENAI_API_KEY=your_openai_key_here NOTION_API_KEY=your_notion_secret_here NOTION_DB_ID=your_notion_database_id_here

# 📁 news_to_notion_bot/requirements.txt openai notion-client beautifulsoup4 requests python-dotenv

# 📊 News to Notion Bot GPT 요약 기반 글로벌 뉴스 → 자동 분류 → Notion DB 저장 시스템 ## ✅ 주요 기능 - 주요 뉴스 사이트 실시간 크롤링 - GPT-4 기반 3줄 요약 및 국가/분야 자동 분류 - Notion 데이터베이스 자동 저장 ## 🛠️ 설치 및 실행 방법 ```bash git clone https://github.com/your-username/news_to_notion_bot.git cd news_to_notion_bot cp .env.example .env # 환경 설정 pip install -r requirements.txt python main.py

OPENAI_API_KEY=sk-xxxxx NOTION_API_KEY=secret_xxxxx NOTION_DB_ID=xxxxx-xxxxx-xxxxx

main.py # 실행 진입점 crawler.py # 뉴스 본문 추출 summarizer.py # GPT 요약 + 분류 notion_api.py # Notion 저장기 .env.example # 환경 설정 예시 requirements.txt # 필요 라이브러리 README.md # 사용 설명서

--- ### 📌 다음 생성 예정 파일 1. `crawler.py`: 뉴스 본문 수집 모듈 2. `summarizer.py`: GPT 요약 및 국가/분야 분류 3. `notion_api.py`: Notion 저장 모듈 “계속”이라고 입력하시면 위 3개 `.py` 파일을 순차적으로 생성하겠습니다.

# 뉴스 사이트에서 기사 링크, 본문 추출 → 요약 → Notion 저장 import requests from bs4 import BeautifulSoup from summarizer import summarize_article from notion_api import save_to_notion def crawl_and_process(site_url): try: res = requests.get(site_url, timeout=10) soup = BeautifulSoup(res.text, "html.parser") links = soup.find_all("a")[:10] # 상위 10개 링크만 테스트용 for link in links: href = link.get("href") title = link.get_text(strip=True) if not href or not title or len(title) < 20 or not href.startswith("http"): continue print(f"요약 중: {title}") try: article = requests.get(href, timeout=10) article_soup = BeautifulSoup(article.text, "html.parser") paragraphs = article_soup.find_all("p") content = " ".join([p.get_text() for p in paragraphs]) summary_data = summarize_article(title, href, content) save_to_notion(title, href, summary_data) except Exception as e: print(f"기사 처리 오류: {e}") except Exception as e: print(f"크롤링 실패: {e}")

# OpenAI GPT 기반 요약 + 국가/분야 분류 import openai import os openai.api_key = os.getenv("OPENAI_API_KEY") def summarize_article(title, url, content): prompt = f""" 아래 뉴스 제목과 내용을 요약하고, 국가(country)와 분야(category)를 추출하세요. 제목: {title} 링크: {url} 본문: {content[:1500]} 아래 JSON 형식으로 응답하세요: {{ "summary": "...", "country": "...", "category": "..." }} """ response = openai.ChatCompletion.create( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) return eval(response.choices[0].message.content)

# Notion DB 저장 모듈 import os from notion_client import Client notion = Client(auth=os.getenv("NOTION_API_KEY")) DB_ID = os.getenv("NOTION_DB_ID") def save_to_notion(title, url, summary_data): notion.pages.create( parent={"database_id": DB_ID}, properties={ "제목": {"title": [{"text": {"content": title}}]}, "링크": {"url": url}, "요약": {"rich_text": [{"text": {"content": summary_data["summary"]}}]}, "국가": {"select": {"name": summary_data["country"]}}, "분야": {"select": {"name": summary_data["category"]}} } )

# 📁 crawler_kor.py import requests from bs4 import BeautifulSoup from summarizer import summarize_article from notion_api import save_to_notion def crawl_kor_news(): sources = { "네이버": "https://news.naver.com/main/main.naver?mode=LSD&mid=shm&sid1=101", # 경제 분야 "인베스팅": "https://kr.investing.com/news/stock-market-news", "매일경제": "https://www.mk.co.kr/news/stock/" } for source, url in sources.items(): print(f"📰 [{source}] 크롤링 중...") try: res = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0"}) soup = BeautifulSoup(res.text, "html.parser") articles = soup.find_all("a", href=True)[:10] for a in articles: link = a["href"] title = a.get_text(strip=True) if not link.startswith("http") or len(title) < 15: continue try: content = extract_kor_article_content(source, link) if content: summary_data = summarize_article(title, link, content) save_to_notion(title, link, summary_data) except Exception as e: print(f"❌ 기사 요약 실패: {e}") except Exception as e: print(f"❌ 크롤링 오류 ({source}): {e}") def extract_kor_article_content(source, url): res = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0"}) soup = BeautifulSoup(res.text, "html.parser") if "naver" in source.lower(): area = soup.select_one("#dic_area") elif "investing" in source.lower(): area = soup.select_one("div.articlePage") elif "매일" in source or "mk" in source.lower(): area = soup.select_one("div.art_txt") or soup.select_one("div.art_txt-lead") else: area = None return area.get_text(strip=True) if area else None

from crawler_kor import crawl_kor_news if __name__ == "__main__": sites = [ ... 기존 글로벌 사이트 ... ] for site in sites: crawl_and_process(site) # 🔽 한국 뉴스 수집 실행 crawl_kor_news()

🟢 컴퓨터 켬 ↓ ▶️ 실행.bat 한 번 클릭 ↓ [6시간 간격 반복 시작] ├ 뉴스 수집 (각 키워드) ├ 중복 검사 ├ GPT 요약 + 분류 └ Notion 저장 ↓ 🔴 컴퓨터 끄면 자동 중단