116 lines
2.7 KiB
Python
116 lines
2.7 KiB
Python
"""
|
||
Utilities
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
import logging
|
||
import traceback
|
||
import datetime
|
||
import hashlib
|
||
|
||
import aioconsole
|
||
|
||
from datatypes import AppConfig
|
||
|
||
|
||
|
||
#
|
||
# PRIVATE
|
||
#
|
||
_stop_event: asyncio.Event
|
||
|
||
|
||
|
||
#
|
||
# PUBLIC
|
||
#
|
||
def setup_logging() -> None:
|
||
"""Setup logging"""
|
||
logging.basicConfig(
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S",
|
||
level=logging.INFO
|
||
)
|
||
logging.getLogger("nio").setLevel(logging.CRITICAL + 1)
|
||
|
||
def setup_app_stop_event() -> None:
|
||
"""Setup app stop asyncio event"""
|
||
global _stop_event
|
||
_stop_event = asyncio.Event()
|
||
|
||
def request_app_stop(reason: str) -> None:
|
||
try:
|
||
log_info(f"{reason}. The application is stopping.")
|
||
_stop_event.set()
|
||
except:
|
||
traceback.print_exc()
|
||
|
||
def get_app_stop_event() -> asyncio.Event:
|
||
return _stop_event
|
||
|
||
def is_terminal_interactive() -> bool:
|
||
"""Returns True if the terminal is interactive"""
|
||
return sys.stdin.isatty() and sys.stdout.isatty()
|
||
|
||
def log_info(text: str) -> None:
|
||
logging.info(text)
|
||
|
||
def log_warning(text: str) -> None:
|
||
logging.warning(text)
|
||
|
||
def log_error(text: str) -> None:
|
||
logging.error(text)
|
||
|
||
async def ainput(text: str = "") -> str:
|
||
# simulate empty input if not TTY
|
||
if not is_terminal_interactive():
|
||
return ""
|
||
return await aioconsole.ainput(text)
|
||
|
||
async def get_password() -> str | None:
|
||
if "MATRIX_PASSWORD" in os.environ:
|
||
return os.environ["MATRIX_PASSWORD"]
|
||
if not is_terminal_interactive():
|
||
return None
|
||
return await ainput("Matrix password: ")
|
||
|
||
def date_to_text(date: datetime.datetime | float, dow: bool = True, seconds: bool = True) -> str:
|
||
''' Returns date as formatted string.
|
||
Day of week can be added.
|
||
Seconds can be added.
|
||
'''
|
||
if type(date) is float:
|
||
date = int(date)
|
||
if type(date) is int:
|
||
date = datetime.datetime.utcfromtimestamp(date)
|
||
|
||
# format string
|
||
format_string = ''
|
||
if dow:
|
||
format_string += '%a, '
|
||
format_string += '%d.%m.%Y, %H:%M'
|
||
if seconds:
|
||
format_string += ':%S'
|
||
# day of week to Russian
|
||
translate_map = [
|
||
('Mon', 'Пн'),
|
||
('Tue', 'Вт'),
|
||
('Wed', 'Ср'),
|
||
('Thu', 'Чт'),
|
||
('Fri', 'Пт'),
|
||
('Sat', 'Сб'),
|
||
('Sun', 'Вс')
|
||
]
|
||
result = date.strftime(format_string) # type: ignore
|
||
for en, ru in translate_map:
|
||
if en in result:
|
||
result = result.replace(en, ru)
|
||
break
|
||
return result
|
||
|
||
def get_hash(data: bytes | str) -> str:
|
||
if type(data) is str:
|
||
data = data.encode(errors="ignore")
|
||
return hashlib.sha256(data).hexdigest() # type: ignore |