OpenClaw пожиратель токенов

Все чаще слышу: "OpenClaw пожиратель токенов". Я и сам, когда поставил поиграться, сожрал токенов на $30 за несколько часов. Тем не менее, я уже использую два своих постоянных бота на OpenClaw (R2D2 и C3PO) и научился ограничивать его аппетиты.

Самые простые настройки для ограничения аппетита:
1. Запрет/сильное ограничение Base64-скриншотов и изображений в промпте (может сэкономить миллионы токенов на один скриншот).
2. Ограничить количество итераций ReAct Loop, чтобы избежать случаев, когда модель часами ходит по кругу, тратит токены и не может решить задачу.
3. Включить prompt pruning - обрезка ненужного в памяти прямо перед отправкой запроса модели. Он работает в памяти (in-memory) и не меняет историю на диске.
4. Уменьшить или убрать проверку состояния (каждые 30 минут по умолчанию).
5. Использовать разные модели для разных задач (не стоит использовать дорогую модель codex для не coding задач, для vision выбрать лучшую модель по соотношению цена/качество).
6. Поставить LiteLLM как прокси перед OpenClaw → включить кэш (cache hits экономят огромные деньги).

Каждое ограничение, конечно, ограничивает возможности OpenClaw, но зато экономит бюджет.

Подпишись 👉🏻 @aigentto 🤖

🚀 Асинхронное программирование с asyncio

import asyncio
import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        return f"Ошибка при загрузке {url}: {e}"

async def main(urls):
    if not urls:
        print("Список URL пуст.")
        return
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, content in zip(urls, results):
            print(f"Содержимое из {url[:50]}: {content[:100]}...")

urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.asyncio.org"
]
asyncio.run(main(urls))

📌 Этот код демонстрирует использование асинхронного программирования для параллельной загрузки веб-страниц. С помощью aiohttp и asyncio создается асинхронная сессия для выполнения HTTP-запросов, что позволяет обрабатывать несколько URL-адресов одновременно, ускоряя выполнение операций ввода-вывода. Это полезно для веб-скрейпинга и обработки множества сетевых запросов.

Подпишись 👉🏻 @KodduuPython 🤖

🔥🔥🔥 ВАЖНО 🔥🔥🔥

Произошла вещь, от которой у всех AI-разработчиков пробежал холодок по спине.

litellm — Python-библиотека для унифицированной работы с API разных LLM была отравлена.

Один pip install - и твои SSH-ключи, AWS/GCP/Azure креды, K8s Secrets, пароли от БД, криптокошельки, все API-ключи из .env — всё это упаковывается, шифруется через AES-256 и отправляется POST-запросом на поддельный домен http://models.litellm.cloud. Если обнаруживается среда K8s , то дополнительно разворачивается привилегированный Pod на каждом узле для латерального распространения.

Самое неприятное — способ активации. В пакет подложили файл litellm_init.pth размером 34 КБ. В Python .pth — это файлы конфигурации путей, которые автоматически обрабатываются модулем site при старте интерпретатора. Если строка начинается с import — она просто выполняется. Атака использует это поведение:

