Примеры использования API Яндекс Форм

import os
import sys

import requests

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def publish_survey(survey_id: str) -> bool:
    """
    Опубликовать форму.
    Форма не будет опубликована в следующих случаях:
    - форма заблокирована;
    - форма уже набрала максимальное количество ответов;
    - у формы включён механизм автопубликации, если заданное время публикации не наступило.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/publish'
    response = requests.post(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось опубликовать форму: {response.status_code} {response.text}')
        return False
    return True


def unpublish_survey(survey_id: str) -> bool:
    """
    Снять с публикации можно любую опубликованную форму.
    В том числе с включённым механизмом автопубликации, если время снятия с публикации не наступило.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/unpublish'
    response = requests.post(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось снять форму с публикации: {response.status_code} {response.text}')
        return False
    return True

def main():
    if len(sys.argv) < 2:
        return

    survey_id = sys.argv[1]
    # Для публикации
    if publish_survey(survey_id):
        print(f'Форма {survey_id} опубликована')

    # Для снятия с публикации
    # if unpublish_survey(survey_id):
    #     print(f'Форма {survey_id} снята с публикации')


if __name__ == '__main__':
    main()

import json
import os
import requests
import sys

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
}


def submit_survey(survey_id: str, **fields) -> int:
    """
    Функция отправляет ответ на форму.
    Принимает на вход json с контентом заполняемых полей.
    Формат полей можно посмотреть в примере запуска скрипта.
    Важно: заполнение формы через публичное API доступно только для форм в составе организации.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/form'
    response = requests.post(url, json=fields, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось отправить ответ на форму: {response.status_code} {response.text}')
        return None

    result = response.json()
    return result.get('answer_id')


def main():
    if len(sys.argv) < 3:
        return

    survey_id = sys.argv[1]
    fields = json.loads(sys.argv[2])
    answer_id = submit_survey(survey_id, **fields)
    if answer_id:
        print(f'Заполнена форма {survey_id}, создан ответ {answer_id}')


if __name__ == '__main__':
    main()

import json
import os
import requests
import sys

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def get_one_answer(answer_id: int):
    """
    Получить один ответ по его коду.
    """
    url = f'{FORMS_PUBLIC_API}/answers'
    fields = {'answer_id': answer_id}
    response = requests.get(url, fields, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось получить ответ на форму: {response.status_code} {response.text}')
        return None
    return response.json()


def main():
    if len(sys.argv) < 2:
        return

    answer_id = int(sys.argv[1])
    answer_data = get_one_answer(answer_id)
    if answer_data:
        survey_id = answer_data['survey']['id']
        print(f'Получен контент ответа {answer_id} для формы {survey_id}', file=sys.stderr)
        print(json.dumps(answer_data, indent=2, ensure_ascii=False))


if __name__ == '__main__':
    main()

import csv
import os
import requests
import sys

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def get_value(item):
    value = item.get('value')
    if value is None:
        return ''
    if isinstance(value, list):
        return ', '.join(value)
    return value


def fetch_answers(survey_id: str, *, page_size: int = 50):
    """
    Получить ответы на форму.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/answers?page_size={page_size}'
    columns = None
    while True:
        response = requests.get(url, headers=HEADERS)
        if response.status_code != 200:
            break

        result = response.json()
        if columns is None:
            columns = result.get('columns') or []
            yield ['ID', 'Created'] + [
                column.get('text')
                for column in columns
            ]

        for answer in result.get('answers') or []:
            yield [answer.get('id'), answer.get('created')] + [
                get_value(item)
                for item in answer.get('data') or []
            ]

        next_page = result.get('next')
        if not next_page:
            break

        next_path = next_page.get('next_url')
        if not next_path:
            break

        url = FORMS_PUBLIC_API + next_path.removeprefix('/v1')


def main():
    if len(sys.argv) < 2:
        return

    survey_id = sys.argv[1]
    print(f'Ответы на форму {survey_id}', file=sys.stderr)
    csvwriter = csv.writer(sys.stdout, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)
    for answer in fetch_answers(survey_id):
        csvwriter.writerow(answer)


if __name__ == '__main__':
    main()

