Refactord the logging + store session files

.gitignore

@@ -6,5 +6,5 @@ src/__pycache__/
 .env.sh
 docker-compose.yml
 docker-compose.yaml
-
+*.sqlite
 manual.xsh

src/already_posted.py

@@ -1,16 +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):
+def mark_as_posted(identifier, path):
     with open(path, "a") as file:
-        sha1 = hashlib.sha1(bytes(id, "utf-8")).hexdigest()
+        sha1 = hashlib.sha1(bytes(identifier, "utf-8")).hexdigest()
         file.write(sha1 + "\n")

src/arguments.py

@@ -1,8 +1,7 @@
 # -*- coding: utf-8 -*-
-import datetime
 import os
 
-from colorama import Back, Fore, Style
+from util import print_log
 
 instagram_user = os.environ.get("I2M_INSTAGRAM_USER")
 user_name = os.environ.get("I2M_USER_NAME")
@@ -76,9 +75,7 @@ def flags(args, defaults):
             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
     return defaults
@@ -111,9 +108,6 @@ def process_arguments(args, defaults):
     )
     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())
 
     # 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)

src/converters.py

@@ -1,7 +1,6 @@
 # -*- coding: utf-8 -*-
-import datetime
 
-from colorama import Back, Fore, Style
+from util import print_log
 
 
 def split_array(arr, size):
@@ -23,37 +22,30 @@ def try_to_get_carousel(array, post):
                         node["edge_sidecar_to_children"]["edges"],
                     )
                 )
-                print(Fore.GREEN + "🎠 > Found carousel!")
-                print(Style.RESET_ALL)
-                print(datetime.datetime.now())
+                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 Exception 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"]:
             try:
                 urls = [node["video_url"]]
-                print(Fore.GREEN + "🎞 > Found video!")
-                print(Style.RESET_ALL)
-                print(datetime.datetime.now())
+                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 Exception 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 Exception as err:
+        print_log("😱💥 > No node :(", color="red")
+        print_log(err)
         return array
     return array

src/main.py

@@ -1,16 +1,14 @@
 # -*- coding: utf-8 -*-
-import datetime
-import json
+
 import os
 import sys
 import time
 
-from colorama import Back, Fore, Style
-from instaloader import Instaloader, LatestStamps, Profile
 from mastodon import Mastodon
 
 from arguments import process_arguments
 from network import get_new_posts
+from util import print_log
 
 default_settings = {
     "instance": None,
@@ -62,10 +60,9 @@ scheduled = settings["scheduled"]
 
 user = {"name": settings["user-name"], "password": settings["user-password"]}
 
-print(Fore.GREEN + "🚀 > Connecting to Mastodon/Pixelfed.."')
-print(Style.RESET_ALL)
-print(datetime.datetime.now())
+print_log("🚀 > Connecting to Mastodon/Pixelfed..", color="green")
 mastodon = Mastodon(access_token=mastodon_token, api_base_url=mastodon_instance)
+
 while True:
     get_new_posts(
         mastodon,

src/network.py

@@ -1,72 +1,68 @@
 # -*- coding: utf-8 -*-
-import datetime
-import hashlib
+
 import time
 
 import requests
-from colorama import Back, Fore, Style
-from instaloader import Instaloader, LatestStamps, Profile
+from instaloader import Instaloader, Profile
 
 from already_posted import already_posted, mark_as_posted
 from converters import split_array, try_to_get_carousel
+from util import print_log
 
 
 def get_instagram_user(user, fetched_user):
-    L = Instaloader()
+    loader = Instaloader()
 
-    print(Fore.GREEN + "🚀 > Connecting to Instagram...")
-    print(Style.RESET_ALL)
-    print(datetime.datetime.now())
+    print_log("🚀 > Connecting to Instagram...", color="green")
 
-    if user["name"] != None:
-        print("USER:", user["name"])
-        L.login(user["name"], user["password"])
-    return Profile.from_username(L.context, fetched_user)
+    if user["name"] is not None:
+        print_log("User " + user["name"])
+        session_file = user["name"] + "_session.sqlite"
+        try:
+            loader.load_session_from_file(user["name"], session_file)
+            assert user["name"] == loader.test_login()
+            print_log("Restored the session")
+        except FileNotFoundError:
+            print_log(
+                "Found no active 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)
 
 
 def get_image(url):
     try:
-        print(Fore.YELLOW + "🚀 > Downloading Image...", url)
-        print(Style.RESET_ALL)
-        print(datetime.datetime.now())
+        print_log("🚀 > Downloading Image... " + url, color="yellow")
 
         response = requests.get(url)
         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 Exception as err:
+        print_log("💥 > Failed to download image.", color="red")
+        print_log(err)
 
 
 def upload_image_to_mastodon(url, mastodon):
     try:
-        print(Fore.YELLOW + "🐘 > Uploading Image...")
-        print(Style.RESET_ALL)
-        print(datetime.datetime.now())
+        print_log("🐘 > Uploading Image...", color="yellow")
         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("✨ > 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 Exception as err:
+        print_log("💥 > failed to upload image to mastodon", color="red")
+        print_log(err)
 
 
-def toot(urls, title, mastodon, fetched_user):
+def toot(urls, title, mastodon):
     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))
@@ -75,13 +71,12 @@ def toot(urls, title, mastodon, fetched_user):
         post_text = post_text[0:1000]
 
         if ids:
-            print(ids)
+            print_log("Post identifiers:" + str(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())
+    except Exception as err:
+        print_log("😿 > Failed to create toot", color="red")
+        print_log(err)
 
 
 def get_new_posts(
@@ -106,22 +101,17 @@ def get_new_posts(
         if stupidcounter < post_limit:
             stupidcounter += 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())
+                print_log("🐘 > Already Posted        "+ post.url, color="yellow")
                 break  # Do not need to go back further in time
-            print("Posting... ", post.url)
-            print(datetime.datetime.now())
+            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)
             else:
-                toot(url_arr, post.caption, mastodon, fetched_user)
+                toot(url_arr, post.caption, mastodon)
             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")

src/util.py

@@ -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
+    )