Initial commit

util.py

@@ -0,0 +1,164 @@
+"""
+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)