import os
import requests
import sys
import time

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def start_export(survey_id: str) -> str:
    """
    Запускает фоновый процесс экспорта ответов.
    Возвращает код запущенной операции.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/answers/export'
    params = {'format': 'xlsx'}
    response = requests.post(url, json=params, headers=HEADERS)
    if response.status_code == 202:
        result = response.json()
        operation_id = result.get('id')
        return operation_id
    print(response.status_code, response.json())


def check_finished(operation_id: str) -> bool:
    url = f'{FORMS_PUBLIC_API}/operations/{operation_id}'
    response = requests.get(url, headers=HEADERS)
    if response.status_code == 200:
        result = response.json()
        return result.get('status') == 'ok'


def download_result(survey_id: str, operation_id: str) -> bytes:
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/answers/export-results'
    params = {'task_id': operation_id}
    response = requests.get(url, params=params, headers=HEADERS)
    if response.status_code == 200:
        return response.content


def main():
    if len(sys.argv) < 2:
        return

    survey_id = sys.argv[1]
    operation_id = start_export(survey_id)
    if operation_id:
        print(f'Запущена операция {operation_id}')

        while not check_finished(operation_id):
            print('...')
            time.sleep(5)

        content = download_result(survey_id, operation_id)
        if content:
            filename = f'{survey_id}.xlsx'
            with open(filename, 'wb') as f:
                f.write(content)

            print(f'Ответы выгружены в файл {filename}')


if __name__ == '__main__':
    main()

import os
import requests
import sys
from pathlib import Path

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def upload_file(survey_id: str, filename: Path) -> str:
    """
    Функция загружает файл, в результате получим ссылку для заполнения поля типа Файл в форме.
    Важно: подгрузить файл можно, только если в настройках формы подключено личное хранилище файлов.
Как подключить хранилище, описано в документации https://yandex.ru/support/forms/storage-for-attached-files#s3-ext

    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/files'
    with open(filename, 'rb') as f:
        files = [
            ('file', (filename.name, f.read(), 'application/octet-stream')),
        ]
    response = requests.post(url, files=files, headers=HEADERS)
    if response.status_code == 201:
        result = response.json()
        return result.get('path')


def download_file(filepath: str) -> bytes:
    """
    Скачать загруженный файл.
    """
    url = f'{FORMS_PUBLIC_API}/files'
    params = {'path': filepath}
    response = requests.get(url, params=params, headers=HEADERS)
    if response.status_code == 200:
        return response.content


def main():
    if len(sys.argv) < 3:
        return

    survey_id = sys.argv[1]
    filename = Path(sys.argv[2])
    uploaded_path = upload_file(survey_id, filename.expanduser())
    print(f'Путь загруженного файла {uploaded_path}')

    content = download_file(uploaded_path)
    print(content)


if __name__ == '__main__':
    main()

import os
import requests

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def create_survey() -> str:
    """
    Создаёт новую форму.
    Возвращает код созданной формы.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/'
    payload = {
        'name': 'Тестовая форма',
        'texts': {
            'submit': 'Отправить',
            'title': 'Спасибо за ответы!',
            'subtitle': 'Ваши ответы приняты.',
        },
    }
    response = requests.post(url, json=payload, headers=HEADERS)
    if response.status_code != 201:
        print(f'Не удалось создать форму: {response.status_code} {response.text}')
        return None

    result = response.json()
    survey_id = result['id']
    return survey_id


def add_question(survey_id: str, payload: dict) -> int:
    """
    Добавляет вопрос в форму.
    Возвращает код созданного вопроса.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/questions/'
    response = requests.post(url, json=payload, headers=HEADERS)
    if response.status_code != 201:
        print(f'Не удалось добавить вопрос: {response.status_code} {response.text}')
        return None

    result = response.json()
    question_id = result['id']
    return question_id


