Add more utils, change twitter model(s), re-work config
This commit is contained in:
parent
1139c406fb
commit
b8db789590
5 changed files with 42 additions and 16 deletions
|
@ -1,2 +1,2 @@
|
||||||
"""JARVIS core shared functionalities."""
|
"""JARVIS core shared functionalities."""
|
||||||
__version__ = "0.1.0"
|
__version__ = "0.3.0"
|
||||||
|
|
|
@ -16,16 +16,8 @@ DEFAULT_PATH = Path("config.yaml")
|
||||||
|
|
||||||
|
|
||||||
class Config(Singleton):
|
class Config(Singleton):
|
||||||
REQUIRED = ["token", "client_id", "mongo", "urls"]
|
REQUIRED = []
|
||||||
OPTIONAL = {
|
OPTIONAL = {}
|
||||||
"sync": False,
|
|
||||||
"log_level": "WARNING",
|
|
||||||
"scales": None,
|
|
||||||
"events": True,
|
|
||||||
"gitlab_token": None,
|
|
||||||
"max_messages": 1000,
|
|
||||||
"twitter": None,
|
|
||||||
}
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_yaml(cls, filepath: Union[Path, str] = DEFAULT_PATH) -> "Config":
|
def from_yaml(cls, filepath: Union[Path, str] = DEFAULT_PATH) -> "Config":
|
||||||
|
|
|
@ -189,19 +189,26 @@ class Starboard(Document):
|
||||||
|
|
||||||
|
|
||||||
@JARVIS_INST.register
|
@JARVIS_INST.register
|
||||||
class Twitter(Document):
|
class TwitterAccount(Document):
|
||||||
|
"""Twitter Account object."""
|
||||||
|
|
||||||
|
handle = fields.StringField(required=True)
|
||||||
|
twitter_id = fields.IntegerField(required=True)
|
||||||
|
last_tweet = fields.IntegerField(required=True)
|
||||||
|
last_sync = fields.DateTimeField(default=datetime.utcnow)
|
||||||
|
|
||||||
|
|
||||||
|
@JARVIS_INST.register
|
||||||
|
class TwitterFollow(Document):
|
||||||
"""Twitter Follow object."""
|
"""Twitter Follow object."""
|
||||||
|
|
||||||
active = fields.BooleanField(default=True)
|
active = fields.BooleanField(default=True)
|
||||||
twitter_id = fields.IntegerField(required=True)
|
twitter_id = fields.IntegerField(required=True)
|
||||||
handle = fields.StringField(required=True)
|
|
||||||
channel = fields.IntegerField(required=True)
|
channel = fields.IntegerField(required=True)
|
||||||
guild = fields.IntegerField(required=True)
|
guild = fields.IntegerField(required=True)
|
||||||
last_tweet = fields.IntegerField(required=True)
|
|
||||||
retweets = fields.BooleanField(default=True)
|
retweets = fields.BooleanField(default=True)
|
||||||
admin = fields.IntegerField(required=True)
|
admin = fields.IntegerField(required=True)
|
||||||
created_at = fields.DateTimeField(default=datetime.utcnow)
|
created_at = fields.DateTimeField(default=datetime.utcnow)
|
||||||
last_sync = fields.DateTimeField()
|
|
||||||
|
|
||||||
|
|
||||||
@JARVIS_INST.register
|
@JARVIS_INST.register
|
||||||
|
|
|
@ -1,8 +1,10 @@
|
||||||
"""JARVIS quality of life utilities."""
|
"""JARVIS quality of life utilities."""
|
||||||
import hashlib
|
import hashlib
|
||||||
|
from datetime import datetime
|
||||||
from typing import Any, Callable, Iterable, List, Optional, Tuple, Union
|
from typing import Any, Callable, Iterable, List, Optional, Tuple, Union
|
||||||
|
|
||||||
from aiohttp import ClientSession
|
from aiohttp import ClientSession
|
||||||
|
from dis_snek.models.discord.embed import Embed
|
||||||
|
|
||||||
from jarvis_core.filters import url
|
from jarvis_core.filters import url
|
||||||
|
|
||||||
|
@ -41,6 +43,29 @@ class Singleton(object):
|
||||||
setattr(self, key, value)
|
setattr(self, key, value)
|
||||||
|
|
||||||
|
|
||||||
|
def build_embed(
|
||||||
|
title: str,
|
||||||
|
description: str,
|
||||||
|
fields: list,
|
||||||
|
color: str = "#FF0000",
|
||||||
|
timestamp: datetime = None,
|
||||||
|
**kwargs: dict,
|
||||||
|
) -> Embed:
|
||||||
|
"""Embed builder utility function."""
|
||||||
|
if not timestamp:
|
||||||
|
timestamp = datetime.now()
|
||||||
|
embed = Embed(
|
||||||
|
title=title,
|
||||||
|
description=description,
|
||||||
|
color=color,
|
||||||
|
timestamp=timestamp,
|
||||||
|
**kwargs,
|
||||||
|
)
|
||||||
|
for field in fields:
|
||||||
|
embed.add_field(**field.to_dict())
|
||||||
|
return embed
|
||||||
|
|
||||||
|
|
||||||
async def hash(
|
async def hash(
|
||||||
data: str, method: Union[Callable, str] = hashlib.sha256, size: int = DEFAULT_BLOCKSIZE
|
data: str, method: Union[Callable, str] = hashlib.sha256, size: int = DEFAULT_BLOCKSIZE
|
||||||
) -> Tuple[str, int, str]:
|
) -> Tuple[str, int, str]:
|
||||||
|
@ -83,6 +108,8 @@ async def hash(
|
||||||
|
|
||||||
def convert_bytesize(b: int) -> str:
|
def convert_bytesize(b: int) -> str:
|
||||||
"""Convert bytes amount to human readable."""
|
"""Convert bytes amount to human readable."""
|
||||||
|
if not b:
|
||||||
|
return "??? B"
|
||||||
b = float(b)
|
b = float(b)
|
||||||
sizes = ["B", "KB", "MB", "GB", "TB", "PB"]
|
sizes = ["B", "KB", "MB", "GB", "TB", "PB"]
|
||||||
size = 0
|
size = 0
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
[tool.poetry]
|
[tool.poetry]
|
||||||
name = "jarvis-core"
|
name = "jarvis-core"
|
||||||
version = "0.2.1"
|
version = "0.3.0"
|
||||||
description = ""
|
description = ""
|
||||||
authors = ["Your Name <you@example.com>"]
|
authors = ["Your Name <you@example.com>"]
|
||||||
|
|
||||||
|
|
Loading…
Add table
Reference in a new issue