468 lines
15 KiB
Python
468 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cron-based scraper for downloading articles from RSS feeds.
|
|
This version replaces the infinite while loop with a single execution
|
|
that can be scheduled via cron job.
|
|
"""
|
|
|
|
import newspaper
|
|
import json
|
|
import feedparser
|
|
import time
|
|
import os
|
|
import requests
|
|
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
|
|
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__)
|
|
|
|
# Robust file path handling - try multiple locations
|
|
def get_feed_file_path():
|
|
"""Get the RSS feed file path, trying multiple locations."""
|
|
possible_paths = [
|
|
"./rss_feeds.json", # Current directory
|
|
"../rss_feeds.json", # Parent directory
|
|
"/home/user/StockDocs/scraper/rss_feeds.json", # Explicit path
|
|
"/app/rss_feeds.json", # Docker path
|
|
"./scraper/rss_feeds.json" # Scraper subdirectory
|
|
]
|
|
|
|
for path in possible_paths:
|
|
if os.path.exists(path):
|
|
print(f"Found feed file at: {path}")
|
|
return path
|
|
|
|
# If no file found, exit the program
|
|
print("Error: RSS feed file not found in any expected location")
|
|
print("Exiting program...")
|
|
exit(1)
|
|
|
|
# Get the feed file path
|
|
FEED_FILE = get_feed_file_path()
|
|
|
|
# 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}...")
|
|
|
|
# Debug: Print current working directory
|
|
logger.debug(f"Current working directory: {os.getcwd()}")
|
|
|
|
try:
|
|
with open(feed_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
logger.info(f"Successfully loaded {feed_file}")
|
|
logger.debug(f"Data type: {type(data)}")
|
|
if isinstance(data, dict) and "rss_feeds" in data:
|
|
logger.info(f"Found rss_feeds section with {len(data['rss_feeds'])} sources")
|
|
return data
|
|
else:
|
|
logger.warning(f"Unexpected data structure. Data keys: {list(data.keys()) if isinstance(data, dict) else 'Not a dict'}")
|
|
return {}
|
|
except FileNotFoundError:
|
|
logger.error(f"{feed_file} not found, returning empty dict.")
|
|
return {}
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Error decoding {feed_file}: {e}, returning empty dict.")
|
|
return {}
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error loading {feed_file}: {e}")
|
|
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"]
|
|
|
|
for site, data in sources.items():
|
|
logger.info(f"Parsing RSS feed: {data['rss_url']}")
|
|
try:
|
|
feed = feedparser.parse(data["rss_url"])
|
|
feed_entries = feed.entries[:limit] if limit else feed.entries
|
|
|
|
for entry in feed_entries:
|
|
if "link" in entry and "title" in entry:
|
|
all_links.append((site, entry.title, entry.link))
|
|
except Exception as e:
|
|
logger.error(f"Error parsing RSS feed: {site} Error: {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 with timeout
|
|
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 reduced parallelism.
|
|
"""
|
|
results = []
|
|
errors = []
|
|
|
|
# Process in smaller batches to reduce resource strain
|
|
batch_size = 5
|
|
|
|
for i in range(0, len(article_list), batch_size):
|
|
batch = article_list[i : i + batch_size]
|
|
logger.info(f"Processing batch {i // batch_size + 1} with {len(batch)} articles")
|
|
|
|
# Use ThreadPoolExecutor instead of ProcessPoolExecutor to avoid
|
|
# process termination issues with browser automation
|
|
with ThreadPoolExecutor(max_workers=3) as executor: # Reduced workers
|
|
futures = [
|
|
executor.submit(pull_article, link, source, title)
|
|
for source, title, link in batch
|
|
]
|
|
|
|
for future in as_completed(futures):
|
|
try:
|
|
result = future.result(timeout=120) # 2 minute timeout
|
|
results.append(result)
|
|
except Exception as e:
|
|
logger.error(f"Error in pull_article: {e}")
|
|
errors.append(e)
|
|
|
|
# Add a small delay between batches to reduce system load
|
|
time.sleep(5)
|
|
|
|
return results, errors
|
|
|
|
|
|
def gather_new_articles():
|
|
"""
|
|
Gather list of all newly downloaded articles and format them for webhook.
|
|
"""
|
|
new_articles = []
|
|
|
|
# Walk through all article directories
|
|
for root, dirs, files in os.walk("articles"):
|
|
for file in files:
|
|
if file != "processed_articles_cache.json": # Skip cache file
|
|
# Get the full file path
|
|
file_path = os.path.join(root, file)
|
|
|
|
# Get the outlet name from the directory path
|
|
outlet = os.path.basename(root)
|
|
|
|
# Create the relative path for the article
|
|
relative_path = os.path.relpath(file_path, "scraper")
|
|
|
|
# Create article data structure
|
|
article_data = {
|
|
"created_at": time.strftime("%Y-%m-%dT%H:%M:%S.%f", time.localtime(os.path.getctime(file_path))),
|
|
"name": file,
|
|
"outlet": outlet,
|
|
"path": f"../{relative_path}"
|
|
}
|
|
|
|
new_articles.append(article_data)
|
|
|
|
return new_articles
|
|
|
|
|
|
def send_to_webhook(articles):
|
|
"""
|
|
Send list of articles to the webhook URL.
|
|
"""
|
|
webhook_url = "http://agents.example.com/webhook-test/49c5b169-c68c-4f8c-90c2-0fcca6e2d387"
|
|
headers = {
|
|
"StockDocsN8NAuthToken": "ganvT4gsgRjWpGE8FMw9uCzFjZrTx8RZCoVm2Dh7skbZecov"
|
|
}
|
|
|
|
try:
|
|
response = requests.post(webhook_url, json=articles, headers=headers, timeout=30)
|
|
if response.status_code == 200:
|
|
print(f"Successfully sent {len(articles)} articles to webhook")
|
|
else:
|
|
print(f"Webhook request failed with status code: {response.status_code}")
|
|
print(f"Response: {response.text}")
|
|
except Exception as e:
|
|
print(f"Error sending to webhook: {e}")
|
|
|
|
|
|
def main():
|
|
"""
|
|
Main scraping function for cron execution.
|
|
This replaces the infinite while loop with a single execution.
|
|
"""
|
|
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")
|
|
return
|
|
|
|
# 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.")
|
|
|
|
# Gather and send new articles to webhook
|
|
new_articles = gather_new_articles()
|
|
if new_articles:
|
|
send_to_webhook(new_articles)
|
|
else:
|
|
logger.info("No new articles to send to webhook")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Major error in main execution: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
raise # Re-raise to ensure the script exits with error code
|
|
|
|
logger.info("Scraping completed successfully.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|