def publish_survey(survey_id: str) -> None:
    """
    Публикует форму.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/publish/'
    response = requests.post(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось опубликовать форму: {response.status_code} {response.text}')


def main():
    survey_id = create_survey()
    print(f'Создана форма {survey_id}')

    text_question = {
        'type': 'string',
        'label': 'Как вас зовут?',
        'placeholder': 'Введите ваше имя',
        'multiline': False,
    }
    question_id = add_question(survey_id, text_question)
    print(f'Добавлен текстовый вопрос {question_id}')

    choice_question = {
        'type': 'enum',
        'label': 'Как вы оцениваете наш продукт?',
        'widget': 'radio',
        'items': [
            {'label': 'Отлично'},
            {'label': 'Хорошо'},
            {'label': 'Удовлетворительно'},
            {'label': 'Плохо'},
        ],
    }
    question_id = add_question(survey_id, choice_question)
    print(f'Добавлен вопрос с выбором ответа {question_id}')

    publish_survey(survey_id)
    print(f'Форма {survey_id} опубликована')


if __name__ == '__main__':
    main()

import os
import requests

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def create_survey() -> str:
    """
    Создаёт новую форму.
    Возвращает код созданной формы.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/'
    payload = {
        'name': 'Тестовая форма',
        'texts': {
            'submit': 'Отправить',
            'title': 'Спасибо за ответы!',
            'subtitle': 'Ваши ответы приняты.',
        },
    }
    response = requests.post(url, json=payload, headers=HEADERS)
    if response.status_code != 201:
        print(f'Не удалось создать форму: {response.status_code} {response.text}')
        return None

    result = response.json()
    survey_id = result['id']
    return survey_id


