[大数据]python读取consul配置及用特定规则替换其中url

安装：

pip install python-consul
pip install pybase62

import re

import base62
import consul


class ConsulInfo:
    def __init__(self, router_adds):
        self.host = "consul.xxxx.local" # consul的url
        self.port = "xxxx" # 端口
        self.token = "xxxx" # 登录consul的token
        self.router_adds = "config/" + router_adds # 读取consul目的文件配置的url
        self.client = consul.Consul(self.host, self.port, token=self.token, verify=False) # 连接consul

    def replace_url(self, url_):
        #替换consul配置中url
        consul_data = self.get_consul_data()
        if not consul_data: return False
        consul_txt = ''
        for index, i in enumerate(consul_data.splitlines()):
        # 读取consul配置文件中每一行
            pattern = re.compile(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+')
            txt_ = re.findall(pattern, i) # 查找每行中是否有http://或者https://开头的url
            if not txt_ or txt_[0].startswith(url_):
            # 如果没有或者是已替换过的url,则直接添加
                consul_txt = consul_txt + i + "\n"
                continue
                
            # 用自己定义的url加上原consul中url经base62转义的值组成新的url
            new_url = url_ + base62.encodebytes(txt_[0].encode()) 
            consul_txt1 = re.sub(pattern, new_url, i)
            consul_txt = consul_txt + consul_txt1 + "\n"
        self.update_consul(bytes(consul_txt, encoding='utf8')) # 更新consul配置
        return True

    def get_consul_data(self):
        # 获取consul配置数据
        _, data = self.client.kv.get(self.router_adds)
        return str(data["Value"], encoding="utf8")

    def update_consul(self, txt):
        # 更新consul配置数据
        response = self.client.kv.put(self.router_adds, value=str(txt, encoding="utf8").encode("utf-8"))
        assert response is True

ConsulInfo("xxxx:debug/data").replace_url("http://22.33.44.55:345/")

运行前：

运行后：