OpenClaw + Яндекс: поиск, диск, GPT

YANDEX_SEARCH_KEY=your_search_api_key
YANDEX_CLOUD_API_KEY=your_cloud_api_key
YANDEX_FOLDER_ID=your_folder_id

mkdir -p skills/yandex
touch skills/yandex/search_skill.py

import os
import requests
from typing import Optional, List
class YandexSearchSkill:     def __init__(self):         self.api_key = os.getenv('YANDEX_SEARCH_KEY')         self.base_url = "https://yandex.ru/search/xml"          def search(self, query: str, top_results: int = 5) -> List[dict]:         """         Поиск информации через Яндекс                  Args:             query: поисковый запрос             top_results: количество результатов (по умолчанию 5)                  Returns:             Список результатов поиска         """         params = {             'text': query,             'apikey': self.api_key,             'lr': 213,  # регион (213 = Россия)             'num': top_results         }                  try:             response = requests.get(self.base_url, params=params, timeout=10)             response.raise_for_status()                          results = []             # Парсинг XML результатов             # Структура зависит от формата ответа API             for doc in response.xml.findall('.//doc'):                 results.append({                     'title': doc.find('title').text,                     'url': doc.find('url').text,                     'snippet': doc.find('snippet').text,                     'domain': doc.find('domain').text                 })                          return results                  except requests.RequestException as e:             return {'error': f'Ошибка поиска: {str(e)}'}

skills:
  - name: yandex_search
    type: search
    enabled: true
    config:
      api_key: ${YANDEX_SEARCH_KEY}
      max_results: 5
      language: ru

openclaw skill test yandex_search \
  --query "лучшие CRM для малого бизнеса 2024"

import os
import requests
from io import BytesIO
from typing import Optional
class YandexDiskSkill:     def __init__(self):         self.api_key = os.getenv('YANDEX_CLOUD_API_KEY')         self.base_url = "https://cloud-api.yandex.net/v1/disk"         self.headers = {             'Authorization': f'OAuth {self.api_key}'         }          def upload_file(self, local_path: str, remote_path: str) -> dict:         """         Загрузить файл на Яндекс Диск                  Args:             local_path: путь к локальному файлу             remote_path: путь на диске (/документы/отчёт.pdf)                  Returns:             Статус загрузки         """         try:             # Получить ссылку для загрузки             params = {'path': remote_path, 'overwrite': True}             resp = requests.get(                 f"{self.base_url}/resources/upload",                 params=params,                 headers=self.headers             )             resp.raise_for_status()                          upload_url = resp.json()['href']                          # Загрузить файл             with open(local_path, 'rb') as f:                 file_resp = requests.put(upload_url, data=f)                 file_resp.raise_for_status()                          return {                 'status': 'success',                 'path': remote_path,                 'message': 'Файл успешно загружен'             }                  except requests.RequestException as e:             return {'status': 'error', 'message': str(e)}          def create_report(self, content: str, filename: str) -> dict:         """         Создать текстовый отчёт на диске         """         try:             # Временно сохраняем в памяти             file_obj = BytesIO(content.encode('utf-8'))                          params = {'path': f'/Отчёты/{filename}'}             resp = requests.get(                 f"{self.base_url}/resources/upload",                 params=params,                 headers=self.headers             )                          upload_url = resp.json()['href']             requests.put(upload_url, data=file_obj)                          return {                 'status': 'success',                 'path': f'/Отчёты/{filename}'             }                  except Exception as e:             return {'status': 'error', 'message': str(e)}

workflows:
  - name: weekly_market_report
    trigger: schedule
    schedule: "0 9   MON"  # Каждый понедельник в 9 утра
    steps:
      - skill: yandex_search
        params:
          query: "тренды рынка маркетинга 2024"
          top_results: 10
      - skill: gpt_analyze
        input: "${previous_step.results}"
        params:
          prompt: "Составь сводку на основе найденных результатов"
      - skill: disk_upload
        input: "${previous_step.analysis}"
        params:
          filename: "market_report_${date}.txt"

import os
import requests
import json
class YandexGPTSkill:     def __init__(self):         self.api_key = os.getenv('YANDEX_CLOUD_API_KEY')         self.folder_id = os.getenv('YANDEX_FOLDER_ID')         self.base_url = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"          def generate(         self,         prompt: str,         model: str = "yandexgpt-3",         temperature: float = 0.7,         max_tokens: int = 2000     ) -> dict:         """         Генерация текста через YandexGPT         """         headers = {             'Authorization': f'Api-Key {self.api_key}',             'Content-Type': 'application/json'         }                  payload = {             'modelUri': f'gpt://{self.folder_id}/{model}/latest',             'completionOptions': {                 'stream': False,                 'temperature': temperature,                 'maxTokens': max_tokens             },             'messages': [                 {                     'role': 'user',                     'text': prompt                 }             ]         }                  try:             response = requests.post(                 self.base_url,                 headers=headers,                 json=payload,                 timeout=30             )             response.raise_for_status()                          result = response.json()             return {                 'status': 'success',                 'text': result['result']['alternatives'][0]['message']['text'],                 'tokens': result['result']['usage']             }                  except requests.RequestException as e:             return {'status': 'error', 'message': str(e)}          def summarize(self, text: str, max_length: int = 500) -> dict:         """         Краткая сводка текста         """         prompt = f"""Сделай краткую сводку следующего текста не более {max_length} символов. Текст: {text}
Сводка:"""                  return self.generate(prompt, max_tokens=max_length // 3)

workflow: competitor_monitoring
  trigger: schedule
  schedule: "0 10   *"  # Каждый день в 10 утра
  
  steps:
    - id: search
      skill: yandex_search
      params:
        query: "купить эквадор кофе цена 2024"
        top_results: 20
    
    - id: filter
      skill: gpt_analyze
      input: "${search.results}"
      params:
        prompt: "Найди и выдели только ценовые предложения конкурентов. Укажи компанию и цену."
    
    - id: report
      skill: disk_upload
      input: "${filter.analysis}"
      params:
        filename: "competitor_prices_${date}.md"
    
    - id: notify
      skill: telegram_send
      input: "${filter.analysis}"
      params:
        message: "Обновление по ценам конкурентов"

workflow: content_reuse
  trigger: manual
  
  steps:
    - id: search
      skill: yandex_search
      params:
        query: "как выбрать CRM для малого бизнеса"
        top_results: 5
    
    - id: extract
      skill: web_scrape
      input: "${search.results[0].url}"
    
    - id: rewrite
      skill: yandex_gpt
      input: "${extract.text}"
      params:
        prompt: "Переработай текст оригинальным стилем, добавь примеры для русского рынка"
    
    - id: save
      skill: disk_upload
      input: "${rewrite.output}"
      params:
        filename: "rewritten_content_${date}.md"

openclaw run workflow trend_analysis \
  --param trend="нейросети в маркетинге" \
  --param industry="digital_marketing" \
  --output-format json