diff --git a/nitter_scraper/__init__.py b/nitter_scraper/__init__.py index 987b5c6..d526447 100644 --- a/nitter_scraper/__init__.py +++ b/nitter_scraper/__init__.py @@ -4,4 +4,4 @@ __all__ = ["get_profile", "get_tweets", "NitterScraper"] -__version__ = "0.5.2" +__version__ = "0.5.3" diff --git a/nitter_scraper/nitter.py b/nitter_scraper/nitter.py index de62f14..bf2dce7 100644 --- a/nitter_scraper/nitter.py +++ b/nitter_scraper/nitter.py @@ -15,6 +15,7 @@ from nitter_scraper.paths import PROJECT_ROOT, TEMPLATES_DIRECTORY # noqa: I202, I100 from nitter_scraper.profile import get_profile # noqa: I202, I100 from nitter_scraper.tweets import get_tweets # noqa: I202, I100 +from nitter_scraper.schema import Tweet class DockerBase(Base): @@ -113,7 +114,16 @@ def get_profile(self, username: str, not_found_ok: bool = False): """ return get_profile(username=username, not_found_ok=not_found_ok, address=self.address) - def get_tweets(self, username: str, pages: int = 25, break_on_tweet_id: Optional[int] = None): + def get_tweets( + self, + username: str, + pages: int = 25, + break_on_tweet_id: Optional[int] = None, + address="https://nitter.net", + headers: Optional[dict[str, str]] = None, + params: Optional[dict[str, str]] = None, + proxies: Optional[dict[str, str]] = None, + ) -> Tweet: """Gets the target users tweets This is a modified version of nitter_scraper.tweets.get_tweets(). @@ -125,10 +135,14 @@ def get_tweets(self, username: str, pages: int = 25, break_on_tweet_id: Optional pages: Max number of pages to lookback starting from the latest tweet. break_on_tweet_id: Gives the ability to break out of a loop if a tweets id is found. address: The address to scrape from. The default is https://nitter.net which should - be used as a fallback address. - + be used as a fallback address. Refer to https://github.com/zedeus/nitter/wiki/Instances + for a list of instances. + headers: HTTP headers to be passed. + params: Search query parameters as found in Nitter URLs upon search. + proxies: Passed to HTMLSession.get(). + Yields: - Tweet Objects + Tweet Objects """ @@ -137,6 +151,9 @@ def get_tweets(self, username: str, pages: int = 25, break_on_tweet_id: Optional pages=pages, break_on_tweet_id=break_on_tweet_id, address=self.address, + headers=headers, + params=params, + proxies=proxies ) def start(self): @@ -145,7 +162,7 @@ def start(self): client = self._get_client() self.container = client.containers.run( - image="zedeus/nitter:2d788704b1d922162b0e7d910f48f9127bc82d7f", + image="zedeus/nitter:2c6cabb4abe79166ce9973d8652fb213c1b0c5a2", auto_remove=True, ports=self.ports, detach=True, diff --git a/nitter_scraper/profile.py b/nitter_scraper/profile.py index e059f6d..33ece48 100644 --- a/nitter_scraper/profile.py +++ b/nitter_scraper/profile.py @@ -94,8 +94,6 @@ def profile_parser(elements: Dict) -> Dict: elements["is_verified"] = True if elements.get("is_verified") else False - elements["is_private"] = True if elements.get("is_private") else False - if elements.get("biography"): elements["biography"] = elements["biography"].text @@ -151,10 +149,6 @@ def html_parser(html: HTML) -> Dict: ".profile-card-fullname .icon-container .verified-icon", first=True ) - elements["is_private"] = html.find( - ".profile-card-fullname .icon-container .icon-lock", first=True - ) - elements["profile_photo"] = html.find(".profile-card-avatar", first=True) elements["banner_photo"] = html.find(".profile-banner a", first=True) @@ -177,7 +171,7 @@ def html_parser(html: HTML) -> Dict: def get_profile( - username: str, not_found_ok: bool = False, address: str = "https://nitter.net" + username: str, not_found_ok: bool = False, address: str = "https://nitter.net", proxies: Optional[dict[str, str]] = None, ) -> Optional[Profile]: """Scrapes nitter for the target users profile information. @@ -188,6 +182,7 @@ def get_profile( address: The address to scrape profile data from. The default scrape location is 'https://nitter.net' which should be used as a backup. This value will normally be replaced by the address of a local docker container instance of nitter. + proxies: Passed to HTMLSession.get() Returns: Profile object if successfully scraped, otherwise None. @@ -199,7 +194,7 @@ def get_profile( """ url = f"{address}/{username}" session = HTMLSession() - response = session.get(url) + response = session.get(url, proxies=proxies) if response.status_code == 200: # user exists elements = html_parser(response.html) diff --git a/nitter_scraper/schema.py b/nitter_scraper/schema.py index 5f53e40..6c10a94 100644 --- a/nitter_scraper/schema.py +++ b/nitter_scraper/schema.py @@ -3,6 +3,30 @@ from pydantic import BaseModel as Base +class Card(Base): + """ + Contains embedded card data. + Attributes: + title: Card's title + description: Card's text + image: Nitter relative URL of the card's image + """ + title: Optional[str] + description: Optional[str] + image: Optional[str] + + @classmethod + def from_dict(cls, elements: Dict[str, Any]) -> "Card": + """Creates Card object from a dictionary of processed text elements. + + Args: + elements: Preprocessed attributes of a tweet object. + + Returns: + Card object. + + """ + return cls(**elements) class Entries(Base): """Contains metadata parsed from the text contents of the tweet. @@ -43,6 +67,7 @@ class Tweet(Base): likes: A count of the times the tweet was liked. entries: Contains the entries object which holds metadata on the tweets text contents. + card: Card attached to the tweet, if any. """ @@ -58,6 +83,7 @@ class Tweet(Base): retweets: int likes: int entries: Entries + card: Optional[Card] @classmethod def from_dict(cls, elements: Dict[str, Any]) -> "Tweet": @@ -88,7 +114,6 @@ class Profile(Base): followers_count: The number of followers this account has. likes_count: Number of likes the follower has received. is_verified: Indicates if the user has been verified. - is_private: Indicates if the users profile is set to private. banner_photo: URL reference to the profiles banner. biography: The users autobiography. user_id: User identification number generated by twitter. @@ -106,7 +131,7 @@ class Profile(Base): followers_count: int likes_count: int is_verified: bool - is_private: bool + # is_private: bool banner_photo: Optional[str] = None biography: Optional[str] = None diff --git a/nitter_scraper/tweets.py b/nitter_scraper/tweets.py index 1e02335..64eeca8 100644 --- a/nitter_scraper/tweets.py +++ b/nitter_scraper/tweets.py @@ -1,11 +1,13 @@ """Module for scraping tweets""" -from datetime import datetime +#from datetime import datetime +from dateutil import parser as dateparser import re from typing import Dict, Optional from requests_html import HTMLSession from nitter_scraper.schema import Tweet # noqa: I100, I202 +from nitter_scraper.schema import Card def link_parser(tweet_link): @@ -19,22 +21,7 @@ def link_parser(tweet_link): def date_parser(tweet_date): - split_datetime = tweet_date.split(",") - - day, month, year = split_datetime[0].strip().split("/") - hour, minute, second = split_datetime[1].strip().split(":") - - data = {} - - data["day"] = int(day) - data["month"] = int(month) - data["year"] = int(year) - - data["hour"] = int(hour) - data["minute"] = int(minute) - data["second"] = int(second) - - return datetime(**data) + return dateparser.parse(tweet_date.replace('ยท', '')) def clean_stat(stat): @@ -50,11 +37,11 @@ def stats_parser(tweet_stats): return stats -def attachment_parser(attachments): +def attachment_parser(attachements): photos, videos = [], [] - if attachments: - photos = [i.attrs["src"] for i in attachments.find("img")] - videos = [i.attrs["src"] for i in attachments.find("source")] + if attachements: + photos = [i.attrs["src"] for i in attachements.find("img")] + videos = [i.attrs["src"] for i in attachements.find("source")] return photos, videos @@ -93,17 +80,34 @@ def parse_tweet(html) -> Dict: data["text"] = content.text # tweet_header = html.find(".tweet-header") #NOTE: Maybe useful later on - + + card = html.find("div.card.large", first=True) + + if card: + data["card"] = {} + card_title = card.find(".card-title", first=True) + if card_title: + data["card"]["title"] = card_title.text + card_description = card.find(".card-description", first=True) + if card_description: + data["card"]["description"] = card_description.text + + card_image = card.find(".card-image img", first=True) + if card_image: + data["card"]["image"] = card_image.attrs["src"] + + data["card"] = Card.from_dict(data["card"]) + else: + data["card"] = None + + stats = stats_parser(html.find(".tweet-stats", first=True)) - if stats.get("comment"): - data["replies"] = clean_stat(stats.get("comment")) + data["replies"] = clean_stat(stats.get("comment")) if stats.get("comment") else 0 - if stats.get("retweet"): - data["retweets"] = clean_stat(stats.get("retweet")) + data["retweets"] = clean_stat(stats.get("retweet"))if stats.get("retweet") else 0 - if stats.get("heart"): - data["likes"] = clean_stat(stats.get("heart")) + data["likes"] = clean_stat(stats.get("heart")) if stats.get("heart") else 0 entries = {} entries["hashtags"] = hashtag_parser(content.text) @@ -125,14 +129,16 @@ def timeline_parser(html): def pagination_parser(timeline, address, username) -> str: next_page = list(timeline.find(".show-more")[-1].links)[0] - return f"{address}/{username}{next_page}" - + return f"{address}/{username}/search{next_page}" def get_tweets( username: str, pages: int = 25, break_on_tweet_id: Optional[int] = None, address="https://nitter.net", + headers: Optional[dict[str, str]] = None, + params: Optional[dict[str, str]] = None, + proxies: Optional[dict[str, str]] = None, ) -> Tweet: """Gets the target users tweets @@ -141,17 +147,24 @@ def get_tweets( pages: Max number of pages to lookback starting from the latest tweet. break_on_tweet_id: Gives the ability to break out of a loop if a tweets id is found. address: The address to scrape from. The default is https://nitter.net which should - be used as a fallback address. + be used as a fallback address. Refer to https://github.com/zedeus/nitter/wiki/Instances + for a list of instances. + headers: HTTP headers to be passed. + params: Search query parameters as found in Nitter URLs upon search. + proxies: Passed to HTMLSession.get(). Yields: Tweet Objects """ - url = f"{address}/{username}" + url = f"{address}/{username}/search" session = HTMLSession() + + if headers: + session.headers.update(headers) def gen_tweets(pages): - response = session.get(url) + response = session.get(url, params=params, proxies=proxies) while pages > 0: if response.status_code == 200: @@ -174,7 +187,12 @@ def gen_tweets(pages): yield tweet - response = session.get(next_url) + response = session.get(next_url, params=params, proxies=proxies) + else: + #print(f'Response status code: {response.status_code}') + #last_error_html = response.html + pages = 0 + break pages -= 1 yield from gen_tweets(pages) diff --git a/pyproject.toml b/pyproject.toml index 740bf6f..0fd40a7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "nitter_scraper" -version = "0.5.2" +version = "0.5.3" description = "Scrape Twitter API without authentication using Nitter." authors = ["dgnsrekt "] readme = "README.md"