update

.gitattributes

@@ -32,4 +32,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.xz filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.jpg filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,12 @@
+
+
+.git/
+.idea/
+
+cache/
+dotenv/
+trained_models/
+**/cache/
+**/__pycache__/
+
+**/*.jpg

README.md

@@ -1,8 +1,8 @@
 ---
-title: Dingtalk
-emoji: 🐢
-colorFrom: red
-colorTo: gray
+title: Ding Talk
+emoji: 🔥
+colorFrom: purple
+colorTo: red
 sdk: docker
 pinned: false
 license: apache-2.0

dingtalk_develop.md

@@ -0,0 +1,47 @@
+## 钉钉机器人开发
+
+### 操作流程
+
+#### 创建机器人
+
+参考文档：
+```text
+https://open.dingtalk.com/document/orgapp/create-an-application
+```
+
+我创建的机器人是像一个用户一样的，你可以发信息给它，然后它会回复你。
+
+步骤大致如下：
+
+首先访问 [钉钉开放平台](https://open-dev.dingtalk.com/)，并登录。
+
+点击上方菜单栏的 [应用开发](https://open-dev.dingtalk.com/fe/app#/corp/app)，
+![app_develop.jpg](docs/pictures/app_develop.jpg)
+
+然后在钉钉应用中点击创建应用，输入 `应用名称`，`应用描述`，`应用图标` 后保存。
+![create_app.jpg](docs/pictures/create_app.jpg)
+![create_app_save.jpg](docs/pictures/create_app_save.jpg)
+
+点击进入创建的应用。
+![in_app.jpg](docs/pictures/in_app.jpg)
+
+在 `添加应用能力` 中 找到 `其它应用能力`中点击添加 `机器人`。
+对机器人配置完成后点发布。
+
+在凭证与基础信息中获取 Client ID, Client Secret
+![client_info.jpg](docs/pictures/client_info.jpg)
+
+
+#### 创建服务后台
+
+参考文档：
+```text
+https://open.dingtalk.com/document/orgapp/robot-receive-message
+https://github.com/open-dingtalk/dingtalk-tutorial-python
+```
+
+
+### 使用效果
+
+聊天示例：
+![talk_example.jpg](docs/pictures/talk_example.jpg)

examples/echo_text_bot.py

@@ -0,0 +1,58 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import logging
+
+import dingtalk_stream
+from dingtalk_stream import AckMessage
+
+from project_settings import project_path
+from project_settings import environment
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--client_id",
+        default=environment.get("client_id"),
+        type=str,
+    )
+    parser.add_argument(
+        "--client_secret",
+        default=environment.get("client_secret"),
+        type=str,
+    )
+    args = parser.parse_args()
+    return args
+
+
+class EchoTextHandler(dingtalk_stream.ChatbotHandler):
+    def __init__(self):
+        super(EchoTextHandler, self).__init__()
+
+    async def process(self, callback: dingtalk_stream.CallbackMessage):
+        incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
+        text = incoming_message.text.content.strip()
+        self.reply_text(text, incoming_message)
+        return AckMessage.STATUS_OK, "OK"
+
+
+def main():
+    args = get_args()
+
+    credential = dingtalk_stream.Credential(
+        client_id=args.client_id,
+        client_secret=args.client_secret,
+    )
+    client = dingtalk_stream.DingTalkStreamClient(credential)
+
+    client.register_callback_handler(
+        dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
+        EchoTextHandler()
+    )
+    client.start_forever()
+    return
+
+
+if __name__ == '__main__':
+    main()

examples/nxlink_bot.py

@@ -0,0 +1,97 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import logging
+
+import dingtalk_stream
+from dingtalk_stream import AckMessage
+
+from project_settings import project_path
+from project_settings import environment
+from toolbox.agent_x.question_answer import AgentX
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--client_id",
+        default=environment.get("client_id"),
+        type=str,
+    )
+    parser.add_argument(
+        "--client_secret",
+        default=environment.get("client_secret"),
+        type=str,
+    )
+    parser.add_argument(
+        "--agent_x_api_key",
+        default=environment.get("agent_x_api_key", default=None),
+        type=str
+    )
+    parser.add_argument(
+        "--agent_x_agent_name",
+        default=environment.get("agent_x_agent_name", default=None),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+class EchoTextHandler(dingtalk_stream.ChatbotHandler):
+    def __init__(self):
+        super(EchoTextHandler, self).__init__()
+
+    async def process(self, callback: dingtalk_stream.CallbackMessage):
+        incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
+        text = incoming_message.text.content.strip()
+        self.reply_text(text, incoming_message)
+        return AckMessage.STATUS_OK, "OK"
+
+
+class AgentXNXLinkHandler(dingtalk_stream.ChatbotHandler):
+    def __init__(self, agent: AgentX):
+        super(AgentXNXLinkHandler, self).__init__()
+        self.agent = agent
+
+    async def process(self, callback: dingtalk_stream.CallbackMessage):
+        incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
+        text = incoming_message.text.content.strip()
+
+        response = self.agent.question_answer(
+            question=text,
+            streaming=True
+        )
+        answer = response["answer"]
+        answer = answer.split("{\"message\":")[0]
+
+        self.reply_text(answer, incoming_message)
+        return AckMessage.STATUS_OK, "OK"
+
+
+def main():
+    args = get_args()
+
+    agent = AgentX(
+        api_key=args.agent_x_api_key,
+        agent_name=args.agent_x_agent_name,
+    )
+
+    agent_x_nxlink_handler = AgentXNXLinkHandler(agent=agent)
+
+    client = dingtalk_stream.DingTalkStreamClient(
+        credential=dingtalk_stream.Credential(
+            client_id=args.client_id,
+            client_secret=args.client_secret,
+        )
+    )
+
+    client.register_callback_handler(
+        dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
+        agent_x_nxlink_handler
+    )
+    client.start_forever()
+    return
+
+
+if __name__ == '__main__':
+    main()

main.py

@@ -0,0 +1,191 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import logging
+import os
+import platform
+import re
+import string
+from typing import List, Tuple
+
+from project_settings import project_path
+
+os.environ["HUGGINGFACE_HUB_CACHE"] = (project_path / "cache/huggingface/hub").as_posix()
+
+logging.basicConfig(
+    level=logging.INFO if platform.system() == "Windows" else logging.INFO,
+    format="%(asctime)s %(levelname)s %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+
+logger = logging.getLogger(__name__)
+
+import dingtalk_stream
+from dingtalk_stream import AckMessage
+import gradio as gr
+from threading import Thread
+import torch
+from transformers.models.gpt2.modeling_gpt2 import GPT2LMHeadModel
+from transformers.models.bert.tokenization_bert import BertTokenizer
+
+from project_settings import environment
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--client_id",
+        default=environment.get("client_id"),
+        type=str,
+    )
+    parser.add_argument(
+        "--client_secret",
+        default=environment.get("client_secret"),
+        type=str,
+    )
+    parser.add_argument(
+        "--model_name",
+        default=(project_path / "trained_models/lib_service_4chan").as_posix() if platform.system() == "Windows" else "qgyd2021/lip_service_4chan",
+        type=str,
+    )
+    parser.add_argument(
+        "--dingtalk_develop_md_file",
+        default="dingtalk_develop.md",
+        type=str,
+    )
+    args = parser.parse_args()
+    return args
+
+
+class LipService4ChanHandler(dingtalk_stream.ChatbotHandler):
+    def __init__(self,
+                 model_name: str = "qgyd2021/lip_service_4chan",
+                 max_input_len: int = 512,
+                 max_new_tokens: int = 512,
+                 top_p: float = 0.9,
+                 temperature: float = 0.35,
+                 repetition_penalty: float = 1.0,
+                 device: str = "cuda" if torch.cuda.is_available() else "cpu",
+                 ):
+        super(LipService4ChanHandler, self).__init__()
+        self.model_name = model_name
+        self.max_input_len = max_input_len
+        self.max_new_tokens = max_new_tokens
+        self.top_p = top_p
+        self.temperature = temperature
+        self.repetition_penalty = repetition_penalty
+        self.device = device
+
+        tokenizer = BertTokenizer.from_pretrained(model_name)
+        model = GPT2LMHeadModel.from_pretrained(model_name)
+        model = model.eval()
+        self.model = model
+        self.tokenizer = tokenizer
+
+    async def process(self, callback: dingtalk_stream.CallbackMessage):
+        incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
+        text = incoming_message.text.content.strip()
+
+        answer = self.get_answer(text)
+        self.reply_text(answer, incoming_message)
+
+        logger.info("incoming message: {}; reply text: {}".format(text, answer))
+
+        return AckMessage.STATUS_OK, "OK"
+
+    @staticmethod
+    def remove_space_between_cn_en(text: str):
+        splits = re.split(" ", text)
+        if len(splits) < 2:
+            return text
+
+        result = ""
+        for t in splits:
+            if t == "":
+                continue
+            if re.search(f"[a-zA-Z0-9{string.punctuation}]$", result) and re.search("^[a-zA-Z0-9]", t):
+                result += " "
+                result += t
+            else:
+                if not result == "":
+                    result += t
+                else:
+                    result = t
+
+        if text.endswith(" "):
+            result += " "
+        return result
+
+    def get_answer(self, text: str):
+        prompt_encoded = self.tokenizer.__call__(text, add_special_tokens=True)
+        input_ids: List[int] = prompt_encoded["input_ids"]
+        input_ids = torch.tensor([input_ids], dtype=torch.long)
+        input_ids = input_ids[:, -self.max_input_len:]
+
+        self.tokenizer.eos_token = self.tokenizer.sep_token
+        self.tokenizer.eos_token_id = self.tokenizer.sep_token_id
+
+        # generate
+        with torch.no_grad():
+            outputs = self.model.generate(
+                input_ids=input_ids,
+                max_new_tokens=self.max_new_tokens,
+                do_sample=True,
+                top_p=self.top_p,
+                temperature=self.temperature,
+                repetition_penalty=self.repetition_penalty,
+                eos_token_id=self.tokenizer.sep_token_id,
+                pad_token_id=self.tokenizer.pad_token_id,
+            )
+            outputs = outputs.tolist()[0][len(input_ids[0]):]
+            answer = self.tokenizer.decode(outputs)
+            answer = answer.strip().replace(self.tokenizer.eos_token, "").strip()
+            answer = self.remove_space_between_cn_en(answer)
+
+        return answer
+
+
+def dingtalk_server(client: dingtalk_stream.DingTalkStreamClient):
+    client.start_forever()
+
+
+def main():
+    args = get_args()
+
+    # ding talk
+    credential = dingtalk_stream.Credential(
+        client_id=args.client_id,
+        client_secret=args.client_secret,
+    )
+    client = dingtalk_stream.DingTalkStreamClient(credential, logger)
+
+    client.register_callback_handler(
+        dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
+        LipService4ChanHandler(
+            model_name=args.model_name
+        )
+    )
+    # client.start_forever()
+
+    # background task
+    thread = Thread(target=dingtalk_server, kwargs={"client": client})
+    thread.start()
+
+    with open(args.dingtalk_develop_md_file, "r", encoding="utf-8") as f:
+        dingtalk_develop_md = f.read()
+
+    # ui
+    with gr.Blocks() as blocks:
+        gr.Markdown(value=dingtalk_develop_md)
+
+    blocks.queue().launch(
+        share=False if platform.system() == "Windows" else False,
+        server_name="127.0.0.1" if platform.system() == "Windows" else "0.0.0.0",
+        server_port=7860
+    )
+
+    return
+
+
+if __name__ == '__main__':
+    main()

project_settings.py

@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import os
+from pathlib import Path
+
+from toolbox.os.environment import EnvironmentManager
+
+
+project_path = os.path.abspath(os.path.dirname(__file__))
+project_path = Path(project_path)
+
+
+environment = EnvironmentManager(
+    path=os.path.join(project_path, "dotenv"),
+    env=os.environ.get("environment", "miyuki"),
+)
+
+
+if __name__ == '__main__':
+    pass

requirements.txt

@@ -0,0 +1,5 @@
+dingtalk-stream==0.17.2
+python-dotenv==1.0.0
+gradio==4.12.0
+transformers==4.39.1
+torch==2.2.1

toolbox/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/agent_x/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/agent_x/question_answer.py

@@ -0,0 +1,362 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import json
+import os
+import random
+import sys
+import time
+from typing import List
+
+pwd = os.path.abspath(os.path.dirname(__file__))
+sys.path.append(os.path.join(pwd, '../../'))
+
+import requests
+
+from project_settings import environment
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--api_key",
+        default=environment.get("agent_x_api_key", default=None),
+        type=str
+    )
+
+    args = parser.parse_args()
+    return args
+
+
+class AgentX(object):
+    def __init__(self,
+                 api_key: str,
+                 agent_name: str = "NXLink智能助手",
+                 url_host: str = "https://api.agentx.so"
+                 ):
+        self.api_key = api_key
+        self.agent_name = agent_name
+        self.url_host = url_host
+
+        self.agent_id = self.get_agent_id()
+
+    def __str__(self):
+        result = "<{}; agent_name: {}; agent_id: {}; api_key: {}>".format(
+            self.__class__.__name__, self.agent_name, self.agent_id, self.api_key)
+        return result
+
+    def get_agent_id(self):
+        url = "{}/api/v1/access/agents".format(self.url_host)
+
+        headers = {
+            "accept": "*/*",
+            "x-api-key": self.api_key
+        }
+        resp = requests.request(
+            "GET",
+            url=url,
+            headers=headers,
+        )
+        if resp.status_code != 200:
+            print(resp.status_code)
+            print(resp.text)
+            exit(0)
+        js = resp.json()
+
+        result = None
+        for e in js:
+            if e["name"] == self.agent_name:
+                result = e["_id"]
+
+        if result is None:
+            raise AssertionError("agent not found")
+        return result
+
+    def get_agent_config(self):
+        url = "{}/api/v1/access/agents/{}".format(self.url_host, self.agent_id)
+
+        headers = {
+            "accept": "*/*",
+            "x-api-key": self.api_key
+        }
+        resp = requests.request(
+            "GET",
+            url=url,
+            headers=headers,
+        )
+        js = resp.json()
+        return js
+
+    def get_conversation_list(self):
+        url = "{}/api/v1/access/agents/{}/conversations".format(self.url_host, self.agent_id)
+
+        headers = {
+            "accept": "*/*",
+            "x-api-key": self.api_key
+        }
+        resp = requests.request(
+            "GET",
+            url=url,
+            headers=headers,
+        )
+        js = resp.json()
+        return js
+
+    def post_message(self, message: str, conversation_id: str, context: int = 0):
+        url = "{}/api/v1/access/conversations/{}/message".format(self.url_host, conversation_id)
+
+        headers = {
+            "accept": "*/*",
+            "Content-type": "application/json",
+            "x-api-key": self.api_key
+        }
+        data = {
+            "message": message,
+            "context": context,
+        }
+        resp = requests.request(
+            "POST",
+            url=url,
+            headers=headers,
+            data=json.dumps(data)
+        )
+        if resp.status_code != 200:
+            print(resp.status_code)
+            print(resp.text)
+            exit(0)
+        js = resp.json()
+        return js
+
+    def post_message_by_sse(self, message: str, conversation_id: str, context: int = 0):
+        url = "{}/api/v1/access/conversations/{}/messagesse".format(self.url_host, conversation_id)
+
+        headers = {
+            "accept": "*/*",
+            "Content-type": "application/json",
+            "x-api-key": self.api_key
+        }
+        data = {
+            "message": message,
+            "context": context,
+        }
+        resp = requests.request(
+            "POST",
+            url=url,
+            headers=headers,
+            data=json.dumps(data),
+            stream=True
+        )
+        # print(resp.headers)
+
+        trace_id = resp.headers["x-trace-id"]
+
+        if resp.status_code == 200:
+            def generator():
+                result = ""
+                buf = b""
+
+                for chunk in resp.iter_content():
+                    buf += chunk
+                    try:
+                        chunk = buf.decode("utf-8")
+                    except UnicodeDecodeError:
+                        continue
+                    result += chunk
+                    buf = b""
+
+                    yield chunk
+            return generator(), trace_id
+
+        else:
+            print(resp.status_code)
+            print(resp.headers["Content-Type"])
+            raise AssertionError
+
+    def get_trace_by_message_id(self, message_id: str):
+        url = "{}/api/v1/access/messages/{}/trace".format(self.url_host, message_id)
+
+        headers = {
+            "accept": "*/*",
+            "x-api-key": self.api_key
+        }
+        resp = requests.request(
+            "GET",
+            url=url,
+            headers=headers,
+        )
+
+        js = resp.json()
+        return js
+
+    def get_trace_by_trace_id(self, trace_id: str):
+        url = "{}/api/v1/access/traces/{}".format(self.url_host, trace_id)
+
+        headers = {
+            "accept": "*/*",
+            "x-api-key": self.api_key
+        }
+        resp = requests.request(
+            "GET",
+            url=url,
+            headers=headers,
+        )
+
+        js = resp.json()
+        return js
+
+    def post_new_conversation_id(self):
+        url = "{}/api/v1/access/agents/{}/conversations/new".format(self.url_host, self.agent_id)
+
+        headers = {
+            "accept": "*/*",
+            "x-api-key": self.api_key
+        }
+        resp = requests.request(
+            "POST",
+            url=url,
+            headers=headers,
+        )
+        js = resp.json()
+
+        conversation_id = js["_id"]
+        return conversation_id
+
+    def delete_conversation(self, conversation_id: str):
+        url = "{}/api/v1/access/conversations/{}".format(self.url_host, conversation_id)
+
+        headers = {
+            "accept": "*/*",
+            "Content-type": "application/json",
+            "x-api-key": self.api_key
+        }
+        resp = requests.request(
+            "DELETE",
+            url=url,
+            headers=headers,
+        )
+        js = resp.json()
+        return js
+
+    def update_context(self, messages: List[dict], conversation_id: str):
+        url = "{}/api/v1/access/conversations/{}/update-context".format(self.url_host, conversation_id)
+
+        headers = {
+            "accept": "*/*",
+            "Content-type": "application/json",
+            "x-api-key": self.api_key
+        }
+        data = {
+            "messages": messages,
+        }
+        resp = requests.request(
+            "PUT",
+            url=url,
+            headers=headers,
+            data=json.dumps(data),
+        )
+        js = resp.json()
+        return js
+
+    def question_answer(self, question: str, conversation_id: str = None, context: List[dict] = None, streaming: bool = False):
+        if conversation_id is None:
+            conversation_id = self.post_new_conversation_id()
+
+        if context is not None:
+            self.update_context(context, conversation_id)
+
+        result = {
+            "answer": None,
+            "reference": None
+        }
+        try:
+            if streaming:
+                resp_stream, trace_id = self.post_message_by_sse(question, conversation_id,
+                                                                 context=0 if context is None else 1)
+                answer = ""
+                for chunk in resp_stream:
+                    print(chunk, end="")
+                    answer += chunk
+                print("\n")
+                result["answer"] = answer
+                # print(answer)
+                # exit(0)
+
+                # [{"title": "", "source": ""}, ...]
+                trace = self.get_trace_by_trace_id(trace_id)
+
+                if trace == "No trace":
+                    reference = "No trace"
+                else:
+                    reference = list()
+                    for t in trace:
+                        reference.append((t["title"], t["source"]))
+
+                result["reference"] = reference
+
+            else:
+                js = self.post_message(question, conversation_id,
+                                       context=0 if context is None else 1)
+                answer = js["text"]
+                result["answer"] = answer
+
+                message_id = js["_id"]
+                trace = self.get_trace_by_message_id(message_id)
+                # print(trace)
+
+                if trace == "No trace":
+                    reference = "No trace"
+                else:
+                    reference = list()
+                    for t in trace:
+                        reference.append((t["title"], t["source"]))
+
+                result["reference"] = reference
+
+        finally:
+            self.delete_conversation(conversation_id)
+
+        return result
+
+
+def main():
+    args = get_args()
+
+    agent = AgentX(
+        api_key=args.api_key,
+        agent_name="Yutong Bus",
+    )
+    print(agent)
+
+    context = [
+        {
+            "user": "你好"
+        },
+        {
+            "assistant": "你好，我们是宇通客车公司，有什么可以帮到您的吗？"
+        },
+        {
+            "user": "需要一辆55座客车。"
+        },
+        {
+            "assistant": "Which country will the bus be used in?"
+        },
+        {
+            "user": "你可以说中文吗。"
+        },
+        {
+            "assistant": "可以的，请问您需要在哪个国家使用客车？"
+        },
+    ]
+
+    question = "你好"
+    time_begin = time.time()
+    response = agent.question_answer(question, context=context, streaming=True)
+    time_cost = time.time() - time_begin
+
+    print(response)
+    print("time cost: {}".format(time_cost))
+    return
+
+
+if __name__ == '__main__':
+    main()

