129 lines
3.7 KiB
Python
129 lines
3.7 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import List
|
|
|
|
import requests
|
|
|
|
from models.member import ChatMember
|
|
from settings import API_BASE
|
|
|
|
|
|
logger = logging.getLogger("[services.core] ")
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
|
|
def _request_endpoint(query_name, body) -> dict:
|
|
logger.debug(f"requesting {query_name}...")
|
|
response = requests.post(
|
|
API_BASE, headers={"Content-Type": "application/json"}, json=body, timeout=30.0
|
|
)
|
|
if response.status_code == 200:
|
|
try:
|
|
r = response.json()
|
|
result = r.get("data", {}).get(query_name, {})
|
|
if result:
|
|
logger.info(f"entries amount in result: {len(result)} ")
|
|
return result
|
|
except ValueError as e:
|
|
logger.error(f"Error decoding JSON response: {e}")
|
|
|
|
return {}
|
|
|
|
|
|
def get_all_authors():
|
|
query_name = "get_authors_all"
|
|
|
|
gql = {
|
|
"query": "query { " + query_name + "{ id slug pic name user } }",
|
|
"variables": None,
|
|
}
|
|
|
|
return _request_endpoint(query_name, gql)
|
|
|
|
|
|
def get_author_by_user(user: str):
|
|
operation = "GetAuthorId"
|
|
query_name = "get_author_id"
|
|
gql = {
|
|
"query": f"query {operation}($user: String!) {{ {query_name}(user: $user){{ id }} }}", # noqa E201, E202
|
|
"operationName": operation,
|
|
"variables": {"user": user.strip()},
|
|
}
|
|
|
|
return _request_endpoint(query_name, gql)
|
|
|
|
|
|
def get_my_followed() -> List[ChatMember]:
|
|
query_name = "get_my_followed"
|
|
|
|
gql = {
|
|
"query": "query { " + query_name + " { authors { id slug pic name } } }",
|
|
"variables": None,
|
|
}
|
|
|
|
result = _request_endpoint(query_name, gql)
|
|
return result.get("authors", [])
|
|
|
|
|
|
class CacheStorage:
|
|
lock = asyncio.Lock()
|
|
period = 5 * 60 # every 5 mins
|
|
client = None
|
|
authors = []
|
|
authors_by_user = {}
|
|
authors_by_id = {}
|
|
|
|
@staticmethod
|
|
async def init():
|
|
"""graphql client connection using permanent token"""
|
|
self = CacheStorage
|
|
async with self.lock:
|
|
task = asyncio.create_task(self.worker())
|
|
logger.info(task)
|
|
|
|
@staticmethod
|
|
async def update_authors():
|
|
self = CacheStorage
|
|
async with self.lock:
|
|
result = get_all_authors()
|
|
logger.info(f"cache loaded {len(result)}")
|
|
if result:
|
|
CacheStorage.authors = result
|
|
for a in result:
|
|
user_id = a.get("user")
|
|
author_id = str(a.get("id"))
|
|
self.authors_by_user[user_id] = a
|
|
self.authors_by_id[author_id] = a
|
|
|
|
@staticmethod
|
|
async def worker():
|
|
"""async task worker"""
|
|
failed = 0
|
|
self = CacheStorage
|
|
while True:
|
|
try:
|
|
logger.info(" - updating profiles data...")
|
|
await self.update_authors()
|
|
failed = 0
|
|
except Exception as er:
|
|
failed += 1
|
|
logger.error(f"{er} - update failed #{failed}, wait 10 seconds")
|
|
if failed > 3:
|
|
logger.error(" - not trying to update anymore")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
break
|
|
if failed == 0:
|
|
when = datetime.now(timezone.utc) + timedelta(seconds=self.period)
|
|
t = format(when.astimezone().isoformat())
|
|
logger.info(
|
|
" ⎩ next update: %s"
|
|
% (t.split("T")[0] + " " + t.split("T")[1].split(".")[0])
|
|
)
|
|
await asyncio.sleep(self.period)
|
|
else:
|
|
await asyncio.sleep(10)
|
|
logger.info(" - trying to update data again")
|