Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from selenium import webdriver
- from selenium.webdriver.common.by import By
- from selenium.webdriver.common.keys import Keys
- import os
- import json
- import logging
- import requests
- import time
- from yt_dlp import YoutubeDL
- from bs4 import BeautifulSoup
- from typing import List, Dict, Optional
- # Load API keys and tokens from environment variables
- API_TOKENS = {
- 'YOUTUBE_API_KEY': os.environ.get('YOUTUBE_API_KEY'),
- 'VIMEO_API_TOKEN': os.environ.get('VIMEO_API_TOKEN'),
- 'DAILYMOTION_API_TOKEN': os.environ.get('DAILYMOTION_API_TOKEN'),
- 'FACEBOOK_API_TOKEN': os.environ.get('FACEBOOK_API_TOKEN'),
- 'INSTAGRAM_API_TOKEN': os.environ.get('INSTAGRAM_API_TOKEN'),
- 'TWITTER_API_TOKEN': os.environ.get('TWITTER_API_TOKEN'),
- 'TWITCH_API_TOKEN': os.environ.get('TWITCH_API_TOKEN'),
- 'TIKTOK_API_TOKEN': os.environ.get('TIKTOK_API_TOKEN'),
- }
- # Function to prompt for missing API tokens
- def prompt_missing_tokens():
- print("Please enter your tokens for the following services (or 'none' to skip all):")
- for token_name in API_TOKENS.keys():
- if API_TOKENS[token_name] is None:
- while True:
- try:
- token = input(f"Please enter your {token_name} token (or 'n' to skip this token, 'none' to skip all): ").strip()
- if token.lower() == 'none':
- print("All tokens will be skipped. Setting all to None.")
- for key in API_TOKENS.keys():
- API_TOKENS[key] = None
- return
- elif token.lower() == 'n':
- print(f"Skipping {token_name}. It will remain None.")
- break
- else:
- os.environ[token_name] = token
- API_TOKENS[token_name] = token
- break
- except EOFError:
- print("Input stream closed. Please try again.")
- break
- except Exception as e:
- print(f"An error occurred: {e}. Please enter the token again.")
- # Function to save scraped data to a text file
- def save_data(url: str, title: str, description: str, keywords: str):
- with open('scraped_data.txt', 'a', encoding='utf-8') as f:
- f.write(f"URL: {url}\nTitle: {title}\nDescription: {description}\nKeywords: {keywords}\n{'-' * 40}\n")
- # Function to extract text from a BeautifulSoup tag
- def get_meta_content(soup, name: str) -> Optional[str]:
- tag = soup.find('meta', attrs={'name': name})
- return tag['content'] if tag and 'content' in tag.attrs else None
- # Function to make API requests
- def make_api_requests(query: str) -> Dict[str, Optional[List[Dict]]]:
- requests_info = {
- 'youtube': {
- 'url': 'https://www.googleapis.com/youtube/v3/search',
- 'params': {'part': 'snippet', 'q': query, 'key': API_TOKENS['YOUTUBE_API_KEY'], 'maxResults': 50},
- },
- 'vimeo': {
- 'url': 'https://api.vimeo.com/videos',
- 'params': {'query': query, 'access_token': API_TOKENS['VIMEO_API_TOKEN']},
- },
- 'dailymotion': {
- 'url': 'https://api.dailymotion.com/videos',
- 'params': {'search': query, 'api_key': API_TOKENS['DAILYMOTION_API_TOKEN']},
- },
- 'facebook': {
- 'url': 'https://graph.facebook.com/v13.0/videos',
- 'params': {'q': query, 'access_token': API_TOKENS['FACEBOOK_API_TOKEN']},
- },
- 'instagram': {
- 'url': 'https://graph.instagram.com/v13.0/media',
- 'params': {'q': query, 'access_token': API_TOKENS['INSTAGRAM_API_TOKEN']},
- },
- 'twitter': {
- 'url': 'https://api.twitter.com/2/tweets/search/recent',
- 'params': {'query': query, 'access_token': API_TOKENS['TWITTER_API_TOKEN']},
- },
- 'twitch': {
- 'url': 'https://api.twitch.tv/helix/search/channels',
- 'params': {'query': query, 'access_token': API_TOKENS['TWITCH_API_TOKEN']},
- },
- 'tiktok': {
- 'url': 'https://api.tiktok.com/v1/discover/search',
- 'params': {'query': query, 'access_token': API_TOKENS['TIKTOK_API_TOKEN']},
- }
- }
- results = {}
- for platform, info in requests_info.items():
- access_token = info['params'].get('access_token')
- if access_token is None and platform not in ['youtube', 'dailymotion']:
- logging.warning(f"No token provided for {platform}. Functionality may be limited.")
- continue # Skip requests that do not have an access token
- try:
- response = requests.get(info['url'], params=info['params'])
- if response.status_code == 200:
- json_response = response.json()
- # For each platform, populate links according to the received data
- if platform == 'youtube':
- links = [f"https://www.youtube.com/watch?v={item['id']['videoId']}" for item in json_response.get('items', [])]
- elif platform == 'vimeo':
- links = [video['link'] for video in json_response.get('data', [])]
- elif platform == 'dailymotion':
- links = [f"https://www.dailymotion.com/video/{video['id']}" for video in json_response.get('list', [])]
- elif platform == 'facebook':
- links = [post['permalink_url'] for post in json_response.get('data', [])]
- elif platform == 'instagram':
- links = [f"https://www.instagram.com/p/{media['id']}/" for media in json_response.get('data', [])]
- elif platform == 'twitter':
- links = [f"https://twitter.com/i/web/status/{tweet['id']}" for tweet in json_response.get('data', [])]
- elif platform == 'twitch':
- links = [channel['url'] for channel in json_response.get('data', [])]
- elif platform == 'tiktok':
- links = [media['url'] for media in json_response.get('data', [])]
- results[platform] = links
- else:
- logging.error(f"Failed to get results from {platform}: {response.status_code}")
- results[platform] = None # Handle errors
- except Exception as e:
- logging.error(f"Error during request to {platform}: {e}")
- return results
- # Function to scrape data from a given URL
- def scrape_url(url: str):
- print(f'Processing URL: {url}')
- try:
- response = requests.get(url)
- if response.status_code == 200:
- html = response.text
- soup = BeautifulSoup(html, 'lxml')
- title: str = soup.title.string if soup.title and soup.title.string else 'No title found'
- description: Optional[str] = get_meta_content(soup, 'description') or 'No description found'
- keywords: Optional[str] = get_meta_content(soup, 'keywords') or 'No keywords found'
- print(f'Title: {title}\nDescription: {description}\nKeywords: {keywords}')
- save_data(url, title, description, keywords)
- else:
- logging.error(f"Failed to retrieve URL: {url} with status code: {response.status_code}")
- except Exception as e:
- logging.error(f"Error occurred while scraping the URL: {e}")
- # Function to download a file
- def download_file(link: str, output_path: str):
- try:
- response = requests.get(link, stream=True)
- if response.status_code == 200:
- with open(output_path, 'wb') as f:
- for chunk in response.iter_content(chunk_size=8192):
- f.write(chunk)
- print(f"Downloaded: {link} to {output_path}")
- else:
- logging.error(f"Failed to download {link} with status code: {response.status_code}")
- except Exception as e:
- logging.error(f"Error downloading {link}: {e}")
- # Function to download multiple links
- def download_multiple_links(links: List[str], output_path: str, num_links: int):
- for i, link in enumerate(links[:num_links]):
- filename = f"downloaded_file_{i + 1}.bin"
- output_file_path = os.path.join(output_path, filename)
- download_file(link.strip(), output_file_path)
- # Function to download multiple videos using yt-dlp
- def download_multiple_videos_with_yt_dlp(links: List[str], output_path: str, num_links: int):
- for link in links[:num_links]:
- download_video_with_yt_dlp(link.strip(), output_path)
- # Function to handle downloading videos using yt-dlp
- def download_video_with_yt_dlp(video_url: str, output_path: str):
- try:
- ydl_opts = {
- 'outtmpl': os.path.join(output_path, '%(title)s.%(ext)s'),
- 'format': 'best',
- }
- with YoutubeDL(ydl_opts) as ydl:
- ydl.download([video_url])
- except Exception as e:
- logging.error(f"Error downloading video from {video_url}: {e}")
- def search_google_videos(query: str, browser: str):
- driver = None
- # Initialize the appropriate WebDriver based on the browser argument
- if browser.lower() == 'chrome':
- driver = webdriver.Chrome() # Ensure chromedriver is installed
- elif browser.lower() == 'firefox':
- driver = webdriver.Firefox() # Ensure geckodriver is installed
- elif browser.lower() == 'edge':
- driver = webdriver.Edge() # Ensure edgedriver is installed
- else:
- print("Unsupported browser. Please use Chrome, Firefox, or Edge.")
- return
- try:
- # Open Google Video search
- driver.get("https://www.google.com/") # Start with Google
- time.sleep(2) # Allow for page to load
- # Find the search box and submit the video search
- search_box = driver.find_element(By.NAME, "q")
- search_box.send_keys(f"videos {query}") # Add 'videos' to the query for better results
- search_box.send_keys(Keys.RETURN)
- time.sleep(2) # Wait for results to load
- # Select 'Videos' from the Google results
- try:
- videos_tab = driver.find_element(By.LINK_TEXT, "Videos") # Click on the videos link
- videos_tab.click()
- time.sleep(2) # Wait for video results to load
- except Exception as e:
- logging.error("Could not find the 'Videos' tab. Proceeding with regular search results.")
- # Initialize a list to store all video URLs
- video_urls = []
- # Find video results
- results = driver.find_elements(By.XPATH, "//a[contains(@href, 'watch')]") # YouTube video links
- for result in results:
- url = result.get_attribute("href")
- if url and ("youtube.com/watch?v=" in url or "vimeo.com/" in url or "dailymotion.com/video/" in url):
- video_urls.append(url)
- # Print all collected video URLs
- print("Collected Video URLs:")
- for url in video_urls:
- print(url)
- except Exception as e:
- logging.error(f"Error occurred during Selenium video search: {e}")
- finally:
- # Ensure the driver quits even in case of an error
- if driver:
- driver.quit()
- def main():
- logging.basicConfig(level=logging.INFO)
- # Ensure output path for downloads exists
- output_path = "downloads"
- os.makedirs(output_path, exist_ok=True)
- while True:
- try:
- command = input("Enter a command (search, scrape, download_file, download_links, download_videos, exit): ").strip().lower()
- if command == 'exit':
- print("Exiting...")
- break
- elif command == 'search':
- search_query = input("Enter the search query: ").strip()
- browser = input("Enter the browser (chrome/firefox/edge) for web search or 'api' for API search: ").strip()
- # Check if using an API or web browser
- if browser.lower() == 'api':
- api_results = make_api_requests(search_query)
- print("API Search Results:")
- for platform, links in api_results.items():
- if links:
- print(f"{platform.capitalize()} Links:")
- for link in links:
- print(link)
- else:
- print(f"No results found for {platform}.")
- else:
- search_google_videos(search_query, browser)
- elif command == 'scrape':
- url = input("Enter a URL to scrape: ").strip()
- scrape_url(url)
- elif command == 'download_file':
- file_url = input("Enter the file link to download: ").strip()
- filename = input("Enter the filename to save as (including extension): ").strip()
- output_file_path = os.path.join(output_path, filename)
- download_file(file_url, output_file_path)
- elif command == 'download_links':
- links = input("Enter the links separated by commas: ").strip().split(',')
- num_links = int(input("Enter the number of links to download: "))
- download_multiple_links(links, output_path, num_links)
- elif command == 'download_videos':
- links = input("Enter the video links separated by commas: ").strip().split(',')
- num_links = int(input("Enter the number of videos to download: "))
- download_multiple_videos_with_yt_dlp(links, output_path, num_links)
- else:
- print("Invalid command. Please try again.")
- except EOFError:
- print("Input stream closed unexpectedly. Exiting...")
- break
- except Exception as e:
- logging.error(f"An unexpected error occurred: {e}")
- prompt_missing_tokens() # Optional prompt to input tokens
- if __name__ == "__main__":
- main() # Run the main function
Advertisement