fetch_articles.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "from bs4 import BeautifulSoup, NavigableString\n",
    "import newspaper\n",
    "import pandas as pd\n",
    "from datetime import datetime\n",
    "import re\n",
    "import os\n",
    "from dataclasses import dataclass, asdict\n",
    "# headers needed to simulate agent\n",
    "headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:97.0) Gecko/20100101 Firefox/97.0'}\n",
    "\n",
    "# data classes\n",
    "@dataclass\n",
    "class LandingPageArticle:\n",
    "    title: str\n",
    "    url: str\n",
    "\n",
    "    def __str__(self):\n",
    "        return f'LandingPageArticle(title={self.title}, url={self.url})'\n",
    "\n",
    "@dataclass\n",
    "class Article:\n",
    "    title: str\n",
    "    url: str\n",
    "    summary: str\n",
    "    last_edit_date: datetime\n",
    "    authors: list[str]\n",
    "    text: str\n",
    "    topic: str | None = None\n",
    "    source: str | None = None\n",
    "    tags: str | None = None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_article_links(zeit_online_url) -> list[LandingPageArticle]:\n",
    "    response = requests.get(zeit_online_url, headers=headers)\n",
    "    if response.status_code != 200:\n",
    "        raise Exception(\"Can't get todays articles\")\n",
    "    soup = BeautifulSoup(response.content, 'html.parser')\n",
    "    articles : list[LandingPageArticle] = []\n",
    "    for article_tag in soup.find_all('article'):\n",
    "        if 'zon-teaser--standard' not in article_tag.get('class', []) or \\\n",
    "            not article_tag.get('data-zplus') or 'zplus' == article_tag.get('data-zplus'):\n",
    "            continue\n",
    "\n",
    "        a_tag = article_tag.find('a', href=True)\n",
    "        if not a_tag or not a_tag.get('href').startswith('https://www.zeit.de/'):\n",
    "            continue\n",
    "        \n",
    "        title = a_tag.get_text(strip=True)\n",
    "        url = a_tag['href']\n",
    "        articles.append(LandingPageArticle(title=title, url=url))\n",
    "    \n",
    "    return articles\n",
    "\n",
    "\n",
    "def scrape_article(article_url: str) -> Article:\n",
    "    response = requests.get(article_url, headers=headers)\n",
    "    if response.status_code != 200:\n",
    "        print(f\"Error scraping article: {article_url}!\")\n",
    "        return None  \n",
    "    soup = BeautifulSoup(response.content, 'html.parser')\n",
    "    try:\n",
    "        article_heading = soup.find('h1', class_='article-heading')\n",
    "        title = article_heading.find('span', class_='article-heading__title').text.strip() if article_heading else None\n",
    "        topic = ''.join(child for child in article_heading.find('span', class_='article-heading__kicker') if isinstance(child, NavigableString)).strip() if article_heading else None\n",
    "\n",
    "        summary_tag = soup.find('div', class_='summary')\n",
    "        summary = summary_tag.text.strip() if summary_tag else None\n",
    "\n",
    "        author_tag = soup.find_all('a', href=lambda href: href and href.startswith('/autoren/'))\n",
    "        authors = [author.get(\"title\", \"\") for author in author_tag] if author_tag else []\n",
    "\n",
    "        metadata_tag = soup.find('div', class_='metadata')\n",
    "        time_tag = metadata_tag.find('time')\n",
    "        source_tag = metadata_tag.find('span', class_='metadata__source')\n",
    "        date = datetime.fromisoformat(time_tag.get('datetime')) if time_tag else None\n",
    "        source = re.sub(r'\\n\\s*,', \n",
    "                        '', \n",
    "                        ''.join([source.text for source in source_tag if source.name != \"a\" or (source.name == \"a\" and not source.get(\"href\", \"\").startswith('/autoren/'))]).strip()).rstrip(',') \\\n",
    "                 if source_tag else None\n",
    "\n",
    "        paragraphs = soup.find_all('p', class_='paragraph article__item')\n",
    "        text = \"\\n\".join([p.text.strip().replace('\\n', '') for p in paragraphs])\n",
    "\n",
    "        tag_list = soup.find('ul', class_='article-tags__list')\n",
    "        tags = [tag.text.strip() for tag in tag_list.find_all('a')] if tag_list else []\n",
    "    except:\n",
    "        print(f\"Error scraping article: {article_url} with newspaper4k!\")\n",
    "        return None\n",
    "    \n",
    "    return Article(title = title,\n",
    "                  topic=topic,\n",
    "                  url = article_url, \n",
    "                  summary = summary,\n",
    "                  authors = authors,\n",
    "                  source= source,\n",
    "                  last_edit_date = date,\n",
    "                  text = text,\n",
    "                  tags=tags)\n",
    "\n",
    "\n",
    "def scrape_article_with_newspaper4k(article_url: str) -> Article:\n",
    "    try:\n",
    "        article = newspaper.article(article_url)\n",
    "    except Exception as e:\n",
    "        print(f\"Error scraping article: {article_url} with newspaper4k!\")\n",
    "        return None\n",
    "    return Article(title = article.title,\n",
    "                              url = article_url, \n",
    "                              summary = article.meta_description,\n",
    "                              authors = article.authors,\n",
    "                              last_edit_date = article.publish_date,\n",
    "                              text = article.text)\n",
    "    \n",
    "def scrape_articles_and_save_as_csv(article_urls: list[str], newspaper: bool = False) -> pd.DataFrame:\n",
    "    scraped_articles = []\n",
    "    for url in article_urls:\n",
    "        scraped_article = scrape_article_with_newspaper4k(url) if newspaper else scrape_article(url)\n",
    "        if scraped_article is not None:\n",
    "            scraped_articles.append(scraped_article)\n",
    "            \n",
    "    df = pd.DataFrame([asdict(scraped_article) for scraped_article in scraped_articles])\n",
    "    if len(df) == 0:\n",
    "        return None\n",
    "    \n",
    "    if newspaper:\n",
    "        df.drop(columns=['topic', 'source', 'tags'], inplace=True)\n",
    "    \n",
    "    os.makedirs('./data/newspaper4k', exist_ok=True)\n",
    "    os.makedirs('./data/self', exist_ok=True)\n",
    "    df.to_csv(f'./data/{\"newspaper4k\" if newspaper else \"self\"}/articles{datetime.now().strftime(\"%Y%m%d%H%M\")}.csv', index=False)\n",
    "    \n",
    "    return df\n",
    "\n",
    "def combine_csv_articles(dir) -> pd.DataFrame:\n",
    "    dataframes = []\n",
    "    for file in os.listdir(dir):\n",
    "        df = pd.read_csv(os.path.join(dir, file), encoding=\"utf-8\")\n",
    "        dataframes.append(df)\n",
    "        \n",
    "    return pd.concat(dataframes, ignore_index=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "All Articles Count: 238\n"
     ]
    }
   ],
   "source": [
    "articles = get_article_links(\"https://www.zeit.de/\")\n",
    "df_current_articles = pd.DataFrame([asdict(article) for article in articles])\n",
    "old_articles_path = './data/all_articles.csv'\n",
    "df_old_articles = pd.read_csv(old_articles_path) if os.path.exists(old_articles_path) else pd.DataFrame(columns=df_current_articles.columns)\n",
    "df_new_articles = df_current_articles[~df_current_articles['url'].isin(set(df_old_articles['url'].tolist()))]\n",
    "\n",
    "new_articles_list = df_new_articles['url'].tolist()\n",
    "scrape_articles_and_save_as_csv(new_articles_list) # own scraping method\n",
    "scrape_articles_and_save_as_csv(new_articles_list, True) # with newspaper4k\n",
    "\n",
    "df_all_articles = pd.concat([df_old_articles, df_new_articles])\n",
    "df_all_articles.to_csv(old_articles_path, index=False)\n",
    "print(\"All Articles Count: \" + str(len(df_all_articles)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>title</th>\n",
       "      <th>url</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "Empty DataFrame\n",
       "Columns: [title, url]\n",
       "Index: []"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_new_articles"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Combine dfs\n",
    "newspaper4k_articles_dir = \"./data/newspaper4k\"\n",
    "self_articles_dir = \"./data/self\"\n",
    "\n",
    "df_all_articles_self = combine_csv_articles(self_articles_dir)\n",
    "df_all_articles_newspaper = combine_csv_articles(newspaper4k_articles_dir)\n",
    "df_all_articles_self.to_csv('./data/all_articles_self.csv', index=False)\n",
    "df_all_articles_newspaper.to_csv('./data/all_articles_newspaper4k.csv', index=False)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}