파이썬 웹 크롤링 실전 - 뉴스 수집 자동화! RSS로 매일 아침 기사 모으기 (feedparser) 📰

https://news.google.com/rss/search?q=검색어&hl=ko&gl=KR&ceid=KR:ko

pip install feedparser requests beautifulsoup4 lxml python-dotenv

pip show feedparser | findstr Version

import feedparser
import requests

HEADERS = {"User-Agent": "Mojangcam-NewsBot/1.0 (+https://mojangcam.tistory.com)"}
URL = "https://news.google.com/rss/search?q=%ED%8C%8C%EC%9D%B4%EC%8D%AC&hl=ko&gl=KR&ceid=KR:ko"

res = requests.get(URL, headers=HEADERS, timeout=10)
res.raise_for_status()

parsed = feedparser.parse(res.content)
print("기사 수:", len(parsed.entries))

for entry in parsed.entries[:3]:
    print("-", entry.title)
    print("  ", entry.link)

import calendar
import re
from datetime import datetime, timezone, timedelta

from bs4 import BeautifulSoup

KST = timezone(timedelta(hours=9))
KEYWORDS = ["파이썬", "AI"]

def clean(html_text: str) -> str:
    """요약문에 섞인 HTML 태그를 제거하고 공백을 정리해요."""
    if not html_text:
        return ""
    text = BeautifulSoup(html_text, "lxml").get_text(" ", strip=True)
    return re.sub(r"\s+", " ", text).strip()

def to_kst(entry) -> str:
    """RSS 발행 시각을 한국 시간 문자열로 바꿔요."""
    t = entry.get("published_parsed") or entry.get("updated_parsed")
    if not t:
        return ""
    dt = datetime.fromtimestamp(calendar.timegm(t), tz=timezone.utc)
    return dt.astimezone(KST).strftime("%Y-%m-%d %H:%M")

def match_keyword(title: str, summary: str) -> bool:
    if not KEYWORDS:
        return True
    blob = f"{title} {summary}".lower()
    return any(k.lower() in blob for k in KEYWORDS)

import sqlite3

DB_PATH = "news.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS seen (
            link       TEXT PRIMARY KEY,
            title      TEXT NOT NULL,
            source     TEXT,
            published  TEXT,
            collected  TEXT NOT NULL
        )
    """)
    conn.commit()
    return conn

NEWS_KEYWORDS=파이썬,AI,반도체
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/여기에_본인_웹훅_주소

.env
news.db
news.csv

import csv
import os

from dotenv import load_dotenv

load_dotenv()

KEYWORDS = [k.strip() for k in os.getenv("NEWS_KEYWORDS", "").split(",") if k.strip()]
WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")
CSV_PATH = "news.csv"

def save_csv(items):
    if not items:
        return
    is_new_file = not os.path.exists(CSV_PATH)
    with open(CSV_PATH, "a", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=list(items[0].keys()))
        if is_new_file:
            writer.writeheader()
        writer.writerows(items)

DISCORD_LIMIT = 2000

def notify_discord(items):
    if not WEBHOOK_URL or not items:
        return
    lines = [f"📰 **새 기사 {len(items)}건**"]
    for item in items:
        line = f"- [{item['제목']}]({item['링크']}) — {item['출처']}"
        # 2,000자를 넘기 직전에 멈춰요
        if len("\n".join(lines + [line])) > DISCORD_LIMIT - 40:
            lines.append(f"…외 {len(items) - (len(lines) - 1)}건")
            break
        lines.append(line)
    res = requests.post(
        WEBHOOK_URL,
        json={"content": "\n".join(lines), "flags": 4},  # flags=4 : 링크 미리보기 끄기
        timeout=10,
    )
    res.raise_for_status()   # 조용히 실패하지 않게!

# news_collector.py
# Python 3.10+ / feedparser 6.0.x / requests 2.34.x / beautifulsoup4 4.15.x
import calendar
import csv
import os
import re
import sqlite3
import time
from datetime import datetime, timezone, timedelta

import feedparser
import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv

load_dotenv()

KEYWORDS = [k.strip() for k in os.getenv("NEWS_KEYWORDS", "").split(",") if k.strip()]
WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")

# 원하는 만큼 추가하세요. 이름은 자유롭게!
FEEDS = {
    "구글뉴스-파이썬": "https://news.google.com/rss/search?q=%ED%8C%8C%EC%9D%B4%EC%8D%AC&hl=ko&gl=KR&ceid=KR:ko",
    "구글뉴스-AI": "https://news.google.com/rss/search?q=AI&hl=ko&gl=KR&ceid=KR:ko",
}

DB_PATH = "news.db"
CSV_PATH = "news.csv"
KST = timezone(timedelta(hours=9))
HEADERS = {"User-Agent": "Mojangcam-NewsBot/1.0 (+https://mojangcam.tistory.com)"}

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS seen (
            link       TEXT PRIMARY KEY,
            title      TEXT NOT NULL,
            source     TEXT,
            published  TEXT,
            collected  TEXT NOT NULL
        )
    """)
    conn.commit()
    return conn

