2026-matrix-csonac/util.py

"""
Utilities
"""

import asyncio
import sys
import os
import json
import logging
import traceback
import datetime
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)

def date_to_text(date: datetime.datetime, 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)
    for en, ru in translate_map:
        if en in result:
            result = result.replace(en, ru)
            break
    return result