#!/Users/sij/miniforge3/envs/instabot/bin/python import os import io from io import BytesIO import copy import re import jwt import json from tqdm import tqdm import pyotp import time import pytz import requests import tempfile import random import subprocess import atexit import urllib.parse import urllib.request import uuid import base64 from time import sleep from datetime import timedelta, datetime as date from PIL import Image from typing import Dict, List from instagrapi import Client as igClient from instagrapi.types import UserShort from urllib.parse import urlparse from instagrapi.exceptions import LoginRequired as ClientLoginRequiredError from openai import OpenAI import argparse from PIL import Image import io import json from ollama import Client as oLlama from SD import SD from dotenv import load_dotenv ########################### ### .env INITIALIZATION ### ########################### load_dotenv() GHOST_API_KEY=os.getenv('GHOST_API_KEY') GHOST_CONTENT_KEY=os.getenv('GHOST_CONTENT_KEY') OPENAI_API_KEY=os.getenv('OPENAI_API_KEY') GHOST_API_URL=os.getenv('GHOST_API_URL') ######################## ### ARGUMENT PARSING ### ######################## parser = argparse.ArgumentParser(description='Instagram bot for generating images and engaging with other users.') parser.add_argument('--account', type=str, help='Select which Instagram account to act as.', default='simulacrai') parser.add_argument('--newsession', action='store_true', help='Start a new session.', default=False) parser.add_argument('--shortsleep', type=int, help='Average short sleep time in seconds.', default=5) parser.add_argument('--longsleep', type=int, help='Average long sleep time in seconds.', default=180) parser.add_argument('--user', type=str, help='Instagram username to fetch media from and comment on.', default='') parser.add_argument('--hashtag', type=str, help='Hashtag to fetch media from and comment on.', default='') parser.add_argument('--image_url', type=str, help='URL to a specific Instagram image to comment on.', default='') parser.add_argument('--count', type=int, help='Number of media items to interact with.', default=1) parser.add_argument('--commenttype', type=str, help='Type of comments to make.', default='') parser.add_argument('--commentmode', type=str, help='Mode of commenting.', default='recent') parser.add_argument('--posttype', type=str, help='Generate images of a specific type.', default=None) parser.add_argument('--postfile', type=str, help='Generate post from an existing image file.', default=None) parser.add_argument('--commentonly', action='store_true', help='Only comment on media, do not generate images.', default=False) parser.add_argument('--postonly', action='store_true', help='Only generate images and post to Instagram, do not comment.', default=False) parser.add_argument('--noghost', action='store_true', help='Skips posting generated images to Ghost.', default=False) parser.add_argument('--local', action='store_true', help='Do not post to Instagram, and disable OpenAI', default=False) parser.add_argument('--openai', action='store_true', help='Allows the use of OpenaI for chat completions and vision', default=False) parser.add_argument('--gpt4', action='store_true', help='Allows the use of OpenAI GPT-4 for chat completions', default=False) parser.add_argument('--ollama', action='store_true', help='Allows the use of Ollama for chat completions', default=False) parser.add_argument('--llava', action='store_true', help='Allows the use of Ollama Llava for image recognition', default=False) parser.add_argument('--dalle', action='store_true', help='Allows the use of OpenAI DALL-E3 for image generation', default=False) parser.add_argument('--wallpaper', action='store_true', help='Set the macOS wallpaper to use the generated images', default=False) parser.add_argument('--fast', action='store_true', help='Appends _fast to workflow names, which are expedited workflows that generally forego ultimate upscaling.', default=False) args = parser.parse_args() ########################### ### FOLDER & PATH SETUP ### ########################### ACCOUNT = args.account VISION_DIR = "/Users/sij/AI/Vision" BASE_DIR = os.path.dirname(os.path.abspath(__file__)) WORKFLOWS_DIR = os.path.join(VISION_DIR, 'workflows') RESULT_DIRECTORY = os.path.join(VISION_DIR, 'ComfyUI/output/') ACCOUNT_DIR = os.path.join(BASE_DIR, ACCOUNT) ACCOUNT_IMAGES_DIR = os.path.join(ACCOUNT_DIR, 'images') ACCOUNT_CONFIG_PATH = os.path.join(ACCOUNT_DIR, f'config.json') DOWNLOADED_IMAGES_PATH = os.path.join(ACCOUNT_DIR, 'downloads') with open(ACCOUNT_CONFIG_PATH, 'r') as config_file: ACCOUNT_CONFIG = json.load(config_file) if not os.path.exists(ACCOUNT_IMAGES_DIR): os.makedirs(ACCOUNT_IMAGES_DIR ) def start_file_server(): command = [ "caddy", "file-server", "--root", ACCOUNT_IMAGES_DIR, "--listen", ":8190", "--browse" ] subprocess.Popen(command) atexit.register(lambda: subprocess.call(["killall", "caddy"])) ################### ### VALIDATION ### ################## if args.account and args.posttype and not args.posttype in ACCOUNT_CONFIG["posts"]: print ("ERROR: NO SUCH POST TYPE IS AVAILABLE FOR THIS ACCOUNT.") if args.account and args.commenttype and not args.commenttype in ACCOUNT_CONFIG["comments"]: print ("ERROR: NO SUCH COMMENTTYPE IS AVAILABLE FOR THIS ACCOUNT.") #################### ### CLIENT SETUP ### #################### cl = igClient(request_timeout=1) if args.local is False and args.ollama is False and args.openai is True: LLM = OpenAI(api_key=OPENAI_API_KEY) else: LLM = oLlama(host='http://localhost:11434') if args.local is False and args.openai is True: VISION_LLM = OpenAI(api_key=OPENAI_API_KEY) else: VISION_LLM = oLlama(host='http://localhost:11434') if args.local is False and args.dalle is True: IMG_GEN = OpenAI(api_key=OPENAI_API_KEY) IMG_MODEL = "dall-e-3" # else: # start_file_server() # IMG_GEN = SD(ACCOUNT_IMAGES_DIR, "") SD_SERVER_ADDRESS = "http://localhost:8188" CLIENT_ID = str(uuid.uuid4()) ############################### ### INSTAGRAM & GHOST SETUP ### ############################### IG_USERNAME = ACCOUNT_CONFIG.get("ig_name") IG_PASSWORD = ACCOUNT_CONFIG.get("ig_pass") IG_SECRET_KEY = ACCOUNT_CONFIG.get("ig_2fa_secret") IG_SESSION_PATH = os.path.join(ACCOUNT_DIR, f'credentials.json') ######################## ### LLM PROMPT SETUP ### ######################## IMG_PROMPT_SYS = ACCOUNT_CONFIG.get("img_prompt_sys") IMG_DESCRIPTION_SYS = ACCOUNT_CONFIG.get("img_description_sys") COMMENT_PROMPT_SYS = ACCOUNT_CONFIG.get("img_comment_sys") HASHTAGS = ACCOUNT_CONFIG.get("preferred_hashtags", []) IMAGE_URL = args.image_url rollover_time = 1702605780 COMPLETED_MEDIA_LOG = os.path.join(ACCOUNT_DIR, f'completed-media.txt') TOTP = pyotp.TOTP(IG_SECRET_KEY) SHORT = args.shortsleep LONG = args.longsleep ####################### ### ⚠️ FUNCTIONS ⚠️ ### ####################### ######################## ### SOCIAL FUNCTIONS ### ######################## def follow_by_username(username) -> bool: """ Follow a user Parameters ---------- username: str Username for an Instagram account Returns ------- bool A boolean value """ userid = cl.user_id_from_username(username) sleep(SHORT) return cl.user_follow(userid) def unfollow_by_username(username) -> bool: """ Unfollow a user Parameters ---------- username: str Username for an Instagram account Returns ------- bool A boolean value """ userid = cl.user_id_from_username(username) sleep(SHORT) return cl.user_unfollow(userid) def get_followers(amount: int = 0) -> Dict[int, UserShort]: """ Get bot's followers Parameters ---------- amount: int, optional Maximum number of media to return, default is 0 - Inf Returns ------- Dict[int, UserShort] Dict of user_id and User object """ sleep(SHORT) return cl.user_followers(cl.user_id, amount=amount) def get_followers_usernames(amount: int = 0) -> List[str]: """ Get bot's followers usernames Parameters ---------- amount: int, optional Maximum number of media to return, default is 0 - Inf Returns ------- List[str] List of usernames """ followers = cl.user_followers(cl.user_id, amount=amount) sleep(SHORT) return [user.username for user in followers.values()] def get_following(amount: int = 0) -> Dict[int, UserShort]: """ Get bot's followed users Parameters ---------- amount: int, optional Maximum number of media to return, default is 0 - Inf Returns ------- Dict[int, UserShort] Dict of user_id and User object """ sleep(SHORT) return cl.user_following(cl.user_id, amount=amount) def get_user_media(username, amount=30): """ Fetch recent media for a given username. Parameters ---------- username: str Username for an Instagram account amount: int, optional Number of media to fetch, default is 20 Returns ------- List of medias """ print(f"Fetching recent media for {username}...") user_id = cl.user_id_from_username(username) medias = cl.user_medias(user_id, amount) final_medias = [] for media in medias: sleep(SHORT) if media.media_type == 1: final_medias.append(media) return final_medias def get_user_image_urls(username, amount=30) -> List[str]: """ Fetch recent media URLs for a given username. Parameters ---------- username: str Username for an Instagram account amount: int, optional Number of media URLs to fetch, default is 20 Returns ------- List[str] List of media URLs """ print(f"Fetching recent media URLs for {username}...") user_id = cl.user_id_from_username(username) medias = cl.user_medias(user_id, amount) urls = [] for media in medias: sleep(SHORT) if media.media_type == 1 and media.thumbnail_url: # Photo urls.append(media.thumbnail_url) # elif media.media_type == 2: # Video # urls.append(media.video_url) # elif media.media_type == 8: # Album # for resource in media.resources: # sleep(SHORT) # if resource.media_type == 1: # urls.append(resource.thumbnail_url) # elif resource.media_type == 2: # urls.append(resource.video_url) return urls def is_valid_url(url): try: result = urlparse(url) return all([result.scheme, result.netloc]) except Exception: return False def get_random_follower(): followers = cl.get_followers_usernames() sleep(SHORT) return random.choice(followers) def get_medias_by_hashtag(hashtag: str, days_ago_max:int = 14, ht_type:str = None, amount:int = args.count): if not ht_type: ht_type = args.commentmode print(f"Fetching {ht_type} media for hashtag: {hashtag}") ht_medias = [] while True: sleep(SHORT) if ht_type == "top": ht_medias.extend(cl.hashtag_medias_top(name=hashtag, amount=amount*10)) elif ht_type == "recent": ht_medias.extend(cl.hashtag_medias_recent(name=hashtag, amount=amount*10)) filtered_medias = filter_medias(ht_medias, days_ago_max=days_ago_max) print(f"Filtered {ht_type} media count obtained for '#{hashtag}': {len(filtered_medias)}") if len(filtered_medias) >= amount: print(f"Desired amount of {amount} filtered media reached.") break return filtered_medias def get_medias_from_all_hashtags(days_ago_max=14, ht_type:str = None, amount:int = args.count): if not ht_type: ht_type = args.commentmode print(f"Fetching {ht_type} media.") filtered_medias = [] while len(filtered_medias) < amount: hashtag = random.choice(HASHTAGS) print(f"Using hashtag: {hashtag}") fetched_medias = [] sleep(SHORT) # Fetch medias based on the hashtag type if ht_type == "top": fetched_medias = cl.hashtag_medias_top(name=hashtag, amount=50) # Fetch a large batch to filter from elif ht_type == "recent": fetched_medias = cl.hashtag_medias_recent(name=hashtag, amount=50) # Same for recent current_filtered_medias = filter_medias(fetched_medias, days_ago_max=days_ago_max) filtered_medias.extend(current_filtered_medias) print(f"Filtered {ht_type} media count obtained for '#{hashtag}': {len(current_filtered_medias)}") # Trim the list if we've collected more than needed if len(filtered_medias) > amount: filtered_medias = filtered_medias[:amount] print(f"Desired amount of {amount} filtered media reached.") break else: print(f"Total filtered media count so far: {len(filtered_medias)}") return filtered_medias def filter_medias( medias: List, like_count_min=None, like_count_max=None, comment_count_min=None, comment_count_max=None, days_ago_max=None, ): # Adjust to use your preferred timezone, for example, UTC days_back = date.now(pytz.utc) - timedelta(days=days_ago_max) if days_ago_max else None return [ media for media in medias if ( (like_count_min is None or media.like_count >= like_count_min) and (like_count_max is None or media.like_count <= like_count_max) and (comment_count_min is None or media.comment_count >= comment_count_min) and (comment_count_max is None or media.comment_count <= comment_count_max) and (days_ago_max is None or (media.taken_at and media.taken_at > days_back)) and not check_media_in_completed_lists(media) ) ] def add_media_to_completed_lists(media): """ Add a media to the completed lists after interacting with it. """ with open(COMPLETED_MEDIA_LOG, 'a') as file: file.write(f"{str(media.pk)}\n") def check_media_in_completed_lists(media): """ Check if a media is in the completed lists. """ with open(COMPLETED_MEDIA_LOG, 'r') as file: completed_media = file.read().splitlines() return str(media.pk) in completed_media def download_and_resize_image(url: str, download_path: str = None, max_dimension: int = 1200) -> str: if not isinstance(url, str): url = str(url) parsed_url = urlparse(url) if not download_path or not os.path.isdir(os.path.dirname(download_path)): # Use a temporary file if download_path is not specified or invalid _, temp_file_extension = os.path.splitext(parsed_url.path) if not temp_file_extension: temp_file_extension = ".jpg" # Default extension if none is found download_path = tempfile.mktemp(suffix=temp_file_extension, prefix="download_") if url and parsed_url.scheme and parsed_url.netloc: try: os.makedirs(os.path.dirname(download_path), exist_ok=True) with requests.get(url) as response: response.raise_for_status() # Raises an HTTPError if the response was an error image = Image.open(BytesIO(response.content)) # Resize the image, preserving aspect ratio if max(image.size) > max_dimension: image.thumbnail((max_dimension, max_dimension)) # Save the image, preserving the original format if possible image_format = image.format if image.format else "JPEG" image.save(download_path, image_format) return download_path except Exception as e: # Handle or log the error as needed print(f"Error downloading or resizing image: {e}") return None ############################### ### ACTIVE SOCIAL FUNCTIONS ### ############################### def comment_on_user_media(user: str, comment_type: str = "default", amount=5): """ Comment on a user's media. """ comment_prompt_usr = ACCOUNT_CONFIG['comments'][comment_type]['img_comment_usr'] medias = get_user_media(user, amount) for media in medias: if not check_media_in_completed_lists(media): sleep(SHORT) if media.thumbnail_url and is_valid_url(media.thumbnail_url): media_path = download_and_resize_image(media.thumbnail_url, f"{DOWNLOADED_IMAGES_PATH}/{media.pk}.jpg") if media_path is not None: encoded_media = encode_image_to_base64(media_path) comment_text = llava(encoded_media, COMMENT_PROMPT_SYS, comment_prompt_usr) if args.llava or not args.openai else gpt4v(encoded_media, COMMENT_PROMPT_SYS, comment_prompt_usr) if comment_text: cl.media_comment(media.pk, comment_text) print(f"Commented on media: {media.pk}") else: print(f"Failed to generate comment for media: {media.pk}") add_media_to_completed_lists(media) sleep(SHORT) else: print(f"We received a nonetype! {media_path}") else: print(f"URL for {media.pk} disappeared it seems...") else: print(f"Media already interacted with: {media.pk}") def comment_by_type(comment_type: str = args.commenttype, amount=3, hashtag: str = None): """ Comment on a hashtag's media. """ if not hashtag: hashtag = random.choice(ACCOUNT_CONFIG['comments'][comment_type]['hashtags']) medias = get_medias_by_hashtag(hashtag=hashtag, days_ago_max=7, amount=amount) for media in medias: if not check_media_in_completed_lists(media): media_path = download_and_resize_image(media.thumbnail_url, f"{DOWNLOADED_IMAGES_PATH}/{media.pk}.jpg") comment_text = None if media_path and os.path.exists(media_path): encoded_media = encode_image_to_base64(media_path) comment_prompt_usr = ACCOUNT_CONFIG['comments'][comment_type]['img_comment_usr'] + " For reference, here is the description that was posted with this image: " + media.caption_text comment_text = llava(encoded_media, comment_prompt_usr) if args.llava or not args.openai else gpt4v(encoded_media, COMMENT_PROMPT_SYS, comment_prompt_usr) if comment_text: cl.media_comment(media.pk, comment_text) print(f"Commented on media: https://instagram.com/p/{media.pk}/") else: print(f"Failed to generate comment for media: https://instagram.com/p/{media.pk}") add_media_to_completed_lists(media) sleep(SHORT) else: print(f"Media already interacted with: {media.pk}") def comment_on_specific_media(media_url, comment_type: str = "default"): """ Comment on a specific media given its URL. """ media_id = cl.media_pk_from_url(media_url) sleep(SHORT) media = cl.media_info(media_id) sleep(SHORT) media_path = download_and_resize_image(media.thumbnail_url, f"{DOWNLOADED_IMAGES_PATH}/{media.pk}.jpg") encoded_media = encode_image_to_base64(media_path) comment_prompt_usr = ACCOUNT_CONFIG['comments'][comment_type]['img_comment_usr'] + " For reference, here is the description that was posted with this image: " + media.caption_text comment_text = llava(encoded_media, comment_prompt_usr) if args.llava or not args.openai else gpt4v(encoded_media, COMMENT_PROMPT_SYS, comment_prompt_usr) if comment_text: cl.media_comment(media.pk, comment_text) print(f"Commented on specific media: https://instagram.com/p/{media.pk}/") else: print(f"Failed to generate comment for specific media: https://instagram.com/p/{media.pk}/") ##################### ### LLM FUNCTIONS ### ##################### def query_llm(llmPrompt: List = [], system_msg: str = "", user_msg: str = "", max_tokens: int = 150): messages = llmPrompt if llmPrompt else [ {"role": "system", "content": system_msg}, {"role": "user", "content": user_msg} ] # options_dict = {"num_predict": max_tokens} response = LLM.chat.create( model="gpt-4", messages=messages, max_tokens=max_tokens ) if args.openai else LLM.chat( model="mixtral", messages=messages, # Assuming `messages` is defined somewhere options={"num_predict": max_tokens} ) print(response) if "choices" in response: first_choice = response.choices[0] if first_choice.message.content: return first_choice.message.content else: message_str = str(first_choice.message) match = re.search(r"content='(.*?)', role=", message_str, re.DOTALL) if match: return match.group(1) else: print("No content found in message string:", message_str) print("Trying again!") query_llm(system_msg, user_msg, max_tokens) elif "message" in response: if "content" in response["message"]: content = response["message"]["content"] print(f"Response: {content}") return content else: print("No choices found in response") return "" def llava(image_base64, prompt): response = VISION_LLM.generate( model = 'llava', prompt = f"This is a chat between a user and an assistant. The assistant is helping the user to describe an image. {prompt}", images = [image_base64] ) print(response) return "" if "pass" in response["response"].lower() else response["response"] def gpt4v(image_base64, prompt_sys: str, prompt_usr: str, max_tokens: int = 150): response_1 = VISION_LLM.chat.completions.create( model="gpt-4-vision-preview", messages=[ { "role": "system", "content": f"This is a chat between a user and an assistant. The assistant is helping the user to describe an image. {prompt_sys}", }, { "role": "user", "content": [ {"type": "text", "text": f"{prompt_usr}"}, { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{image_base64}" }, }, ], } ], max_tokens=max_tokens, stream=False ) if response_1 and response_1.choices: if len(response_1.choices) > 0: first_choice = response_1.choices[0] if first_choice.message and first_choice.message.content: comment_content = first_choice.message.content if "PASS" in comment_content: return "" print(f"Generated comment: {comment_content}") response_2 = VISION_LLM.chat.completions.create( model="gpt-4-vision-preview", messages=[ { "role": "system", "content": f"This is a chat between a user and an assistant. The assistant is helping the user to describe an image. {prompt_sys}", }, { "role": "user", "content": [ {"type": "text", "text": f"{prompt_usr}"}, { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{image_base64}" }, }, ], }, { "role": "assistant", "content": comment_content }, { "role": "user", "content": "Please refine it, and remember to ONLY include the caption or comment, nothing else! That means no preface, no postscript, no notes, no reflections, and not even any acknowledgment of this follow-up message. I need to be able to use your output directly on social media. Do include emojis though." } ], max_tokens=max_tokens, stream=False ) if response_2 and response_2.choices: if len(response_2.choices) > 0: first_choice = response_2.choices[0] if first_choice.message and first_choice.message.content: final_content = first_choice.message.content print(f"Generated comment: {final_content}") if "PASS" in final_content: return "" else: return final_content print("Vision response did not contain expected data.") print(f"Vision response: {response_1}") sleep(15) try_again = gpt4v(image_base64, prompt_sys, prompt_usr, max_tokens) return try_again def encode_image_to_base64(image_path): if os.path.exists(image_path): with Image.open(image_path) as image: output_buffer = BytesIO() image.save(output_buffer, format='JPEG') byte_data = output_buffer.getvalue() base64_str = base64.b64encode(byte_data).decode('utf-8') return base64_str else: print(f"Error: File does not exist at {image_path}") def get_image(status_data, key): """Extract the filename and subfolder from the status data and read the file.""" try: outputs = status_data.get("outputs", {}) images_info = outputs.get(key, {}).get("images", []) if not images_info: raise Exception("No images found in the job output.") image_info = images_info[0] # Assuming the first image is the target filename = image_info.get("filename") subfolder = image_info.get("subfolder", "") # Default to empty if not present file_path = os.path.join(RESULT_DIRECTORY, subfolder, filename) with open(file_path, 'rb') as file: return file.read() except KeyError as e: raise Exception(f"Failed to extract image information due to missing key: {e}") except FileNotFoundError: raise Exception(f"File {filename} not found at the expected path {file_path}") def update_prompt(workflow: dict, post: dict, positive: str, found_key=[None], path=None): if path is None: path = [] try: if isinstance(workflow, dict): for key, value in workflow.items(): current_path = path + [key] if isinstance(value, dict): if value.get('class_type') == 'SaveImage' and value.get('inputs', {}).get('filename_prefix') == 'API_': found_key[0] = key update_prompt(value, post, positive, found_key, current_path) elif isinstance(value, list): # Recursive call with updated path for each item in a list for index, item in enumerate(value): update_prompt(item, post, positive, found_key, current_path + [str(index)]) if value == "API_PPrompt": workflow[key] = post.get(value, "") + positive # print(f"Updated API_PPrompt to: {workflow[key]}") elif value == "API_SPrompt": workflow[key] = post.get(value, "") # print(f"Updated API_SPrompt to: {workflow[key]}") elif value == "API_NPrompt": workflow[key] = post.get(value, "") # print(f"Updated API_NPrompt to: {workflow[key]}") elif key == "seed" or key == "noise_seed": workflow[key] = random.randint(1000000000000, 9999999999999) # print(f"Updated seed to: {workflow[key]}") elif (key == "width" or key == "max_width" or key == "scaled_width") and (value == 1023 or value == 1025): workflow[key] = post.get(value, "") elif (key == "dimension" or key == "height" or key == "max_height" or key == "scaled_height") and (value == 1023 or value == 1025): workflow[key] = post.get(value, "") except Exception as e: print(f"Error in update_prompt at path {' -> '.join(path)}: {e}") raise return found_key[0] ################################## ### IMAGE GENERATION FUNCTIONS ### ################################## def generate_and_post_image(chosen_post = None): """ The main function that orchestrates the workflow from prompt update, image generation, to Instagram posting. """ if chosen_post and chosen_post in ACCOUNT_CONFIG['posts']: post = ACCOUNT_CONFIG['posts'][chosen_post] print(f"Loaded post for {chosen_post}") else: print(f"Unable to load post for {chosen_post}") chosen_post = choose_post(ACCOUNT_CONFIG['posts']) post = ACCOUNT_CONFIG['posts'][chosen_post] print(f"Defaulted to {chosen_post}") image_concept = query_llm(llmPrompt = post['llmPrompt'], max_tokens = 180) print(f"Image concept: {image_concept}") workflow_name = random.choice(post['workflows']) workflow_data = None if args.fast: workflow_data = load_json(None, f"{workflow_name}_fast") if workflow_data is None: workflow_data = load_json(None, workflow_name) jpg_file_path = image_gen(workflow_data) instagram_description = llava(system_msg = IMG_DESCRIPTION_SYS, user_msg = image_concept, max_tokens = 150) instagram_description = re.sub(r'^["\'](.*)["\']$', r'\1', instagram_description) description_filename = f"{prompt_id}.txt" description_path = os.path.join(ACCOUNT_IMAGES_DIR, description_filename) with open(description_path, "w") as desc_file: desc_file.write(instagram_description) upload_photo(jpg_file_path, instagram_description) print(f"Image successfully uploaded to Instagram with description: {instagram_description}") print(f"") def image_gen(workflow_data: dict): saved_file_key = update_prompt(workflow = workflow_data, post = post, positive=image_concept) print(f"") print(f"Saved file key: {saved_file_key}") prompt_id = queue_prompt(workflow_data) print(f"Prompt ID: {prompt_id}") status_data = poll_status(prompt_id) image_data = get_image(status_data, saved_file_key) print(f"Image data obtained") jpg_file_path = save_as_jpg(image_data, prompt_id, 1440, 90) print(f"Image successfully saved to {jpg_file_path}") return jpg_file_path def image_gen2(prompt: str, model: str): response = IMG_GEN.images.generate( model=model, prompt=prompt, size="1024x1024", quality="standard", n=1, ) image_url = response.data[0].url image_path = download_and_resize_image(image_url) return image_path def queue_prompt(prompt: dict): response = requests.post(f"{SD_SERVER_ADDRESS}/prompt", json={"prompt": prompt, "client_id": CLIENT_ID}) if response.status_code == 200: return response.json().get('prompt_id') else: raise Exception(f"Failed to queue prompt. Status code: {response.status_code}, Response body: {response.text}") def poll_status(prompt_id): """Poll the job status until it's complete and return the status data.""" start_time = time.time() # Record the start time while True: elapsed_time = int(time.time() - start_time) # Calculate elapsed time in seconds status_response = requests.get(f"{SD_SERVER_ADDRESS}/history/{prompt_id}") # Use \r to return to the start of the line, and end='' to prevent newline print(f"\rGenerating {prompt_id}. Elapsed time: {elapsed_time} seconds", end='') if status_response.status_code != 200: raise Exception("Failed to get job status") status_data = status_response.json() job_data = status_data.get(prompt_id, {}) if job_data.get("status", {}).get("completed", False): print() print(f"{prompt_id} completed in {elapsed_time} seconds.") return job_data time.sleep(1) def get_image(status_data, key): """Extract the filename and subfolder from the status data and read the file.""" try: outputs = status_data.get("outputs", {}) images_info = outputs.get(key, {}).get("images", []) if not images_info: raise Exception("No images found in the job output.") image_info = images_info[0] # Assuming the first image is the target filename = image_info.get("filename") subfolder = image_info.get("subfolder", "") # Default to empty if not present file_path = os.path.join(RESULT_DIRECTORY, subfolder, filename) with open(file_path, 'rb') as file: return file.read() except KeyError as e: raise Exception(f"Failed to extract image information due to missing key: {e}") except FileNotFoundError: raise Exception(f"File {filename} not found at the expected path {file_path}") ################################ ### PRIMARY ACTIVE FUNCTIONS ### ################################ def load_post(chosen_post: str = "default"): if chosen_post in ACCOUNT_CONFIG['posts']: post = ACCOUNT_CONFIG['posts'][chosen_post] print(f"Loaded post for {chosen_post}") else: print(f"Unable to load post for {chosen_post}. Choosing a default post.") chosen_post = choose_post(ACCOUNT_CONFIG['posts']) post = ACCOUNT_CONFIG['posts'][chosen_post] print(f"Defaulted to {chosen_post}") return post def handle_image_workflow(chosen_post=None): """ Orchestrates the workflow from prompt update, image generation, to either saving the image and description locally or posting to Instagram based on the local flag. """ post = load_post(chosen_post) workflow_name = random.choice(post['workflows']) print(f"Workflow name: {workflow_name}") print(f"Generating image concept for {chosen_post} and {workflow_name} now.") image_concept = query_llm(llmPrompt=post['llmPrompt'], max_tokens=120) print(f"Image concept for {chosen_post}: {image_concept}") workflow_data = None if args.fast: workflow_data = load_json(None, f"{workflow_name}_fast") if workflow_data is None: workflow_data = load_json(None, workflow_name) if args.dalle and not args.local: jpg_file_path = image_gen(image_concept, "dall-e-3") else: saved_file_key = update_prompt(workflow=workflow_data, post=post, positive=image_concept) print(f"Saved file key: {saved_file_key}") prompt_id = queue_prompt(workflow_data) print(f"Prompt ID: {prompt_id}") status_data = poll_status(prompt_id) image_data = get_image(status_data, saved_file_key) jpg_file_path = save_as_jpg(image_data, prompt_id, 1440, 90) handle_single_image(jpg_file_path, chosen_post) def handle_single_image(jpg_file_path: str, chosen_post: str, prompt: str = None): if not prompt: prompt = ACCOUNT_CONFIG['posts'][chosen_post]['Vision_Prompt'] encoded_string = encode_image_to_base64(jpg_file_path) print(f"Image successfully encoded from {jpg_file_path}") instagram_description = llava(encoded_string, prompt) if args.llava or not args.openai else gpt4v(encoded_string, prompt, 150) instagram_description = re.sub(r'^["\'](.*)["\']$', r'\1', instagram_description) ghost_tags = ACCOUNT_CONFIG['posts'][chosen_post]['ghost_tags'] # Save description to file and upload or save locally description_filename = jpg_file_path.rsplit('.', 1)[0] + ".txt" description_path = os.path.join(ACCOUNT_IMAGES_DIR, description_filename) with open(description_path, "w") as desc_file: desc_file.write(instagram_description) if not args.local: post_url = upload_photo(jpg_file_path, instagram_description) print(f"Image posted at {post_url}") if not args.noghost: title_prompt = f"Generate a short 3-5 word title for this image, which already includes the following description: {instagram_description}" img_title = llava(encoded_string, title_prompt) ghost_text = f"{instagram_description}\n\nInstagram link" ghost_url = post_to_ghost(img_title, jpg_file_path, ghost_text, ghost_tags) print(f"Ghost post: {ghost_url}") else: img_title = None else: print(f"Image and description saved locally due to args.local being True.") if args.wallpaper: change_wallpaper(jpg_file_path) print(f"Wallpaper changed.") def choose_post(posts): total_frequency = sum(posts[post_type]['frequency'] for post_type in posts) random_choice = random.randint(1, total_frequency) current_sum = 0 for post_type, post_info in posts.items(): current_sum += post_info['frequency'] if random_choice <= current_sum: return post_type def load_json(json_payload, workflow): if json_payload: return json.loads(json_payload) elif workflow: workflow_path = os.path.join(WORKFLOWS_DIR, f"{workflow}.json" if not workflow.endswith('.json') else workflow) with open(workflow_path, 'r') as file: return json.load(file) else: raise ValueError("No valid input provided.") ################################## ### IMAGE PROCESSING FUNCTIONS ### ################################## def resize_and_convert_image(image_path, max_size=2160, quality=80): with Image.open(image_path) as img: # Resize image ratio = max_size / max(img.size) new_size = tuple([int(x * ratio) for x in img.size]) img = img.resize(new_size, Image.Resampling.LANCZOS) # Convert to JPEG img_byte_arr = io.BytesIO() img.save(img_byte_arr, format='JPEG', quality=quality) img_byte_arr = img_byte_arr.getvalue() return img_byte_arr def save_as_jpg(image_data, prompt_id, max_size=2160, quality=80): filename_png = f"{prompt_id}.png" image_path_png = os.path.join(ACCOUNT_IMAGES_DIR, filename_png) try: # Save the raw PNG data to a file with open(image_path_png, 'wb') as file: file.write(image_data) # Open the PNG, resize it, and save it as JPEG with Image.open(image_path_png) as img: # Resize image if necessary if max(img.size) > max_size: ratio = max_size / max(img.size) new_size = tuple([int(x * ratio) for x in img.size]) img = img.resize(new_size, Image.Resampling.LANCZOS) # Prepare the path for the converted image new_file_name = f"{prompt_id}.jpeg" new_file_path = os.path.join(ACCOUNT_IMAGES_DIR, new_file_name) # Convert to JPEG and save img.convert('RGB').save(new_file_path, format='JPEG', quality=quality) # Optionally, delete the temporary PNG file os.remove(image_path_png) return new_file_path except Exception as e: print(f"Error processing image: {e}") return None def upload_photo(path, caption, title: str=None): print(f"Uploading photo from {path}...") media = cl.photo_upload(path, caption) post_url = f"https://www.instagram.com/p/{media.code}/" return post_url def format_duration(seconds): """Return a string representing the duration in a human-readable format.""" if seconds < 120: return f"{int(seconds)} sec" elif seconds < 6400: return f"{int(seconds // 60)} min" else: return f"{seconds / 3600:.2f} hr" ######################## ### HELPER FUNCTIONS ### ######################## import subprocess def change_wallpaper(image_path): command = """ osascript -e 'tell application "Finder" to set desktop picture to POSIX file "{}"' """.format(image_path) subprocess.run(command, shell=True) def sleep(seconds): """Sleep for a random amount of time, approximately the given number of seconds.""" sleepupto(seconds*0.66, seconds*1.5) def sleepupto(min_seconds, max_seconds=None): interval = random.uniform(min_seconds if max_seconds is not None else 0, max_seconds if max_seconds is not None else min_seconds) start_time = time.time() end_time = start_time + interval with tqdm(total=interval, desc=f"Sleeping for {format_duration(interval)}", unit=" sec", ncols=75, bar_format='{desc}: {bar} {remaining}') as pbar: while True: current_time = time.time() elapsed_time = current_time - start_time remaining_time = end_time - current_time if elapsed_time >= interval: break duration = min(1, interval - elapsed_time) # Adjust sleep time to not exceed interval time.sleep(duration) pbar.update(duration) # Update remaining time display pbar.set_postfix_str(f"{format_duration(remaining_time)} remaining") ######################## ### GHOST FUNCTIONS ### ######################## def generate_jwt_token(): id, secret = GHOST_API_KEY.split(':') iat = int(date.now().timestamp()) header = {'alg': 'HS256', 'typ': 'JWT', 'kid': id} payload = {'iat': iat, 'exp': iat + 5 * 60, 'aud': '/admin/'} token = jwt.encode(payload, bytes.fromhex(secret), algorithm='HS256', headers=header) return token.decode('utf-8') if isinstance(token, bytes) else token def post_to_ghost(title, image_path, html_content, ghost_tags): jwt_token = generate_jwt_token() # Prepare headers for the request ghost_headers = {'Authorization': f'Ghost {jwt_token}'} # Open the image file in binary mode with open(image_path, 'rb') as f: files = {'file': (image_path.split('/')[-1], f, 'image/jpeg')} # Adjust the MIME type if necessary image_response = requests.post(f"{GHOST_API_URL}/images/upload/", headers=ghost_headers, files=files) image_response.raise_for_status() # Make sure the request was successful image_url = image_response.json()['images'][0]['url'] # Extract the URL of the uploaded image # Embed the image URL in the Mobiledoc content mobiledoc = { "version": "0.3.1", "atoms": [], "cards": [["html", {"cardName": "html", "html": html_content}]], "markups": [], "sections": [[10, 0]] } image_card = ["image", {"src": image_url}] mobiledoc["cards"].append(image_card) mobiledoc["sections"].append([10, len(mobiledoc["cards"]) - 1]) mobiledoc = json.dumps(mobiledoc) post_data = { 'posts': [{ 'title': title, 'mobiledoc': mobiledoc, 'status': 'published' }] } post_response = requests.post(f"{GHOST_API_URL}/posts/", json=post_data, headers=ghost_headers) post_response.raise_for_status() post_url = post_response.json()['posts'][0]['url'] return post_url ######################################################## ######################################################## ######################################################## if __name__ == "__main__": current_unix_time = int(date.now().timestamp()) if args.local: if args.postfile: handle_single_image(args.postfile, args.posttype) else: while True: handle_image_workflow(args.posttype) else: time_since_rollover = current_unix_time - rollover_time time_remaining = 30 - (time_since_rollover % 30) if time_remaining < 4: print("Too close to end of TOTP counter. Waiting.") sleepupto(5, 5) if not args.newsession and os.path.exists(IG_SESSION_PATH): cl.load_settings(IG_SESSION_PATH) print("Loaded past session.") elif args.newsession and cl.login(IG_USERNAME, IG_PASSWORD, verification_code=TOTP.now()): cl.dump_settings(IG_SESSION_PATH) print("Logged in and saved new session.") else: raise Exception(f"Failed to login as {IG_USERNAME}.") if IMAGE_URL: comment_on_specific_media(IMAGE_URL, args.commenttype) elif args.user: comment_on_user_media(args.user, args.count) else: if args.postfile: handle_single_image(args.postfile, args.posttype) else: while True: if args.commentonly: print(f"Generating a comment about {args.commenttype}") comment_by_type(args.commenttype, args.count, args.hashtag) sleep(180) elif args.postonly: print(f"Generating a post about {args.posttype}") handle_image_workflow(args.posttype) sleep(360) else: if random.random() < 0.75: if args.commenttype: print(f"Generating a comment about {args.commenttype}") comment_by_type(args.commenttype, 1) sleep(180) else: chosen_comment = random.choice(list(ACCOUNT_CONFIG["comments"].keys())) print(f"Generating a comment about {chosen_comment}") comment_by_type(chosen_comment) sleep(180) else: if args.posttype: print(f"Generating a post about {args.posttype}") handle_image_workflow(args.posttype) sleep(360) else: chosen_post = random.choice(list(ACCOUNT_CONFIG["posts"].keys())) print(f"Generating a post about {chosen_post}") handle_image_workflow(chosen_post) sleep(360)