Merge pull request #21 from Tishka17/feature/post_manager

Channel post manager
Tishka17 · Sep 25, 2023 · e66af59 · e66af59
2 parents cc0a01c + 76e84b7
commit e66af59
Show file tree

Hide file tree

Showing 12 changed files with 403 additions and 0 deletions.
diff --git a/README.md b/README.md
@@ -158,3 +158,29 @@ The same behavior is supported in sulguk. Otherwise, you can set the language on
 #### Tags which contents is ignored:
 
 `<head>`, `<link>`, `<meta>`, `<script>`, `<style>`, `<template>`, `<title>`
+
+
+## Command line utility for channel management
+
+1. Install with addons
+```shell
+pip install 'sulguk[cli]'
+```
+
+2. Set environment variable `BOT_TOKEN`
+
+```shell
+export BOT_TOKEN="your telegram token"
+```
+
+3. Send HTML file as a message to your channel. Additional files will be sent as comments to the first one. You can provide a channel name or a public link
+
+```shell
+sulguk send @chat_id file.html
+```
+
+4. If you want to, edit using the link from shell or from your tg client. Edition of comments is supported as well.
+
+```shell
+sulguk edit 'https://t.me/channel/1?comment=42' file.html
+```
diff --git a/pyproject.toml b/pyproject.toml
@@ -21,6 +21,12 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = []
+[project.optional-dependencies]
+cli = [
+    "aiogram",
+]
+[project.scripts]
+sulguk = "sulguk.post_manager.cli:cli"
 
 [project.urls]
 "Homepage" = "https://github.com/tishka17/sulguk"

