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(): global _message, _ticker, _title _message = DavFile(client=CLIENT, path="message.txt") _ticker = DavFile(client=CLIENT, path="ticker.txt") _title = DavFile(client=CLIENT, path="title.txt") DavFile.refresh(60) _logger.debug("text router started") @router.get("/message") async def get_message(): 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) ): return list(ticker_commands) @router.get("/title") async def get_title(): return _md.convert( str(_title) )