toolbox/json/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/json/misc.py

@@ -0,0 +1,63 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+from typing import Callable
+
+
+def traverse(js, callback: Callable, *args, **kwargs):
+    if isinstance(js, list):
+        result = list()
+        for l in js:
+            l = traverse(l, callback, *args, **kwargs)
+            result.append(l)
+        return result
+    elif isinstance(js, tuple):
+        result = list()
+        for l in js:
+            l = traverse(l, callback, *args, **kwargs)
+            result.append(l)
+        return tuple(result)
+    elif isinstance(js, dict):
+        result = dict()
+        for k, v in js.items():
+            k = traverse(k, callback, *args, **kwargs)
+            v = traverse(v, callback, *args, **kwargs)
+            result[k] = v
+        return result
+    elif isinstance(js, int):
+        return callback(js, *args, **kwargs)
+    elif isinstance(js, str):
+        return callback(js, *args, **kwargs)
+    else:
+        return js
+
+
+def demo1():
+    d = {
+        "env": "ppe",
+        "mysql_connect": {
+            "host": "$mysql_connect_host",
+            "port": 3306,
+            "user": "callbot",
+            "password": "NxcloudAI2021!",
+            "database": "callbot_ppe",
+            "charset": "utf8"
+        },
+        "es_connect": {
+            "hosts": ["10.20.251.8"],
+            "http_auth": ["elastic", "ElasticAI2021!"],
+            "port": 9200
+        }
+    }
+
+    def callback(s):
+        if isinstance(s, str) and s.startswith('$'):
+            return s[1:]
+        return s
+
+    result = traverse(d, callback=callback)
+    print(result)
+    return
+
+
+if __name__ == '__main__':
+    demo1()

