niming_igapi/ig/IG.py

import os
import logging
from typing import List

from instagrapi import Client

from utils.tbProcessor import easyExceptionHandler
from utils.const import DEVICE

# logging
iglog = logging.getLogger("ig")
iglog.setLevel(level=logging.DEBUG)

cl:Client = None

# init
def init(askcl:Client) -> None:
    global cl
    cl = askcl
    
    
# login
def login() -> int:
    # Env
    ACCOUNT_USERNAME = os.getenv("ACCOUNT_USERNAME", None).strip()
    ACCOUNT_PASSWORD = os.getenv("ACCOUNT_PASSWORD", None).strip()
    
    # session
    session = None
    if os.path.exists("./session.json"):    
        session = cl.load_settings("session.json")

    cl.delay_range = [2, 5]
    # login with sessionid
    cl.set_device(DEVICE)
    sessionSuccess = True
    if session:
        iglog.info("Trying logging in with session")
        try:
            cl.set_settings(session)
            cl.login(ACCOUNT_USERNAME, ACCOUNT_PASSWORD)
            cl.get_timeline_feed()
        except:
            sessionSuccess = False
    else:
        sessionSuccess = False

    # login with username and password
    if not sessionSuccess:
        iglog.info("Trying logging in with username and password")
        try:
            old_session = cl.get_settings()
            cl.set_settings({})
            cl.set_uuids(old_session["uuids"])
            cl.login(ACCOUNT_USERNAME, ACCOUNT_PASSWORD)
            cl.get_timeline_feed()
        except:
            iglog.error("Cannot log in")
            return 0
    
    cl.dump_settings("session.json")
    
    # return
    username = cl.account_info().dict()["username"]
    iglog.info("Logged as %s"%username)
    return 1


# Get account info
def account_info() -> dict | None:
    iglog.info("Fetching account info")
    try:
        info = cl.account_info().dict()
        return info
    except Exception as e:
        easyExceptionHandler(e)
        return None


# Get media info
def media_info(code:str) -> dict | None:
    try:
        pk = cl.media_pk_from_code(code)
        info = cl.media_info(pk).dict()
        return info
    except Exception as e:
        easyExceptionHandler(e)
        return None
    

# Upload media
def upload_media(content:str, paths:List[str]) -> dict | None:
    try:
        # uplaod
        if len(paths) == 0:
            return None
        elif len(paths) == 1:
            media = cl.photo_upload(path=paths[0], caption=content).dict()
        else:
            media = cl.album_upload(paths=paths, caption=content).dict()
            
        return media
    except Exception as e:
        easyExceptionHandler(e)
        return None
    

# Delete Media
def delete_media(code:str) -> int:
    try:
        media_pk = str(cl.media_pk_from_code(code))
        media_id = cl.media_id(media_pk)
        cl.media_delete(media_id)
        return 0
    except Exception as e:
        easyExceptionHandler(e)
        return 1