Build an Automated Economic Times News Scraper in Python

📰 Build an Automated Economic Times News Scraper in Python





If you have ever wanted to collect news data for analysis, machine learning, or just to build your own custom news feed, web scraping is the way to go. In this project, we are going to look at a beginner-friendly Python script that automates the extraction of news articles from the Economic Times.


🚀 What This Project Does

This script is designed to be a continuous, automated news collector. Here is exactly what it handles behind the scenes:

  • Reads RSS Feeds: It taps into the official Economic Times RSS feeds across multiple categories (Markets, Tech, Startups, Politics, etc.).
  • Scrapes Full Articles: It doesn't just stop at the summary. It visits the actual article link to extract the full body text.
  • Extracts Metadata: It grabs the title, author, published time, and even the thumbnail image.
  • Saves to JSON: It neatly organizes and saves the scraped data into specific category files (e.g., tech_industry.json).

⚙️ The Setup & Requirements

Before running the code, you will need to install two powerful Python libraries. Open your terminal and run:

pip install requests beautifulsoup4
💡 Beginner Friendly Settings:
The script is built with a dedicated "Beginner Settings" block at the very top. You can easily change how many articles it downloads (ARTICLES_PER_CATEGORY), where to save them (OUTPUT_DIR), and whether it should run forever or just once (RUN_CONTINUOUSLY).

💻 The Python Code

Below is the full source code for the scraper, pulled directly from my GitHub repository. Feel free to copy it or clone the repo!


"""
Economic Times RSS + article scraper.

What this script does:
1. Reads article links from Economic Times RSS feeds.
2. Opens each article page with requests.
3. Extracts useful fields such as title, author, timestamps, body text, and thumbnail.
4. Saves each category into its own JSON file inside the output folder.

This script is intentionally written to be beginner-friendly.
Most users only need to edit the settings in the "BEGINNER SETTINGS" section below.
"""

import json
import os
import re
import time
import xml.etree.ElementTree as ET
from datetime import datetime

import requests
from bs4 import BeautifulSoup


# ---------------------------------------------------------------------------
# BEGINNER SETTINGS: edit these values first
# ---------------------------------------------------------------------------

# Number of RSS articles to try for each category in one cycle.
ARTICLES_PER_CATEGORY = 2

# Set this to False if you only want to test the scraper once and stop.
RUN_CONTINUOUSLY = True

# Sleep time between categories.
WAIT_BETWEEN_CATEGORIES = 3.0

# Sleep time after one full cycle finishes.
# This is only used when RUN_CONTINUOUSLY = True.
WAIT_BETWEEN_CYCLES = 200

# Maximum number of characters to save for the long article body.
# Use None if you do not want to trim the content.
MAX_DESCRIPTION_LENGTH = 1500

# Request settings.
REQUEST_TIMEOUT = 12
REQUEST_RETRIES = 3

# Folder where JSON files will be stored.
OUTPUT_DIR = "NEWS_DATA_ET"

# Keep as None to scrape all categories.
# Example: ["top_news", "markets", "economy"]
SELECTED_CATEGORIES = None


BASE_ET = "https://economictimes.indiatimes.com"

