Horhik/Instagram2Fedi
1
0
Fork 0
mirror of https://github.com/Horhik/Instagram2Fedi.git synced 2025-04-08 13:26:19 +00:00
Code Issues Projects Releases Wiki Activity

Merge 27f9922266 into f5cceda106

Browse source
This commit is contained in:
J. C. Müller 2025-03-23 20:04:42 +00:00 • committed by GitHub
commit 5a5a71c1f5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 367 additions and 258 deletions

4
.gitignore vendored
View file

@ -1,8 +1,10 @@
.idea/
*~
already_posted.txt
config/
src/__pycache__/
.venv
.venv/
.env.sh
docker-compose.yml
docker-compose.yaml
manual.xsh

13
Dockerfile
View file

@ -1,24 +1,21 @@
FROM python:3.9
FROM python:3.13
COPY . /app
COPY ./requirements.txt /app/requirements.txt
WORKDIR /app
RUN pip install -r requirements.txt
ENV USE_DOCKER=1
COPY ./src /app/src
ENV YOUR_CONTAINER_NAME="$YOUR_CONTAINER_NAME"
ENV I2M_INSTAGRAM_USER="$I2M_INSTAGRAM_USER"
ENV I2M_INSTANCE="$I2M_INSTANCE"
ENV I2M_TOKEN="$I2M_TOKEN"
ENV I2M_CHECK_INTERVAL "$I2M_CHECK_INTERVAL"
ENV I2M_POST_INTERVAL "$I2M_POST_INTERVAL"
ENV I2M_USE_MASTODON "$I2M_USE_MASTODON"
ENV I2M_FETCH_COUNT "$I2M_FETCH_COUNT"
ENV PYTHONUNBUFFERED 1
#ENTRYPOINT ["python", "/app/src/main.py", "--instagram-user", I2M_INSTAGRAM_USER, "--instance", I2M_INSTANCE, "--token", I2M_TOKEN, "--check-interval", I2M_CHECK_INTERVAL, "--post-interval", I2M_POST_INTERVAL, "--fetch-count", I2M_FETCH_COUNT, "--use-mastodon", I2M_USE_MASTODON]
#ENTRYPOINT ["echo", "--instagram-user", I2M_INSTAGRAM_USER, "--instance", I2M_INSTANCE, "--token", I2M_TOKEN, "--check-interval", I2M_CHECK_INTERVAL, "--post-interval", I2M_POST_INTERVAL, "--fetch-count", I2M_FETCH_COUNT, "--use-mastodon", I2M_USE_MASTODON]

8
LICENSE
View file

@ -631,8 +631,8 @@ to attach them to the start of each source file to most effectively
state the exclusion of warranty; and each file should have at least
the "copyright" line and a pointer to where the full notice is found.
"Instagram2Fedi": a decentralized messenger over freenet.
Copyright (C) 2021 Osokin George
"Instagram2Fedi": an automated tool to cross-post from Instagram to Mastodon or Pixelfed
Copyright (C) 2024 George Osokin, Marc "T0K_" Damie
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
@ -652,7 +652,7 @@ Also add information on how to contact you by electronic and paper mail.
If the program does terminal interaction, make it output a short
notice like this when it starts in an interactive mode:
Instagram2Fedi Copyright (C) 2021 George Osokin
Instagram2Fedi Copyright (C) 2024 George Osokin, Marc "T0K_" Damie
This program comes with ABSOLUTELY NO WARRANTY; for details type `show w'.
This is free software, and you are welcome to redistribute it
under certain conditions; type `show c' for details.
@ -672,5 +672,3 @@ may consider it more useful to permit linking proprietary applications with
the library. If this is what you want to do, use the GNU Lesser General
Public License instead of this License. But first, please read
<https://www.gnu.org/licenses/why-not-lgpl.html>.

39
README.md
View file

