From f57f499ad75f54841146aeeed2a1948f5f23c717 Mon Sep 17 00:00:00 2001 From: FujiwaraChoki <78088687+FujiwaraChoki@users.noreply.github.com> Date: Sun, 1 Mar 2026 20:47:59 +0100 Subject: [PATCH 1/4] Fix outreach scraper execution and CSV handling (#129) --- src/classes/Outreach.py | 122 ++++++++++++++++++++++++++++------------ 1 file changed, 86 insertions(+), 36 deletions(-) diff --git a/src/classes/Outreach.py b/src/classes/Outreach.py index a518f20..96fdae7 100644 --- a/src/classes/Outreach.py +++ b/src/classes/Outreach.py @@ -3,19 +3,24 @@ import io import re import csv import time +import glob +import shlex import zipfile import yagmail import requests import subprocess +import platform from cache import * from status import * from config import * + class Outreach: """ Class that houses the methods to reach out to businesses. """ + def __init__(self) -> None: """ Constructor for the Outreach class. @@ -32,6 +37,15 @@ class Outreach: # Set email credentials self.email_creds = get_email_credentials() + def _find_scraper_dir(self) -> str: + candidates = sorted(glob.glob("google-maps-scraper-*")) + for candidate in candidates: + if os.path.isdir(candidate) and os.path.exists( + os.path.join(candidate, "go.mod") + ): + return candidate + return "" + def is_go_installed(self) -> bool: """ Check if go is installed. @@ -56,8 +70,7 @@ class Outreach: Returns: None """ - # Check if the scraper is already unzipped, if not, unzip it - if os.path.exists("google-maps-scraper-0.9.7"): + if self._find_scraper_dir(): info("=> Scraper already unzipped. Skipping unzip.") return @@ -72,18 +85,31 @@ class Outreach: Returns: None """ - # Check if the scraper is already built, if not, build it - if os.path.exists("google-maps-scraper.exe"): + binary_name = ( + "google-maps-scraper.exe" + if platform.system() == "Windows" + else "google-maps-scraper" + ) + if os.path.exists(binary_name): print(colored("=> Scraper already built. Skipping build.", "blue")) return - os.chdir("google-maps-scraper-0.9.7") - os.system("go mod download") - os.system("go build") - os.system("mv google-maps-scraper ../google-maps-scraper") - os.chdir("..") + scraper_dir = self._find_scraper_dir() + if not scraper_dir: + raise FileNotFoundError( + "Could not locate extracted google-maps-scraper directory." + ) - def run_scraper_with_args_for_30_seconds(self, args: str, timeout = 300) -> None: + subprocess.run(["go", "mod", "download"], cwd=scraper_dir, check=True) + subprocess.run(["go", "build"], cwd=scraper_dir, check=True) + + built_binary = os.path.join(scraper_dir, binary_name) + if not os.path.exists(built_binary): + raise FileNotFoundError(f"Expected built scraper binary at: {built_binary}") + + os.replace(built_binary, binary_name) + + def run_scraper_with_args_for_30_seconds(self, args: str, timeout=300) -> None: """ Run the scraper with the specified arguments for 30 seconds. @@ -96,19 +122,22 @@ class Outreach: """ # Run the scraper with the specified arguments info(" => Running scraper...") - command = "google-maps-scraper " + args + binary_name = ( + "google-maps-scraper.exe" + if platform.system() == "Windows" + else "google-maps-scraper" + ) + command = [os.path.join(os.getcwd(), binary_name)] + shlex.split(args) try: - scraper_process = subprocess.call(command.split(" "), shell=True, timeout=float(timeout)) + scraper_process = subprocess.run(command, timeout=float(timeout)) - if scraper_process == 0: - subprocess.call("taskkill /f /im google-maps-scraper.exe", shell=True) + if scraper_process.returncode == 0: print(colored("=> Scraper finished successfully.", "green")) else: - subprocess.call("taskkill /f /im google-maps-scraper.exe", shell=True) print(colored("=> Scraper finished with an error.", "red")) - + except subprocess.TimeoutExpired: + print(colored("=> Scraper timed out.", "red")) except Exception as e: - subprocess.call("taskkill /f /im google-maps-scraper.exe", shell=True) print(colored("An error occurred while running the scraper:", "red")) print(str(e)) @@ -127,19 +156,19 @@ class Outreach: items = f.readlines() items = [item.strip() for item in items[1:]] return items - + def set_email_for_website(self, index: int, website: str, output_file: str): """Extracts an email address from a website and updates a CSV file with it. - This method sends a GET request to the specified website, searches for the - first email address in the HTML content, and appends it to the specified - row in a CSV file. If no email address is found, no changes are made to - the CSV file. + This method sends a GET request to the specified website, searches for the + first email address in the HTML content, and appends it to the specified + row in a CSV file. If no email address is found, no changes are made to + the CSV file. - Args: - index (int): The row index in the CSV file where the email should be appended. - website (str): The URL of the website to extract the email address from. - output_file (str): The path to the CSV file to update with the extracted email.""" + Args: + index (int): The row index in the CSV file where the email should be appended. + website (str): The URL of the website to extract the email address from. + output_file (str): The path to the CSV file to update with the extracted email.""" # Extract and set an email for a website email = "" @@ -163,7 +192,7 @@ class Outreach: with open(output_file, "w", newline="", errors="ignore") as csvfile: csvwriter = csv.writer(csvfile) csvwriter.writerows(items) - + def start(self) -> None: """ Start the outreach process. @@ -191,7 +220,16 @@ class Outreach: message_body = get_outreach_message_body_file() # Run - self.run_scraper_with_args_for_30_seconds(f"-input niche.txt -results \"{output_path}\"", timeout=get_scraper_timeout()) + self.run_scraper_with_args_for_30_seconds( + f'-input niche.txt -results "{output_path}"', timeout=get_scraper_timeout() + ) + + if not os.path.exists(output_path): + error( + f" => Scraper output not found at {output_path}. Check scraper logs and configuration." + ) + os.remove("niche.txt") + return # Get the items from the file items = self.get_items_from_file(output_path) @@ -203,10 +241,15 @@ class Outreach: time.sleep(2) # Create a yagmail SMTP client outside the loop - yag = yagmail.SMTP(user=self.email_creds["username"], password=self.email_creds["password"], host=self.email_creds["smtp_server"], port=self.email_creds["smtp_port"]) + yag = yagmail.SMTP( + user=self.email_creds["username"], + password=self.email_creds["password"], + host=self.email_creds["smtp_server"], + port=self.email_creds["smtp_port"], + ) # Get the email for each business - for item in items: + for index, item in enumerate(items, start=1): try: # Check if the item"s website is valid website = item.split(",") @@ -215,8 +258,8 @@ class Outreach: if website != "": test_r = requests.get(website) if test_r.status_code == 200: - self.set_email_for_website(items.index(item), website, output_path) - + self.set_email_for_website(index, website, output_path) + # Send emails using the existing SMTP connection receiver_email = item.split(",")[-1] @@ -224,11 +267,18 @@ class Outreach: warning(f" => No email provided. Skipping...") continue - subject = message_subject.replace("{{COMPANY_NAME}}", item[0]) - body = open(message_body, "r").read().replace("{{COMPANY_NAME}}", item[0]) + company_name = item.split(",")[0] + subject = message_subject.replace( + "{{COMPANY_NAME}}", company_name + ) + body = ( + open(message_body, "r") + .read() + .replace("{{COMPANY_NAME}}", company_name) + ) info(f" => Sending email to {receiver_email}...") - + yag.send( to=receiver_email, subject=subject, @@ -240,4 +290,4 @@ class Outreach: warning(f" => Website {website} is invalid. Skipping...") except Exception as err: error(f" => Error: {err}...") - continue \ No newline at end of file + continue From f011975c887d9bfdf86909cb950802f6ee615ba1 Mon Sep 17 00:00:00 2001 From: FujiwaraChoki <78088687+FujiwaraChoki@users.noreply.github.com> Date: Sun, 1 Mar 2026 20:48:17 +0100 Subject: [PATCH 2/4] Fix songs download fallback and audio selection (#130) * Fix songs download fallback and audio selection * Retry songs download when directory exists but empty --- src/utils.py | 66 ++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 54 insertions(+), 12 deletions(-) diff --git a/src/utils.py b/src/utils.py index 67684f0..1b1f731 100644 --- a/src/utils.py +++ b/src/utils.py @@ -7,6 +7,11 @@ import platform from status import * from config import * +DEFAULT_SONG_ARCHIVE_URLS = [ + "https://filebin.net/bb9ewdtckolsf3sg/drive-download-20240209T180019Z-001.zip" +] + + def close_running_selenium_instances() -> None: """ Closes any running Selenium instances. @@ -28,6 +33,7 @@ def close_running_selenium_instances() -> None: except Exception as e: error(f"Error occurred while closing running Selenium instances: {str(e)}") + def build_url(youtube_video_id: str) -> str: """ Builds the URL to the YouTube video. @@ -40,6 +46,7 @@ def build_url(youtube_video_id: str) -> str: """ return f"https://www.youtube.com/watch?v={youtube_video_id}" + def rem_temp_files() -> None: """ Removes temporary files in the `.mp` directory. @@ -56,6 +63,7 @@ def rem_temp_files() -> None: if not file.endswith(".json"): os.remove(os.path.join(mp_dir, file)) + def fetch_songs() -> None: """ Downloads songs into songs/ directory to use with geneated videos. @@ -72,28 +80,53 @@ def fetch_songs() -> None: if get_verbose(): info(f" => Created directory: {files_dir}") else: - # Skip if songs are already downloaded - return + existing_audio_files = [ + name + for name in os.listdir(files_dir) + if os.path.isfile(os.path.join(files_dir, name)) + and name.lower().endswith((".mp3", ".wav", ".m4a", ".aac", ".ogg")) + ] + if len(existing_audio_files) > 0: + return - # Download songs - response = requests.get(get_zip_url() or "https://filebin.net/bb9ewdtckolsf3sg/drive-download-20240209T180019Z-001.zip") + configured_url = get_zip_url().strip() + download_urls = [configured_url] if configured_url else [] + download_urls.extend(DEFAULT_SONG_ARCHIVE_URLS) - # Save the zip file - with open(os.path.join(files_dir, "songs.zip"), "wb") as file: - file.write(response.content) + archive_path = os.path.join(files_dir, "songs.zip") + downloaded = False - # Unzip the file - with zipfile.ZipFile(os.path.join(files_dir, "songs.zip"), "r") as file: - file.extractall(files_dir) + for download_url in download_urls: + try: + response = requests.get(download_url, timeout=60) + response.raise_for_status() + + with open(archive_path, "wb") as file: + file.write(response.content) + + with zipfile.ZipFile(archive_path, "r") as file: + file.extractall(files_dir) + + downloaded = True + break + except Exception as err: + warning(f"Failed to fetch songs from {download_url}: {err}") + + if not downloaded: + raise RuntimeError( + "Could not download a valid songs archive from any configured URL" + ) # Remove the zip file - os.remove(os.path.join(files_dir, "songs.zip")) + if os.path.exists(archive_path): + os.remove(archive_path) success(" => Downloaded Songs to ../Songs.") except Exception as e: error(f"Error occurred while fetching songs: {str(e)}") + def choose_random_song() -> str: """ Chooses a random song from the songs/ directory. @@ -102,9 +135,18 @@ def choose_random_song() -> str: str: The path to the chosen song. """ try: - songs = os.listdir(os.path.join(ROOT_DIR, "Songs")) + songs_dir = os.path.join(ROOT_DIR, "Songs") + songs = [ + name + for name in os.listdir(songs_dir) + if os.path.isfile(os.path.join(songs_dir, name)) + and name.lower().endswith((".mp3", ".wav", ".m4a", ".aac", ".ogg")) + ] + if len(songs) == 0: + raise RuntimeError("No audio files found in Songs directory") song = random.choice(songs) success(f" => Chose song: {song}") return os.path.join(ROOT_DIR, "Songs", song) except Exception as e: error(f"Error occurred while choosing random song: {str(e)}") + raise From af3bf5dba4aac9b236a780016d14d4700e683b17 Mon Sep 17 00:00:00 2001 From: FujiwaraChoki <78088687+FujiwaraChoki@users.noreply.github.com> Date: Sun, 1 Mar 2026 20:48:28 +0100 Subject: [PATCH 3/4] Fix Twitter posting flow for X compose UI (#131) * Fix Twitter posting flow for X compose UI * Avoid unnecessary tweet generation for manual posts --- src/classes/Twitter.py | 120 +++++++++++++++++++++++++---------------- 1 file changed, 74 insertions(+), 46 deletions(-) diff --git a/src/classes/Twitter.py b/src/classes/Twitter.py index 2607c6f..2ced56f 100644 --- a/src/classes/Twitter.py +++ b/src/classes/Twitter.py @@ -2,29 +2,35 @@ import re import g4f import sys import time +import os +import json from cache import * from config import * from status import * from constants import * -from typing import List +from typing import List, Optional from datetime import datetime from termcolor import colored from selenium_firefox import * from selenium import webdriver from selenium.common import exceptions -from selenium.webdriver.common import keys from selenium.webdriver.common.by import By from selenium.webdriver.firefox.service import Service from selenium.webdriver.firefox.options import Options from webdriver_manager.firefox import GeckoDriverManager +from selenium.webdriver.support.ui import WebDriverWait +from selenium.webdriver.support import expected_conditions as EC class Twitter: """ Class for the Bot, that grows a Twitter account. """ - def __init__(self, account_uuid: str, account_nickname: str, fp_profile_path: str, topic: str) -> None: + + def __init__( + self, account_uuid: str, account_nickname: str, fp_profile_path: str, topic: str + ) -> None: """ Initializes the Twitter Bot. @@ -43,11 +49,16 @@ class Twitter: # Initialize the Firefox profile self.options: Options = Options() - + # Set headless state of browser if get_headless(): self.options.add_argument("--headless") + if not os.path.isdir(fp_profile_path): + raise ValueError( + f"Firefox profile path does not exist or is not a directory: {fp_profile_path}" + ) + # Set the profile path self.options.add_argument("-profile") self.options.add_argument(fp_profile_path) @@ -56,9 +67,12 @@ class Twitter: self.service: Service = Service(GeckoDriverManager().install()) # Initialize the browser - self.browser: webdriver.Firefox = webdriver.Firefox(service=self.service, options=self.options) + self.browser: webdriver.Firefox = webdriver.Firefox( + service=self.service, options=self.options + ) + self.wait: WebDriverWait = WebDriverWait(self.browser, 30) - def post(self, text: str = None) -> None: + def post(self, text: Optional[str] = None) -> None: """ Starts the Twitter Bot. @@ -71,47 +85,62 @@ class Twitter: bot: webdriver.Firefox = self.browser verbose: bool = get_verbose() - bot.get("https://twitter.com") + bot.get("https://x.com/compose/post") - time.sleep(2) - - post_content: str = self.generate_post() + post_content: str = text if text is not None else self.generate_post() now: datetime = datetime.now() - print(colored(f" => Posting to Twitter:", "blue"), post_content[:30] + "...") + print(colored(" => Posting to Twitter:", "blue"), post_content[:30] + "...") + body = post_content - try: - bot.find_element(By.XPATH, "//a[@data-testid='SideNav_NewTweet_Button']").click() - except exceptions.NoSuchElementException: - time.sleep(3) - bot.find_element(By.XPATH, "//a[@data-testid='SideNav_NewTweet_Button']").click() + text_box = None + text_box_selectors = [ + (By.CSS_SELECTOR, "div[data-testid='tweetTextarea_0'][role='textbox']"), + (By.XPATH, "//div[@data-testid='tweetTextarea_0']//div[@role='textbox']"), + (By.XPATH, "//div[@role='textbox']"), + ] - time.sleep(2) - body = post_content if text is None else text + for selector in text_box_selectors: + try: + text_box = self.wait.until(EC.element_to_be_clickable(selector)) + text_box.click() + text_box.send_keys(body) + break + except Exception: + continue - try: - bot.find_element(By.XPATH, "//div[@role='textbox']").send_keys(body) - except exceptions.NoSuchElementException: - time.sleep(2) - bot.find_element(By.XPATH, "//div[@role='textbox']").send_keys(body) + if text_box is None: + raise RuntimeError( + "Could not find tweet text box. Ensure you are logged into X in this Firefox profile." + ) - time.sleep(1) - bot.find_element(By.CLASS_NAME, "notranslate").send_keys(keys.Keys.ENTER) - bot.find_element(By.XPATH, "//button[@data-testid='tweetButton']").click() + post_button = None + post_button_selectors = [ + (By.XPATH, "//button[@data-testid='tweetButtonInline']"), + (By.XPATH, "//button[@data-testid='tweetButton']"), + (By.XPATH, "//span[text()='Post']/ancestor::button"), + ] + + for selector in post_button_selectors: + try: + post_button = self.wait.until(EC.element_to_be_clickable(selector)) + post_button.click() + break + except Exception: + continue + + if post_button is None: + raise RuntimeError("Could not find the Post button on X compose screen.") if verbose: print(colored(" => Pressed [ENTER] Button on Twitter..", "blue")) - time.sleep(4) + time.sleep(2) # Add the post to the cache - self.add_post({ - "content": post_content, - "date": now.strftime("%m/%d/%Y, %H:%M:%S") - }) + self.add_post({"content": body, "date": now.strftime("%m/%d/%Y, %H:%M:%S")}) success("Posted to Twitter successfully!") - def get_posts(self) -> List[dict]: """ Gets the posts from the cache. @@ -121,12 +150,10 @@ class Twitter: """ if not os.path.exists(get_twitter_cache_path()): # Create the cache file - with open(get_twitter_cache_path(), 'w') as file: - json.dump({ - "posts": [] - }, file, indent=4) + with open(get_twitter_cache_path(), "w") as file: + json.dump({"accounts": []}, file, indent=4) - with open(get_twitter_cache_path(), 'r') as file: + with open(get_twitter_cache_path(), "r") as file: parsed = json.load(file) # Find our account @@ -140,7 +167,9 @@ class Twitter: # Return the posts return posts - + + return [] + def add_post(self, post: dict) -> None: """ Adds a post to the cache. @@ -156,17 +185,16 @@ class Twitter: with open(get_twitter_cache_path(), "r") as file: previous_json = json.loads(file.read()) - + # Find our account accounts = previous_json["accounts"] for account in accounts: if account["id"] == self.account_uuid: account["posts"].append(post) - + # Commit changes with open(get_twitter_cache_path(), "w") as f: f.write(json.dumps(previous_json)) - def generate_post(self) -> str: """ @@ -180,9 +208,9 @@ class Twitter: messages=[ { "role": "user", - "content": f"Generate a Twitter post about: {self.topic} in {get_twitter_language()}. The Limit is 2 sentences. Choose a specific sub-topic of the provided topic." + "content": f"Generate a Twitter post about: {self.topic} in {get_twitter_language()}. The Limit is 2 sentences. Choose a specific sub-topic of the provided topic.", } - ] + ], ) if get_verbose(): @@ -193,11 +221,11 @@ class Twitter: sys.exit(1) # Apply Regex to remove all * - completion = re.sub(r"\*", "", completion).replace("\"", "") - + completion = re.sub(r"\*", "", completion).replace('"', "") + if get_verbose(): info(f"Length of post: {len(completion)}") if len(completion) >= 260: - return self.generate_post() + return completion[:257].rsplit(" ", 1)[0] + "..." return completion From ddc2287dbba221ba5ece9d37e6bfc1c661f0166c Mon Sep 17 00:00:00 2001 From: FujiwaraChoki <78088687+FujiwaraChoki@users.noreply.github.com> Date: Sun, 1 Mar 2026 20:48:38 +0100 Subject: [PATCH 4/4] Validate affiliate and Firefox profile inputs early (#132) --- src/classes/AFM.py | 64 ++++++++++++++++++++++++++++++++++------------ 1 file changed, 48 insertions(+), 16 deletions(-) diff --git a/src/classes/AFM.py b/src/classes/AFM.py index 044934f..bceb88b 100644 --- a/src/classes/AFM.py +++ b/src/classes/AFM.py @@ -1,4 +1,6 @@ import g4f +import os +from urllib.parse import urlparse from status import * from config import * @@ -11,11 +13,20 @@ from selenium.webdriver.firefox.service import Service from selenium.webdriver.firefox.options import Options from webdriver_manager.firefox import GeckoDriverManager + class AffiliateMarketing: """ - This class will be used to handle all the affiliate marketing related operations. + This class will be used to handle all the affiliate marketing related operations. """ - def __init__(self, affiliate_link: str, fp_profile_path: str, twitter_account_uuid: str, account_nickname: str, topic: str) -> None: + + def __init__( + self, + affiliate_link: str, + fp_profile_path: str, + twitter_account_uuid: str, + account_nickname: str, + topic: str, + ) -> None: """ Initializes the Affiliate Marketing class. @@ -38,19 +49,32 @@ class AffiliateMarketing: if get_headless(): self.options.add_argument("--headless") + if not os.path.isdir(fp_profile_path): + raise ValueError( + f"Firefox profile path does not exist or is not a directory: {fp_profile_path}" + ) + # Set the profile path self.options.add_argument("-profile") self.options.add_argument(fp_profile_path) - + # Set the service self.service: Service = Service(GeckoDriverManager().install()) # Initialize the browser - self.browser: webdriver.Firefox = webdriver.Firefox(service=self.service, options=self.options) + self.browser: webdriver.Firefox = webdriver.Firefox( + service=self.service, options=self.options + ) # Set the affiliate link self.affiliate_link: str = affiliate_link + parsed_link = urlparse(self.affiliate_link) + if parsed_link.scheme not in ["http", "https"] or not parsed_link.netloc: + raise ValueError( + f"Affiliate link is invalid. Expected a full URL, got: {self.affiliate_link}" + ) + # Set the Twitter account UUID self.account_uuid: str = twitter_account_uuid @@ -72,8 +96,10 @@ class AffiliateMarketing: self.browser.get(self.affiliate_link) # Get the product name - product_title: str = self.browser.find_element(By.ID, AMAZON_PRODUCT_TITLE_ID).text - + product_title: str = self.browser.find_element( + By.ID, AMAZON_PRODUCT_TITLE_ID + ).text + # Get the features of the product features: any = self.browser.find_elements(By.ID, AMAZON_FEATURE_BULLETS_ID) @@ -82,7 +108,7 @@ class AffiliateMarketing: if get_verbose(): info(f"Features: {features}") - + # Set the product title self.product_title: str = product_title @@ -102,12 +128,7 @@ class AffiliateMarketing: # Generate the response response: str = g4f.ChatCompletion.create( model=parse_model(get_model()), - messages=[ - { - "role": "user", - "content": prompt - } - ] + messages=[{"role": "user", "content": prompt}], ) # Return the response @@ -121,13 +142,19 @@ class AffiliateMarketing: pitch (str): The pitch for the product. """ # Generate the response - pitch: str = self.generate_response(f"I want to promote this product on my website. Generate a brief pitch about this product, return nothing else except the pitch. Information:\nTitle: \"{self.product_title}\"\nFeatures: \"{str(self.features)}\"") + "\nYou can buy the product here: " + self.affiliate_link + pitch: str = ( + self.generate_response( + f'I want to promote this product on my website. Generate a brief pitch about this product, return nothing else except the pitch. Information:\nTitle: "{self.product_title}"\nFeatures: "{str(self.features)}"' + ) + + "\nYou can buy the product here: " + + self.affiliate_link + ) self.pitch: str = pitch # Return the response return pitch - + def share_pitch(self, where: str) -> None: """ This method will be used to share the pitch on the specified platform. @@ -137,7 +164,12 @@ class AffiliateMarketing: """ if where == "twitter": # Initialize the Twitter class - twitter: Twitter = Twitter(self.account_uuid, self.account_nickname, self._fp_profile_path, self.topic) + twitter: Twitter = Twitter( + self.account_uuid, + self.account_nickname, + self._fp_profile_path, + self.topic, + ) # Share the pitch twitter.post(self.pitch)