ovdashboard/api/ovdashboard_api/routers/text.py

import logging
import re
from typing import Iterator

from fastapi import APIRouter, Depends
from markdown import Markdown
from pydantic import BaseModel

from .. import CLIENT
from ..config import SETTINGS
from ..dav_file import DavFile

router = APIRouter(prefix="/text", tags=["text"])

_logger = logging.getLogger(__name__)
_md = Markdown()

_message = ""
_ticker = ""
_title = ""


@router.on_event("startup")
async def on_startup() -> None:
    global _message, _ticker, _title

    _message = DavFile(CLIENT.resource("message.txt"))
    _ticker = DavFile(CLIENT.resource("ticker.txt"))
    _title = DavFile(CLIENT.resource("title.txt"))
    DavFile.refresh(60)

    _logger.debug("text router started")


@router.get("/message")
async def get_message() -> str:
    return _md.convert(
        str(_message)
    )


async def get_ticker_lines() -> Iterator[str]:
    return (
        line.strip()
        for line in str(_ticker).split("\n")
        if line.strip()
    )


async def get_ticker_content_lines(
    ticker_lines: Iterator[str] = Depends(get_ticker_lines),
) -> Iterator[str]:
    return (
        line
        for line in ticker_lines
        if not line.startswith(".")
    )


@router.get("/ticker/content")
async def get_ticker_content(
    ticker_content_lines: Iterator[str] = Depends(get_ticker_content_lines),
) -> str:
    return _md.convert(
        SETTINGS.ticker_separator.join(ticker_content_lines)
    )

_re_ticker_command_line = re.compile(
    r"^\.([a-z]+)\s+(.*)$",
    flags=re.IGNORECASE,
)


class TickerCommand(BaseModel):
    command: str
    argument: str


async def get_ticker_commands(
    ticker_lines: Iterator[str] = Depends(get_ticker_lines),
) -> Iterator[TickerCommand]:
    return (
        TickerCommand(
            command=match.group(1),
            argument=match.group(2),
        )
        for line in ticker_lines
        if (match := _re_ticker_command_line.match(line))
    )


@router.get("/ticker/commands", response_model=list[TickerCommand])
async def get_ticker_commands(
    ticker_commands: Iterator[str] = Depends(get_ticker_commands)
) -> list[TickerCommand]:
    return list(ticker_commands)


@router.get("/title")
async def get_title() -> str:
    return _md.convert(
        str(_title)
    )