def clean(html_text: str) -> str:
    if not html_text:
        return ""
    text = BeautifulSoup(html_text, "lxml").get_text(" ", strip=True)
    return re.sub(r"\s+", " ", text).strip()

def to_kst(entry) -> str:
    t = entry.get("published_parsed") or entry.get("updated_parsed")
    if not t:
        return ""
    dt = datetime.fromtimestamp(calendar.timegm(t), tz=timezone.utc)
    return dt.astimezone(KST).strftime("%Y-%m-%d %H:%M")

def match_keyword(title: str, summary: str) -> bool:
    if not KEYWORDS:
        return True
    blob = f"{title} {summary}".lower()
    return any(k.lower() in blob for k in KEYWORDS)

def fetch_feed(url: str):
    res = requests.get(url, headers=HEADERS, timeout=10)
    res.raise_for_status()
    return feedparser.parse(res.content)

def collect():
    conn = init_db()
    new_items = []

    for feed_name, url in FEEDS.items():
        try:
            parsed = fetch_feed(url)
        except requests.RequestException as e:
            print(f"[SKIP] {feed_name} 요청 실패: {e}")
            continue

        if parsed.bozo and not parsed.entries:
            print(f"[SKIP] {feed_name} 파싱 실패: {parsed.bozo_exception}")
            continue

        for entry in parsed.entries:
            link = entry.get("link", "").strip()
            title = clean(entry.get("title", ""))
            if not link or not title:
                continue

            summary = clean(entry.get("summary", ""))[:200]
            if not match_keyword(title, summary):
                continue

            cur = conn.execute("SELECT 1 FROM seen WHERE link = ?", (link,))
            if cur.fetchone():
                continue

            source = entry.get("source", {}).get("title") or feed_name
            published = to_kst(entry)
            collected = datetime.now(KST).strftime("%Y-%m-%d %H:%M")

            conn.execute(
                "INSERT INTO seen (link, title, source, published, collected)"
                " VALUES (?, ?, ?, ?, ?)",
                (link, title, source, published, collected),
            )
            new_items.append({
                "제목": title,
                "출처": source,
                "발행일": published,
                "링크": link,
                "요약": summary,
                "수집시각": collected,
            })

        time.sleep(1)  # 서버 배려! 피드 사이에 1초 쉬어요

    conn.commit()
    conn.close()
    return new_items

def save_csv(items):
    if not items:
        return
    is_new_file = not os.path.exists(CSV_PATH)
    with open(CSV_PATH, "a", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=list(items[0].keys()))
        if is_new_file:
            writer.writeheader()
        writer.writerows(items)

DISCORD_LIMIT = 2000

def notify_discord(items):
    if not WEBHOOK_URL or not items:
        return
    lines = [f"📰 **새 기사 {len(items)}건**"]
    for item in items:
        line = f"- [{item['제목']}]({item['링크']}) — {item['출처']}"
        # 2,000자를 넘기 직전에 멈춰요
        if len("\n".join(lines + [line])) > DISCORD_LIMIT - 40:
            lines.append(f"…외 {len(items) - (len(lines) - 1)}건")
            break
        lines.append(line)
    res = requests.post(
        WEBHOOK_URL,
        json={"content": "\n".join(lines), "flags": 4},  # flags=4 : 링크 미리보기 끄기
        timeout=10,
    )
    res.raise_for_status()   # 조용히 실패하지 않게!

def main():
    items = collect()
    print(f"[{datetime.now(KST):%Y-%m-%d %H:%M}] 새 기사 {len(items)}건")
    for item in items:
        print(f" - ({item['출처']}) {item['제목']}")
    save_csv(items)
    notify_discord(items)

if __name__ == "__main__":
    main()

python news_collector.py

[2026-08-20 08:09] 새 기사 2건
 - (예제일보) 파이썬 3.15 정식 출시 임박 - lazy import 확정
 - (예제경제) AI 반도체 수출 규제 논의

[2026-08-20 08:09] 새 기사 0건

import schedule

schedule.every().day.at("08:00").do(main)   # 매일 아침 8시
schedule.every(30).minutes.do(main)          # 또는 30분마다

while True:
    schedule.run_pending()
    time.sleep(1)

from urllib.robotparser import RobotFileParser

rp = RobotFileParser()
rp.set_url("https://example.com/robots.txt")
rp.read()

print(rp.can_fetch("Mojangcam-NewsBot/1.0", "https://example.com/news/1"))  # True
print(rp.can_fetch("Mojangcam-NewsBot/1.0", "https://example.com/admin/x"))  # False