def add_question(survey_id: str, payload: dict) -> int:
    """
    Добавляет вопрос в форму.
    Возвращает код созданного вопроса.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/questions/'
    response = requests.post(url, json=payload, headers=HEADERS)
    if response.status_code != 201:
        print(f'Не удалось добавить вопрос: {response.status_code} {response.text}')
        return None

    result = response.json()
    question_id = result['id']
    return question_id


def move_question(survey_id: str, question_id: int, payload: dict) -> None:
    """
    Перемещает вопрос.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/questions/{question_id}/move/'
    response = requests.post(url, json=payload, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось переместить вопрос {question_id}: {response.status_code} {response.text}')


def main():
    survey_id = create_survey()
    print(f'Создана форма {survey_id}')

    questions = [
        # Страница 1
        # Текстовый вопрос
        {
            'type': 'string',
            'label': 'Как вас зовут?',
            'placeholder': 'Введите ваше имя',
        },
        # Логический вопрос
        {
            'type': 'boolean',
            'label': 'Вы согласны с условиями использования?',
        },
        # Числовой вопрос
        {
            'type': 'integer',
            'label': 'Сколько лет вы пользуетесь нашим продуктом?',
            'placeholder': 'Введите число',
        },
        # Выбор ответа
        {
            'type': 'enum',
            'label': 'Как вы оцениваете наш продукт?',
            'widget': 'radio',
            'items': [
                {'label': 'Отлично'},
                {'label': 'Хорошо'},
                {'label': 'Удовлетворительно'},
                {'label': 'Плохо'},
            ],
        },
        # Страница 2
        # Выбор даты
        {
            'type': 'date',
            'label': 'Укажите дату рождения',
        },
        # Выбор периода
        {
            'type': 'daterange',
            'label': 'Укажите период отпуска',
        },
        # Загрузка файла
        {
            'type': 'file',
            'label': 'Прикрепите резюме',
        },
        # Текст
        {
            'type': 'comment',
            'label': 'Раздел: Дополнительная информация',
            'header': True,
        },
        # Страница 3
        # Оценка по шкале
        {
            'type': 'matrix',
            'label': 'Оцените продукт по критериям',
            'rows': [
                {'label': 'Удобство'},
                {'label': 'Надёжность'},
            ],
            'columns': [
                {'label': 'Плохо'},
                {'label': 'Нормально'},
                {'label': 'Отлично'},
            ],
        },
        # Выбор города из списка
        {
            'type': 'suggest',
            'label': 'Выберите город',
            'data_source': {'name': 'city'},  # Или "country" для стран
        },
        # Серия вопросов
        {
            'type': 'series',
            'label': 'Блок вопросов о работе',
        },
        # Дочерние вопросы для серии
        {
            'type': 'string',
            'label': 'Ваша должность',
            'placeholder': 'Например: разработчик',
        },
        {
            'type': 'string',
            'label': 'Название компании',
            'placeholder': 'Введите название',
        },
    ]

    ids = []
    for question in questions:
        question_id = add_question(survey_id, question)
        print(f'Добавлен вопрос {question["type"]} {question_id}')
        ids.append(question_id)

    # Страница 2: вопросы 5–8
    move_question(survey_id, ids[4], {'create_page': True, 'page': 2})
    move_question(survey_id, ids[5], {'page': 2, 'position': 2})
    move_question(survey_id, ids[6], {'page': 2, 'position': 3})
    move_question(survey_id, ids[7], {'page': 2, 'position': 4})

    # Страница 3: вопросы 9–11
    move_question(survey_id, ids[8],  {'create_page': True, 'page': 3})
    move_question(survey_id, ids[9],  {'page': 3, 'position': 2})
    move_question(survey_id, ids[10], {'page': 3, 'position': 3})

    print('Вопросы разбиты по страницам')

    # Вопросы 12 и 13 перемещаем внутрь серии
    series_id = ids[10]
    move_question(survey_id, ids[11], {'question': series_id})
    move_question(survey_id, ids[12], {'question': series_id, "position": 2})


    print(f'Вопросы добавлены в серию {series_id}')


if __name__ == '__main__':
    main()

import os
import requests
import sys

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def create_keyset(survey_id: str, total: int) -> int:
    """
    Создаёт набор ключей для формы.
    Возвращает код созданного набора.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/keysets/'
    payload = {
        'name': 'Тестовый набор ключей',
        'total': total,
        'is_enabled': True,
    }
    response = requests.post(url, json=payload, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось создать набор ключей: {response.status_code} {response.text}')
        return None

    result = response.json()
    keyset_id = result['id']
    return keyset_id


def download_keyset(survey_id: str, keyset_id: int) -> bytes:
    """
    Скачивает набор ключей в формате XLSX.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/keysets/{keyset_id}/download/'
    response = requests.get(url, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось скачать ключи: {response.status_code} {response.text}')
        return None

    return response.content


def main():
    if len(sys.argv) < 3:
        return

    survey_id = sys.argv[1]
    total = int(sys.argv[2])
    keyset_id = create_keyset(survey_id, total=total)
    print(f'Создан набор ключей {keyset_id}')

    content = download_keyset(survey_id, keyset_id)
    filename = f'keys_{keyset_id}.xlsx'
    with open(filename, 'wb') as f:
        f.write(content)

    print(f'Ключи сохранены в файл {filename}')


if __name__ == '__main__':
    main()

import os
import requests
import sys

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_TOKEN = os.environ.get('FORMS_TOKEN')
ORG_ID = os.environ.get('ORG_ID')
HEADERS = {
    'Authorization': f'OAuth {FORMS_TOKEN}',
    'X-Org-Id': ORG_ID,
}


def add_question(survey_id: str) -> int:
    """
    Добавляет текстовый вопрос в форму.
    Возвращает код созданного вопроса.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/questions/'
    payload = {
        'type': 'string',
        'label': 'Комментарий с картинкой',
        'placeholder': 'Введите текст',
    }
    response = requests.post(url, json=payload, headers=HEADERS)
    if response.status_code != 201:
        print(f'Не удалось добавить вопрос: {response.status_code} {response.text}')
        return None

    result = response.json()
    question_id = result['id']
    return question_id


def upload_image(survey_id: str, image_path: str) -> dict:
    """
    Загружает картинку в форму.
    Возвращает объект с полями id, links, name.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/images/'
    with open(image_path, 'rb') as f:
        response = requests.post(url, files={'image': f}, headers=HEADERS)
    if response.status_code != 201:
        print(f'Не удалось загрузить картинку: {response.status_code} {response.text}')
        return None

    return response.json()


def attach_image(survey_id: str, question_id: int, image: dict) -> bool:
    """
    Прикрепляет загруженную картинку к вопросу.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/questions/{question_id}/'
    payload = {
        'image': {
            'id': image['id'],
            'links': image['links'],
            'name': image['name'],
        },
    }
    response = requests.patch(url, json=payload, headers=HEADERS)
    if response.status_code != 200:
        print(f'Не удалось прикрепить картинку: {response.status_code} {response.text}')
        return False
    return True

def main():
    if len(sys.argv) < 3:
        return

    survey_id = sys.argv[1]
    image_path = sys.argv[2]
    question_id = add_question(survey_id)
    if question_id:
        print(f'Добавлен вопрос {question_id}')

    image = upload_image(survey_id, image_path)
    if image:
        print(f'Загружена картинка {image["id"]}')

    if attach_image(survey_id, question_id, image):
        print(f'Картинка прикреплена к вопросу {question_id}')


if __name__ == '__main__':
    main()