Merge pull request #47 from code-yeongyu/feature/twitter-video-downloader

Add cli interface with class `TwitterVideoTools`

Author: code-yeongyu

README.md

@@ -15,12 +15,36 @@
 pip install twitter-video-tools
 ```
 
-### with Poetry (Recommended)
+### with Poetry
 
 ```sh
 poetry add twitter-video-tools
 ```
 
+## Usage
+
+### Command line
+
+```sh
+python3 -m twitter_video_tools [link]
+```
+
+Supported link types:
+
+- Video tweet: <https://twitter.com/twtvtOfficial/status/1599748329927499777>
+- Video from monsnode: <https://monsnode.com/v1506575871309589251>
+- Specific user's uploaded videos: <https://twitter.com/twtvtOfficial/media>
+- Specific user's liked videos: <https://twitter.com/twtvtOfficial/likes>
+
+### Python Embedding
+
+```python
+from twitter_video_tools import TwitterVideoTools
+
+with TwitterVideoTools() as twitter_video_tools:
+    twitter_video_tools.download_from_user('twtvtOfficial')
+```
+
 ## Contribution
 
 ### Prerequisites

poetry.lock

@@ -60,6 +60,7 @@ playwright = "^1.27.1"
 bs4 = "^0.0.1"
 requests = "^2.28.1"
 yt-dlp = "^2022.11.11"
+typer = {extras = ["all"], version = "^0.7.0"}
 
 [tool.poetry.group.dev.dependencies]
 monkeytype = "^22.2.0"

pyproject.toml

@@ -1,7 +1,45 @@
 # pylint: disable=useless-import-alias
 
-from .monsnode_parser import MonsnodeParser as MonsnodeParser
-from .platform_video_downloader import \
-    PlatformVideoDownloader as PlatformVideoDownloader
-from .twitter_crawler import TwitterCrawler as TwitterCrawler
-from .video_downloader import VideoDownloader as VideoDownloader
+from typing import Optional
+
+import typer
+from rich import print as rich_print
+
+from .twitter_video_tools import TwitterVideoTools as TwitterVideoTools
+from .utils import URLValidator
+
+
+def _cli_main(
+    link: str = typer.Argument(..., help='Video tweet link or target user\'s likes or media.'),
+    username: Optional[str] = typer.Option(None, help='Your twitter credentials username.'),
+    password: Optional[str] = typer.Option(None, help='Your twitter credentials password.'),
+    until_link: Optional[str] = typer.Option(
+        None,
+        help='Keeps finding videos until this link is found. None for no limit. Only for user\'s likes or media.',
+    ),
+) -> None:
+    url_validator = URLValidator(link)
+
+    if not url_validator.is_valid_link():
+        rich_print(f'\'[underline]{link}[/underline]\' [bold red]is an invalid link.[/bold red]')
+        return
+    twitter_video_tools = TwitterVideoTools(username, password)
+
+    if url_validator.is_valid_twitter_media_link():
+        target_username = link.split('/')[3]
+        twitter_video_tools.download_from_user(target_username, until_link)
+    elif url_validator.is_valid_twitter_liked_link():
+        target_username = link.split('/')[3]
+        twitter_video_tools.download_from_liked_tweets(target_username, until_link)
+    else:
+        twitter_video_tools.download([link])
+
+
+def main() -> None:
+    typer.run(_cli_main)
+
+
+__all__ = [
+    'main',
+    'TwitterVideoTools',
+]

twitter_video_tools/__init__.py

@@ -0,0 +1,4 @@
+import twitter_video_tools
+
+if __name__ == '__main__':
+    twitter_video_tools.main()

twitter_video_tools/__main__.py

@@ -4,8 +4,6 @@
 from playwright.sync_api import sync_playwright
 from yt_dlp.utils import YoutubeDLError
 
-from twitter_video_tools.monsnode_parser import MonsnodeParser
-
 from .monsnode_parser import MonsnodeParser
 from .twitter_crawler import TwitterCrawler
 from .utils.youtube_dl_wrapper import youtube_dl_wrapper

twitter_video_tools/platform_video_downloader.py

@@ -72,6 +72,25 @@ def get_recent_liked_tweet(self, username: str) -> str:
         self._goto_liked_tweets(username)
         return self._get_tweets_in_current_screen()[0]
 
+    def _get_tweets_in_current_screen(self) -> list[str]:
+        links: list[str] = []
+
+        while True:
+            articles = self.page.locator('article')
+            article_length = articles.count()
+            try:
+                links = [
+                    'https://twitter.com' +
+                    (articles.nth(i).locator('div').locator('a').nth(3).get_attribute('href', timeout=500) or '')
+                    for i in range(article_length)
+                ]
+                break
+            except Error:    # if articles in the page are not reachable
+                self.page.mouse.wheel(0, 500)    #  scrolling down to refresh the articles
+                self.page.mouse.wheel(0, -500)    #  scrolling down to refresh the articles
+
+        return links
+
     def get_video_of_tweet(self, link: str, timeout: Optional[float] = 10000) -> list[tuple[str, str]]:
         video_links: list[str] = []
 
@@ -97,30 +116,54 @@ def _goto_liked_tweets(self, username: str) -> None:
         self.page.goto(f'https://twitter.com/{username}/likes')
         self.page.wait_for_selector('article')
 
-    def _get_video_tweets_in_current_screen(self) -> list[str]:
+    def get_all_media_tweets(self, username: str, scroll_timeout: float = 0.8) -> list[str]:
+        """Get the username's all liked tweets
+        Returns the list of links of liked tweets
+        """
+        return self.get_media_tweets_until(
+            username, 'nothing', scroll_timeout
+        )    # 'nothing' was intended because the given `until_link` would be never found on the links list
+
+    def get_media_tweets_until(self, username: str, until_link: str, scroll_timeout: float = 0.8) -> list[str]:
+        """Scrolling down the list of media tweets until the given `until_link` found
+        Returns the list of links of media tweets
+        """
+        self._goto_media_tweets(username)
         links: list[str] = []
 
+        previous_height = self.page_current_height
         while True:
-            articles = self.page.locator('article:has(video)')
-            article_length = articles.count()
-            try:
-                links = [
-                    'https://twitter.com' +
-                    (articles.nth(i).locator('div').locator('a').nth(3).get_attribute('href', timeout=500) or '')
-                    for i in range(article_length)
-                ]
+            # 1. scroll down
+            # 2. get the link of tweets in the current screen(tweets are not reachable if it's not rendering)
+            # 3. break if page reaches to the bottom or the given `until_link` found
+
+            self.page.mouse.wheel(0, 1500)
+            time.sleep(scroll_timeout)    # wait for mouse wheel to scroll down
+            is_page_bottom = self.page_current_height == previous_height
+            if is_page_bottom:
+                break
+            previous_height = self.page_current_height
+
+            new_links = self._get_video_tweets_in_current_screen()
+            links.extend(new_links)
+            links = list(set(links))
+
+            print(f'Found {len(links)} media tweets.')
+
+            if until_link in links:
                 break
-            except Error:    # if articles in the page are not reachable
-                self.page.mouse.wheel(0, 500)    #  scrolling down to refresh the articles
-                self.page.mouse.wheel(0, -500)    #  scrolling down to refresh the articles
 
         return links
 
-    def _get_tweets_in_current_screen(self) -> list[str]:
+    def _goto_media_tweets(self, username: str) -> None:
+        self.page.goto(f'https://twitter.com/{username}/media')
+        self.page.wait_for_selector('article')
+
+    def _get_video_tweets_in_current_screen(self) -> list[str]:
         links: list[str] = []
 
         while True:
-            articles = self.page.locator('article')
+            articles = self.page.locator('article:has(video)')
             article_length = articles.count()
             try:
                 links = [