164 lines
4.4 KiB
Python
164 lines
4.4 KiB
Python
"""
|
|
Utilities
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import os
|
|
import json
|
|
import logging
|
|
import traceback
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
import aiofiles
|
|
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 set_next_batch(config: AppConfig, next_batch: str | None) -> bool:
|
|
"""Save `next_batch` to session directory."""
|
|
try:
|
|
next_batch_file_path = Path.cwd() / config.store_dir / "next_batch.txt"
|
|
if next_batch is None:
|
|
next_batch_file_path.unlink(True)
|
|
return True
|
|
async with aiofiles.open(next_batch_file_path, "w") as f:
|
|
await f.write(next_batch)
|
|
return True
|
|
except:
|
|
return False
|
|
|
|
async def get_next_batch(config: AppConfig) -> str | None:
|
|
"""Get `next_batch`"""
|
|
try:
|
|
next_batch_file_path = Path.cwd() / config.store_dir / "next_batch.txt"
|
|
async with aiofiles.open(next_batch_file_path, "r") as f:
|
|
next_batch = (await f.read()).strip()
|
|
if not next_batch:
|
|
next_batch = None
|
|
return next_batch
|
|
except:
|
|
return None
|
|
|
|
async def get_session_data(config: AppConfig) -> tuple[str, str] | tuple[None, None]:
|
|
"""
|
|
Get (access_token, device_id) or (None, None)
|
|
"""
|
|
try:
|
|
token_file_path = Path.cwd() / config.store_dir / "auth.json"
|
|
async with aiofiles.open(token_file_path, "r") as f:
|
|
j = json.loads((await f.read()).strip())
|
|
return j["access_token"], j["device_id"]
|
|
except:
|
|
return None, None
|
|
|
|
async def set_session_data(config: AppConfig, token_device_pair: tuple[str, str] | None) -> bool:
|
|
"""
|
|
Set new (access_token, device_id) pair; use None to remove it.
|
|
|
|
Returns:
|
|
True on success
|
|
"""
|
|
try:
|
|
token_file_path = Path.cwd() / config.store_dir / "auth.json"
|
|
if token_device_pair is None:
|
|
token_file_path.unlink(True)
|
|
return True
|
|
async with aiofiles.open(token_file_path, "w") as f:
|
|
await f.write(json.dumps({"access_token": token_device_pair[0], "device_id": token_device_pair[1]}))
|
|
return True
|
|
except:
|
|
return False
|
|
|
|
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 get_hostname_from_url(url: str) -> str | None:
|
|
"""Returns `matrix.domain.net` for `https://matrix.domain.net/bla/bla/bla`"""
|
|
try:
|
|
return urlparse(url).hostname
|
|
except:
|
|
return None
|
|
|
|
def check_and_remove_html(possible_html: str) -> tuple[bool, str]:
|
|
"""Checks if `possible_html` is a valid HTML text and returns (is_html, text_without_tags)"""
|
|
has_tags = False
|
|
text_fragments = []
|
|
class Extractor(HTMLParser):
|
|
def handle_starttag(self, tag, attrs):
|
|
nonlocal has_tags
|
|
has_tags = True
|
|
def handle_data(self, data):
|
|
text_fragments.append(data)
|
|
|
|
parser = Extractor(convert_charrefs=True)
|
|
parser.feed(possible_html)
|
|
|
|
try:
|
|
if has_tags:
|
|
return (True, " ".join("".join(text_fragments).split()))
|
|
except:
|
|
traceback.print_exc()
|
|
|
|
return (False, possible_html) |