diff --git a/backend/ig/IG.py b/backend/ig/IG.py index d600e1d..0cd5ec9 100644 --- a/backend/ig/IG.py +++ b/backend/ig/IG.py @@ -3,9 +3,10 @@ import logging from typing import List from instagrapi import Client +import pyotp from backend.utils import ld_picturemaker -from config.config import DEBUG, ACCOUNT_USERNAME, ACCOUNT_PASSWORD +from config.config import DEBUG, ACCOUNT_USERNAME, ACCOUNT_PASSWORD, _2FA from utils.err import easyExceptionHandler #from utils.const import DEVICE @@ -46,7 +47,15 @@ def login() -> int: old_session = cl.get_settings() cl.set_settings({}) cl.set_uuids(old_session["uuids"]) - cl.login(ACCOUNT_USERNAME, ACCOUNT_PASSWORD) + + if _2FA == "TOTP": + from config.config import _2FA_TOTPSEED + totp = pyotp.TOTP(_2FA_TOTPSEED) + totpcode = totp.now() + cl.login(ACCOUNT_USERNAME, ACCOUNT_PASSWORD, verification_code=totpcode) + else: + cl.login(ACCOUNT_USERNAME, ACCOUNT_PASSWORD) + cl.get_timeline_feed() except: iglog.error("Cannot log in") diff --git a/config/config.py.example b/config/config.py.example index 7acedfa..4a24620 100644 --- a/config/config.py.example +++ b/config/config.py.example @@ -30,6 +30,9 @@ RELOGIN_LIMIT = 10*60 # 10 mins - re-login limit # IG ACCOUNT_USERNAME = "" ACCOUNT_PASSWORD = "" +## IG.2FA +_2FA = "" # we only support totp now :P +_2FA_TOTPSEED = "" # type define {mine:ext} FILE_MINE_TYPE = { @@ -48,7 +51,8 @@ FILE_MINE_TYPE = { #################### # Interface config # #################### -INTERFACE = "interface/example.py" +#INTERFACE = "interface/example.py" +INTERFACE = "interface/tcivs.py" #################### # PictureMaker # diff --git a/interface/tcivs.py b/interface/tcivs.py new file mode 100644 index 0000000..a41bffd --- /dev/null +++ b/interface/tcivs.py @@ -0,0 +1,82 @@ +import datetime +import requests +import io + +""" +Interface + +The bridge between web application(niming) and igapi. +An interface should have a get() function. +""" + +HOST="http://172.16.20.145:3000" + +def get(index:int, media:bool=True) -> dict | None: + """ + Every interface should have this function. + Backend calls this function to get data of a post. + + Args: + index (int): ID of the post + media (bool): Send media or not + + Returns: + dict: Data of a post. Formatted as shown in Note 1 (an_example_of_context) + None: An error occurred, and nothing was returned. + """ + + # get data + ## fake server in localhost + ### where is the api of web application? + res = requests.get(f"{HOST}/api/post?cursor={index}") + if res.status_code != 200: + return None + + rj = res.json()[0] + media_arr = [] + + # get media + if media and rj["enclosure"] is not None: + for m in rj["enclosure"]: + _m = requests.get(f"{HOST}/{m}") + if _m.status_code == 200: + media_arr.append(io.BytesIO(_m.content)) + + # return + result = { + "id": rj["id"], + "metadata": { + "create_time": datetime.datetime.strptime(rj["post_at"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=datetime.timezone.utc), + "author": rj["signing"], + "tags": [], + "category": "", + # ext + "parent": None #rj["parent"] # parent id + }, + "content": { + "text": rj["content"], + "media": media_arr + } + } + + return result + + +# Note 1 +an_example_of_context = { + "id": int, + "metadata": { + "create_time": datetime.datetime, + "author": str, + "tags": list[str], + "category": str, + # ext + "parent": int | None # parent id + }, + "content": { + "text": str, + "media": [ + io.BytesIO + ] + } +} \ No newline at end of file