public repository of the spon_api. Initial publication
This commit is contained in:
Orihara Izaya
@@ -0,0 +1,16 @@
|
||||
"""
|
||||
spon_api is a package for scraping article text, metadata, and comments from DER
|
||||
SPIEGEL. It consists of three modules:
|
||||
- `archive` which fetches the article list for a specific day
|
||||
- `article` which fetches text and metadata for a specific article
|
||||
- `comments` which fetches text and metadata of comments for a specific article
|
||||
"""
|
||||
|
||||
from spon_api import archive, article, comments
|
||||
|
||||
__author__ = "Gerrit Anders"
|
||||
__license__ = "GPL 3.0"
|
||||
__version__ = "1.2"
|
||||
__maintainer__ = "Gerrit Anders"
|
||||
__email__ = "g.anders@iwm-tuebingen.de"
|
||||
__status__ = "Maintenance"
|
||||
@@ -0,0 +1,125 @@
|
||||
"""
|
||||
The `archive.py` module handles scraping archive data from specified URLs.
|
||||
The retrieval is done by date, with each article on that date being fetched.
|
||||
"""
|
||||
|
||||
import datetime as dt
|
||||
import dateparser
|
||||
import requests
|
||||
import tldextract as url_extract
|
||||
from lxml.html import HtmlElement, fromstring
|
||||
from typing import Optional, Any
|
||||
|
||||
|
||||
class Archive:
|
||||
"""
|
||||
A class used to represent an archive of articles for a specific date.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
date : dt.date
|
||||
The date for which the archive should be fetched.
|
||||
doc : Optional[HtmlElement]
|
||||
The parsed HTML document of the archive page.
|
||||
|
||||
Methods
|
||||
-------
|
||||
fetch():
|
||||
Fetches the archive page for the specified date.
|
||||
|
||||
parse() -> list[dict[str, Any]]:
|
||||
Parses the fetched archive page to extract article details.
|
||||
"""
|
||||
|
||||
def __init__(self, date: dt.date):
|
||||
"""
|
||||
Constructs the Archive object.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
date : dt.date
|
||||
The date for which the archive should be fetched.
|
||||
"""
|
||||
self.date: dt.date = date
|
||||
self.doc: Optional[HtmlElement] = None
|
||||
|
||||
def fetch(self) -> None:
|
||||
"""
|
||||
Fetches the archive page for the specified date and sets the `doc`
|
||||
attribute with the parsed HTML document.
|
||||
|
||||
Raises
|
||||
------
|
||||
requests.exceptions.RequestException
|
||||
If the HTTP request to fetch the archive page fails.
|
||||
"""
|
||||
archive_url: str = f'https://www.spiegel.de/nachrichtenarchiv/artikel-{self.date.strftime("%d.%m.%Y")}.html'
|
||||
response: requests.Response = requests.get(archive_url)
|
||||
self.doc: HtmlElement = fromstring(response.text)
|
||||
|
||||
def parse(self) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Parses the fetched archive page to extract details of each article.
|
||||
|
||||
Returns
|
||||
-------
|
||||
list[dict[str, str]]
|
||||
A list of dictionaries, each containing details of an article including URL,
|
||||
headline, and whether it's a paid article.
|
||||
|
||||
Raises
|
||||
------
|
||||
AssertionError
|
||||
"""
|
||||
assert self.doc is not None, "Fetch the document first before parsing."
|
||||
|
||||
articles: list[dict[str, str]] = []
|
||||
for article in self.doc.xpath("//article"):
|
||||
article_details: dict[str, Any] = {}
|
||||
|
||||
url: str = article.xpath(".//a/@href")[0]
|
||||
|
||||
if url_extract.extract(url).registered_domain != "spiegel.de":
|
||||
continue
|
||||
|
||||
headline: str = article.xpath(".//a/@title")[0]
|
||||
spiegel_plus_icon: str = article.xpath(
|
||||
'boolean(.//span[@data-flag-name="Spplus-paid"])'
|
||||
)
|
||||
string_date_published: str = article.xpath("string(./footer/span[1])")
|
||||
channel: str = article.xpath("string(./footer/span[3])")
|
||||
|
||||
article_details["url"] = url
|
||||
article_details["headline"] = headline
|
||||
article_details["is_paid"] = bool(spiegel_plus_icon)
|
||||
article_details["date_published"] = self._parse_date(string_date_published)
|
||||
article_details["channel"] = channel
|
||||
|
||||
articles.append(article_details)
|
||||
|
||||
return articles
|
||||
|
||||
@staticmethod
|
||||
def _parse_date(date_string: str) -> Optional[dt.datetime]:
|
||||
"""
|
||||
Parses a date string using the `dateparser` library.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
date_string : str
|
||||
The date string to be parsed.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Optional[dt.datetime]
|
||||
The parsed datetime object or None if parsing fails.
|
||||
"""
|
||||
try:
|
||||
parsed_date: dt.datetime = dateparser.parse(
|
||||
date_string=date_string, languages=["de"], settings={}
|
||||
)
|
||||
|
||||
return parsed_date
|
||||
|
||||
except ValueError:
|
||||
return None
|
||||
@@ -0,0 +1,113 @@
|
||||
"""
|
||||
The `article.py` module handles fetching and parsing individual articles.
|
||||
The articles are extracted by url, with the url pointing to an article.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import requests
|
||||
from lxml.html import HtmlElement, fromstring
|
||||
from typing import Dict, Optional
|
||||
|
||||
|
||||
class Article:
|
||||
"""A class used to represent a specific article.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
url : str
|
||||
The URL of the article.
|
||||
doc : Optional[HtmlElement]
|
||||
The parsed HTML document of the article.
|
||||
|
||||
Methods
|
||||
-------
|
||||
fetch() -> None:
|
||||
Fetches the article page and sets the `doc` attribute with the parsed HTML document.
|
||||
parse() -> dict[str, Any]:
|
||||
Parses the fetched article HTML to extract necessary details.
|
||||
"""
|
||||
|
||||
def __init__(self, url: str):
|
||||
"""Constructs an Article object.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
url : str
|
||||
The URL of the article.
|
||||
"""
|
||||
self.url: str = url
|
||||
self.doc: Optional[HtmlElement] = None
|
||||
|
||||
def fetch(self):
|
||||
"""Fetches the article HTML content and sets the `doc` attribute with the parsed HTML document.
|
||||
|
||||
Raises
|
||||
------
|
||||
requests.exceptions.RequestException
|
||||
If the HTTP request to fetch the article page fails.
|
||||
"""
|
||||
resp: requests.Response = requests.get(self.url)
|
||||
self.doc: HtmlElement = fromstring(resp.text)
|
||||
|
||||
def parse(self) -> dict:
|
||||
"""Parses the fetched article HTML to extract necessary details.
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict[str, Any]
|
||||
A dictionary containing article details such as URL, ID, channel, subchannel, headline,
|
||||
intro, text, topics, author, comments_enabled, dates (created, modified, published), and breadcrumbs.
|
||||
|
||||
Raises
|
||||
------
|
||||
AssertionError
|
||||
If `fetch` method was not called before parsing.
|
||||
"""
|
||||
assert self.doc is not None, "Fetch the document first before parsing."
|
||||
|
||||
linked_data_content: str = self.doc.xpath(
|
||||
'//script[@type="application/ld+json"]/text()'
|
||||
)[0]
|
||||
linked_data: dict = json.loads(linked_data_content)
|
||||
linked_data_by_type: dict[str, dict] = {
|
||||
ld_entry["@type"]: ld_entry for ld_entry in linked_data
|
||||
}
|
||||
news_linked_data: dict = linked_data_by_type["NewsArticle"]
|
||||
|
||||
settings: dict = json.loads(
|
||||
self.doc.xpath('//script[@type="application/settings+json"]/text()')[0]
|
||||
)
|
||||
info: dict = settings["editorial"]["info"]
|
||||
|
||||
text_nodes: list[HtmlElement] = self.doc.cssselect(
|
||||
"main .word-wrap > p, main .word-wrap > h3, main .word-wrap > ul > li, main .word-wrap > ol > li"
|
||||
)
|
||||
full_text: str = "\n".join([node.text_content() for node in text_nodes])
|
||||
text: str = re.sub(r"\n+", "\n", full_text).strip()
|
||||
|
||||
article_information: dict = {
|
||||
"url": self.doc.xpath('string(//link[@rel="canonical"]/@href)'),
|
||||
"id": info["article_id"],
|
||||
"channel": info["channel"],
|
||||
"subchannel": info["subchannel"],
|
||||
"headline": {"main": info["headline"], "social": info["headline_social"]},
|
||||
"intro": info["intro"],
|
||||
"text": text,
|
||||
"topics": info["topics"],
|
||||
"author": settings["editorial"]["author"],
|
||||
"comments_enabled": settings["editorial"]["attributes"][
|
||||
"is_comments_enabled"
|
||||
],
|
||||
"date_created": news_linked_data["dateCreated"],
|
||||
"date_modified": news_linked_data["dateModified"],
|
||||
"date_published": news_linked_data["datePublished"],
|
||||
"breadcrumbs": [
|
||||
breadcrumb["item"]["name"]
|
||||
for breadcrumb in linked_data_by_type["BreadcrumbList"][
|
||||
"itemListElement"
|
||||
]
|
||||
],
|
||||
}
|
||||
|
||||
return article_information
|
||||
@@ -0,0 +1,138 @@
|
||||
"""
|
||||
The `comments.py` module handles fetching and parsing comments for individual articles.
|
||||
Comments are extracted based on the article ID, with options for nesting replies.
|
||||
"""
|
||||
|
||||
from collections import defaultdict
|
||||
import requests
|
||||
from typing import Optional, Any
|
||||
from .config import TALK_ENDPOINT_URL
|
||||
|
||||
|
||||
class Comments:
|
||||
"""A class used to represent and handle comments for a specific article.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
article_id : str
|
||||
The ID of the article for which comments are to be fetched.
|
||||
nesting : bool
|
||||
Whether to nest replies under their parent comments or not.
|
||||
page_size_limit : int
|
||||
The maximum number of comments to fetch per request.
|
||||
comments : Optional[list[dict[str, Any]]]
|
||||
List to store fetched comments data.
|
||||
|
||||
Methods
|
||||
-------
|
||||
fetch() -> None:
|
||||
Fetches comments from the server and stores them in the `comments` attribute.
|
||||
parse() -> list[dict[str, Any]]:
|
||||
Parses the fetched comments into a structured format, optionally nesting replies.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, article_id: str, nesting: bool = True, page_size_limit: int = 1000
|
||||
):
|
||||
"""Constructs a Comments object.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
article_id : str
|
||||
The ID of the article for which comments are to be fetched.
|
||||
nesting : bool, optional
|
||||
Whether to nest replies under their parent comments or not (default is True).
|
||||
page_size_limit : int, optional
|
||||
The maximum number of comments to fetch per request (default is 1000).
|
||||
"""
|
||||
self.article_id: str = article_id
|
||||
self.nesting: bool = nesting
|
||||
self.page_size_limit: int = page_size_limit
|
||||
self.comments: Optional[list] = None
|
||||
|
||||
def fetch(self):
|
||||
"""Fetches comments from the server and stores them in the `comments` attribute.
|
||||
|
||||
Raises
|
||||
------
|
||||
requests.RequestException
|
||||
If the request to fetch comments fails.
|
||||
"""
|
||||
query: str = """
|
||||
query GetComments($assetId: ID!, $cursor: Cursor, $limit: Int) {
|
||||
asset(id: $assetId) {
|
||||
comments(deep: true, query: {sortBy: CREATED_AT, sortOrder: ASC, limit: $limit, cursor: $cursor}) {
|
||||
endCursor
|
||||
hasNextPage
|
||||
nodes {
|
||||
id
|
||||
parent {id}
|
||||
body
|
||||
action_summaries {
|
||||
__typename
|
||||
count
|
||||
}
|
||||
tags {
|
||||
tag {
|
||||
...
|
||||
}
|
||||
...
|
||||
}
|
||||
...
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
self.comments: list = []
|
||||
current_cursor = None
|
||||
while True:
|
||||
response: requests.Response = requests.post(
|
||||
TALK_ENDPOINT_URL,
|
||||
json={
|
||||
"query": query,
|
||||
"variables": {
|
||||
"assetId": self.article_id,
|
||||
"cursor": current_cursor,
|
||||
"limit": self.page_size_limit,
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
response_data: dict[str, Any] = response.json()["data"]
|
||||
comments_page_data: dict[str, Any] = response_data["asset"]["comments"]
|
||||
self.comments.extend(comments_page_data["nodes"])
|
||||
current_cursor = comments_page_data["endCursor"]
|
||||
if not comments_page_data["hasNextPage"]:
|
||||
break
|
||||
|
||||
def parse(self) -> list[dict]:
|
||||
"""Parses the fetched comments into a structured format, optionally nesting replies.
|
||||
|
||||
Returns
|
||||
-------
|
||||
List[Dict[str, Any]]
|
||||
A list of parsed comments, with replies nested under their parent comments if nesting is enabled.
|
||||
|
||||
Raises
|
||||
------
|
||||
AssertionError
|
||||
If `fetch` method was not called before parsing.
|
||||
"""
|
||||
assert self.comments is not None, "Fetch the comments first before parsing."
|
||||
|
||||
top_level_comments: list[dict[str, Any]] = []
|
||||
replies_dict: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||||
for comment in self.comments:
|
||||
comment["replies"]: list = []
|
||||
parent_ref: Optional[dict[str, Any]] = comment.pop("parent")
|
||||
if parent_ref is None:
|
||||
top_level_comments.append(comment)
|
||||
else:
|
||||
if self.nesting:
|
||||
replies_dict[parent_ref["id"]].append(comment)
|
||||
else:
|
||||
comment["parent_id"] = parent_ref["id"]
|
||||
top_level_comments.append(comment)
|
||||
|
||||
return top_level_comments
|
||||
@@ -0,0 +1,7 @@
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
from pathlib import Path
|
||||
|
||||
env_path = Path(__file__).parent / ".env"
|
||||
load_dotenv(dotenv_path=env_path)
|
||||
TALK_ENDPOINT_URL = os.getenv("TALK_ENDPOINT_URL")
|
||||
Reference in New Issue
Block a user