# Format: { "category_name": "full_rss_url" }
CATEGORY_RSS_MAP = {
    # Top News
    "top_news": f"{BASE_ET}/rssfeedsdefault.cms",
    "latest_news": f"{BASE_ET}/News/rssfeeds/1715249553.cms",

    # Markets
    "markets": f"{BASE_ET}/markets/rssfeeds/1977021501.cms",
    "stocks": f"{BASE_ET}/markets/stocks/rssfeeds/2146842.cms",
    "ipo": f"{BASE_ET}/markets/ipos/fpos/rssfeeds/14655708.cms",
    "cryptocurrency": f"{BASE_ET}/markets/cryptocurrency/rssfeeds/82519373.cms",
    "commodities": f"{BASE_ET}/markets/commodities/rssfeeds/1808152121.cms",
    "forex": f"{BASE_ET}/markets/forex/rssfeeds/1150221130.cms",
    "bonds": f"{BASE_ET}/markets/bonds/rssfeeds/2146846.cms",

    # News
    "india_news": f"{BASE_ET}/news/india/rssfeeds/81582957.cms",
    "economy": f"{BASE_ET}/news/economy/rssfeeds/1373380680.cms",
    "politics": f"{BASE_ET}/news/politics-and-nation/rssfeeds/1052732854.cms",
    "international": f"{BASE_ET}/news/international/rssfeeds/858478126.cms",
    "company": f"{BASE_ET}/news/company/rssfeeds/2143429.cms",
    "defence": f"{BASE_ET}/news/defence/rssfeeds/46687796.cms",
    "science": f"{BASE_ET}/news/science/rssfeeds/39872847.cms",
    "environment": f"{BASE_ET}/news/environment/rssfeeds/2647163.cms",
    "sports": f"{BASE_ET}/news/sports/rssfeeds/26407562.cms",
    "elections": f"{BASE_ET}/news/elections/rssfeeds/65869819.cms",

    # Industry
    "industry": f"{BASE_ET}/industry/rssfeeds/13352306.cms",
    "tech_industry": f"{BASE_ET}/tech/rssfeeds/13357270.cms",
    "healthcare": f"{BASE_ET}/industry/healthcare/biotech/rssfeeds/13358050.cms",
    "services": f"{BASE_ET}/industry/services/rssfeeds/13354120.cms",
    "media_ent": f"{BASE_ET}/industry/media/entertainment/rssfeeds/13357212.cms",
    "transportation": f"{BASE_ET}/industry/transportation/rssfeeds/13353990.cms",
    "renewables": f"{BASE_ET}/industry/renewables/rssfeeds/81585238.cms",
    "banking_finance": f"{BASE_ET}/industry/banking/finance/rssfeeds/13358250.cms",

    # Tech and Startups
    "startups": f"{BASE_ET}/tech/startups/rssfeeds/78570540.cms",
    "funding": f"{BASE_ET}/tech/funding/rssfeeds/78570550.cms",
    "information_tech": f"{BASE_ET}/tech/information-tech/rssfeeds/78570530.cms",
    "tech_internet": f"{BASE_ET}/tech/technology/rssfeeds/78570561.cms",
    "education": f"{BASE_ET}/education/rssfeeds/913168846.cms",

    # Personal Finance and Wealth
    "wealth": f"{BASE_ET}/wealth/rssfeeds/837555174.cms",
    "mutual_funds": f"{BASE_ET}/mf/rssfeeds/359241701.cms",
    "personal_finance": f"{BASE_ET}/wealth/personal-finance-news/rssfeeds/49674901.cms",
    "insurance": f"{BASE_ET}/wealth/insure/rssfeeds/47119917.cms",
    "tax": f"{BASE_ET}/wealth/tax/rssfeeds/47119912.cms",

    # Small Business
    "small_biz": f"{BASE_ET}/small-biz/rssfeeds/5575607.cms",
    "entrepreneurship": f"{BASE_ET}/small-biz/entrepreneurship/rssfeeds/11993034.cms",
    "gst": f"{BASE_ET}/small-biz/gst/rssfeeds/58475404.cms",

    # Jobs and Careers
    "jobs": f"{BASE_ET}/jobs/rssfeeds/107115.cms",

    # Opinion
    "opinion": f"{BASE_ET}/opinion/rssfeeds/897228639.cms",
    "et_editorial": f"{BASE_ET}/opinion/et-editorial/rssfeeds/3376910.cms",

    # Magazines and Lifestyle
    "travel": f"{BASE_ET}/magazines/travel/rssfeeds/640246854.cms",
    "et_magazine": f"{BASE_ET}/magazines/et-magazine/rssfeeds/7771003.cms",

    # NRI
    "nri": f"{BASE_ET}/nri/rssfeeds/7771250.cms",
}

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0.0.0 Safari/537.36"
    ),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Referer": "https://www.google.com/",
    "DNT": "1",
    "Upgrade-Insecure-Requests": "1",
}

session = requests.Session()
session.headers.update(HEADERS)

os.makedirs(OUTPUT_DIR, exist_ok=True)


def get_category_list() -> list[str]:
    """Return the categories that should be scraped in this run."""
    if SELECTED_CATEGORIES is None:
        return list(CATEGORY_RSS_MAP.keys())

    invalid_categories = [name for name in SELECTED_CATEGORIES if name not in CATEGORY_RSS_MAP]
    if invalid_categories:
        raise ValueError(
            "Unknown categories in SELECTED_CATEGORIES: "
            + ", ".join(invalid_categories)
        )

    return list(SELECTED_CATEGORIES)


def db_path(category: str) -> str:
    """Return the JSON file path for a category."""
    return os.path.join(OUTPUT_DIR, f"{category}.json")


def db_load(category: str) -> list[dict]:
    """Load one category JSON file. Returns an empty list if it does not exist."""
    path = db_path(category)
    if not os.path.exists(path):
        return []

    try:
        with open(path, "r", encoding="utf-8") as file:
            data = json.load(file)
            return data if isinstance(data, list) else []
    except (json.JSONDecodeError, OSError):
        return []


