From 691d920c2bedaeec58b6b17364fe53e9cce02cc9 Mon Sep 17 00:00:00 2001 From: Morgan McMillian Date: Wed, 25 Dec 2024 06:29:41 -0800 Subject: [PATCH] add cli for various odd tasks --- .gitignore | 1 + pyproject.toml | 3 +- src/pnut_matrix/cmd.py | 334 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 337 insertions(+), 1 deletion(-) create mode 100755 src/pnut_matrix/cmd.py diff --git a/.gitignore b/.gitignore index c731d5f..37bf7d9 100644 --- a/.gitignore +++ b/.gitignore @@ -60,6 +60,7 @@ target/ # my other cruft *.yaml +*.yml *.db *.sublime-project *.sublime-workspace diff --git a/pyproject.toml b/pyproject.toml index 8fcd659..6b77621 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,4 +34,5 @@ build-backend = "setuptools.build_meta" [project.entry-points.console_scripts] pnutservice = "pnut_matrix.pnutservice:main" -matrixappsvc = "pnut_matrix.appservice:main" +appservice = "pnut_matrix.appservice:main" +ascmd = "pnut_matrix.cmd:cmd" diff --git a/src/pnut_matrix/cmd.py b/src/pnut_matrix/cmd.py new file mode 100755 index 0000000..bd40a1a --- /dev/null +++ b/src/pnut_matrix/cmd.py @@ -0,0 +1,334 @@ +#!/usr/bin/env python3 + +import logging +import asyncio +import asyncclick as click +import requests +import pnutpy +import json +import yaml + +from mautrix.client import ClientAPI +from mautrix.types import TextMessageEventContent, Format, MessageType, EventType +from mautrix.errors import MatrixConnectionError +from mautrix.errors.request import MNotFound, MForbidden + +PNUT_API="https://api.pnut.io/v1" + +@click.group() +@click.option('--debug', '-d', is_flag=True) +@click.option('--config', '-c', required=True) +@click.pass_context +async def cmd(ctx, debug, config): + if debug: + logging.basicConfig(level=logging.DEBUG) + + ctx.ensure_object(dict) + with open(config, "rb") as config_file: + ctx.obj['config'] = yaml.load(config_file, Loader=yaml.SafeLoader) + +@cmd.command() +@click.pass_context +async def get_streams(ctx): + config = ctx.obj['config'] + logging.debug(config) + endpoint = "/streams" + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['PNUT_APPTOKEN'] + } + + url = f"{PNUT_API}{endpoint}" + r = requests.get(url, headers=headers) + + if r.status_code == 200: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.argument('key') +@click.pass_context +def rm_stream(ctx, key): + config = ctx.obj['config'] + logging.debug(config) + endpoint = f"/streams/{key}" + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['PNUT_APPTOKEN'] + } + + url = f"{PNUT_API}{endpoint}" + r = requests.delete(url, headers=headers) + + if r.status_code == 200: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.pass_context +def rm_streams(ctx): + config = ctx.obj['config'] + logging.debug(config) + endpoint = f"/streams" + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['PNUT_APPTOKEN'] + } + + url = f"{PNUT_API}{endpoint}" + r = requests.delete(url, headers=headers) + + if r.status_code == 200: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.argument('key') +@click.argument('object_types') +@click.pass_context +def new_stream(ctx, key, object_types): + config = ctx.obj['config'] + endpoint = f"/streams" + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['PNUT_APPTOKEN'] + } + otypes = [x.strip() for x in object_types.split(',')] + data = { + 'type': "long_poll", + 'object_types': otypes, + } + + url = f"{PNUT_API}{endpoint}" + r = requests.post(url, headers=headers, json=data) + + if r.status_code == 201: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.argument('key') +@click.argument('object_types') +@click.pass_context +def update_stream(ctx, key, object_types): + config = ctx.obj['config'] + endpoint = f"/streams/{key}" + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['PNUT_APPTOKEN'] + } + otypes = [x.strip() for x in object_types.split(',')] + data = { + 'object_types': otypes, + } + + url = f"{PNUT_API}{endpoint}" + r = requests.put(url, headers=headers, json=data) + + if r.status_code == 200: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +# TODO: this needs debugging +@cmd.command() +@click.pass_context +def pnut_app_token(ctx): + config = ctx.obj['config'] + logging.debug(config) + # endpoint = f"/oauth/access_token" + # headers = { + # "Content-Type": "application/json", + # } + # data = { + # 'client_id': config['PNUTCLIENT_ID'], + # 'client_secret': config['PNUTCLIENT_SECRET'], + # 'grant_type': "client_credentials" + # } + # + # url = f"{PNUT_API}{endpoint}" + # r = requests.post(url, headers=headers, json=data) + # + # if r.status_code == 200: + # click.echo(json.dumps(r.json(), indent=4)) + # + # else: + # click.echo(r.status_code) + # click.echo(r.text) + +@cmd.command() +@click.argument('username') +@click.pass_context +def new_matrix_asuser(ctx, username): + config = ctx.obj['config'] + endpoint = "/_matrix/client/v3/register" + url = config['MATRIX_HOST'] + endpoint + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['MATRIX_AS_TOKEN'] + } + params = {'kind': 'user'} + data = {'type': 'm.login.application_service','username': username} + + r = requests.post(url, headers=headers, json=data, params=params) + if r.status_code == 200: + click.echo("SUCCESS!") + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.argument('invitee') +@click.pass_context +def new_pnut_admin_room(ctx, invitee): + config = ctx.obj['config'] + endpoint = "/_matrix/client/v3/createRoom" + url = config['MATRIX_HOST'] + endpoint + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['MATRIX_AS_TOKEN'] + } + data = { + 'visibility': "private", + 'is_direct': False, + 'name': "Pnut Bridge Admin Room", + 'room_alias_name': f"{config['MATRIX_PNUT_PREFIX']}admin", + 'invite': [invitee], + 'power_level_content_override': { + 'users': { + f"{config['MATRIX_AS_ID']}": 100, + f"{invitee}": 100 + } + } + } + logging.debug(data) + r = requests.post(url, headers=headers, json=data) + + if r.status_code == 200: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.pass_context +def new_pnut_global_room(ctx): + config = ctx.obj['config'] + endpoint = "/_matrix/client/v3/createRoom" + url = config['MATRIX_HOST'] + endpoint + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['MATRIX_AS_TOKEN'] + } + # data = { + # 'visibility': "public", + # 'name': "Pnut Global Stream", + # 'room_alias_name': "pnut_global", + # 'power_level_content_override': { + # "events_default": 2, + # "invite": 2, + # } + # } + data = { + 'visibility': "public", + 'name': "Pnut Global Stream", + 'room_alias_name': f"{config['MATRIX_PNUT_PREFIX']}global" + } + logging.debug(data) + r = requests.post(url, headers=headers, json=data) + + if r.status_code == 200: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.argument('room_id') +@click.argument('matrix_id') +@click.argument('power_level') +@click.pass_context +def elevate_matrix_user(ctx, room_id, matrix_id, power_level): + config = ctx.obj['config'] + endpoint = f"/_matrix/client/v3/rooms/{room_id}/state/" + url = config['MATRIX_HOST'] + endpoint + "m.room.power_levels" + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + config['MATRIX_AS_TOKEN'] + } + data = { + 'users': { + f"{config['MATRIX_AS_ID']}": 100, + f"{matrix_id}": int(power_level) + } + } + logging.debug(data) + r = requests.put(url, headers=headers, json=data) + + if r.status_code == 200: + click.echo(json.dumps(r.json(), indent=4)) + + else: + click.echo(r.status_code) + click.echo(r.text) + +@cmd.command() +@click.pass_context +async def list_joined_rooms(ctx): + config = ctx.obj['config'] + matrix_api = ClientAPI(config['MATRIX_AS_ID'], + base_url=config['MATRIX_HOST'], + token=config['MATRIX_AS_TOKEN']) + room_list = await matrix_api.get_joined_rooms() + for room_id in room_list: + click.echo(room_id) + + try: + room_name = await matrix_api.get_state_event(room_id, + EventType.ROOM_NAME) + click.echo(f'name: {room_name.name}') + except MNotFound: + pass + + try: + room_alias = await matrix_api.get_state_event(room_id, + EventType.ROOM_CANONICAL_ALIAS) + click.echo(f'alias: {room_alias.canonical_alias}') + except MNotFound: + pass + + # members = await matrix_api.get_joined_members(room_id) + # click.echo(members) + click.echo('-----------') + +@cmd.command() +@click.argument('room_id') +@click.pass_context +async def part_room(ctx, room_id): + config = ctx.obj['config'] + matrix_api = ClientAPI(config['MATRIX_AS_ID'], + base_url=config['MATRIX_HOST'], + token=config['MATRIX_AS_TOKEN']) + + # TODO: need to clear alias + await matrix_api.leave_room(room_id) + +if __name__ == '__main__': + cmd(_anyio_backend="asyncio")