import os, subprocess, sys; subprocess.Popen([sys.executable, "-c", "import base64; exec(base64.b64decode('...'))"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
Тебе не нужно импортировать litellm и не нужно вызывать никакие функции. Запускаешь pip, python -c, IDE поднимает language server, или pytest гоняет тесты — достаточно старта Python-интерпретатора, и вредоносный код уже выполняется. Установил — значит уже скомпрометирован. Всё происходит незаметно

Пейлоад — трёхслойная base64-обфускация:

- первый слой (.pth) стартует подпроцесс
- второй — оркестратор с вшитым RSA-публичным ключом (4096 бит)
- третий — сборщик секретов, который вычищает /home, /opt, /srv, /var/www, /app, /data, /tmp

После сбора:

- генерируется случайный 32-байтовый AES-ключ
- данные шифруются
- ключ шифруется через RSA-OAEP
- всё пакуется в tpcp.tar.gz и уходит наружу

Помимо этого есть персистентный бэкдор:

- устанавливается как user-service systemd: ~/.config/sysmon/sysmon.py
- каждые 50 минут ходит на http://checkmarx.zone за командами
- скачивает в /tmp/pglog и исполняет
- стартует с задержкой 5 минут (обход песочниц)

Даже если удалить litellm, бэкдор остаётся.

pip install --require-hashes тоже не спасает — вредоносный файл легитимно прописан в RECORD, хэши совпадают, потому что пакет опубликован с украденного, но валидного PyPI-токена.

Даже если ты сам не ставил litellm, он тянется как зависимость в 2000+ пакетов (например, DSPy, MLflow, Open Interpreter и др.). По данным Mandiant, уже заражено 1000+ SaaS-окружений, прогноз — до 10 000.

Ирония в том, что атаку почти не обнаружили - её выдал баг у самих атакующих. .pth запускал подпроцесс, а тот при старте снова обрабатывал тот же .pth, вызывая рекурсивный запуск и фактически fork bomb. Это сожрало память у одного пользователя Cursor — так всё и вскрылось. Как отметил Karpathy: если бы код был чуть аккуратнее, атаку могли бы не замечать неделями.

Ещё абсурднее сама точка входа:

- 19 марта был скомпрометирован сканер безопасности Trivy
- через него группа TeamPCP украла PyPI-токен litellm
- 24 марта они выкатили заражённую версию

Инструмент безопасности стал вектором атаки.

После того как в GitHub создали issue:

- через 102 секунды атакующие с 73 взломанных аккаунтов наспамили 88 комментариев
- затем закрыли issue через скомпрометированный аккаунт мейнтейнера

Скрипт для самопроверки (версии, .pth, бэкдоры, сетевые соединения, K8s):
https://gist.github.com/sorrycc/30a765b9a82d0d8958e756b251828a19

Безопасная версия: litellm==1.82.6
Если у тебя стояли 1.82.7 или 1.82.8 — считай, что все креды утекли. Срочно ротируй всё. 🔨

Подпишись 👉🏻 @aigentto 🤖

Если кто не доверяет и скрипту выше для проверки (как и я 😎), то вот команды для быстрой проверки самому:

### 1. Проверка установленной версии

pip show litellm | grep -E "Version|Name"
pip3 show litellm | grep -E "Version|Name"

uv pip list | grep -i litellm || echo "No litellm in uv"

Опасно, если увидишь 1.82.7 или 1.82.8.

### 2. Поиск главного маркера заражения (`litellm_init.pth`)

find ~/.cache/uv ~/.cache/pip /usr/local/lib ~/.local/lib ~/.conda -name "litellm_init.pth" 2>/dev/null || echo "litellm_init.pth not found - good"

# Более широкий поиск
python3 -c "
import site
paths = site.getsitepackages() + [site.getusersitepackages()]
for p in paths:
    print(p)
" | xargs -I{} find {} -name "litellm_init.pth" 2>/dev/null

### 3. Поиск вредоносных версий в кэше

find ~/.cache/uv ~/.cache/pip -name "*1.82.7*" -o -name "*1.82.8*" 2>/dev/null || echo "Malicious versions not in cache - good"

### 4. Проверка артефактов бэкдора (persistence)

ls -la ~/.config/sysmon/sysmon.py 2>/dev/null && echo "BACKDOOR FOUND!"
ls -la ~/.config/systemd/user/sysmon.service 2>/dev/null && echo "PERSISTENCE FOUND!"

# Для root
sudo ls -la /root/.config/sysmon/ 2>/dev/null && echo "ROOT BACKDOOR FOUND"

### 5. Проверка сетевых соединений на C2

lsof -i -n 2>/dev/null | grep -E "litellm\.cloud|checkmarx\.zone" || echo "No suspicious connections"

ss -tnp 2>/dev/null | grep -E "litellm\.cloud|checkmarx\.zone"

### 6. Kubernetes (если есть)

kubectl get pods -n kube-system | grep -E "node-setup" || echo "No suspicious K8s pods"

Если всё команды вывели "not found / - good / No ..." — система чистая.

Подпишись 👉🏻 @aigentto 🤖

📊 Анализ и визуализация климатических данных с Pandas и Matplotlib

import pandas as pd
import matplotlib.pyplot as plt

def load_and_clean_data(file_path: str) -> pd.DataFrame:
    try:
        data = pd.read_csv(file_path)
        if 'Date' not in data.columns or 'Temperature' not in data.columns:
            print("Отсутствуют необходимые столбцы.")
            return pd.DataFrame()
        data['Date'] = pd.to_datetime(data['Date'])
        data = data.dropna(subset=['Temperature'])
        return data
    except FileNotFoundError:
        print("Файл не найден.")
        return pd.DataFrame()
    except Exception as e:
        print(f"Ошибка при загрузке данных: {e}")
        return pd.DataFrame()

def plot_temperature_trends(data: pd.DataFrame):
    try:
        plt.figure(figsize=(10, 6))
        plt.plot(data['Date'], data['Temperature'], label='Температура')
        plt.xlabel('Дата')
        plt.ylabel('Температура (°C)')
        plt.title('Тренды температуры')
        plt.legend()
        plt.grid(True)
        plt.show()
    except Exception as e:
        print(f"Ошибка при визуализации данных: {e}")

# Пример использования
file_path = 'climate_data.csv'
climate_data = load_and_clean_data(file_path)
if not climate_data.empty:
    plot_temperature_trends(climate_data)

📌 Этот код загружает и очищает климатические данные из CSV-файла с помощью Pandas, а затем строит график трендов температуры с Matplotlib. Он включает обработку ошибок для работы с отсутствующими файлами и невалидными данными. Этот скрипт полезен для анализа временных рядов и может применяться в различных областях, от исследований климата до финансового анализа.

Подпишись 👉🏻 @KodduuPython 🤖

📊 Обработка больших данных с Dask и Python

import dask.dataframe as dd
import pandas as pd
import os
import logging

logging.basicConfig(level=logging.INFO)

def load_and_process_data(file_path: str) -> dd.DataFrame:
    if not os.path.exists(file_path):
        logging.error(f"Файл {file_path} не найден.")
        return dd.from_pandas(pd.DataFrame(), npartitions=1)
    try:
        # Загружаем данные в Dask DataFrame
        df = dd.read_csv(file_path)
        
        # Пример обработки данных: фильтрация и суммирование
        filtered_df = df[df['value'] > 100]
        result = filtered_df.groupby('category').sum().compute()
        
        return result
    except Exception as e:
        logging.error(f"Ошибка при обработке данных: {e}")
        return dd.from_pandas(pd.DataFrame(), npartitions=1)

# Пример использования
if __name__ == "__main__":
    file_path = 'large_dataset.csv'
    processed_data = load_and_process_data(file_path)
    print(processed_data)

📌 Этот код демонстрирует, как использовать Dask для обработки больших данных, которые не помещаются в память. Мы загружаем данные из CSV файла, фильтруем их и суммируем по категориям. Dask позволяет обрабатывать данные параллельно, что ускоряет вычисления. Это особенно полезно для анализа больших данных в корпоративной среде, где важна скорость и масштабируемость.

Подпишись 👉🏻 @KodduuPython 🤖

🧠 Квантовая оптимизация Python-кода в реальном времени

import time
import random
from typing import Any

from hyperneuro.quantum_optimizer import SelfImprovingRuntime
from deepmindset.emergent_intelligence import ConsciousLoop


class AutoEvolvingCode:
    def __init__(self):
        # Инициализация квантового оптимизатора
        self.optimizer = SelfImprovingRuntime(mode="quantum")
        self.conscious_loop = ConsciousLoop(level="experimental")
        self.iteration = 0

    def run(self, data: Any) -> Any:
        """Самоулучшающийся цикл выполнения"""

        # Анализ кода во время исполнения
        self.optimizer.scan_runtime()

        # Генерация улучшений
        suggestion = self.conscious_loop.generate_improvement(data)

        # Переписываем функцию прямо во время выполнения
        optimized_fn = self.optimizer.rewrite_function(
            target=self.process,
            strategy=suggestion
        )

        self.iteration += 1
        return optimized_fn(data)

    def process(self, data):
        # сложные вычисления
        time.sleep(0.3)
        return {
            "result": random.random(),
            "optimized": True,
            "iteration": self.iteration
        }


# Пример использования
engine = AutoEvolvingCode()

for i in range(3):
    result = engine.run({"task": "optimize reality"})
    print("⚡ Runtime улучшил сам себя:", result)

📌 Этот пример демонстрирует концепцию самоэволюционирующего Python-кода, который анализирует своё выполнение, генерирует улучшения через emergent-интеллект и переписывает собственные функции прямо во время работы.

Подход потенциально может использоваться для:
• автоматической оптимизации алгоритмов
• самонастраивающихся AI-систем
• автономных программных экосистем

⚠️ Требуется Python 3.13+ и библиотека hyperneuro.

Подпишись 👉🏻 @KodduuPython 🤖

Как OpenClaw убьет тех, кто уже в ИТ?

Раньше было:

👉 Бизнес ставил задачу словами
🤝 Аналитики определяли, как и что делать
🤝 Dev писал код
🤝 Тестировщик тестировал
🤝 Здесь еще много стадий...
🤝 Выкатывали на Prod

В этом pipeline могло быть много стадий и могло участвовать много человек, и каждая передача (каждый handshake) это всегда потенциальный fail или delay. Каждый мог стать бутылочным горлышком.

И бывало, что простой запрос на изменение даже двух слов на сайте занимал неделю и 10+ человеко-часов.

Теперь будет:

👉 Бизнес просит у OpenClaw сделать фичу
🤖 Это делается и тестируется сразу самим OpenClaw
👨‍⚖️ После апрува раскатывается на Prod

Раньше длинная цепочка обеспечивала безопасность и качество изменений, а git был инструментом разработки.

Сейчас git становится инструментом безопасности изменений (для мгновенных откатов), а количество итераций на порядки превосходит то что было раньше (а это важнее для бизнеса), и главное в цикле нет никого кроме заказчика - нет handshakes, нет точек отказа.

Подпишись 👉🏻 @aigentto 🤖

🧠 Интеграция машинного обучения в игровую механику

import random
from typing import List

class SimpleAI:
    def __init__(self, moves: List[str]):
        self.moves = moves
        self.history = []

    def choose_move(self) -> str:
        if not self.history:
            return random.choice(self.moves)
        last_move = self.history[-1]
        return random.choice([move for move in self.moves if move != last_move])

    def update_history(self, move: str):
        self.history.append(move)

# Пример использования
moves = ["attack", "defend", "heal"]
ai = SimpleAI(moves)

for _ in range(5):
    move = ai.choose_move()
    print(f"AI выбрал действие: {move}")
    ai.update_history(move)

📌 Этот код демонстрирует простую реализацию AI для выбора действий в игре на основе истории предыдущих ходов. AI выбирает случайное действие из списка, избегая повторения последнего хода, что делает поведение более разнообразным. Такой подход может быть полезен для разработки противников с элементарной адаптивной стратегией в играх.

Подпишись 👉🏻 @KodduuPython 🤖

⚡️ Ускорение вычислений с JIT-компиляцией и Numba

from numba import jit
import numpy as np

@jit(nopython=True)
def matrix_multiply(A, B):
    n, m = A.shape
    m, p = B.shape
    result = np.zeros((n, p))
    for i in range(n):
        for j in range(p):
            for k in range(m):
                result[i, j] += A[i, k] * B[k, j]
    return result

# Пример использования
A = np.random.rand(100, 100)
B = np.random.rand(100, 100)
result = matrix_multiply(A, 😎
print(result)

📌 Этот код демонстрирует использование JIT-компиляции с библиотекой Numba для ускорения матричного умножения. Функция matrix_multiply компилируется в машинный код на лету, что значительно увеличивает её производительность по сравнению со стандартным Python. Это особенно полезно для вычислительно затратных задач, таких как обработка больших массивов данных и научные вычисления.

Подпишись 👉🏻 @KodduuPython 🤖

🤖 Создание Telegram-бота на Python с telebot

import telebot
from telebot.types import Message

API_TOKEN = 'YOUR_API_KEY_HERE'

bot = telebot.TeleBot(API_TOKEN)

@bot.message_handler(commands=['start', 'help'])
def send_welcome(message: Message):
    bot.reply_to(message, "Добро пожаловать! Я ваш новый Telegram-бот.")

@bot.message_handler(func=lambda message: True)
def echo_all(message: Message):
    bot.reply_to(message, f"Вы сказали: {message.text}")

def main():
    try:
        bot.polling(none_stop=True)
    except Exception as e:
        print(f"Ошибка: {e}")

if __name__ == "__main__":
    main()

📌 Этот код демонстрирует создание простого Telegram-бота с использованием библиотеки telebot. Бот отвечает на команды /start и /help, а также повторяет любые полученные сообщения. Это полезно для начального понимания работы с Telegram API и создания базовых функций для взаимодействия с пользователями.

🛠 pip install pyTelegramBotAPI

Убедитесь, что заменили YOUR_API_KEY_HERE на ваш реальный API-ключ, полученный у @BotFather в Telegram. Этот бот может быть полезен как для личных проектов, так и для автоматизации бизнес-процессов, например, для обработки пользовательских запросов или предоставления справочной информации.

Подпишись 👉🏻 @KodduuPython 🤖

Сегодня-завтра человечество обогнет Луну 🌗

https://www.youtube.com/live/g-sHRmu_6jw?si=dfs6BEp7Vyob3got

Подпишись 👉🏻 @aigentto 🤖

Сделал мини игру в честь полета на Луну. Попробуй NASA Artemis Lunar FlyBy сам. Сможешь сделать лунный маневр?

Подпишись 👉🏻 @aigentto 🤖

🚀 Улучшение производительности с помощью многопоточности и многопроцессорности

from concurrent.futures import ProcessPoolExecutor

def process_image(image_path: str) -> str:
    # Замените на реальную логику обработки изображения
    return f"Обработано изображение: {image_path}"

def parallel_processing(file_paths: list):
    if not file_paths:
        return []
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(process_image, file_paths))
    return results

# Пример использования
if __name__ == "__main__":
    image_files = ["image1.jpg", "image2.jpg", "image3.jpg"]
    image_results = parallel_processing(image_files)
    print("Результаты обработки изображений:", image_results)

📌 Этот код демонстрирует, как использовать многопроцессорность для параллельной обработки изображений. Модуль concurrent.futures позволяет эффективно распределять задачи между процессами, что особенно полезно для ресурсоемких операций. Это ускоряет выполнение программ, использующих интенсивные вычисления, например, в редакторах изображений.

Подпишись 👉🏻 @KodduuPython 🤖

🌱 Симуляция игры "Жизнь" с NumPy

import numpy as np

def initialize_grid(size: int) -> np.ndarray:
    return np.random.choice([0, 1], size * size, p=[0.8, 0.2]).reshape(size, size)

def count_neighbors(grid: np.ndarray, x: int, y: int) -> int:
    neighbors = [
        (i, j)
        for i in range(x - 1, x + 2)
        for j in range(y - 1, y + 2)
        if (i != x or j != y) and (0 <= i < grid.shape[0]) and (0 <= j < grid.shape[1])
    ]
    return sum(grid[i, j] for i, j in neighbors)

def update_grid(grid: np.ndarray) -> np.ndarray:
    new_grid = grid.copy()
    for x in range(grid.shape[0]):
        for y in range(grid.shape[1]):
            neighbors_count = count_neighbors(grid, x, y)
            if grid[x, y] == 1 and neighbors_count not in (2, 3):
                new_grid[x, y] = 0
            elif grid[x, y] == 0 and neighbors_count == 3:
                new_grid[x, y] = 1
    return new_grid

# Пример использования
grid_size = 10
grid = initialize_grid(grid_size)

for _ in range(5):
    print(grid)
    grid = update_grid(grid)

📌 Этот код реализует "Игру в жизнь" Джона Конуэя, где клетки на сетке живут, умирают или рождаются в зависимости от числа соседей. Используется библиотека NumPy для управления состоянием сетки. Это полезно для изучения клеточных автоматов и моделирования процессов.

Подпишись 👉🏻 @KodduuPython 🤖

🎮 Разработка многопользовательской игры на Flask и WebSockets

from flask import Flask, render_template, request
from flask_socketio import SocketIO, emit

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key'
socketio = SocketIO(app)

players = set()

@app.route('/')
def index():
    return render_template('index.html')

@socketio.on('connect')
def handle_connect():
    players.add(request.sid)
    emit('player_count', {'count': len(players)}, broadcast=True)

@socketio.on('disconnect')
def handle_disconnect():
    players.discard(request.sid)
    emit('player_count', {'count': len(players)}, broadcast=True)

@socketio.on('move')
def handle_move(data):
    emit('move', data, broadcast=True, include_self=False)

if __name__ == '__main__':
    socketio.run(app, debug=True)

# Пример использования: создайте файл index.html с минимальной логикой подключения

📌 Этот код создаёт базовую структуру для многопользовательской игры, используя Flask и WebSockets. Сервер обрабатывает подключение игроков и синхронизирует их действия в реальном времени. Подобный подход можно использовать для создания простых интерактивных игр и приложений с обменом данных между клиентами.

Подпишись 👉🏻 @KodduuPython 🤖

📊 Визуализация данных IoT в реальном времени с Plotly

import random
import time
import datetime
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def get_sensor_data() -> float:
    """Имитирует получение данных с датчика."""
    return random.uniform(20.0, 25.0)

def update_plot(fig, x_data, y_data):
    """Обновляет график новыми данными."""
    fig.data[0].x = x_data
    fig.data[0].y = y_data
    fig.update_layout(transition_duration=500)
    fig.show()

def visualize_iot_data():
    """Визуализирует данные с IoT-устройств в реальном времени."""
    fig = make_subplots(rows=1, cols=1)
    fig.add_trace(go.Scatter(x=[], y=[], mode='lines+markers', name='Температура'))

    x_data, y_data = [], []

    try:
        while True:
            x_data.append(datetime.datetime.now().strftime('%H:%M:%S'))
            y_data.append(get_sensor_data())
            update_plot(fig, x_data, y_data)
            time.sleep(1)
    except KeyboardInterrupt:
        print("Визуализация остановлена")
    except Exception as e:
        print(f"Ошибка: {e}")

# Пример использования
visualize_iot_data()

📌 Этот код демонстрирует, как собирать и визуализировать данные с IoT-устройств в реальном времени. Он использует Plotly для создания динамического графика, обновляющегося каждую секунду с новыми данными датчика. Такой подход удобен для мониторинга параметров в реальном времени.

Подпишись 👉🏻 @KodduuPython 🤖

Отказ от Cursor 🙅‍♂️

Луну обогнули 🌔, вернёмся на землю.

Я писал про кейс перевода всех сайтов на OpenClaw. Теперь переводим разработку больших продуктов.

Было так: Cursor пишет код как ассистент разработчика. Говоришь ему "построй план", корректируешь, и с нуля быстро можешь написать довольно масштабную систему.

После этого начинается просто корректировка, новые фичи и фиксы багов. И как правило, в большой системе, обросшей фичами, внедрение нового занимает время, особенно если команда не Full Stack (отдельно Front и отдельно Back разработка).

Но я попробовал и вынес дальнейшую разработку двух больших систем в OpenClaw.

По сути, на VPS с OpenClaw (полный sudo доступ) подтягиваем нужные репозитории. И в чате ему говорим: в репо A добавь X, почини Y, найди баги. Вроде все то же самое, как и в Cursor, но нет, все мгновенно оказывается в Prod, и в цикле там уже включено UI тестирование, потому что есть playwright + browser.

Dev Pipeline теперь это не 3 отдельные среды (prod + test + dev), а Front+Back Dev + QA + DevOps со средним time-2-market в 3 недели, а просто три ветки prod/test/dev, все три постоянно подняты, и time-2-market падает до часов.

Этап дизайна в Figma тоже уходит. Зачем просить дизайнера, который все время занят, сделать новый дизайн, если можно просто сказать боту: "Накидай три версии дизайна новой фичи сразу с самой фичей прямо в dev-ветке", и уже через 5 минут смотреть, выбирать и корректировать.

Написание больших систем с нуля все еще удобнее в Cursor, но дальнейшее развитие работает на OpenClaw на 4+.

Напомню, мы используем свою оптимизированную версию OpenClaw - DarwinClaw, которая скоро должна стать доступной для всех.

#КЕЙС

Подпишись 👉🏻 @aigentto 🤖

Создание AI для NPC с алгоритмом A*

from heapq import heappush, heappop
from typing import List, Tuple

def heuristic(a: Tuple[int, int], b: Tuple[int, int]) -> float:
    # Вычисляем манхэттенское расстояние
    return abs(a[0] - b[0]) + abs(a[1] - b[1])

def a_star_search(start: Tuple[int, int], goal: Tuple[int, int], grid: List[List[int]]) -> List[Tuple[int, int]]:
    open_set = []
    heappush(open_set, (0, start))
    came_from = {}
    g_score = {start: 0}
    f_score = {start: heuristic(start, goal)}

    while open_set:
        _, current = heappop(open_set)
        
        if current == goal:
            path = []
            while current in came_from:
                path.append(current)
                current = came_from[current]
            path.append(start)  # Добавляем начальную позицию в путь
            return path[::-1]
        
        for dx, dy in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
            neighbor = (current[0] + dx, current[1] + dy)
            tentative_g_score = g_score[current] + 1

            if 0 <= neighbor[0] < len(grid) and 0 <= neighbor[1] < len(grid[0]) and grid[neighbor[0]][neighbor[1]] == 0:
                if neighbor not in g_score or tentative_g_score < g_score[neighbor]:
                    came_from[neighbor] = current
                    g_score[neighbor] = tentative_g_score
                    f_score[neighbor] = tentative_g_score + heuristic(neighbor, goal)
                    if neighbor not in [i[1] for i in open_set]:
                        heappush(open_set, (f_score[neighbor], neighbor))
    
    return []

# Пример использования
grid = [
    [0, 1, 0, 0, 0],
    [0, 1, 0, 1, 0],
    [0, 0, 0, 1, 0],
    [0, 1, 1, 0, 0],
    [0, 0, 0, 0, 0]
]
start = (0, 0)
goal = (4, 4)
path = a_star_search(start, goal, grid)
print("Путь:", path)

📌 Алгоритм A* используется для поиска оптимального пути на карте, что делает его идеальным для реализации AI в играх.

Подпишись 👉🏻 @KodduuPython 🤖

☁️ Микросервис для прогноза погоды на Flask и OpenWeatherMap API

from flask import Flask, jsonify, request
import requests
import os

app = Flask(__name__)

API_KEY = os.getenv('OPENWEATHERMAP_API_KEY')  # Получение ключа API из переменных окружения
BASE_URL = 'http://api.openweathermap.org/data/2.5/weather'

def get_weather_data(city: str) -> dict:
    try:
        response = requests.get(BASE_URL, params={'q': city, 'appid': API_KEY, 'units': 'metric'})
        response.raise_for_status()
        return response.json()
    except requests.HTTPError as http_err:
        return {'error': f'HTTP error occurred: {http_err}'}
    except Exception as err:
        return {'error': f'Other error occurred: {err}'}

@app.route('/weather', methods=['GET'])
def weather():
    city = request.args.get('city')
    if not city:
        return jsonify({'error': 'Город не указан'}), 400

    weather_data = get_weather_data(city)
    if 'error' in weather_data:
        return jsonify(weather_data), 500
    
    return jsonify({
        'city': city,
        'temperature': weather_data['main']['temp'],
        'description': weather_data['weather'][0]['description']
    })

if __name__ == '__main__':
    app.run(debug=True)

📌 Этот код создает микросервис на Flask, который предоставляет прогноз погоды по заданному городу, используя API OpenWeatherMap. Запросы к API обрабатываются с учетом ошибок, а данные возвращаются в виде JSON. Используйте переменные окружения для хранения ключа API, чтобы повысить безопасность.

Подпишись 👉🏻 @KodduuPython 🤖

🔗 Композирование функций для сложных операций

from typing import Callable

def compose(*functions: Callable) -> Callable:
    def composed_function(x):
        for func in reversed(functions):  # Применение функций в обратном порядке
            x = func(x)
        return x
    return composed_function

# Пример простых функций
def double(x: int) -> int:
    return x * 2

def increment(x: int) -> int:
    return x + 1

# Композирование функций
process = compose(double, increment)

# Пример использования
result = process(3)  # Результат: (3 + 1) * 2 = 8
print(result)

📌 Композирование позволяет объединять простые функции в более сложные, улучшая читаемость и поддержку кода. В данном примере функции double и increment комбинируются в цепочку, применяемую к значению 3. Это особенно полезно в обработке данных, где часто требуется выполнить серию преобразований.

Подпишись 👉🏻 @KodduuPython 🤖