Init

.dockerignore

@@ -0,0 +1,3 @@
+.venv/
+.git/
+files/

.env.example

@@ -0,0 +1,4 @@
+OPENAI_API_KEY=key
+TG_API_KEY=kei
+TG_ADMIN_CHAT_ID=id
+TG_USER_ACCESS=id1,id2

.flake8

@@ -0,0 +1,2 @@
+[flake8]
+ignore = E501

.gitignore

@@ -0,0 +1,4 @@
+api_key.txt
+.venv/
+.env
+__pycache__

.vscode/launch.json

@@ -0,0 +1,22 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python: Current File",
+            "type": "python",
+            "request": "launch",
+            "program": "main.py",
+            "console": "integratedTerminal",
+            "justMyCode": true,
+            "args": [
+                "telegram",
+                "<",
+                "${workspaceFolder}/input.txt",
+            ],
+            "envFile": "${workspaceFolder}/.env"
+        }
+    ]
+}

Dockerfile

@@ -0,0 +1,13 @@
+# syntax=docker/dockerfile:1
+FROM ubuntu:jammy
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install --no-install-recommends -y python3-pip ddgr sudo apt-utils nmap mrt \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY main.py .
+CMD ["python3", "main.py", "telegram"]

README.md

@@ -0,0 +1,48 @@
+# AI Chatbot
+
+This project is a chatbot that uses OpenAI's GPT-3/4 model for generating responses. The model is capable of handling function calling and executing tasks based on instructions provided in the conversation. The chatbot can be run as a console-based chat application or as a Telegram bot.
+
+
+## Getting Started
+
+### Prerequisites
+
+- Python 3.7 or higher
+- OpenAI Python package
+- Telegram Python package
+
+### Environment Variables
+
+Before running the application, you need to set the following environment variables:
+
+- `OPENAI_API_KEY`: Your OpenAI API key
+- `TG_API_KEY`: Your Telegram API key
+- `TG_ADMIN_CHAT_ID`: The chat ID of the Telegram admin
+- `TG_USER_ACCESS`: A comma-separated list of authorized user IDs
+
+## Usage
+
+### Running as a Console-based Chat
+
+To run the application as a console-based chat, execute the script with no arguments:
+
+```bash
+python main.py
+```
+
+### Running as a Telegram Bot
+
+To run the application as a Telegram bot, execute the script with the "telegram" argument:
+
+```bash
+python main.py telegram
+```
+
+## License
+
+This project is licensed under the MIT License.
+
+## Acknowledgments
+
+- OpenAI for the GPT-3/4 model
+- Python-Telegram-Bot developers for the Telegram Bot wrapper

docker-compose.yml

@@ -0,0 +1,11 @@
+services:
+  bot:
+    build: .
+    environment:
+      - OPENAI_API_KEY
+      - TG_API_KEY
+      - TG_ADMIN_CHAT_ID
+      - TG_USER_ACCESS
+    volumes:
+      - ${PWD}:/app
+    restart: always

files/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

function/__init__.py

@@ -0,0 +1,8 @@
+# flake8: noqa
+from .function import *
+from .bash_executor import *
+from .expression_executor import *
+from .file_sender import *
+from .message_sender import *
+from .python_executor import *
+from .website_content_fetcher import *

function/bash_executor.py

@@ -0,0 +1,43 @@
+import logging
+import subprocess
+from typing import Any, AnyStr
+
+from .function import Function
+
+
+class BashExecutor(Function):
+    specification = {
+        "name": "execute_bash_code",
+        "description": "Execute bash code or terminal command and return the output.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "string",
+                    "description": "The bash code or terminal command to execute.",
+                },
+                "continue": {
+                    "type": "string",
+                    "enum": ["true", "false"],
+                    "description": "The boolean flag that determines whether to process output from the function. If set to true, the function's output will be processed. If set to false, the function's output will not be processed. ",
+                },
+            },
+            "required": ["code", "continue"],
+        },
+    }
+
+    def get_name(self) -> AnyStr:
+        return self.specification.get("name")
+
+    def func(self, data) -> Any:
+        code = data if isinstance(data, str) else data.get("code")
+        logging.warning(f"# {code}")
+        try:
+            process = subprocess.run(code, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=60)
+            if process.stderr:
+                return process.stderr
+            return process.stdout if process.stdout else "Done!"
+        except Exception as e:
+            logging.debug(data)
+            logging.error(e)
+            return str(e)

function/expression_executor.py

