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

import os
import requests
import sys

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_OAUTH_TOKEN = os.environ.get('FORMS_OAUTH_TOKEN')


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


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


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_OAUTH_TOKEN = os.environ.get('FORMS_OAUTH_TOKEN')


def submit_survey(survey_id: str, **fields) -> int:
    """

    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/form'
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    response = requests.post(url, json=fields, headers=headers)
    if response.status_code == 200:
        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_OAUTH_TOKEN = os.environ.get('FORMS_OAUTH_TOKEN')


def get_one_answer(answer_id: int):
    """

    """
    url = f'{FORMS_PUBLIC_API}/answers'
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    fields = {'answer_id': answer_id}
    response = requests.get(url, fields, headers=headers)
    if response.status_code == 200:
        return response.json()


def main():
    if len(sys.argv) < 2 or not sys.argv[1].isdigit():
        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_HOST = 'https://api.forms.yandex.net'
FORMS_OAUTH_TOKEN = os.environ.get('FORMS_OAUTH_TOKEN')


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):
    """

    """
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    path = f'/v1/surveys/{survey_id}/answers?page_size={page_size}'
    columns = None
    while True:
        url = f'{FORMS_PUBLIC_API_HOST}{path}'
        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

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


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 json
import os
import requests
import sys
import time

FORMS_PUBLIC_API = 'https://api.forms.yandex.net/v1'
FORMS_OAUTH_TOKEN = os.environ.get('FORMS_OAUTH_TOKEN')


def start_export(survey_id: str) -> str:
    """
    Запускает фоновый процесс экспорта ответов.
    Возвращает код запущенной операции.
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/answers/export'
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    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}'
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    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'
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    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_OAUTH_TOKEN = os.environ.get('FORMS_OAUTH_TOKEN')


def upload_file(survey_id: str, filename: Path) -> str:
    """
    Подгрузить файл, в результате получим ссылку для применения на сабмите
    """
    url = f'{FORMS_PUBLIC_API}/surveys/{survey_id}/files'
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    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'
    headers = {'Authorization': f'OAuth {FORMS_OAUTH_TOKEN}'}
    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()