def db_save(category: str, articles: list[dict]) -> None:
    """Save articles atomically so partial writes do not corrupt the JSON file."""
    path = db_path(category)
    temp_path = path + ".tmp"

    with open(temp_path, "w", encoding="utf-8") as file:
        json.dump(articles, file, indent=4, ensure_ascii=False)

    os.replace(temp_path, path)


def db_append(category: str, article: dict) -> bool:
    """
    Add a new article at the top of the category file.

    Returns:
        True  -> article was saved
        False -> article was already present
    """
    articles = db_load(category)
    existing_urls = {item.get("url") for item in articles}

    if article["url"] in existing_urls:
        return False

    articles.insert(0, article)
    db_save(category, articles)
    return True


def db_pop(category: str) -> dict | None:
    """Remove and return the newest article from a category file."""
    articles = db_load(category)
    if not articles:
        return None

    latest_article = articles.pop(0)
    db_save(category, articles)
    return latest_article


def safe_get(url: str, timeout: int = REQUEST_TIMEOUT) -> requests.Response | None:
    """
    Fetch a URL with simple retries.

    Some news sites temporarily return 403/429/503, so we retry a few times.
    """
    for attempt in range(REQUEST_RETRIES):
        try:
            response = session.get(url, timeout=timeout)
            response.raise_for_status()
            return response
        except requests.exceptions.HTTPError as error:
            status_code = error.response.status_code if error.response else None
            if status_code in (403, 429, 503):
                time.sleep(3 * (attempt + 1))
            else:
                break
        except requests.exceptions.RequestException:
            time.sleep(2 * (attempt + 1))

    return None


def fetch_rss_stubs(category: str, limit: int) -> list[dict]:
    """Read an RSS feed and return lightweight article records."""
    rss_url = CATEGORY_RSS_MAP[category]
    response = safe_get(rss_url)
    if not response:
        return []

    try:
        root = ET.fromstring(response.content)
    except ET.ParseError:
        return []

    namespace = {"media": "http://search.yahoo.com/mrss/"}
    items = root.findall(".//item")[:limit]
    stubs = []

    for item in items:
        def text_of(tag: str, default: str = "") -> str:
            element = item.find(tag)
            return element.text.strip() if element is not None and element.text else default

        thumbnail = None
        for media_tag in ("media:content", "media:thumbnail"):
            element = item.find(media_tag, namespace)
            if element is not None:
                thumbnail = element.get("url")
                if thumbnail:
                    break

        keywords = [category_tag.text.strip() for category_tag in item.findall("category") if category_tag.text]
        raw_description = text_of("description")
        clean_description = BeautifulSoup(raw_description, "html.parser").get_text(" ", strip=True)

        url = text_of("link") or text_of("guid")
        if not url:
            continue

        stubs.append(
            {
                "title": text_of("title"),
                "url": url,
                "short_desc": clean_description[:300],
                "published_time": text_of("pubDate"),
                "keywords": keywords,
                "thumbnail": thumbnail,
            }
        )

    return stubs


def _try_json_ld(soup: BeautifulSoup) -> str:
    """Try articleBody from JSON-LD data first because it is often cleanest."""
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string or "")
            nodes = data if isinstance(data, list) else [data]
            for node in nodes:
                for item in node.get("@graph", [node]):
                    body = item.get("articleBody", "")
                    if body and len(body) > 80:
                        return body
        except Exception:
            pass
    return ""


def _try_dom_selectors(soup: BeautifulSoup) -> str:
    """Try a few selectors commonly seen on Economic Times article pages."""
    for attrs in [
        {"class": "artText"},
        {"class": "article-body"},
        {"id": "article-content"},
        {"role": "article"},
        {"itemprop": "articleBody"},
    ]:
        element = soup.find(attrs=attrs)
        if element:
            text = element.get_text(" ", strip=True)
            if len(text) > 100:
                return text

    # Fallback heuristic: use the div with the largest number of direct <p> tags.
    best_div = None
    best_count = 0
    for div in soup.find_all("div"):
        paragraphs = div.find_all("p", recursive=False)
        if len(paragraphs) > best_count:
            best_count = len(paragraphs)
            best_div = div

    if best_div and best_count >= 3:
        seen = set()
        blocks = []
        for paragraph in best_div.find_all("p"):
            text = paragraph.get_text(" ", strip=True)
            if len(text) > 40 and text not in seen:
                seen.add(text)
                blocks.append(text)

        result = " ".join(blocks)
        if len(result) > 100:
            return result

    return ""