diff --git a/src/sulguk/post_manager/__init__.py b/src/sulguk/post_manager/__init__.py
diff --git a/src/sulguk/post_manager/chat_info.py b/src/sulguk/post_manager/chat_info.py
@@ -0,0 +1,19 @@
+from logging import getLogger
+from typing import Union
+
+from aiogram import Bot
+from aiogram.exceptions import TelegramBadRequest
+
+from .exceptions import ChatNotFound
+
+logger = getLogger(__name__)
+
+
+async def get_chat(bot: Bot, chat_id: Union[str, int]):
+    try:
+        return await bot.get_chat(chat_id)
+    except TelegramBadRequest as e:
+        if "chat not found" in e.message:
+            logger.error("Chat %s not found", chat_id)
+            raise ChatNotFound
+        raise
diff --git a/src/sulguk/post_manager/cli.py b/src/sulguk/post_manager/cli.py
@@ -0,0 +1,40 @@
+import asyncio
+import logging
+import os
+
+from aiogram import Bot
+
+from .editor import edit
+from .exceptions import ManagerError
+from .params import parse_args
+from .sender import send
+
+
+async def main():
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s - %(levelname)s - %(name)s - %(message)s",
+    )
+    logging.getLogger("aiogram").setLevel(logging.WARNING)
+    bot = Bot(token=os.getenv("BOT_TOKEN"))
+    args = parse_args()
+    try:
+        if args.command == "edit":
+            await edit(bot, args)
+        else:
+            await send(bot, args)
+    except ManagerError:
+        logging.error("There were errors during execution. See above")
+    finally:
+        await bot.session.close()
+
+
+def cli():
+    try:
+        asyncio.run(main())
+    except (KeyboardInterrupt, SystemExit):
+        pass
+
+
+if __name__ == '__main__':
+    cli()
diff --git a/src/sulguk/post_manager/editor.py b/src/sulguk/post_manager/editor.py
@@ -0,0 +1,36 @@
+import logging
+
+from aiogram import Bot
+from aiogram.exceptions import TelegramBadRequest
+
+from .chat_info import get_chat
+from .file import load_file
+from .params import EditArgs
+
+logger = logging.getLogger(__name__)
+
+
+async def edit(bot: Bot, args: EditArgs):
+    chat = await get_chat(bot, args.destination.group_id)
+    if not args.destination.post_id:
+        raise ValueError("No post provided to edit")
+    if args.destination.comment_id:
+        chat_id = chat.linked_chat_id
+        message_id = args.destination.comment_id
+    else:
+        chat_id = chat.id
+        message_id = args.destination.post_id
+
+    data = load_file(args.file)
+    try:
+        await bot.edit_message_text(
+            chat_id=chat_id,
+            message_id=message_id,
+            text=data.text,
+            entities=data.entities,
+        )
+    except TelegramBadRequest as e:
+        if "message is not modified" in e.message:
+            logger.debug("Nothing changed")
+            return
+        raise
diff --git a/src/sulguk/post_manager/exceptions.py b/src/sulguk/post_manager/exceptions.py
@@ -0,0 +1,10 @@
+class ManagerError(Exception):
+    pass
+
+
+class LinkedMessageNotFound(ManagerError):
+    pass
+
+
+class ChatNotFound(ManagerError):
+    pass
diff --git a/src/sulguk/post_manager/file.py b/src/sulguk/post_manager/file.py
@@ -0,0 +1,15 @@
+import logging
+
+from sulguk import transform_html, RenderResult
+from .exceptions import ManagerError
+
+logger = logging.getLogger(__name__)
+
+
+def load_file(filename) -> RenderResult:
+    try:
+        with open(filename) as f:
+            return transform_html(f.read())
+    except FileNotFoundError:
+        logger.error("File `%s` not found", filename)
+        raise ManagerError
diff --git a/src/sulguk/post_manager/links.py b/src/sulguk/post_manager/links.py
@@ -0,0 +1,89 @@
+from dataclasses import dataclass
+from typing import Optional, Union
+from urllib.parse import urlparse, parse_qs
+
+from aiogram.types import Message, Chat
+
+
+@dataclass
+class Link:
+    group_id: Union[str, int]
+    post_id: Optional[int] = None
+    comment_id: Optional[int] = None
+
+
+class LinkParseError(ValueError):
+    pass
+
+
+def parse_group_id(group_id: str) -> Union[int, str]:
+    try:
+        return int(group_id)
+    except ValueError:
+        if group_id.startswith("@"):
+            return group_id
+        return "@" + group_id
+
+
+def parse_link(link: str) -> Link:
+    if not link.startswith("https://"):
+        return Link(group_id=parse_group_id(link))
+    parsed = urlparse(link)
+    if parsed.scheme != "https":
+        raise LinkParseError(f"Invalid scheme: {parsed.scheme}")
+    if parsed.hostname != "t.me":
+        raise LinkParseError(f"Invalid hostname: {parsed.hostname}")
+    path = parsed.path[1:].split("/")  # remove starting /
+    if len(path) == 1:
+        return Link(parse_group_id(path[0]))
+    elif len(path) == 2:
+        params = parse_qs(parsed.query)
+        try:
+            post_id = int(path[1])
+        except ValueError:
+            raise LinkParseError(f"Invalid post id: {path[1]}")
+        comment_id_raw = params.get("comment")
+        if not comment_id_raw:
+            comment_id = None
+        elif len(comment_id_raw) == 1:
+            try:
+                comment_id = int(comment_id_raw[0])
+            except ValueError:
+                raise LinkParseError(f"Invalid comment id: {path[1]}")
+        else:
+            raise LinkParseError(f"Cannot parse comment id: {parsed.query}")
+        return Link(
+            group_id=parse_group_id(path[0]),
+            post_id=post_id,
+            comment_id=comment_id,
+        )
+    else:
+        raise LinkParseError(f"Invalid path: {parsed.path}")
+
+
+def unparse_link(link: Link) -> str:
+    if isinstance(link.group_id, str):
+        group_id = link.group_id.lstrip("@")
+    else:
+        group_id = link.group_id
+    result = f"https://t.me/{group_id}/"
+    if link.post_id:
+        result += f"{link.post_id}"
+    if link.comment_id:
+        result += f"?comment={link.comment_id}"
+    return result
+
+
+def make_link(
+        chat: Chat,
+        message: Optional[Message] = None,
+        comment: Optional[Message] = None,
+) -> Link:
+    link = Link(
+        group_id=chat.username,
+    )
+    if message:
+        link.post_id = message.message_id
+    if comment:
+        link.comment_id = comment.message_id
+    return link
diff --git a/src/sulguk/post_manager/params.py b/src/sulguk/post_manager/params.py
@@ -0,0 +1,46 @@
+from argparse import ArgumentParser
+from typing import List, Literal, Union
+
+from .links import Link, parse_link
+
+
+class SendArgs:
+    command: Literal["send"]
+    mode: Literal["poll", "getChat"]
+    destination: Link
+    file: List[str]
+
+
+class EditArgs:
+    command: Literal["edit"]
+    destination: Link
+    file: str
+
+
+def init_parser():
+    root = ArgumentParser(prog='Sulguk message manager')
+    subparsers = root.add_subparsers(dest="command")
+    sender = subparsers.add_parser("send")
+    sender.add_argument(
+        "destination", type=parse_link
+    )
+    sender.add_argument(
+        "file", nargs='+'
+    )
+    sender.add_argument(
+        "-m", "--mode", choices=["poll", "getChat"],
+        default="poll",
+    )
+    editor = subparsers.add_parser("edit")
+    editor.add_argument(
+        "destination", type=parse_link
+    )
+    editor.add_argument(
+        "file",
+    )
+    return root
+
+
+def parse_args() -> Union[SendArgs, EditArgs]:
+    parser = init_parser()
+    return parser.parse_args()