See social network users cryptocurrency addresses and balances. Powered by transparent blockchains.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

190 lines
6.5 KiB

from contextlib import contextmanager
from ipaddress import IPv4Address
from tempfile import _TemporaryFileWrapper as TemporaryFile
from tempfile import NamedTemporaryFile
import time
from typing import ClassVar, Optional
import docker
from docker.client import DockerClient
from docker.models.containers import Container
from jinja2 import Environment, FileSystemLoader
from loguru import logger
from pydantic import BaseModel as Base
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
class DockerBase(Base):
"""Provides helper methods for connecting to the docker client."""
client: ClassVar[DockerClient] = None
def _get_client(cls):
if cls.client is None:
cls.client = docker.from_env()"Docker connection successful.")
return cls.client
class Nitter(DockerBase):
"""Nitter Docker container object
host (IPv4Address): The host address the docker container will bind too.
port (int): The port the docker container will listen to.
container_name: the container name of the already running container.
tempfile (TemporaryFile): A TemporaryFile file generated from a template.
container (Container): Local representation of a container object.
Holds the started instance of a docker container.
address (str): The full address of the docker container.
ports (dict[int, int]): Binds the listening port to the nitter docker container's
internal port 8080.
config_filepath (str): Path name to the generated tempfile.
volumes (dict[str, dict[str, str]]): used to configure a bind volume.
host: IPv4Address
port: int
container_name: str
tempfile: TemporaryFile = None
container: Optional[Container]
class Config:
arbitrary_types_allowed = True
def address(self):
return f"http://{}:{self.port}"
def config_filepath(self):
if self.tempfile:
def ports(self):
return {8080: self.port}
def volumes(self):
volumes = {"bind": "/src/nitter.conf", "mode": "ro"}
return {self.config_filepath: volumes}
def _render_config(self):
env = Environment(loader=FileSystemLoader(TEMPLATES_DIRECTORY))
template = env.get_template("nitter.conf")
return template.render(self.dict())
def _create_configfile(self):
config = self._render_config()
self.tempfile = NamedTemporaryFile(dir=PROJECT_ROOT)
def get_profile(self, username: str, not_found_ok: bool = False):
"""Scrapes nitter for the target users profile information.
This is a modified version of nitter_scraper.profile.get_profile().
This version automatically uses the address of the docker container as the primary
address to scrape profile data.
username: The target profiles username.
not_found_ok: If not_found_ok is false (the default), a ValueError is raised if
the target profile doesn't exist. If not_found_ok is true, None will be returned
Profile object if successfully scraped, otherwise None.
ValueError: If the target profile does not exist and the not_found_ok argument is
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, with_replies: Optional[bool]=False):
"""Gets the target users tweets
This is a modified version of nitter_scraper.tweets.get_tweets().
This version automatically uses the address of the docker container as the primary
address to scrape profile data.
username: Targeted users username.
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 which should
be used as a fallback address.
Tweet Objects
return get_tweets(
def get(self):
client = self._get_client()
self.container = client.containers.get(
)"Got container {} {self.container.short_id}. Status {self.container.status} at {self.address}.")
def start(self):
"""Starts the docker the container"""
client = self._get_client()
self.container =
time.sleep(1)"Running container {} {self.container.short_id}.")
def stop(self):
"""Stops the docker the container""""Stopping container {} {self.container.short_id}.")
if self.container:
self.container.stop(timeout=5)"Container {} {self.container.short_id} Destroyed.")
def NitterScraper(host: str = "", port: int = 8008, container_name: str = "ff1fc8e9f922"):
"""The NitterScraper context manager.
Takes care of configuring, starting, and stopping a docker instance of nitter.
host: The host address the docker container will bind too.
port: The port the docker container will listen to.
Nitter: An object representing a started nitter docker container.
nitter = Nitter(host=host, port=port, container_name=container_name)
yield nitter