def _try_readability(soup: BeautifulSoup) -> str:
    """Use all visible paragraphs while filtering obvious noise."""
    noise = re.compile(
        r"(cookie|subscribe|sign[ -]?in|follow us|download app|"
        r"advertisement|copyright|all rights reserved|terms of use|"
        r"privacy policy|read more|also read|trending|newsletter|"
        r"ET prime|login to read|exclusive for subscribers)",
        re.I,
    )

    seen = set()
    blocks = []
    for paragraph in soup.find_all("p"):
        text = paragraph.get_text(" ", strip=True)
        if len(text) < 40 or noise.search(text) or text in seen:
            continue
        seen.add(text)
        blocks.append(text)

    return " ".join(blocks) if len(blocks) >= 2 else ""


def _try_meta(soup: BeautifulSoup) -> str:
    """Fallback to page description tags if article content is hard to extract."""
    for attr, name in [("property", "og:description"), ("name", "description")]:
        tag = soup.find("meta", attrs={attr: name})
        if tag and tag.get("content", "").strip():
            return tag["content"].strip()
    return ""


def get_content(soup: BeautifulSoup) -> str:
    """Try multiple extraction strategies and return the first useful result."""
    for extractor in (_try_json_ld, _try_dom_selectors, _try_readability, _try_meta):
        result = extractor(soup)
        if result and len(result) > 60:
            return result

    return "Content could not be extracted (likely paywalled or JS-rendered)."


def get_author(soup: BeautifulSoup) -> str:
    """Extract author from meta tags or JSON-LD."""
    for attr, name in [
        ("name", "author"),
        ("property", "author"),
        ("property", "article:author"),
    ]:
        tag = soup.find("meta", attrs={attr: name})
        if tag and tag.get("content", "").strip():
            return tag["content"].strip()

    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string or "")
            nodes = data if isinstance(data, list) else [data]
            for node in nodes:
                author = node.get("author")
                if isinstance(author, dict):
                    return author.get("name", "")
                if isinstance(author, list) and author:
                    return author[0].get("name", "")
        except Exception:
            pass

    return "Economic Times"


def get_modified_time(soup: BeautifulSoup) -> str:
    """Read the last modified timestamp if the page exposes it."""
    tag = soup.find("meta", property="article:modified_time")
    return tag["content"].strip() if tag and tag.get("content") else ""


def get_category_from_page(soup: BeautifulSoup, url: str, fallback: str) -> str:
    """Get a human-readable category name from the page or article URL."""
    tag = soup.find("meta", property="article:section")
    if tag and tag.get("content"):
        return tag["content"].strip()

    try:
        section = url.split("indiatimes.com/")[1].split("/")[0]
        if section and "articleshow" not in section:
            return section.replace("-", " ").title()
    except Exception:
        pass

    return fallback.replace("_", " ").title()


def get_thumbnail_from_page(soup: BeautifulSoup) -> str | None:
    """Get the article image from Open Graph metadata."""
    tag = soup.find("meta", property="og:image")
    return tag["content"].strip() if tag and tag.get("content") else None


def trim_long_description(text: str) -> str:
    """Trim long article text if a max length is configured."""
    if MAX_DESCRIPTION_LENGTH is None or len(text) <= MAX_DESCRIPTION_LENGTH:
        return text

    return text[:MAX_DESCRIPTION_LENGTH].rsplit(" ", 1)[0] + "..."


def enrich(stub: dict, category: str) -> dict | None:
    """Turn a lightweight RSS stub into a full article record."""
    url = stub.get("url", "")
    if not url:
        return None

    response = safe_get(url)
    if not response:
        return None

    soup = BeautifulSoup(response.text, "html.parser")
    long_description = trim_long_description(get_content(soup))

    return {
        "title": stub["title"],
        "url": url,
        "category": get_category_from_page(soup, url, category),
        "author": get_author(soup),
        "published_time": stub.get("published_time", ""),
        "modified_time": get_modified_time(soup),
        "keywords": stub.get("keywords", []),
        "short_desc": stub.get("short_desc", ""),
        "long_desc": long_description,
        "thumbnail": stub.get("thumbnail") or get_thumbnail_from_page(soup),
        "source": "Economic Times",
    }


def print_cycle_summary(cycle: int, categories: list[str]) -> None:
    """Print how many saved articles exist in each category file."""
    print(f"\n{'-' * 60}")
    print(f"Cycle #{cycle} complete.")
    print("DB snapshot:")
    for category in categories:
        article_count = len(db_load(category))
        print(f"  {category:<20} -> {article_count:>4} articles  ({db_path(category)})")
    print(f"{'-' * 60}")


