Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion nitter_scraper/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,4 +4,4 @@

__all__ = ["get_profile", "get_tweets", "NitterScraper"]

__version__ = "0.5.2"
__version__ = "0.5.3"
27 changes: 22 additions & 5 deletions nitter_scraper/nitter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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().
Expand All @@ -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

"""

Expand All @@ -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):
Expand All @@ -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,
Expand Down
11 changes: 3 additions & 8 deletions nitter_scraper/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)
Expand All @@ -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.

Expand All @@ -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.
Expand All @@ -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)
Expand Down
29 changes: 27 additions & 2 deletions nitter_scraper/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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.

"""

Expand All @@ -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":
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand Down
86 changes: 52 additions & 34 deletions nitter_scraper/tweets.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -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):
Expand All @@ -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


Expand Down Expand Up @@ -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)
Expand All @@ -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

Expand All @@ -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:
Expand All @@ -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)
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -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 <dgnsrekt@pm.me>"]
readme = "README.md"
Expand Down