Skip to content
Snippets Groups Projects
Commit 1880dc20 authored by Peder Hovdan Andresen's avatar Peder Hovdan Andresen
Browse files

Initial commit

parents
Branches master
No related tags found
No related merge requests found
from discord.ext.commands import Bot
from .config import load
bot = Bot(command_prefix="?")
from discord.ext import commands
from discord.ext import tasks
from discord import TextChannel, DMChannel
from typing import Union, List
class BaseCog(commands.Cog):
EMBED_CHAR_LIMIT = 1024
def __init__(self, bot: commands.Bot) -> None:
self.bot = bot
async def send_text_message(self,
ctx: Union[commands.Context, TextChannel, DMChannel, int],
text: str
) -> None:
"""
Sends an arbitrarily long string as one or more messages to a channel.
String is split into multiple messages if length of string exceeds
Discord text message character limit.
Parameters
----------
ctx : `commands.Context`
Discord Context
text : `str`
String to post to channel
channel_id : `Optional[int]`, optional
Optional channel ID if target channel is not part of the context.
"""
channel: Union[commands.Context, TextChannel, DMChannel] = None
# Obtain channel
if isinstance(ctx, int):
channel = self.bot.get_channel(channel_id)
elif isinstance(ctx, (commands.Context, TextChannel, DMChannel)):
channel = ctx
else:
raise discord.DiscordException("Argument 'ctx' must be a Discord Context, text channel or a channel ID.")
# Split string into chunks
chunks = await self._split_string_to_chunks(text)
# Send chunked messages
for chunk in chunks:
await channel.send(chunk)
async def _split_string_to_chunks(self, text: str, limit: int=None) -> List[str]:
"""Splits a string into (default: 1024) char long chunks."""
if not limit or limit>self.CHAR_LIMIT:
limit = self.CHAR_LIMIT
return [text[i:i+limit] for i in range(0, len(text), limit)]
async def _split_string_by_lines(self, text: str, limit: int=None, strict: bool=False) -> List[str]:
"""Splits a string into `limit`-sized chunks. DEFAULT: 1024
The string is split into n<=limit sized chunks based on
occurences of newline chars, whereas `BaseCog._split_string_to_chunks()`
splits string into n<=limit chunks, with no regard for splitting
words or sentences based on newline chars.
"""
if not limit or limit > self.EMBED_CHAR_LIMIT:
limit = self.EMBED_CHAR_LIMIT
if len(text) < limit: # no need to split
return [text]
chunk = "" # Lines in a chunk
chunk_len = 0 # Size of current chunk
chunks = []
for line in text.splitlines(keepends=True):
line_len = len(line)
# Handle lines whose length exceeds limit
if line_len > limit:
if strict:
raise discord.DiscordException(
"Unable to split string. Line length exceeds limit!"
)
else: # Fall back on _split_string_to_chunks()
return await self._split_string_to_chunks(text, limit)
if chunk_len + line_len > limit:
chunks.append(chunk)
chunk = ""
chunk_len = 0
chunk += line
chunk_len += line_len
else:
if chunk:
chunks.append(chunk)
return chunks
\ No newline at end of file
"""
BUNCHA PSEUDO CODE HERE NOW
"""
from discord.ext import commands
from discord.ext import tasks
from osmon import OSMon
from .base_cog import BaseCog
class MonitorCog(BaseCog):
def __init__(self, bot: commands.Bot) -> None:
self.bot = bot
self.monitor = OSMon(some_config_file_path_here)
@tasks.loop(minutes=5)
async def ping_servers(self) -> None:
await self.check_servers()
async def check_servers(self) -> None:
failed = await self.monitor.check()
if not failed:
return
await self._report_failed(failed)
async def _report_failed(self, failed: List[OpenStackServers]) -> None:
await self.send_text_message(
self.bot.config.report_channel,
"The following servers failed: xxxxxxxxxxx"
"These measures were taken: yyyyyyyyyyyyyyy"
)
\ No newline at end of file
from __future__ import annotations
import yaml
from pathlib import Path
from dataclasses import dataclass
# String concatenation
def join(loader, node) -> str:
seq = loader.construct_sequence(node)
return ''.join(str(i) for i in seq)
# Parse value as Path
def path(loader, node) -> Path:
seq = loader.construct_sequence(node)
return Path('/'.join(str(i) for i in seq))
## register the tag handler
yaml.add_constructor('!join', join)
yaml.add_constructor('!path', path)
@dataclass
class Config:
bot_id: int
bot_token: str
bot_secret: str
openstack_credentials: OpenStackCredentials
@classmethod
def from_dict(cls, d: dict) -> Config:
def load_config(path: str="config.yml") -> dict:
with open(path, "r") as f:
config = yaml.load(f)
return config
discord:
id:
token:
secret:
osmon:
network: imt3003
auth_url: url
application_credential_id: id
application_credential_secret: secret
manager_prefix: manager
www_prefix: www
timeout: 5
emails:
- pedeha@ntnu.no
- patricek@stud.ntnu.no
- anderscw@stud.ntnu.no
smtp_url: smtp.gmail.com
smtp_username: username
smtp_password: password
ignored:
- www3
This diff is collapsed.
[tool.poetry]
name = "bigtechmonitor"
version = "0.1.0"
description = ""
authors = ["PederHA <peder.andresen@gmail.com>"]
[tool.poetry.dependencies]
python = "^3.8"
"discord.py" = "^1.6.0"
pyyaml = "^5.4.1"
[tool.poetry.dev-dependencies]
[build-system]
requires = ["poetry>=0.12"]
build-backend = "poetry.masonry.api"
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment