test_lark_one_click_deploy.py

test_lark_one_click_deploy.py

import asyncio
import logging
import webbrowser

from urllib.parse import quote

import httpx
from uuid import uuid4

from connectai.lark.websocket import WS_LARK_PROXY_SERVER, WS_LARK_PROXY_PROTOCOL


LARK_DEPLOY_SERVER = 'https://deploy.ai2e.cn/feishu/app/internal'


class DeployClient(object):

    def __init__(self, store):
        self.store = store
        self.init_channel()

    def random_string(self):
        return str(uuid4()).replace('-', '')

    def __getattr__(self, name, default=''):
        value = self.store.get(name) or default
        if not value and name in ['encrypt_key', 'verification_token']:
            value = self.random_string()
            self.store.set(name, value)
        return value

    def __setattr__(self, name, value):
        if name not in ['store', 'deploy_channel', 'request_channel']:
            self.store.set(name, value)
        super().__setattr__(name, value)

    def init_channel(self):
        self.deploy_channel = self.random_string()
        self.request_channel = self.random_string()


    def get_hool_url(self):
        return f"{WS_LARK_PROXY_PROTOCOL}://{WS_LARK_PROXY_SERVER}/hook/{self.deploy_channel}?request_id={self.request_channel}"

    def get_url(self, deploy=True):
        return f"{WS_LARK_PROXY_PROTOCOL}://{WS_LARK_PROXY_SERVER}/sub/{self.deploy_channel if deploy else self.request_channel}"

    def get_callback_url(self):
        return f"{WS_LARK_PROXY_PROTOCOL}://{WS_LARK_PROXY_SERVER}/hook/{self.app_id}"

    async def start(self):
        await asyncio.gather(
            self.start_brower(),
            self.start_client(),
        )
    
    async def start_brower(self):
        hook_url = self.get_hool_url()
        deploy_url = f"{LARK_DEPLOY_SERVER}?redirect_uri={quote(hook_url)}&app_id={self.app_id}&name={self.name}&desc={self.desc}&avatar={self.avatar}"
        logging.error("debug deploy_url %r", deploy_url)
        webbrowser.open(deploy_url)

    async def start_client(self):
        channel_url, request_url = self.get_url(), self.get_url(False)
        # wait app_id and so app_secret
        response = httpx.get(channel_url, timeout=60).json()
        # 模拟拿到app_secret
        # save data, and jump publish url
        if response.get('body', {}).get('app_secret', ''):
            self.app_secret = response.get('body', {}).get('app_secret', '')
            self.app_id = response.get('body', {}).get('app_id', '')
        # 重置两个channel
        # self.init_channel()
        hook_url = self.get_hool_url()
        events = [
            "20",
            "im.message.message_read_v1",
            "im.message.reaction.created_v1",
            "im.message.reaction.deleted_v1",
            "im.message.recalled_v1",
            "im.message.receive_v1",
        ]
        scope_ids = [
            "8002",
            "100032",
            "6081",
            "14",
            "1",
            "21001",
            "20001",
            "20011",
            "3001",
            "20012",
            "20010",
            "3000",
            "20008",
            "1000",
            "20009",
        ]
        callback_url = self.get_callback_url()
        redirect_url = f"{LARK_DEPLOY_SERVER}/publish?redirect_uri={quote(hook_url)}&app_id={self.app_id}&events={','.join(events)}&encrypt_key={self.encrypt_key}&&verification_token={self.verification_token}&scopes={','.join(scope_ids)}&hook_url={callback_url}"
        html = f'''
<h1>redirect to publish</h1>
<script>
location.href = '{redirect_url}'
</script>
        '''
        response = httpx.post(request_url, data=html, headers={
            'Content-Type': 'text/html;charset=utf-8',
            'If-None-Match': '-1'
        })
        while True:
            mixin_url = f"{channel_url},{self.app_id}"
            response = httpx.get(mixin_url, timeout=600).json()
            logging.debug('get mixin_url %r %r', mixin_url, response)
            if 'challenge' in response.get('body', {}):
                request_id = response['headers']['x-request-id']
                challenge_url = f"{WS_LARK_PROXY_PROTOCOL}://{WS_LARK_PROXY_SERVER}/sub/{request_id}"
                response = httpx.post(challenge_url, json=dict(challenge=response['body']['challenge']))
            else:
                # 需要返回数据才算结束
                html = f'''
<h1>success to publish</h1>
<script>
window.close()
</script>
                '''
                response = httpx.post(request_url, data=html, headers={
                    'Content-Type': 'text/html;charset=utf-8',
                    'If-None-Match': '-1'
                })
                break


if __name__ == "__main__":
    import click
    from dotenv import find_dotenv, load_dotenv, set_key, dotenv_values

    dotfile = find_dotenv() or '.env'
    config = dotenv_values(dotfile)

    class DotEnvStore(object):
        def __init__(self, **kwargs):
            self.dotfile = find_dotenv()
            for k, v in kwargs.items():
                if v:
                    self.set(k, v)
            self.load()

        def load(self):
            self.config = dotenv_values(self.dotfile)

        def get(self, name, default=None):
            return self.config.get(name) or default

        def set(self, name, value):
            set_key(self.dotfile, name, value)
            self.load()

        def __str__(self):
            return '\n'.join([f'{k}: {v}' for k, v in self.config.items()])


    @click.command()
    @click.option('--app_id', required=False, default=config.get('app_id'))
    @click.option('--name', prompt="Name", help='Your App Name', default=config.get('app_name'))
    @click.option('--desc', prompt="Description", help='Your App Description', default=config.get('app_description'))
    def main(app_id, name, desc):
        store = DotEnvStore(app_id=app_id, name=name, desc=desc)
        print('app_id', store.get('app_id'))
        client = DeployClient(store)
        asyncio.run(client.start())

    main()