soluzka

index.py

Nov 8th, 2024
93
0
Never
2
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 13.99 KB | Software | 0 0
  1. from selenium import webdriver
  2. from selenium.webdriver.common.by import By
  3. from selenium.webdriver.common.keys import Keys
  4. import os
  5. import json
  6. import logging
  7. import requests
  8. import time
  9. from yt_dlp import YoutubeDL
  10. from bs4 import BeautifulSoup
  11. from typing import List, Dict, Optional
  12.  
  13. # Load API keys and tokens from environment variables
  14. API_TOKENS = {
  15.     'YOUTUBE_API_KEY': os.environ.get('YOUTUBE_API_KEY'),
  16.     'VIMEO_API_TOKEN': os.environ.get('VIMEO_API_TOKEN'),
  17.     'DAILYMOTION_API_TOKEN': os.environ.get('DAILYMOTION_API_TOKEN'),
  18.     'FACEBOOK_API_TOKEN': os.environ.get('FACEBOOK_API_TOKEN'),
  19.     'INSTAGRAM_API_TOKEN': os.environ.get('INSTAGRAM_API_TOKEN'),
  20.     'TWITTER_API_TOKEN': os.environ.get('TWITTER_API_TOKEN'),
  21.     'TWITCH_API_TOKEN': os.environ.get('TWITCH_API_TOKEN'),
  22.     'TIKTOK_API_TOKEN': os.environ.get('TIKTOK_API_TOKEN'),
  23. }
  24.  
  25. # Function to prompt for missing API tokens
  26. def prompt_missing_tokens():
  27.     print("Please enter your tokens for the following services (or 'none' to skip all):")
  28.     for token_name in API_TOKENS.keys():
  29.         if API_TOKENS[token_name] is None:
  30.             while True:
  31.                 try:
  32.                     token = input(f"Please enter your {token_name} token (or 'n' to skip this token, 'none' to skip all): ").strip()
  33.                     if token.lower() == 'none':
  34.                         print("All tokens will be skipped. Setting all to None.")
  35.                         for key in API_TOKENS.keys():
  36.                             API_TOKENS[key] = None
  37.                         return
  38.                     elif token.lower() == 'n':
  39.                         print(f"Skipping {token_name}. It will remain None.")
  40.                         break
  41.                     else:
  42.                         os.environ[token_name] = token
  43.                         API_TOKENS[token_name] = token
  44.                         break
  45.                 except EOFError:
  46.                     print("Input stream closed. Please try again.")
  47.                     break
  48.                 except Exception as e:
  49.                     print(f"An error occurred: {e}. Please enter the token again.")
  50.  
  51. # Function to save scraped data to a text file
  52. def save_data(url: str, title: str, description: str, keywords: str):
  53.     with open('scraped_data.txt', 'a', encoding='utf-8') as f:
  54.         f.write(f"URL: {url}\nTitle: {title}\nDescription: {description}\nKeywords: {keywords}\n{'-' * 40}\n")
  55.  
  56. # Function to extract text from a BeautifulSoup tag
  57. def get_meta_content(soup, name: str) -> Optional[str]:
  58.     tag = soup.find('meta', attrs={'name': name})
  59.     return tag['content'] if tag and 'content' in tag.attrs else None
  60.  
  61. # Function to make API requests
  62. def make_api_requests(query: str) -> Dict[str, Optional[List[Dict]]]:
  63.     requests_info = {
  64.         'youtube': {
  65.             'url': 'https://www.googleapis.com/youtube/v3/search',
  66.             'params': {'part': 'snippet', 'q': query, 'key': API_TOKENS['YOUTUBE_API_KEY'], 'maxResults': 50},
  67.         },
  68.         'vimeo': {
  69.             'url': 'https://api.vimeo.com/videos',
  70.             'params': {'query': query, 'access_token': API_TOKENS['VIMEO_API_TOKEN']},
  71.         },
  72.         'dailymotion': {
  73.             'url': 'https://api.dailymotion.com/videos',
  74.             'params': {'search': query, 'api_key': API_TOKENS['DAILYMOTION_API_TOKEN']},
  75.         },
  76.         'facebook': {
  77.             'url': 'https://graph.facebook.com/v13.0/videos',
  78.             'params': {'q': query, 'access_token': API_TOKENS['FACEBOOK_API_TOKEN']},
  79.         },
  80.         'instagram': {
  81.             'url': 'https://graph.instagram.com/v13.0/media',
  82.             'params': {'q': query, 'access_token': API_TOKENS['INSTAGRAM_API_TOKEN']},
  83.         },
  84.         'twitter': {
  85.             'url': 'https://api.twitter.com/2/tweets/search/recent',
  86.             'params': {'query': query, 'access_token': API_TOKENS['TWITTER_API_TOKEN']},
  87.         },
  88.         'twitch': {
  89.             'url': 'https://api.twitch.tv/helix/search/channels',
  90.             'params': {'query': query, 'access_token': API_TOKENS['TWITCH_API_TOKEN']},
  91.         },
  92.         'tiktok': {
  93.             'url': 'https://api.tiktok.com/v1/discover/search',
  94.             'params': {'query': query, 'access_token': API_TOKENS['TIKTOK_API_TOKEN']},
  95.         }
  96.     }
  97.  
  98.     results = {}
  99.     for platform, info in requests_info.items():
  100.         access_token = info['params'].get('access_token')
  101.         if access_token is None and platform not in ['youtube', 'dailymotion']:
  102.             logging.warning(f"No token provided for {platform}. Functionality may be limited.")
  103.             continue  # Skip requests that do not have an access token
  104.  
  105.         try:
  106.             response = requests.get(info['url'], params=info['params'])
  107.             if response.status_code == 200:
  108.                 json_response = response.json()
  109.                 # For each platform, populate links according to the received data
  110.                 if platform == 'youtube':
  111.                     links = [f"https://www.youtube.com/watch?v={item['id']['videoId']}" for item in json_response.get('items', [])]
  112.                 elif platform == 'vimeo':
  113.                     links = [video['link'] for video in json_response.get('data', [])]
  114.                 elif platform == 'dailymotion':
  115.                     links = [f"https://www.dailymotion.com/video/{video['id']}" for video in json_response.get('list', [])]
  116.                 elif platform == 'facebook':
  117.                     links = [post['permalink_url'] for post in json_response.get('data', [])]
  118.                 elif platform == 'instagram':
  119.                     links = [f"https://www.instagram.com/p/{media['id']}/" for media in json_response.get('data', [])]
  120.                 elif platform == 'twitter':
  121.                     links = [f"https://twitter.com/i/web/status/{tweet['id']}" for tweet in json_response.get('data', [])]
  122.                 elif platform == 'twitch':
  123.                     links = [channel['url'] for channel in json_response.get('data', [])]
  124.                 elif platform == 'tiktok':
  125.                     links = [media['url'] for media in json_response.get('data', [])]
  126.  
  127.                 results[platform] = links
  128.             else:
  129.                 logging.error(f"Failed to get results from {platform}: {response.status_code}")
  130.                 results[platform] = None  # Handle errors
  131.         except Exception as e:
  132.             logging.error(f"Error during request to {platform}: {e}")
  133.  
  134.     return results
  135.  
  136. # Function to scrape data from a given URL
  137. def scrape_url(url: str):
  138.     print(f'Processing URL: {url}')
  139.     try:
  140.         response = requests.get(url)
  141.         if response.status_code == 200:
  142.             html = response.text
  143.             soup = BeautifulSoup(html, 'lxml')
  144.  
  145.             title: str = soup.title.string if soup.title and soup.title.string else 'No title found'
  146.             description: Optional[str] = get_meta_content(soup, 'description') or 'No description found'
  147.             keywords: Optional[str] = get_meta_content(soup, 'keywords') or 'No keywords found'
  148.  
  149.             print(f'Title: {title}\nDescription: {description}\nKeywords: {keywords}')
  150.             save_data(url, title, description, keywords)
  151.         else:
  152.             logging.error(f"Failed to retrieve URL: {url} with status code: {response.status_code}")
  153.     except Exception as e:
  154.         logging.error(f"Error occurred while scraping the URL: {e}")
  155.  
  156. # Function to download a file
  157. def download_file(link: str, output_path: str):
  158.     try:
  159.         response = requests.get(link, stream=True)
  160.         if response.status_code == 200:
  161.             with open(output_path, 'wb') as f:
  162.                 for chunk in response.iter_content(chunk_size=8192):
  163.                     f.write(chunk)
  164.             print(f"Downloaded: {link} to {output_path}")
  165.         else:
  166.             logging.error(f"Failed to download {link} with status code: {response.status_code}")
  167.     except Exception as e:
  168.         logging.error(f"Error downloading {link}: {e}")
  169.  
  170. # Function to download multiple links
  171. def download_multiple_links(links: List[str], output_path: str, num_links: int):
  172.     for i, link in enumerate(links[:num_links]):
  173.         filename = f"downloaded_file_{i + 1}.bin"
  174.         output_file_path = os.path.join(output_path, filename)
  175.         download_file(link.strip(), output_file_path)
  176.  
  177. # Function to download multiple videos using yt-dlp
  178. def download_multiple_videos_with_yt_dlp(links: List[str], output_path: str, num_links: int):
  179.     for link in links[:num_links]:
  180.         download_video_with_yt_dlp(link.strip(), output_path)
  181.  
  182. # Function to handle downloading videos using yt-dlp
  183. def download_video_with_yt_dlp(video_url: str, output_path: str):
  184.     try:
  185.         ydl_opts = {
  186.             'outtmpl': os.path.join(output_path, '%(title)s.%(ext)s'),
  187.             'format': 'best',
  188.         }
  189.         with YoutubeDL(ydl_opts) as ydl:
  190.             ydl.download([video_url])
  191.     except Exception as e:
  192.         logging.error(f"Error downloading video from {video_url}: {e}")
  193.  
  194. def search_google_videos(query: str, browser: str):
  195.     driver = None
  196.  
  197.     # Initialize the appropriate WebDriver based on the browser argument
  198.     if browser.lower() == 'chrome':
  199.         driver = webdriver.Chrome()  # Ensure chromedriver is installed
  200.     elif browser.lower() == 'firefox':
  201.         driver = webdriver.Firefox()  # Ensure geckodriver is installed
  202.     elif browser.lower() == 'edge':
  203.         driver = webdriver.Edge()  # Ensure edgedriver is installed
  204.     else:
  205.         print("Unsupported browser. Please use Chrome, Firefox, or Edge.")
  206.         return
  207.  
  208.     try:
  209.         # Open Google Video search
  210.         driver.get("https://www.google.com/")  # Start with Google
  211.         time.sleep(2)  # Allow for page to load
  212.  
  213.         # Find the search box and submit the video search
  214.         search_box = driver.find_element(By.NAME, "q")
  215.         search_box.send_keys(f"videos {query}")  # Add 'videos' to the query for better results
  216.         search_box.send_keys(Keys.RETURN)
  217.  
  218.         time.sleep(2)  # Wait for results to load
  219.  
  220.         # Select 'Videos' from the Google results
  221.         try:
  222.             videos_tab = driver.find_element(By.LINK_TEXT, "Videos")  # Click on the videos link
  223.             videos_tab.click()
  224.             time.sleep(2)  # Wait for video results to load
  225.         except Exception as e:
  226.             logging.error("Could not find the 'Videos' tab. Proceeding with regular search results.")
  227.  
  228.         # Initialize a list to store all video URLs
  229.         video_urls = []
  230.  
  231.         # Find video results
  232.         results = driver.find_elements(By.XPATH, "//a[contains(@href, 'watch')]")  # YouTube video links
  233.         for result in results:
  234.             url = result.get_attribute("href")
  235.             if url and ("youtube.com/watch?v=" in url or "vimeo.com/" in url or "dailymotion.com/video/" in url):
  236.                 video_urls.append(url)
  237.  
  238.         # Print all collected video URLs
  239.         print("Collected Video URLs:")
  240.         for url in video_urls:
  241.             print(url)
  242.  
  243.     except Exception as e:
  244.         logging.error(f"Error occurred during Selenium video search: {e}")
  245.     finally:
  246.         # Ensure the driver quits even in case of an error
  247.         if driver:
  248.             driver.quit()
  249.  
  250. def main():
  251.     logging.basicConfig(level=logging.INFO)
  252.  
  253.     # Ensure output path for downloads exists
  254.     output_path = "downloads"
  255.     os.makedirs(output_path, exist_ok=True)
  256.  
  257.     while True:
  258.         try:
  259.             command = input("Enter a command (search, scrape, download_file, download_links, download_videos, exit): ").strip().lower()
  260.             if command == 'exit':
  261.                 print("Exiting...")
  262.                 break
  263.             elif command == 'search':
  264.                 search_query = input("Enter the search query: ").strip()
  265.                 browser = input("Enter the browser (chrome/firefox/edge) for web search or 'api' for API search: ").strip()
  266.                
  267.                 # Check if using an API or web browser
  268.                 if browser.lower() == 'api':
  269.                     api_results = make_api_requests(search_query)
  270.                     print("API Search Results:")
  271.                     for platform, links in api_results.items():
  272.                         if links:
  273.                             print(f"{platform.capitalize()} Links:")
  274.                             for link in links:
  275.                                 print(link)
  276.                         else:
  277.                             print(f"No results found for {platform}.")
  278.                 else:
  279.                     search_google_videos(search_query, browser)
  280.             elif command == 'scrape':
  281.                 url = input("Enter a URL to scrape: ").strip()
  282.                 scrape_url(url)
  283.             elif command == 'download_file':
  284.                 file_url = input("Enter the file link to download: ").strip()
  285.                 filename = input("Enter the filename to save as (including extension): ").strip()
  286.                 output_file_path = os.path.join(output_path, filename)
  287.                 download_file(file_url, output_file_path)
  288.             elif command == 'download_links':
  289.                 links = input("Enter the links separated by commas: ").strip().split(',')
  290.                 num_links = int(input("Enter the number of links to download: "))
  291.                 download_multiple_links(links, output_path, num_links)
  292.             elif command == 'download_videos':
  293.                 links = input("Enter the video links separated by commas: ").strip().split(',')
  294.                 num_links = int(input("Enter the number of videos to download: "))
  295.                 download_multiple_videos_with_yt_dlp(links, output_path, num_links)
  296.             else:
  297.                 print("Invalid command. Please try again.")
  298.         except EOFError:
  299.             print("Input stream closed unexpectedly. Exiting...")
  300.             break
  301.         except Exception as e:
  302.             logging.error(f"An unexpected error occurred: {e}")
  303.             prompt_missing_tokens()  # Optional prompt to input tokens
  304.  
  305. if __name__ == "__main__":
  306.     main()  # Run the main function
Tags: download app
Advertisement
Comments
Add Comment
Please, Sign In to add comment