Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from core.cache.api import Cache
- from core.database.api import Database
- from core.database.models import Agent
- from datetime import datetime, timedelta
- from instagram import Account, AsyncWebAgent, AsyncWebAgentAccount
- import json
- import logging
- from random import randint
- class Logic:
- delay_groups = [
- ("update_account_meta", "make_account_decision"),
- ("update_follows", "update_followers", "update_likes", "update_comments"),
- ("get_follows", "get_followers", "get_media", "get_likes", "get_comments"),
- ("follow", "unfollow", "like", "unlike", "add_comment", "delete_comment"),
- ("check_account",),
- ]
- logger = logging.getLogger("tasks")
- database=None
- cache=None
- @classmethod
- def set_database(cls, database: Database):
- cls.datatabase = database
- @classmethod
- def set_cache(cls, cache: Cache):
- cls.cache = cache
- # Common methods
- @classmethod
- async def get_target(cls, name: str, db_agent=None, only_agent=False):
- target = None
- if not db_agent is None:
- target = await cls.cache.get_target(db_agent=db_agent, name=name)
- if not target and not only_agent:
- target = await cls.cache.get_target(name=name)
- return target
- @classmethod
- async def add_target(cls, name: str, value, db_agent=None, first=False):
- return await cls.cache.add_target(
- name=name,
- value=value,
- db_agent=db_agent,
- first=first,
- )
- @classmethod
- async def get_delay(cls, action: str, db_agent=None):
- if not db_agent is None:
- last = datetime(1970, 1, 1)
- delay_group = []
- for group in cls.delay_groups:
- if action in group:
- delay_group = group
- for a in delay_group:
- data = await cls.cache.get_agent_action_time(db_agent=db_agent, name=a)
- if not data is None and data > last:
- last = data
- break
- delay = None
- if not db_agent is None:
- try:
- delay = randint(
- getattr(db_agent.settings, f"{action}_min_delay").value,
- getattr(db_agent.settings, f"{action}_max_delay").value,
- )
- except (AttributeError, TypeError, ValueError):
- pass
- if delay is None:
- delay = randint(
- int(cls.database.get_setting(key=f"{action}_min_delay").value),
- int(cls.database.get_setting(key=f"{action}_max_delay").value),
- )
- if db_agent is None:
- return delay
- if last + timedelta(seconds=delay) > datetime.now():
- await cls.cache.set_agent_action_time(
- db_agent=db_agent,
- name=action,
- dt=last + timedelta(seconds=delay),
- )
- return (last - datetime.now() + timedelta(seconds=delay)).total_seconds()
- await cls.cache.set_agent_action_time(db_agent=db_agent, name=action, dt=datetime.now())
- return 0
- # Logic
- @classmethod
- async def start(cls, db_agent: Agent, loop):
- pass
- @classmethod
- async def stop(cls, db_agent: Agent, loop):
- pass
- @classmethod
- async def follow(cls, db_agent: Agent, account: Account):
- pass
- @classmethod
- async def unfollow(cls, db_agent: Agent, account: Account):
- pass
- @classmethod
- async def get_followers_count(cls):
- raise NotImplementedError
- @classmethod
- async def set_followers(cls, db_agent: Agent, account: Account, followers: list):
- pass
- @classmethod
- async def get_follows_count(cls):
- raise NotImplementedError
- @classmethod
- async def set_follows(cls, db_agent: Agent, account: Account, follows: list):
- pass
- @classmethod
- async def get_account_media_count(cls):
- raise NotImplementedError
- @classmethod
- async def set_account_media(cls, db_agent: Agent, account: Account, media: list):
- pass
- @classmethod
- async def check_account(cls, db_agent: Agent, account: Account, data):
- pass
- @classmethod
- async def update_account_meta(cls, account: Account):
- pass
- @classmethod
- async def make_account_decision(cls, account: Account):
- pass
- @classmethod
- async def add_account_for_scrap(cls, account: Account, db_agent=None, first=False):
- pass
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement