Python script to run as a cron job or manually that tracks who followed/unfollowed you

instagram_follower_tracker.py

import logging
import os
import sys
import time
import pickle
from datetime import datetime
from pathlib import Path

import pandas as pd
from dotenv import dotenv_values
from instagrapi import Client
from instagrapi.exceptions import ClientError
from pandas.errors import EmptyDataError


current_file_path = Path(__file__).parent.resolve()

logging.basicConfig(
    filename=current_file_path.joinpath("follower_check.log"),
    filemode="a",
    format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s",
    datefmt="%d/%b/%Y %H:%M:%S",
    level=logging.DEBUG,
)
disabled_loggers = ["instagrapi", "urllib3", "public_request"]

for d_logger in disabled_loggers:
    logging.getLogger(d_logger).setLevel(logging.WARNING)

logger = logging.getLogger(__name__)


def login_to_api(settings_path: Path):
    config = dotenv_values(current_file_path.joinpath(".env"))
    username = config.get("INSTAGRAM_USERNAME")
    password = config.get("INSTAGRAM_PASSWORD")

    if username is None or password is None:
        logger.error("Username or password not found in .env")
        notify("Instagram Follower Check Error", "Username or password not found in .env")
        sys.exit(1)

    try:
        cl = Client()
        if settings_path.exists():
            cl.load_settings(settings_path)
        cl.login(username, password)
    except Exception as e:
        logger.error("Failed to login. Error: %s", str(e))
        notify(
            "Instagram Follower Check Error",
            f"Failed to login for follower check {str(e)}",
        )
        sys.exit(1)

    return cl


def get_followers():
    logger.info("Logging in to Instagram")

    client_path = current_file_path.joinpath(".client.pkl")
    settings_path = current_file_path.joinpath(".client.settings.json")

    if client_path.exists():
        with open(client_path, "rb") as f:
            client = pickle.load(f)
        if settings_path.exists():
            client.load_settings(str(settings_path))
    else:
        client = login_to_api(settings_path)
        with open(client_path, "wb") as f:
            pickle.dump(client, f)

    client.dump_settings(settings_path)

    try:
        followers = client.user_followers(str(client.user_id))
    except ClientError:
        try:
            client = login_to_api(settings_path)
        except ClientError as e:
            logger.error("Unable to login")
            notify(
                "Instagram Follower Check Error",
                f"Failed to login for follower check {str(e)}",
            )
        followers = client.user_followers(str(client.user_id))
    logger.info("Got followers")

    usernames = [user.username for user in followers.values()]
    sorted_usernames = sorted(usernames)
    return sorted_usernames


def notify(title, text):
    logger.info("Notifying %s", text)
    time.sleep(3)
    os.system(
        f'osascript -e \'display notification "{text}" with title "{title}" '
        f'sound name "default"\''
    )


def main():
    followers = get_followers()

    csv_path = current_file_path.joinpath("followers.csv")

    logger.info("Reading old csv")
    today = datetime.now().isoformat()
    try:
        df = pd.read_csv(csv_path, index_col="followers")
    except EmptyDataError:
        df = pd.DataFrame()

    logger.info("Got followers, creating new dataframe")
    following_col = ["following"] * len(followers)
    new_df = pd.DataFrame({today: following_col, "followers": followers})
    new_df.set_index("followers", inplace=True)

    if df.empty:
        logger.info("No old csv found. Writing new csv")
        new_df.to_csv(current_file_path.joinpath("followers.csv"))
        return

    last_col_name = df.iloc[:, -1].name

    logger.info("Merging old and new dataframes on followers index")
    merged_df = pd.merge(df, new_df, how="outer", left_index=True, right_index=True)
    merged_df.fillna("not following", inplace=True)

    logger.info("Diffing dataframes")
    diff = merged_df[merged_df[today] != merged_df[last_col_name]]

    if len(diff.index) > 0:
        logger.info("Follower list changed, notifying")
        changed_followers = diff.index.tolist()
        old_statuses = diff[last_col_name].tolist()
        new_statuses = diff[today].tolist()

        for username, o, n in zip(changed_followers, old_statuses, new_statuses):
            logger.debug("Follower: %s, old_status: %s, new_status: %s", username, o, n)
            if o == "not following" and n == "following":
                notify("New Follower", username)
            elif o == "following" and n == "not following":
                notify("Unfollowed", username)
    else:
        logger.info("Follower list did not change")

    logger.info("Writing new csv")
    merged_df.to_csv(csv_path)


if __name__ == "__main__":
    main()