pubsub-purge.py

#!/usr/bin/python3
import asyncio
import getpass
import sys

import aioxmpp

async def amain(address, password, override_peer, no_verify, pubsub_service, pubsub_node):
    client = aioxmpp.PresenceManagedClient(
        address,
        aioxmpp.make_security_layer(password, no_verify=no_verify),
        override_peer=override_peer,
    )

    disco = client.summon(aioxmpp.DiscoClient)
    pubsub = client.summon(aioxmpp.PubSubClient)

    async with client.connected() as stream:
        print("connected")
        await pubsub.purge(pubsub_service, pubsub_node)


def jid(s):
    return aioxmpp.JID.fromstr(s)


def main():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "address",
        help="The JID/address to log in with",
        type=jid,
    )

    parser.add_argument(
        "--host", "-H",
        default=None,
        help="Override the host to connect to",
    )

    parser.add_argument(
        "--port", "-p",
        type=int,
        default=5222,
        help="If --host is given, override the port, too (default: 5222)"
    )

    parser.add_argument(
        "--no-verify",
        action="store_true",
        help="Disable TLS verification (don’t)",
        default=False,
    )

    parser.add_argument(
        "pubsub_service",
        help="Address of the pubsub service",
        type=jid,
    )

    parser.add_argument(
        "pubsub_node",
        help="Name of the pubsub node to purge",
    )

    args = parser.parse_args()

    override_peer = []
    if args.host is not None:
        override_peer.append((args.host, args.port, aioxmpp.connector.STARTTLSConnector()))

    password = getpass.getpass()

    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(amain(
            args.address, password,
            override_peer,
            args.no_verify,
            args.pubsub_service,
            args.pubsub_node,
        ))
    finally:
        loop.close()


if __name__ == "__main__":
    sys.exit(main() or 0)