first commit

.github/workflows/python-cli.yml

@@ -0,0 +1,56 @@
+name: CI
+
+on:
+  push:
+    branches: [ master, main ]
+    tags:
+      - 'v*.*.*'
+  pull_request:
+    branches: [ master, main ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.8', '3.11']
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install ".[dev]"
+    - name: Check code with black
+      run: |
+        black --check .
+    - name: Lint code
+      run: |
+        pylama
+    - name: Test with pytest
+      run: |
+        #pytest
+
+  deploy:
+    needs: test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+        with:
+          python-version: '3.x'
+      - id: check-tag
+        run: |
+          if [[ "${{ github.event.ref }}" =~ ^refs/tags/v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
+              echo ::set-output name=match::true
+          fi
+      - name: Create PyPI release
+        uses: casperdcl/deploy-pypi@v2
+        with:
+          password: ${{ secrets.PYPI_TOKEN }}
+          pip: true
+          # only upload if a tag is pushed (otherwise just build & check)
+          upload: ${{ github.event_name == 'push' && steps.check-tag.outputs.match == 'true' }}

.gitignore

@@ -0,0 +1,12 @@
+__pycache__
+.cache
+.mypy_cache
+.pytest_cache
+build
+dist
+.eggs
+*.egg-info
+
+*~
+.#*
+*#

README.md

@@ -0,0 +1,35 @@
+# dcstatus
+
+[![Latest Release](https://img.shields.io/pypi/v/dcstatus.svg)](https://pypi.org/project/dcstatus)
+[![CI](https://github.com/deltachat-bot/dcstatus/actions/workflows/python-ci.yml/badge.svg)](https://github.com/deltachat-bot/dcstatus/actions/workflows/python-ci.yml)
+[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
+
+Bot to get Delta Chat releases status information.
+
+## Install
+
+```sh
+pip install dcstatus
+```
+
+Then, to setup [Playwright](https://playwright.dev/python/docs/intro), run:
+
+```sh
+playwright install
+```
+
+## Usage
+
+Configure the bot:
+
+```sh
+dcstatus init [email protected] PASSWORD
+```
+
+Start the bot:
+
+```sh
+dcstatus serve
+```
+
+Run `dcstatus --help` to see all available options.

dcstatus/__init__.py

@@ -0,0 +1,11 @@
+"""bot's main entry point"""
+
+from .hooks import cli
+
+
+def main() -> None:
+    """Run the application."""
+    try:
+        cli.start()
+    except KeyboardInterrupt:
+        pass

dcstatus/__main__.py

@@ -0,0 +1,5 @@
+"""Support for package execution."""
+
+from . import main
+
+main()

dcstatus/changelog.py

@@ -0,0 +1,61 @@
+"""Extraction of release versions and DC core versions from apps CHANGELOG"""
+
+import re
+from re import Pattern
+
+from .web import session
+
+ORG_URL = "https://raw.githubusercontent.com/deltachat"
+
+
+def fetch_changelog(
+    url: str, app_regex: Pattern, core_regex: Pattern, count: int
+) -> list[tuple[str, str]]:
+    versions = []
+    app = ""
+    core = ""
+    with session.get(url) as resp:
+        for line in resp.text.splitlines():
+            line = line.strip()
+            if match := app_regex.match(line):
+                if app:
+                    versions.append((app, "unknown"))
+                app = match.group("app").strip()
+            if match := core_regex.match(line):
+                core = match.group("core").strip()
+                if app:  # otherwise it is an orphan core from Unreleased section
+                    versions.append((app, core))
+                app = ""
+            if len(versions) >= count:
+                break
+    return versions
+
+
+def get_android_changelog(count: int) -> list[tuple[str, str]]:
+    url = f"{ORG_URL}/deltachat-android/refs/heads/main/CHANGELOG.md"
+    app_regex = re.compile(r"## v(?P<app>\d+\.\d+\.\d+.*)")
+    core_regex = re.compile(
+        r"(\*|-) (using core|update to core|update core( to)?) ?(?P<core>.+)",
+        re.IGNORECASE,
+    )
+    return fetch_changelog(url, app_regex, core_regex, count)
+
+
+def get_ios_changelog(count: int) -> list[tuple[str, str]]:
+    url = f"{ORG_URL}/deltachat-ios/refs/heads/main/CHANGELOG.md"
+    app_regex = re.compile(r"## v(?P<app>\d+\.\d+\.\d+.*)")
+    core_regex = re.compile(
+        r"(\*|-) (update to core|update core( to)?|using core) ?(?P<core>.+)",
+        re.IGNORECASE,
+    )
+    return fetch_changelog(url, app_regex, core_regex, count)
+
+
+def get_desktop_changelog(count: int) -> list[tuple[str, str]]:
+    url = f"{ORG_URL}/deltachat-desktop/refs/heads/main/CHANGELOG.md"
+    app_regex = re.compile(r"## \[(?P<app>\d+\.\d+\.\d+)\].*")
+    core_regex = re.compile(
+        r"(\*|-) update `@deltachat/stdio-rpc-server`.* to `?(?P<core>.+)`",
+        re.IGNORECASE,
+    )
+    return fetch_changelog(url, app_regex, core_regex, count)

dcstatus/hooks.py

@@ -0,0 +1,108 @@
+"""Event Hooks"""
+
+import os
+from argparse import Namespace
+
+from cachelib import FileSystemCache, NullCache
+from deltabot_cli import BotCli
+from deltachat2 import Bot, ChatType, CoreEvent, EventType, MsgData, NewMsgEvent, events
+from rich.logging import RichHandler
+
+from .status import get_status
+
+cli = BotCli("dcstatus")
+cli.add_generic_option(
+    "--no-time",
+    help="do not display date timestamp in log messages",
+    action="store_false",
+)
+cache = NullCache()
+
+
+@cli.on_init
+def on_init(bot: Bot, args: Namespace) -> None:
+    bot.logger.handlers = [
+        RichHandler(show_path=False, omit_repeated_times=False, show_time=args.no_time)
+    ]
+    for accid in bot.rpc.get_all_account_ids():
+        if not bot.rpc.get_config(accid, "displayname"):
+            bot.rpc.set_config(accid, "displayname", "Delta Chat Status")
+            status = "I'm a bot, send /help to me for more info"
+            bot.rpc.set_config(accid, "selfstatus", status)
+            bot.rpc.set_config(accid, "delete_device_after", "3600")
+
+
+@cli.on_start
+def on_start(_bot: Bot, args: Namespace) -> None:
+    global cache
+    path = os.path.join(args.config_dir, "cache")
+    if not os.path.exists(path):
+        os.makedirs(path)
+    cache = FileSystemCache(path, default_timeout=60 * 60 * 3)
+
+
+@cli.on(events.RawEvent)
+def log_event(bot: Bot, accid: int, event: CoreEvent) -> None:
+    if event.kind == EventType.INFO:
+        bot.logger.debug(event.msg)
+    elif event.kind == EventType.WARNING:
+        bot.logger.warning(event.msg)
+    elif event.kind == EventType.ERROR:
+        bot.logger.error(event.msg)
+    elif event.kind == EventType.MSG_DELIVERED:
+        bot.rpc.delete_messages(accid, [event.msg_id])
+    elif event.kind == EventType.SECUREJOIN_INVITER_PROGRESS:
+        if event.progress == 1000 and not is_bot(bot, accid, event.contact_id):
+            bot.logger.debug("QR scanned by contact id=%s", event.contact_id)
+            chatid = bot.rpc.create_chat_by_contact_id(accid, event.contact_id)
+            send_help(bot, accid, chatid)
+
+
+@cli.on(events.NewMessage(command="/help"))
+def _help(bot: Bot, accid: int, event: NewMsgEvent) -> None:
+    bot.rpc.markseen_msgs(accid, [event.msg.id])
+    send_help(bot, accid, event.msg.chat_id)
+
+
+@cli.on(events.NewMessage(command="/status"))
+def _status(bot: Bot, accid: int, event: NewMsgEvent) -> None:
+    bot.rpc.markseen_msgs(accid, [event.msg.id])
+    text = "Delta Chat Status"
+    html = get_status(cache)
+    bot.rpc.send_msg(accid, event.msg.chat_id, MsgData(text=text, html=html))
+
+
+@cli.on(events.NewMessage(is_info=False))
+def on_message(bot: Bot, accid: int, event: NewMsgEvent) -> None:
+    if bot.has_command(event.command):
+        return
+
+    msg = event.msg
+    chat = bot.rpc.get_basic_chat_info(accid, msg.chat_id)
+    if chat.chat_type == ChatType.SINGLE:
+        bot.rpc.markseen_msgs(accid, [msg.id])
+        send_help(bot, accid, event.msg.chat_id)
+
+
+@cli.after(events.NewMessage)
+def delete_msgs(bot, accid, event):
+    bot.rpc.delete_messages(accid, [event.msg.id])
+
+
+def send_help(bot: Bot, accid: int, chatid: int) -> None:
+    text = (
+        "👋 hi, I'm a bot, you can send /status to me"
+        " to get the status of Delta Chat releases."
+    )
+    bot.rpc.send_msg(accid, chatid, MsgData(text=text))
+
+
+def is_bot(bot: Bot, accid: int, contactid: int) -> bool:
+    return bot.rpc.get_contact(accid, contactid).is_bot
+
+
+if __name__ == "__main__":
+    try:
+        cli.start()
+    except KeyboardInterrupt:
+        pass