toolbox/os/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/os/environment.py

@@ -0,0 +1,114 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import json
+import os
+
+from dotenv import load_dotenv
+from dotenv.main import DotEnv
+
+from toolbox.json.misc import traverse
+
+
+class EnvironmentManager(object):
+    def __init__(self, path, env, override=False):
+        filename = os.path.join(path, '{}.env'.format(env))
+        self.filename = filename
+
+        load_dotenv(
+            dotenv_path=filename,
+            override=override
+        )
+
+        self._environ = dict()
+
+    def open_dotenv(self, filename: str = None):
+        filename = filename or self.filename
+        dotenv = DotEnv(
+            dotenv_path=filename,
+            stream=None,
+            verbose=False,
+            interpolate=False,
+            override=False,
+            encoding="utf-8",
+        )
+        result = dotenv.dict()
+        return result
+
+    def get(self, key, default=None, dtype=str):
+        result = os.environ.get(key)
+        if result is None:
+            if default is None:
+                result = None
+            else:
+                result = default
+        else:
+            result = dtype(result)
+        self._environ[key] = result
+        return result
+
+
+_DEFAULT_DTYPE_MAP = {
+    'int': int,
+    'float': float,
+    'str': str,
+    'json.loads': json.loads
+}
+
+
+class JsonConfig(object):
+    """
+    将 json 中, 形如 `$float:threshold` 的值, 处理为:
+    从环境变量中查到 threshold, 再将其转换为 float 类型.
+    """
+    def __init__(self, dtype_map: dict = None, environment: EnvironmentManager = None):
+        self.dtype_map = dtype_map or _DEFAULT_DTYPE_MAP
+        self.environment = environment or os.environ
+
+    def sanitize_by_filename(self, filename: str):
+        with open(filename, 'r', encoding='utf-8') as f:
+            js = json.load(f)
+
+        return self.sanitize_by_json(js)
+
+    def sanitize_by_json(self, js):
+        js = traverse(
+            js,
+            callback=self.sanitize,
+            environment=self.environment
+        )
+        return js
+
+    def sanitize(self, string, environment):
+        """支持 $ 符开始的, 环境变量配置"""
+        if isinstance(string, str) and string.startswith('$'):
+            dtype, key = string[1:].split(':')
+            dtype = self.dtype_map[dtype]
+
+            value = environment.get(key)
+            if value is None:
+                raise AssertionError('environment not exist. key: {}'.format(key))
+
+            value = dtype(value)
+            result = value
+        else:
+            result = string
+        return result
+
+
+def demo1():
+    import json
+
+    from project_settings import project_path
+
+    environment = EnvironmentManager(
+        path=os.path.join(project_path, 'server/callbot_server/dotenv'),
+        env='dev',
+    )
+    init_scenes = environment.get(key='init_scenes', dtype=json.loads)
+    print(init_scenes)
+    print(environment._environ)
+    return
+
+
+if __name__ == '__main__':
+    demo1()

toolbox/os/other.py

@@ -0,0 +1,9 @@
+import os
+import inspect
+
+
+def pwd():
+    """你在哪个文件调用此函数, 它就会返回那个文件所在的 dir 目标"""
+    frame = inspect.stack()[1]
+    module = inspect.getmodule(frame[0])
+    return os.path.dirname(os.path.abspath(module.__file__))