📰 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:
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!
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:
🚀 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!
