IG: support TOTP

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")

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   #

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
+        ]
+    }
+}