@ -1,8 +1,10 @@
_Guys... Instagram is sh*t. Even [bibliogram](https://www.reddit.com/r/privacy/comments/wrczxc/bibliogram_is_being_discontinued/) is being discontinued. If you're able to migrate you proile to any fediverse instance or contact to person, whose instagram you'd like to crosspost and ask him to post to fediverse to, it wil be the best desicion_
_Guys... Instagram is sh*t. Even [bibliogram](https://www.reddit.com/r/privacy/comments/wrczxc/bibliogram_is_being_discontinued/)
is being discontinued. If you're able to migrate you profile to any fediverse instance or contact to person, whose
instagram you'd like to cross-post and ask him to post to fediverse to, it wil be the best decision_
# Instagram2Fedi <span><img width="50px" src="https://upload.wikimedia.org/wikipedia/commons/9/93/Fediverse_logo_proposal.svg"></span>
Simple tool for crossposting posts from instagram to Mastodon/Pixelfed.
Simple tool for cross-posting posts from instagram to Mastodon/Pixelfed.
## Using without docker
See [Docs.md](./Docs.md)
@ -18,20 +20,30 @@ services:
build:
context: .
image: "horhik/instagram2fedi:latest"
volumes:
- ${HOME}/.config/instaloader:/root/.config/instaloader
- ./config:/app/config
environment:
- YOUR_CONTAINER_NAME=<whatever>
- I2M_INSTAGRAM_USER=<instgram username>
- I2M_INSTANCE=<mastodon or pixelfed instance>
- I2M_TOKEN=<your token here>
- I2M_CHECK_INTERVAL=3600 #1 hour
- I2M_POST_INTERVAL=3600 #1 hour
- YOUR_CONTAINER_NAME= #<whatever>
- I2M_INSTAGRAM_USER= #<instagram username to be fetched>
- I2M_INSTANCE= #<Mastodon or Pixelfed instance> Leave blank if you use src/create_credentials.py.
- I2M_TOKEN= #<SECRET MASTODON TOKEN filename> Leave blank if you use src/create_credentials.py.
- I2M_CHECK_INTERVAL=3600 #1 hour
- I2M_POST_INTERVAL=3600 #1 hour
- I2M_USE_MASTODON=4 #max carouse - is 4, if there's no limit set to -1
- I2M_FETCH_COUNT=5 # how many instagram posts to fetch per check_interval -
- I2M_USER_NAME=admin # Your instagram login name
- I2M_USER_PASSWORD=admin # Your instagram password
- I2M_FETCH_COUNT=5 # how many instagram posts to fetch per check_interval
- I2M_USER_NAME= #<Your instagram login name>
- I2M_USER_PASSWORD= #<Your instagram password> Not needed if ~/.config/instaloader/session-${I2M_USER_NAME} exists.
```
** Note: ** _Since somewhen it's seems not possible to fetch any data from instagram anonymously (maybe i'm wrong and there's a solution, I'll be very happy to know about it). Due that you unfortunately have to had an instagram accound and provide login and password to this script_
> ** Note: ** _Since some time it seems to be not possible to fetch any data from Instagram anonymously (maybe I'm wrong
and there's a solution, I'll be very happy to know about it). Due to that you unfortunately need an Instagram account
and provide login and password to this script_
> ** Note: ** _Instagram may block login attempts from this script when using username and password,
and may also suspend the account. Instead of using username and password from within the script, use your
browser to log into Instagram, e.g. Firefox, then extract the session information from the browser using the command
`instaloader --load-cookies=firefox`. You can leave the password I2M_USER_PASSWORD blank.
2. And edit environment variables
@ -52,6 +64,3 @@ source ./run.sh
![screenshot](./img.png)

19
default.docker-compose.yaml
View file

@ -4,14 +4,17 @@ services:
build:
context: .
#image: "horhik/instagram2fedi:latest"
volumes:
- ${HOME}/.config/instaloader:/root/.config/instaloader
- ./config:/app/config
environment:
- YOUR_CONTAINER_NAME=instagram2fedi
- I2M_INSTAGRAM_USER= #<fetched instagram user name>
- I2M_INSTANCE= #<Mastodon or pixelfed instance>
- I2M_TOKEN= # SECRET TOKEN
- I2M_CHECK_INTERVAL=3600 #1 hour
- I2M_POST_INTERVAL=3600 #1 hour
- I2M_INSTAGRAM_USER= #<instagram username to be fetched>
- I2M_INSTANCE= #<Mastodon or Pixelfed instance> Leave blank if you use src/create_credentials.py.
- I2M_TOKEN= #<SECRET MASTODON TOKEN filename> Leave blank if you use src/create_credentials.py.
- I2M_CHECK_INTERVAL=3600 #1 hour
- I2M_POST_INTERVAL=3600 #1 hour
- I2M_USE_MASTODON=4 #max carouse - is 4, if there's no limit set to -1
- I2M_FETCH_COUNT=5 # how many instagram posts to fetch per check_interval -
- I2M_USER_NAME= # Your instagram login name
- I2M_USER_PASSWORD= # Your instagram password
- I2M_FETCH_COUNT=5 # how many instagram posts to fetch per check_interval
- I2M_USER_NAME= #<Your instagram login name>
- I2M_USER_PASSWORD= #<Your instagram password> Not needed if ~/.config/instaloader/session-${I2M_USER_NAME} exists.

24
requirements.txt
View file

@ -1,14 +1,14 @@
blurhash==1.1.4
certifi==2022.6.15
charset-normalizer==2.1.0
colorama==0.4.5
decorator==5.1.1
idna==3.3
instaloader==4.9.3
Mastodon.py==1.5.1
python-dateutil==2.8.2
browser-cookie3==0.20.1
certifi==2025.1.31
charset-normalizer==3.4.1
colorama==0.4.6
decorator==5.2.1
idna==3.10
instaloader==4.14.1
Mastodon.py==2.0.1
python-dateutil==2.9.0.post0
python-magic==0.4.27
pytz==2022.1
requests==2.28.1
six==1.16.0
urllib3==1.26.11
requests==2.32.3
six==1.17.0
urllib3==2.3.0

14
src/already_posted.py
View file

@ -1,14 +1,16 @@
import hashlib
def already_posted(id, path):
def already_posted(identifier, path):
with open(path) as file:
content = file.read().split("\n")
sha1 = hashlib.sha1(bytes(id, "utf-8")).hexdigest()
sha1 = hashlib.sha1(bytes(identifier, "utf-8")).hexdigest()
if sha1 in content:
return True
return False
def mark_as_posted(id, path):
with open(path, 'a') as file:
sha1 = hashlib.sha1(bytes(id, "utf-8")).hexdigest()
file.write(sha1+'\n')
def mark_as_posted(identifier, path):
with open(path, "a") as file:
sha1 = hashlib.sha1(bytes(identifier, "utf-8")).hexdigest()
file.write(sha1 + "\n")

139
src/arguments.py
View file

@ -1,107 +1,108 @@
# -*- coding: utf-8 -*-
import os
import datetime
from colorama import Fore, Back, Style
"""Functions to process the arguments and the environment variables."""
import os
from util import print_log
instagram_user = os.environ.get("I2M_INSTAGRAM_USER")
user_name = os.environ.get("I2M_USER_NAME")
user_password = os.environ.get("I2M_USER_PASSWORD")
instance = os.environ.get("I2M_INSTANCE")
token = os.environ.get("I2M_TOKEN")
check_interval = os.environ.get("I2M_CHECK_INTERVAL") #1 hour
post_interval = os.environ.get("I2M_POST_INTERVAL") #1 hour
use_mastodon = os.environ.get("I2M_USE_MASTODON") #max carousel is 4, if there's no limit set to -1
fetch_count = os.environ.get("I2M_FETCH_COUNT") # how many instagram posts to fetch per check_interval
if os.environ.get("I2M_SCHEDULED") == "True":
scheduled_run = True # run continuously (if False) or a single time (if True)
else:
scheduled_run = False
if os.environ.get("I2M_VERBOSE") == "True": # verbose output
verbose_output = True
else:
verbose_output = False
check_interval = os.environ.get("I2M_CHECK_INTERVAL") # 1 hour
post_interval = os.environ.get("I2M_POST_INTERVAL") # 1 hour
use_mastodon = os.environ.get(
"I2M_USE_MASTODON"
) # max carousel is 4, if there's no limit set to -1
fetch_count = os.environ.get(
"I2M_FETCH_COUNT"
) # how many instagram posts to fetch per check_interval
if verbose_output:
print('instagram', instagram_user)
print('instagram', instance)
print(token)
print(check_interval)
print(post_interval)
print(use_mastodon)
print(fetch_count)
print(user_name)
print(user_password)
print(scheduled_run)
print(verbose_output)
# run continuously (if False) or a single time (if True)
SCHEDULED_RUN = os.environ.get("I2M_SCHEDULED", '').lower() == "true"
VERBOSE_OUTPUT = os.environ.get("I2M_VERBOSE", '').lower() == "true"
if VERBOSE_OUTPUT:
print_log("instagram", instagram_user)
print_log("instagram", instance)
print_log(token)
print_log(check_interval)
print_log(post_interval)
print_log(use_mastodon)
print_log(fetch_count)
print_log(user_name)
print_log(user_password)
print_log(SCHEDULED_RUN)
print_log(VERBOSE_OUTPUT)
def flags(args, defaults):
"""Process the flags and update the setting dictionary."""
count = 1
while (len(args) > count):
if(args[count] == "--instance"):
while len(args) > count:
if args[count] == "--instance":
defaults["instance"] = args[count + 1]
elif (args[count] == "--instagram-user"):
elif args[count] == "--instagram-user":
defaults["instagram-user"] = args[count + 1]
elif (args[count] == "--token"):
elif args[count] == "--token":
defaults["token"] = args[count + 1]
elif (args[count] == "--check-interval"):
elif args[count] == "--check-interval":
defaults["check-interval"] = int(args[count + 1])
elif (args[count] == "--post-interval"):
elif args[count] == "--post-interval":
defaults["post-interval"] = int(args[count + 1])
elif (args[count] == "--fetch-count"):
elif args[count] == "--fetch-count":
defaults["fetch-count"] = int(args[count + 1])
elif (args[count] == "--use-mastodon"):
elif args[count] == "--use-mastodon":
defaults["carousel-limit"] = int(args[count + 1])
elif (args[count] == "--use-docker"):
elif args[count] == "--use-docker":
defaults["use-docker"] = args[count + 1]
elif (args[count] == "--user-name"):
elif args[count] == "--user-name":
defaults["user-name"] = args[count + 1]
elif (args[count] == "--user-password"):
elif args[count] == "--user-password":
defaults["user-password"] = args[count + 1]
elif (args[count] == "--scheduled"):
elif args[count] == "--scheduled":
defaults["scheduled"] = True
count -= 1
elif (args[count] == "--verbose"):
elif args[count] == "--verbose":
defaults["verbose"] = True
count -= 1
else:
print(Fore.RED + '❗ -> Wrong Argument Name!...')
print(Style.RESET_ALL)
print(datetime.datetime.now())
print_log("❗ -> Wrong Argument Name!...", color="red")
count +=2
count += 2
return defaults
def check_defaults(arg):
return arg if arg != '' and arg else None
def process_arguments(args, defaults):
defaults["instance"] = instance if instance !='' and instance else None
defaults["instagram-user"] = instagram_user if instagram_user != '' and instagram_user else None
# Users login and password
defaults["user-name"] = check_defaults(user_name)
defaults["user-password"] = check_defaults(user_password)
defaults["token"] = token if token != '' and token else None
defaults["check-interval"] = int(check_interval) if check_interval != '' and check_interval else None
defaults["post-interval"] = int(post_interval) if post_interval != '' and post_interval else None
defaults["fetch-count"] = int(fetch_count) if fetch_count != '' and fetch_count else None
defaults["carousel-limit"] = int(use_mastodon) if use_mastodon != '' and use_mastodon else None
defaults["scheduled"] = bool(scheduled_run) if scheduled_run else False
defaults["verbose"] = bool(verbose_output) if verbose_output else False
#print(Fore.RED + '❗ -> Missing Argument ')
#print(Style.RESET_ALL)
#print(datetime.datetime.now())
"""Process the arguments and update the setting dictionary."""
if instance:
defaults["instance"] = instance
if instagram_user:
defaults["instagram-user"] = instagram_user
if user_name:
defaults["user-name"] = user_name
if user_password:
defaults["user-password"] = user_password
if token:
defaults["token"] = token
if check_interval:
defaults["check-interval"] = int(check_interval)
if post_interval:
defaults["post-interval"] = int(post_interval)
if fetch_count:
defaults["fetch-count"] = int(fetch_count)
if use_mastodon:
defaults["carousel-limit"] = int(use_mastodon)
if SCHEDULED_RUN:
defaults["scheduled"] = SCHEDULED_RUN
if VERBOSE_OUTPUT:
defaults["verbose"] = VERBOSE_OUTPUT
# Command line arguments more prioritized, if smth has been written in .env and in cmd args, then Instagram2Fedi will take values from `cmd args`
# Command line arguments more prioritized,
# if smth has been written in .env and in cmd args,
# then Instagram2Fedi will take values from `cmd args`
new_defaults = flags(args, defaults)
return new_defaults

60
src/converters.py
View file

@ -1,52 +1,56 @@
# -*- coding: utf-8 -*-
from colorama import Fore, Back, Style
import datetime
"""Functions to handle videos and caroussel"""
from util import print_log
def split_array(arr, size):
"""Split an array into sub-arrays of specific size."""
count = len(arr) // size + 1
new_arr = []
for i in range(count):
new_arr.append(arr[i*size:(i+1)*size])
new_arr.append(arr[i * size : (i + 1) * size])
return new_arr
def try_to_get_carousel(array, post):
"""Extract videos or caroussel from an Instagram post."""
try:
node = vars(post)['_node']
if 'edge_sidecar_to_children' in node:
node = vars(post)["_node"]
if "edge_sidecar_to_children" in node:
try:
urls = list(map(lambda arr: arr['node']['display_url'], node['edge_sidecar_to_children']['edges']))
print(Fore.GREEN + "🎠 > Found carousel!")
print(Style.RESET_ALL)
print(datetime.datetime.now())
urls = list(
map(
lambda arr: arr["node"]["display_url"],
node["edge_sidecar_to_children"]["edges"],
)
)
print_log("🎠 > Found carousel!", color="green")
return urls
except Exception as e:
print(Fore.RED + "🎠💥 > No carousel :( \n", e)
print(Style.RESET_ALL)
print(datetime.datetime.now())
except KeyError as err:
print_log("🎠💥 > No carousel ", color="red")
print_log(err)
return array
else:
print(Fore.YELLOW + "🎠💥 > No carousel\n")
print_log("🎠💥 > No carousel", color="yellow")
# We can also have video in a separate key
if 'is_video' in node and node ['is_video']:
if "is_video" in node and node["is_video"]:
try:
urls = [node['video_url']]
print(Fore.GREEN + "🎞 > Found video!")
print(Style.RESET_ALL)
print(datetime.datetime.now())
urls = [node["video_url"]]
print_log("🎞 > Found video!", color="green")
return urls
except Exception as e:
print(Fore.RED + "🎞💥 > No video :( \n", e)
print(Style.RESET_ALL)
print(datetime.datetime.now())
except KeyError as err:
print_log("🎞💥 > No video :(", color="red")
print_log(err)
return array
else:
print(Fore.YELLOW + "🎠💥 > No video\n")
print_log("🎠💥 > No video", color="yellow")
except Exception as e:
print(Fore.RED + "😱💥 > No node :( \n", e)
print(Style.RESET_ALL)
print(datetime.datetime.now())
except KeyError as err:
print_log("😱💥 > No node :(", color="red")
print_log(err)
return array
return array

32
src/create_credentials.py Normal file
View file

@ -0,0 +1,32 @@
"""
Run this script once, to create the Mastodon login credentials for Instagram2Fedi.
More info: https://mastodonpy.readthedocs.io/en/stable/
"""
import os
from mastodon import Mastodon
CONFIG_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "config"))
CLIENT_CREDENTIALS = os.path.join(CONFIG_DIR, "client_credentials.secret")
USER_CREDENTIALS = os.path.join(CONFIG_DIR, "user_credentials.secret")
if __name__ == "__main__":
os.makedirs(CONFIG_DIR, exist_ok=True)
base_url = input("Enter the Mastodon base URL (e.g. mastodon.social):")
Mastodon.create_app(
'Instagram2Fedi',
api_base_url=f'https://{base_url}',
to_file=CLIENT_CREDENTIALS
)
mastodon = Mastodon(client_id=CLIENT_CREDENTIALS, )
print("Open the following URL in the browser and paste the code you get:")
print(mastodon.auth_request_url())
mastodon.log_in(
code=input("Enter the OAuth authorization code:"),
to_file=USER_CREDENTIALS
)
print(f"Client and user credentials stored in: {CONFIG_DIR}")

79
src/main.py
View file

@ -1,30 +1,29 @@
# -*- coding: utf-8 -*-
"""Script to cross-post from Instagram to Mastodon/Pixelfed"""
import os
import sys
import time
import datetime
import json
from mastodon import Mastodon
from colorama import Fore, Back, Style
from instaloader import Profile, Instaloader, LatestStamps
from arguments import process_arguments
from create_credentials import CONFIG_DIR, USER_CREDENTIALS
from network import get_new_posts
from util import print_log
default_settings = {
"instance": None,
"instagram-user": None,
"user-name": "",
"user-name": None,
"user-password": None,
"token": None,
"token": USER_CREDENTIALS,
"check-interval": 3600,
"post-interval": 60,
"fetch-count" : 10,
"fetch-count": 10,
"carousel-limit": 4,
"scheduled": False,
"verbose": False
"verbose": False,
}
settings = process_arguments(sys.argv, default_settings)
@ -32,50 +31,52 @@ settings = process_arguments(sys.argv, default_settings)
verbose = settings["verbose"]
if verbose:
print("ARGUMENTS")
print(sys.argv)
print('SETTINGS' , settings)
print_log(f"SETTINGS {settings}")
agree = [1, True, "true", "True", "yes", "Yes"]
if (os.environ.get("USE_DOCKER")):
id_filename = "/app/already_posted.txt"
elif (os.environ.get("USE_KUBERNETES")):
id_filename = "/data/already_posted.txt"
if os.environ.get("USE_KUBERNETES"):
ID_FILENAME = "/data/already_posted.txt"
else:
id_filename = "./already_posted.txt"
os.makedirs(CONFIG_DIR, exist_ok=True)
ID_FILENAME = os.path.join(CONFIG_DIR, "already_posted.txt")
with open(id_filename, "a") as f:
with open(ID_FILENAME, "a", encoding="utf-8") as f:
f.write("\n")
fetched_user = settings["instagram-user"]
mastodon_instance = settings["instance"]
mastodon_token = settings["token"]
mastodon_token = os.path.abspath(settings["token"])
post_limit = settings["fetch-count"]
time_interval_sec = settings["check-interval"] #1d
post_interval = settings["post-interval"]#1m
time_interval_sec = settings["check-interval"]
post_interval = settings["post-interval"]
using_mastodon = settings["carousel-limit"] > 0;
using_mastodon = settings["carousel-limit"] > 0
mastodon_carousel_size = settings["carousel-limit"]
scheduled = settings["scheduled"]
user = {"name": settings["user-name"], "password": settings["user-password"]}
user = {
"name": settings["user-name"],
"password": settings["user-password"]
}
print_log("🚀 > Connecting to Mastodon/Pixelfed..", color="green")
if not os.path.isfile(mastodon_token):
print_log(f'Could not find Mastodon token file: {mastodon_token}', color='red')
sys.exit(1)
mastodon_client = Mastodon(access_token=mastodon_token, api_base_url=mastodon_instance)
print(Fore.GREEN + '🚀 > Connecting to Mastodon/Pixelfed...')
print(Style.RESET_ALL)
print(datetime.datetime.now())
mastodon = Mastodon(
access_token = mastodon_token,
api_base_url = mastodon_instance
# api_base_url = 'https://pixelfed.tokyo/'
)
while True:
get_new_posts(mastodon, mastodon_carousel_size, post_limit, id_filename, using_mastodon, mastodon_carousel_size, post_interval, fetched_user, user)
get_new_posts(
mastodon_client,
mastodon_carousel_size,
post_limit,
ID_FILENAME,
using_mastodon,
mastodon_carousel_size,
post_interval,
fetched_user,
user,
)
if scheduled:
break
print_log(f"⏳ > Sleeping for {time_interval_sec/3600.0:.1f} hours...", color="green")
time.sleep(time_interval_sec)

176
src/network.py
View file

@ -1,110 +1,152 @@
# -*- coding: utf-8 -*-
from colorama import Fore, Back, Style
import requests
"""Functions to interact with Instagram and Mastodon."""
import time
import datetime
from pathlib import Path
import requests
from instaloader import Instaloader, Profile
from instaloader.exceptions import QueryReturnedBadRequestException, ConnectionException
from mastodon import MastodonError
from already_posted import already_posted, mark_as_posted
from converters import split_array, try_to_get_carousel
import hashlib
from instaloader import Profile, Instaloader, LatestStamps
from util import print_log
def get_instagram_user(user, fetched_user):
L = Instaloader()
"""Fetch the target Instagram account.
An authentication attempt is performed if some credentials were provided"""
loader = Instaloader()
print_log("🚀 > Connecting to Instagram...", color="green")
print(Fore.GREEN + 'TEST 🚀 > Connecting to Instagram...')
print(Style.RESET_ALL)
print(datetime.datetime.now())
if user["name"] is not None:
print_log("User " + user["name"])
session_file = str(Path.home().joinpath('.config', 'instaloader', f'session-{user["name"]}'))
try:
loader.load_session_from_file(user["name"], session_file)
try:
assert user["name"] == loader.test_login()
except QueryReturnedBadRequestException:
print_log(
"Instagram requires a human verification... "
+ "connect via a browser to solve a captcha.",
color="red",
)
input("Press ENTER once the captcha is solved.")
assert user["name"] == loader.test_login()
except ConnectionException:
print_log(
"Invalid session (probably flagged as bot by Instagram)...",
color="red",
)
raise
print_log("Restored the session")
except (FileNotFoundError, ConnectionException):
print_log(
"Found no valid session... authentication attempt", color="yellow"
)
loader.login(user["name"], user["password"])
print_log("Authentication successful", color="green")
loader.save_session_to_file(session_file)
return Profile.from_username(loader.context, fetched_user)
if user["name"] != None:
print("USER USER USER!!!!!!!!!!!!!", user["name"])
L.login(user["name"], user["password"])
return Profile.from_username(L.context, fetched_user)
def get_image(url):
try:
print(Fore.YELLOW + "🚀 > Downloading Image...", url)
print(Style.RESET_ALL)
print(datetime.datetime.now())
"""Download an image from Instagram."""
response = requests.get(url)
try:
print_log("🚀 > Downloading Image... " + url, color="yellow")
response = requests.get(url, timeout=60)
response.raw.decode_content = True
print(Fore.GREEN + "✨ > Downloaded!")
print(Style.RESET_ALL)
print(datetime.datetime.now())
print_log("✨ > Downloaded!", color="green")
return response.content
except Exception as e:
print(Fore.RED + "💥 > Failed to download image. \n", e)
print(Style.RESET_ALL)
print(datetime.datetime.now())
except requests.exceptions.RequestException as err:
print_log("💥 > Failed to download image: " +url, color="red")
print_log(err)
raise
def upload_image_to_mastodon(url, mastodon):
def upload_image_to_mastodon(url, mastodon_client):
"""Upload an Instagram image to Mastodon."""
try:
print(Fore.YELLOW + "🐘 > Uploading Image...")
print(Style.RESET_ALL)
print(datetime.datetime.now())
media = mastodon.media_post(media_file = get_image(url), mime_type = "image/jpeg") # sending image to mastodon
print(Fore.GREEN + "✨ > Uploaded!")
print(Style.RESET_ALL)
print(datetime.datetime.now())
print_log("🐘 > Uploading Image...", color="yellow")
media = mastodon_client.media_post(
media_file=get_image(url), mime_type="image/jpeg"
) # sending image to mastodon
print_log("✨ > Uploaded!", color="green")
return media["id"]
except Exception as e:
print(Fore.RED + "💥 > failed to upload image to mastodon. \n", e)
print(Style.RESET_ALL)
print(datetime.datetime.now())
except MastodonError as err:
print_log("💥 > Failed to upload image to Mastodon", color="red")
print_log(err)
raise
except requests.exceptions.RequestException as _err:
print_log("💥 > Image not downloaded... cancel the post.", color="red")
raise
def toot(urls, title, mastodon_client):
"""Create toots from Instagram posts."""
def toot(urls, title, mastodon, fetched_user ):
try:
print(Fore.YELLOW + "🐘 > Creating Toot...", title)
print(Style.RESET_ALL)
print(datetime.datetime.now())
print_log("🐘 > Creating Toot..." + title, color="yellow")
ids = []
for url in urls:
ids.append(upload_image_to_mastodon(url, mastodon))
post_text = str(title) + "\n" # creating post text
ids.append(upload_image_to_mastodon(url, mastodon_client))
post_text = str(title) + " #bot #crosspost" + "\n" # creating post text
post_text = post_text.replace("@", "[at]")
post_text = post_text[0:1000]
if(ids):
print(ids)
mastodon.status_post(post_text, media_ids = ids)
except Exception as e:
print(Fore.RED + "😿 > Failed to create toot \n", e)
print(Style.RESET_ALL)
print(datetime.datetime.now())
if ids:
print_log("Post identifiers:" + str(ids))
mastodon_client.status_post(post_text, media_ids=ids)
except (MastodonError, requests.exceptions.RequestException):
print_log("😿 > Failed to create toot", color="red")
def get_new_posts(
mastodon_client,
_mastodon_carousel_size, # TODO: remove or use it
post_limit,
already_posted_path,
using_mastodon,
carousel_size,
post_interval,
fetched_user,
user,
):
"""Fetch new Instagram posts and create toots."""
def get_new_posts(mastodon, mastodon_carousel_size, post_limit, already_posted_path, using_mastodon, carousel_size, post_interval, fetched_user, user):
# fetching user profile to get new posts
profile = get_instagram_user(user, fetched_user)
# get list of all posts
posts = profile.get_posts()
stupidcounter = 0
stupid_counter = 0
for post in posts:
url_arr = try_to_get_carousel([post.url], post)
# checking only `post_limit` last posts
if stupidcounter < post_limit:
stupidcounter += 1
if stupid_counter < post_limit:
stupid_counter += 1
if already_posted(str(post.mediaid), already_posted_path):
print(Fore.YELLOW + "🐘 > Already Posted ", post.url)
print(Style.RESET_ALL)
print(datetime.datetime.now())
continue
print("Posting... ", post.url)
print(datetime.datetime.now())
print_log("🐘 > Already Posted " + post.url, color="yellow")
break # Do not need to go back further in time
print_log("Posting... " + post.url)
if using_mastodon:
urls_arr = split_array(url_arr, carousel_size)
for urls in urls_arr:
toot(urls, post.caption, mastodon, fetched_user)
toot(urls, post.caption, mastodon_client)
else:
toot(url_arr, post.caption, mastodon, fetched_user)
toot(url_arr, post.caption, mastodon_client)
mark_as_posted(str(post.mediaid), already_posted_path)
time.sleep(post_interval)
else:
break
print(Fore.GREEN + "✨ > Fetched All")
print(Style.RESET_ALL)
print(datetime.datetime.now())
print_log("✨ > Fetched All", color="green")

18
src/util.py Normal file
View file

@ -0,0 +1,18 @@
from colorama import Fore, Style
import datetime
color_dict = {
"red": Fore.RED,
"green": Fore.GREEN,
"blue": Fore.BLUE,
"yellow": Fore.YELLOW,
"white": Fore.WHITE,
}
def print_log(message, color="white"):
if color not in color_dict:
raise ValueError("Unknown log color: " + color)
print(
f"[{datetime.datetime.now()}] {color_dict[color]}{message}{Style.RESET_ALL}"
)