100 lines
No EOL
2.6 KiB
Python
100 lines
No EOL
2.6 KiB
Python
"""
|
|
Backend API for frontend to call
|
|
"""
|
|
import logging
|
|
from typing import Tuple
|
|
from cachetools import TTLCache, cached
|
|
|
|
from config.config import ACCINFO_CACHE, RELOGIN_LIMIT
|
|
from backend import backend
|
|
from backend.utils import ld_interface
|
|
from backend.db import dbhelper
|
|
from backend.ig import IG
|
|
|
|
# logger
|
|
bkapilog = logging.getLogger("backend.api")
|
|
bkapilog.setLevel(level=logging.INFO)
|
|
|
|
# account info
|
|
cache_accinfo = TTLCache(maxsize=1, ttl=ACCINFO_CACHE)
|
|
@cached(cache_accinfo)
|
|
def IG_account_info() -> dict | None:
|
|
result = IG.account_info()
|
|
return result
|
|
|
|
|
|
# login
|
|
cache_login = TTLCache(maxsize=1, ttl=RELOGIN_LIMIT)
|
|
@cached(cache_login)
|
|
def _IG_login() -> int:
|
|
result = IG.login()
|
|
return result
|
|
def IG_login() -> str:
|
|
if len(cache_login): # cooldown
|
|
bkapilog.info("IG_login: cooldown")
|
|
return "Cooldown"
|
|
|
|
# login
|
|
lgres = _IG_login()
|
|
if lgres == 1:
|
|
bkapilog.info("IG_login: login successed")
|
|
return None
|
|
else:
|
|
bkapilog.error("IG_login: login failed")
|
|
return "Login Failed"
|
|
|
|
|
|
# get queue content
|
|
def BACKEND_queue() -> dict:
|
|
t = backend.queue.items()
|
|
reply = { _[0]:str(_[1]["aid"]) for _ in t }
|
|
return reply
|
|
|
|
|
|
# task: upload
|
|
def upload(aid:int) -> Tuple[str, int]:
|
|
# check - visible
|
|
article = ld_interface.inf.get(index=aid, media=False)
|
|
if article is None:
|
|
return "Article not found", 1
|
|
|
|
# check - already in queue
|
|
if backend.queue["upload-"+str(aid)]:
|
|
return "Request is already in queue", 1
|
|
|
|
# check - there is a requet in queue that wants to delete same target
|
|
if backend.queue["delete-"+str(aid)]:
|
|
backend.queue.pop("delete-"+str(aid))
|
|
return "Canceled delete article request", 0
|
|
|
|
# check - already uploaded
|
|
uploaded = dbhelper.solo_article_fetcher(aid=aid)
|
|
if uploaded:
|
|
return "Already posted", 1
|
|
|
|
# put into queue
|
|
backend.queue["upload-"+str(aid)] = {"aid":aid}
|
|
|
|
return "Put into queue", 0
|
|
|
|
|
|
# task: delete
|
|
def delete(aid:int) -> Tuple[str, int]:
|
|
# check - already in queue
|
|
if backend.queue["delete-"+str(aid)]:
|
|
return "Request is already in queue", 1
|
|
|
|
# check - there is a requet in queue that wants to upload same target
|
|
if backend.queue["upload-"+str(aid)]:
|
|
backend.queue.pop("upload-"+str(aid))
|
|
return "Canceled upload post request", 0
|
|
|
|
# check - never uploaded
|
|
uploaded = dbhelper.solo_article_fetcher(aid=aid)
|
|
if not uploaded:
|
|
return "Has not been posted yet", 1
|
|
|
|
# put into queue
|
|
backend.queue["delete-"+str(aid)] = {"aid":aid}
|
|
|
|
return "Put into queue", 0 |