@@ -0,0 +1,40 @@
+import logging
+from typing import Any, AnyStr
+
+from .function import Function
+
+
+class ExpressionExecutor(Function):
+    specification = {
+        "name": "execute_python_expression",
+        "description": "Execute Python expression and return the output.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "string",
+                    "description": "The Python expression to execute.",
+                },
+                "continue": {
+                    "type": "string",
+                    "enum": ["true", "false"],
+                    "description": "The boolean flag that determines whether to process output from the function. If set to true, the function's output will be processed. If set to false, the function's output will not be processed. ",
+                },
+            },
+            "required": ["code", "continue"],
+        },
+    }
+
+    def get_name(self) -> AnyStr:
+        return self.specification.get("name")
+
+    def func(self, data) -> Any:
+        code = data if isinstance(data, str) else data.get("code")
+        logging.warning(f"= {code}\n")
+
+        try:
+            output = eval(code)
+            return output
+        except Exception as e:
+            logging.error(e)
+            return str(e)

function/file_sender.py

@@ -0,0 +1,45 @@
+import logging
+import sys
+from io import StringIO
+
+from typing import Any, AnyStr
+
+from .function import Function
+
+
+class FileSender(Function):
+    specification = {
+        "name": "send_file",
+        "description": "Sends a file to a Telegram chat",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "filepath": {
+                    "type": "string",
+                    "description": "The file path to send"
+                },
+            },
+            "required": ["filepath"],
+        },
+    }
+
+    def get_name(self) -> AnyStr:
+        return self.specification.get("name")
+
+    def func(self, data) -> Any:
+        code = data if isinstance(data, str) else data.get("code")
+        logging.warning(f">>> {code}")
+
+        # Redirect stdout to a string
+        old_stdout = sys.stdout
+        redirected_output = sys.stdout = StringIO()
+
+        try:
+            exec(code)
+        except Exception as e:
+            logging.error(e)
+        finally:
+            # Reset stdout
+            sys.stdout = old_stdout
+
+        return redirected_output.getvalue()

function/function.py

@@ -0,0 +1,11 @@
+from typing import Any, List, Protocol, Dict, AnyStr
+
+
+class Function(Protocol):
+    specification: Dict
+
+    def get_name(self) -> AnyStr:
+        pass
+
+    def func(self, quz: List[int]) -> Any:
+        pass

function/message_sender.py

@@ -0,0 +1,35 @@
+import requests
+from typing import Any, AnyStr
+
+from .function import Function
+
+
+class MessageSender(Function):
+    specification = {
+        "name": "send_message_to_admin",
+        "description": "Sends a message to a admin",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "msg": {
+                    "type": "string",
+                    "description": "The text of the message"
+                },
+            },
+            "required": ["msg"],
+        },
+    }
+
+    def __init__(self, tg_api_key, tg_admin_chat_id):
+        self.tg_api_key = tg_api_key
+        self.tg_admin_chat_id = tg_admin_chat_id
+
+    def get_name(self) -> AnyStr:
+        return self.specification.get("name")
+
+    def func(self, data) -> Any:
+        msg = data if isinstance(data, str) else data.get("msg")
+        url = f"https://api.telegram.org/bot{self.tg_api_key}/sendMessage"
+        data = {"parse_mode": "Markdown", "disable_notification": "true", "chat_id": self.tg_admin_chat_id, "text": msg}
+        response = requests.post(url, data=data)
+        return "Sent!" if response.status_code == 200 else "Got error!"

function/python_executor.py

@@ -0,0 +1,50 @@
+import logging
+import sys
+from io import StringIO
+
+from typing import Any, AnyStr
+
+from .function import Function
+
+
+class PythonExecutor(Function):
+    specification = {
+        "name": "execute_python_code",
+        "description": "Execute Python code and return the output.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "string",
+                    "description": "The Python code to execute.",
+                },
+                "continue": {
+                    "type": "string",
+                    "enum": ["true", "false"],
+                    "description": "The boolean flag that determines whether to process output from the function. If set to true, the function's output will be processed. If set to false, the function's output will not be processed. ",
+                },
+            },
+            "required": ["code", "continue"],
+        },
+    }
+
+    def get_name(self) -> AnyStr:
+        return self.specification.get("name")
+
+    def func(self, data) -> Any:
+        code = data if isinstance(data, str) else data.get("code")
+        logging.warning(f">>> {code}")
+
+        # Redirect stdout to a string
+        old_stdout = sys.stdout
+        redirected_output = sys.stdout = StringIO()
+
+        try:
+            exec(code)
+        except Exception as e:
+            logging.error(e)
+        finally:
+            # Reset stdout
+            sys.stdout = old_stdout
+
+        return redirected_output.getvalue()