418 lines
14 KiB
Python
418 lines
14 KiB
Python
import newspaper
|
|
import json
|
|
import feedparser
|
|
import time
|
|
import os
|
|
import logging
|
|
from datetime import datetime
|
|
from selenium import webdriver
|
|
from selenium.webdriver.firefox.options import Options as FirefoxOptions
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError
|
|
import nltk
|
|
from nltk.downloader import Downloader
|
|
|
|
# Setup logging with timestamps
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FEED_FILE = os.getenv("FEED_FILE", "./rss_short_feed.json")
|
|
MAX_FEED_WORKERS = int(os.getenv("MAX_FEED_WORKERS", "10"))
|
|
MAX_ARTICLE_WORKERS = int(os.getenv("MAX_ARTICLE_WORKERS", "10"))
|
|
|
|
# Ensure necessary NLTK resources are downloaded
|
|
d = Downloader()
|
|
if not d.is_installed("punkt_tab"):
|
|
nltk.download("punkt_tab")
|
|
|
|
articles = []
|
|
|
|
|
|
def load_rss_feed_sources(feed_file=FEED_FILE):
|
|
"""
|
|
Loads the RSS feed sources from a JSON file.
|
|
"""
|
|
logger.info(f"Loading RSS feed sources from {feed_file}...")
|
|
|
|
try:
|
|
with open(feed_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
logger.info(f"Successfully loaded {feed_file}")
|
|
return data
|
|
except FileNotFoundError:
|
|
logger.error(f"{feed_file} not found, returning empty list.")
|
|
return []
|
|
except json.JSONDecodeError:
|
|
logger.error(f"Error decoding {feed_file} , returning empty list.")
|
|
return []
|
|
|
|
|
|
def mine_all_articles(rss_feed_sources, limit=None):
|
|
"""
|
|
Mines all articles from the given RSS feed sources.
|
|
Returns a list of (site, title, link) tuples.
|
|
"""
|
|
all_links = []
|
|
|
|
# Check if rss_feed_sources is a valid dict with rss_feeds key
|
|
if not isinstance(rss_feed_sources, dict):
|
|
logger.warning(f"rss_feed_sources is not a dict, it's {type(rss_feed_sources)}")
|
|
return all_links
|
|
|
|
if "rss_feeds" not in rss_feed_sources:
|
|
logger.warning("rss_feeds key not found in rss_feed_sources")
|
|
return all_links
|
|
|
|
sources = rss_feed_sources["rss_feeds"]
|
|
|
|
# Parse RSS feeds in parallel for better performance
|
|
def parse_single_feed(site, data):
|
|
"""Parse a single RSS feed with timeout and error handling"""
|
|
try:
|
|
logger.info(f"Parsing RSS feed: {data['rss_url']}")
|
|
# Add more aggressive timeout settings with fallback
|
|
# Use a wrapper to ensure we don't hang indefinitely
|
|
import signal
|
|
|
|
def timeout_handler(signum, frame):
|
|
raise TimeoutError(f"Timeout parsing feed: {site}")
|
|
|
|
# Set up signal-based timeout (this is a fallback for truly hanging requests)
|
|
old_handler = signal.signal(signal.SIGALRM, timeout_handler)
|
|
signal.alarm(10) # 10 second alarm
|
|
|
|
feed = feedparser.parse(data["rss_url"], timeout=8) # 8 second timeout
|
|
signal.alarm(0) # Cancel the alarm
|
|
signal.signal(signal.SIGALRM, old_handler)
|
|
|
|
feed_entries = feed.entries[:limit] if limit else feed.entries
|
|
|
|
entries = []
|
|
for entry in feed_entries:
|
|
if "link" in entry and "title" in entry:
|
|
entries.append((site, entry.title, entry.link))
|
|
return entries
|
|
except TimeoutError as e:
|
|
logger.error(f"Timeout parsing RSS feed: {site} Error: {str(e)}")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"Error parsing RSS feed: {site} Error: {str(e)}")
|
|
return []
|
|
|
|
# Use ThreadPoolExecutor for parallel RSS feed parsing
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
max_workers = min(10, len(sources)) # Limit concurrent workers
|
|
|
|
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
# Submit all feed parsing tasks
|
|
future_to_site = {
|
|
executor.submit(parse_single_feed, site, data): site
|
|
for site, data in sources.items()
|
|
}
|
|
|
|
# Collect results as they complete
|
|
for future in as_completed(future_to_site, timeout=30): # 30 second overall timeout
|
|
try:
|
|
entries = future.result()
|
|
all_links.extend(entries)
|
|
except Exception as e:
|
|
site = future_to_site[future]
|
|
logger.error(f"Error processing feed for {site}: {str(e)}")
|
|
|
|
return all_links
|
|
|
|
|
|
def generate_filename_from_url(url):
|
|
"""
|
|
Generates a filename from the given URL by replacing slashes with underscores.
|
|
"""
|
|
# Use only the last part of the URL or replace slashes
|
|
return url.replace("https://", "").replace("http://", "").replace("/", "_")
|
|
|
|
|
|
def generate_safe_filename(name):
|
|
# Remove/replace characters not allowed in filenames
|
|
import re
|
|
|
|
safe = re.sub(r'[\\/*?:"<>|]', "_", name)
|
|
return safe
|
|
|
|
|
|
def save_article_to_file(article, filename, source="Unfiltered"):
|
|
"""
|
|
Saves the given article text to a file with the specified filename.
|
|
"""
|
|
# articles dir should already be there
|
|
# os.makedirs("articles", exist_ok=True)
|
|
|
|
outputDir = "articles/" + source
|
|
os.makedirs(outputDir, exist_ok=True) if source else None
|
|
|
|
# Sanitize filename: use only the last part of the URL or replace slashes
|
|
safe_filename = generate_filename_from_url(filename)
|
|
file_path = os.path.join(outputDir, safe_filename)
|
|
# Save the source as the first line in the file for later retrieval
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
f.write(f"SOURCE:{source}\n")
|
|
f.write(article)
|
|
|
|
# Only log when a new file is actually created (not cached)
|
|
logger.info(f"New article saved: {safe_filename} from {source}")
|
|
|
|
|
|
def get_article_with_selenium(url):
|
|
"""
|
|
Gets article text using Selenium Firefox driver with proper error handling
|
|
and cleanup.
|
|
"""
|
|
driver = None
|
|
try:
|
|
# Configure Firefox options
|
|
options = FirefoxOptions()
|
|
options.add_argument("--headless")
|
|
options.set_preference("dom.ipc.processCount", 1) # Reduce process count
|
|
|
|
# Initialize driver
|
|
driver = webdriver.Firefox(options=options)
|
|
driver.set_page_load_timeout(30) # 30 seconds timeout
|
|
|
|
# Navigate to URL
|
|
driver.get(url)
|
|
|
|
# Wait for page to load (explicit wait instead of sleep)
|
|
try:
|
|
WebDriverWait(driver, 15).until(
|
|
EC.presence_of_element_located((By.TAG_NAME, "body"))
|
|
)
|
|
except:
|
|
pass # Continue even if wait times out
|
|
|
|
time.sleep(2) # Brief additional wait
|
|
|
|
html = driver.page_source
|
|
|
|
# Parse with Newspaper4k
|
|
article = newspaper.article(url, input_html=html, language="en")
|
|
article.nlp()
|
|
logger.info(f"Successfully extracted article with Selenium from {url}")
|
|
return article.text
|
|
|
|
except Exception as e:
|
|
logger.error(f"Selenium failed for {url}: {str(e)}")
|
|
return ""
|
|
finally:
|
|
# Always quit the driver
|
|
if driver:
|
|
try:
|
|
driver.quit()
|
|
except:
|
|
pass # Ignore errors in cleanup
|
|
|
|
|
|
def get_article_with_playwright(url):
|
|
"""
|
|
Gets article text using Playwright with proper error handling.
|
|
"""
|
|
try:
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
with sync_playwright() as p:
|
|
# Use Chromium instead of Firefox for better compatibility
|
|
browser = p.chromium.launch(headless=True, timeout=30000)
|
|
page = browser.new_page()
|
|
|
|
# Set user agent to avoid bot detection
|
|
page.set_extra_http_headers(
|
|
{
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
|
|
}
|
|
)
|
|
|
|
page.goto(url, wait_until="load")
|
|
|
|
# Wait for content to load
|
|
time.sleep(3)
|
|
|
|
html = page.content()
|
|
browser.close()
|
|
|
|
# Parse with Newspaper4k
|
|
article = newspaper.article(url, input_html=html, language="en")
|
|
article.nlp()
|
|
logger.info(f"Successfully extracted article with Playwright from {url}")
|
|
return article.text
|
|
|
|
except Exception as e:
|
|
logger.error(f"Playwright failed for {url}: {str(e)}")
|
|
return ""
|
|
|
|
|
|
def pull_article(link, source, title=None, save_to_file=True):
|
|
"""
|
|
Pulls an article from a given link with fallback mechanisms.
|
|
"""
|
|
filename = title if title else link
|
|
safe_filename = generate_filename_from_url(filename)
|
|
|
|
# Check if already cached
|
|
if os.path.exists(os.path.join("articles", source, safe_filename)):
|
|
logger.info(f"Article already cached: {filename}")
|
|
with open(
|
|
os.path.join("articles", source, safe_filename), "r", encoding="utf-8"
|
|
) as f:
|
|
return f.read()
|
|
|
|
text = ""
|
|
|
|
try:
|
|
# Try newspaper4k first
|
|
article = newspaper.article(link)
|
|
article.download()
|
|
article.parse()
|
|
text = article.text
|
|
|
|
if not text or len(text) < 200:
|
|
raise ValueError(
|
|
"\tArticle text too short, falling back to Playwright/Selenium."
|
|
)
|
|
logger.info(f"Successfully pulled article with newspaper4k from {link}")
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"newspaper4k extraction failed for {link}: {e}, falling back to Playwright."
|
|
)
|
|
|
|
try:
|
|
text = get_article_with_playwright(link)
|
|
logger.info(f"Successfully pulled article from {link} with Playwright")
|
|
|
|
if not text or len(text) < 200:
|
|
logger.warning(f"Playwright article too short, falling back to Selenium.")
|
|
# Fallback to Selenium with better error handling
|
|
text = get_article_with_selenium(link)
|
|
logger.info(f"Successfully pulled article from {link} with Selenium")
|
|
except Exception as e:
|
|
logger.error(f"Playwright failed for {link}: {e}")
|
|
|
|
# Fallback to Selenium
|
|
try:
|
|
text = get_article_with_selenium(link)
|
|
logger.info(f"Successfully pulled article from {link} with Selenium")
|
|
except Exception as e:
|
|
logger.error(f"Selenium failed for {link}: {e}")
|
|
return ""
|
|
|
|
if save_to_file:
|
|
save_article_to_file(text, filename, source)
|
|
|
|
return text
|
|
|
|
|
|
def safe_pull_articles(article_list):
|
|
"""
|
|
Safely pull articles with improved error handling and increased parallelism.
|
|
"""
|
|
if not article_list:
|
|
return [], []
|
|
|
|
results = []
|
|
errors = []
|
|
|
|
# Use ThreadPoolExecutor for parallel article pulling with higher concurrency
|
|
# Use configurable worker setting
|
|
max_workers = min(MAX_ARTICLE_WORKERS, len(article_list)) # Cap at configured workers, but don't exceed article count
|
|
batch_size = max(1, min(20, len(article_list) // 4)) # Dynamic batch size
|
|
|
|
logger.info(f"Starting parallel article pulling with {max_workers} workers and batch size {batch_size}")
|
|
|
|
# Process all articles in parallel with proper error handling
|
|
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
# Submit all tasks at once for maximum parallelism
|
|
futures = [
|
|
executor.submit(pull_article, link, source, title)
|
|
for source, title, link in article_list
|
|
]
|
|
|
|
# Collect results as they complete
|
|
for i, future in enumerate(as_completed(futures, timeout=300)): # 5 minute timeout total
|
|
try:
|
|
result = future.result(timeout=120) # 2 minute timeout per article
|
|
if result: # Only count non-empty results
|
|
results.append(result)
|
|
# Log progress every 100 articles
|
|
if (i + 1) % 100 == 0:
|
|
logger.info(f"Processed {i + 1} articles out of {len(article_list)}")
|
|
except Exception as e:
|
|
logger.error(f"Error processing article: {e}")
|
|
errors.append(e)
|
|
|
|
return results, errors
|
|
|
|
|
|
def main():
|
|
"""
|
|
Main scraping loop.
|
|
"""
|
|
while True:
|
|
logger.info("=========================================")
|
|
logger.info("Starting new scraping iteration...")
|
|
|
|
try:
|
|
# Pull the RSS feed sources from the JSON file
|
|
rss_feed_sources = load_rss_feed_sources()
|
|
|
|
# Mine all articles from the RSS feed sources
|
|
rss_feed_links = mine_all_articles(rss_feed_sources)
|
|
|
|
# Randomize the order of the links to help with load balancing
|
|
import random
|
|
|
|
random.shuffle(rss_feed_links)
|
|
|
|
logger.info(f"Found {len(rss_feed_links)} articles to process")
|
|
|
|
if not rss_feed_links:
|
|
logger.info("No articles found, sleeping for 15 minutes")
|
|
time.sleep(15 * 60)
|
|
continue
|
|
|
|
# Process articles with better error handling and resource management
|
|
results, errors = safe_pull_articles(rss_feed_links)
|
|
|
|
logger.info(f"Attempted to Pull {len(results)} articles in parallel.")
|
|
logger.info(
|
|
f"Encountered {len(errors)} errors during article pulling. "
|
|
+ "Outputting errors to a local file."
|
|
)
|
|
|
|
# Output errors to a local file
|
|
if errors:
|
|
with open("errors.txt", "w", encoding="utf-8") as f:
|
|
for error in errors:
|
|
f.write(str(error) + "\n")
|
|
logger.info(f"Errors logged to errors.txt")
|
|
|
|
# Print all results to a log file
|
|
with open("results.txt", "w", encoding="utf-8") as f:
|
|
for result in results:
|
|
f.write(result + "\n")
|
|
logger.info("All articles pulled successfully.")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Major error in main loop: {e}")
|
|
# Continue to next iteration even if there's a major error
|
|
|
|
# Sleep for a while before the next iteration
|
|
logger.info("Sleeping for 15 minutes before the next iteration...")
|
|
time.sleep(15 * 60) # Sleep for 15 minutes
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|