def run_cycle(cycle: int, categories: list[str]) -> None:
    """Run one scraping cycle across all selected categories."""
    print(f"\n{'=' * 60}")
    print(f"Cycle #{cycle}  |  {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{'=' * 60}")

    for index, category in enumerate(categories, start=1):
        print(f"\n[{index}/{len(categories)}] Category: {category.upper()}")

        stubs = fetch_rss_stubs(category, ARTICLES_PER_CATEGORY)
        if not stubs:
            print("  No RSS articles found for this category.")
        else:
            existing_urls = {article.get("url") for article in db_load(category)}

            for stub in stubs:
                article_url = stub["url"]
                article_title = stub.get("title", "Untitled article")

                # Skip already saved articles before fetching the full page again.
                if article_url in existing_urls:
                    print(f"  Skipping duplicate: {article_title[:60]}")
                    continue

                print(f"  Fetching: {article_url}")
                article = enrich(stub, category)

                if article:
                    saved = db_append(category, article)
                    if saved:
                        existing_urls.add(article_url)
                        print(f"  Saved: {article['title'][:60]}")
                    else:
                        print(f"  Duplicate after save check: {article['title'][:60]}")
                else:
                    print(f"  Failed: {article_title[:60]}")

        if index < len(categories):
            print(f"\n  Waiting {WAIT_BETWEEN_CATEGORIES}s before the next category...")
            time.sleep(WAIT_BETWEEN_CATEGORIES)

    print_cycle_summary(cycle, categories)


def print_startup_summary(categories: list[str]) -> None:
    """Show the current settings before the scraper starts."""
    run_mode = "continuous" if RUN_CONTINUOUSLY else "single cycle"

    print("\nEconomic Times Scraper")
    print(f"Output folder           : {os.path.abspath(OUTPUT_DIR)}")
    print(f"Selected categories     : {len(categories)}")
    print(f"Articles per category   : {ARTICLES_PER_CATEGORY}")
    print(f"Run mode                : {run_mode}")
    print(f"Wait between categories : {WAIT_BETWEEN_CATEGORIES}s")
    print(f"Wait between cycles     : {WAIT_BETWEEN_CYCLES}s")
    print(f"Request timeout         : {REQUEST_TIMEOUT}s")
    print(f"Request retries         : {REQUEST_RETRIES}\n")


def main() -> None:
    """Entry point for the scraper."""
    categories = get_category_list()
    print_startup_summary(categories)

    cycle = 1
    while True:
        try:
            run_cycle(cycle, categories)

            if not RUN_CONTINUOUSLY:
                print("\nRun complete. Exiting because RUN_CONTINUOUSLY = False.")
                break

            cycle += 1
            print(f"\nSleeping for {WAIT_BETWEEN_CYCLES}s before the next cycle...\n")
            time.sleep(WAIT_BETWEEN_CYCLES)
        except KeyboardInterrupt:
            print("\nStopped by user.")
            break


if __name__ == "__main__":
    main()

📊 Expected Output

When you run the script, it will print a clean, readable log in your terminal showing exactly what it is doing. It skips duplicates automatically so you never save the same article twice!

============================================================ Cycle #1 | 2024-05-18 10:00:00 ============================================================ [1/33] Category: TOP_NEWS Fetching: https://economictimes.indiatimes.com/... Saved: Sensex crashes 800 points amid global sell-off... Fetching: https://economictimes.indiatimes.com/... Saved: RBI announces new guidelines for digital lending... Waiting 3.0s before the next category... ------------------------------------------------------------ Cycle #1 complete. DB snapshot: top_news -> 2 articles (NEWS_DATA_ET\top_news.json) ------------------------------------------------------------ Sleeping for 200s before the next cycle...

The JSON Data Structure

If you open one of the generated JSON files (like top_news.json), you will see beautifully structured data ready for your next big project:

[ { "title": "Sensex crashes 800 points amid global sell-off", "url": "https://economictimes.indiatimes.com/...", "category": "Markets", "author": "ET Markets Web Team", "published_time": "Sat, 18 May 2024 09:30:00 IST", "modified_time": "2024-05-18T09:35:00Z", "keywords": [ "Sensex", "Nifty", "Stock Market" ], "short_desc": "Indian equity indices plunged on Saturday...", "long_desc": "Indian equity indices plunged on Saturday following weak global cues...", "thumbnail": "https://img.etimg.com/thumb/msid-12345.jpg", "source": "Economic Times" } ]

🚀 Next Steps

This data is perfect for building a custom news dashboard, running Natural Language Processing (NLP) to perform sentiment analysis on the stock market, or feeding into an AI model. Happy scraping!

Post a Comment

Do Leave Your Comments...

Previous Post Next Post

Contact Form