Всем привет.
Пытаюсь добавить турбо страницы через скрипт от яндекса, но ловлю ошибку.
File "turbo.py", line 14
<rss version="2.0" xmlns:yandex="http://news.yandex.ru" xmlns:turbo="http://turbo.yandex.ru">
^
SyntaxError: EOL while scanning string literal
Вот сам скрипт, а тут
описание от яндекса.
# coding=utf-8
import json
import pprint
import time
from urlparse import urlparse
import requests
from requests import HTTPError
HOST_ADDRESS = 'https://example.com'
RSS_STRING = '\
<?xml version="1.0" encoding="UTF-8"?>\
<rss version="2.0" xmlns:yandex="http://news.yandex.ru" xmlns:turbo="http://turbo.yandex.ru">
<channel>
<item turbo="true">
<title>Заголовок страницы</title>
<link>https://example.com</link>
<turbo:content>
<![CDATA[
<header>
<h1>Ресторан «Полезный завтрак»</h1>
<h2>Вкусно и полезно</h2>
<figure>
<img src="https://avatars.mds.yandex.net/get-sbs-sd/403988/e6f459c3-8ada-44bf-a6c9-dbceb60f3757/orig">
</figure>
<menu>
<a href="https://example.com/page1.html">Пункт меню 1</a>
<a href="https://example.com/page2.html">Пункт меню 2</a>
</menu>
</header>
<p>Как хорошо начать день? Вкусно и полезно позавтракать!</p>
<p>Приходите к нам на завтрак. Фотографии наших блюд ищите <a href="#">на нашем сайте</a>.</p>
<h2>Меню</h2>
<figure>
<img src="https://avatars.mds.yandex.net/get-sbs-sd/369181/49e3683c-ef58-4067-91f9-786222aa0e65/orig">
<figcaption>Омлет с травами</figcaption>
</figure>
<p>В нашем меню всегда есть свежие, вкусные и полезные блюда.</p>
<p>Убедитесь в этом сами.</p>
<button formaction="tel:+7(123)456-78-90"
data-background-color="#5B97B0"
data-color="white"
data-primary="true">Заказать столик</button>
<div data-block="widget-feedback" data-stick="false">
<div data-block="chat" data-type="whatsapp" data-url="https://whatsapp.com"></div>
<div data-block="chat" data-type="telegram" data-url="http://telegram.com/"></div>
<div data-block="chat" data-type="vkontakte" data-url="https://vk.com/"></div>
<div data-block="chat" data-type="facebook" data-url="https://facebook.com"></div>
<div data-block="chat" data-type="viber" data-url="https://viber.com"></div>
</div>
<p>Наш адрес: <a href="#">Nullam dolor massa, porta a nulla in, ultricies vehicula arcu.</a></p>
<p>Фотографии — http://unsplash.com</p>
]]>
</turbo:content>
</item>
</channel>
</rss>'
OAUTH_TOKEN = 'OAuth-токен'
AUTH_HEADER = {
'Authorization': 'OAuth %s' % OAUTH_TOKEN
}
SESSION = requests.Session()
SESSION.headers.update(AUTH_HEADER)
API_VERSION = 'v4'
API_BASE_URL = 'https://api.webmaster.yandex.net'
API_URL = API_BASE_URL + '/' + API_VERSION
def validate_api_response(response, required_key_name=None):
content_type = response.headers['Content-Type']
content = json.loads(response.text) if 'application/json' in content_type else None
if response.status_code == 200:
if required_key_name and required_key_name not in content:
raise HTTPError('Unexpected API response. Missing required key: %s' % required_key_name, response=response)
elif content and 'error_message' in content:
raise HTTPError('Error API response. Error message: %s' % content['error_message'], response=response)
else:
response.raise_for_status()
return content
def url_to_host_id(url):
parsed_url = urlparse(url)
scheme = parsed_url.scheme
if not scheme:
raise ValueError('No protocol (https or http) in url')
if scheme != 'http' and scheme != 'https':
raise ValueError('Illegal protocol: %s' % scheme)
port = parsed_url.port
if not port:
port = 80 if scheme == 'http' else 443
hostname = parsed_url.hostname
hostname = hostname.encode('idna').rstrip('.').lower()
return scheme + ':' + hostname + ':' + str(port)
def get_user_id():
r = SESSION.get(API_URL + '/user/')
c = validate_api_response(r, 'user_id')
return c['user_id']
def get_user_host_ids(user_id):
path = '/user/{user_id}/hosts'.format(user_id=user_id)
r = SESSION.get(API_URL + path)
c = validate_api_response(r, 'hosts')
host_ids = [host_info['host_id'] for host_info in c['hosts']]
return host_ids
def is_user_host_id(user_id, host_id):
host_ids = get_user_host_ids(user_id)
return host_id in host_ids
def get_rss_upload_path(user_id, host_id):
path = '/user/{user_id}/hosts/{host_id}/turbo/uploadAddress/?mode={mode}'.format(
user_id=user_id, host_id=host_id, mode='DEBUG')
r = SESSION.get(API_URL + path)
c = validate_api_response(r, 'upload_address')
parsed_url = c['upload_address']
return parsed_url
def upload_rss(upload_path, rss_data):
headers = {
'Content-Type': 'application/rss+xml'
}
r = SESSION.post(url=upload_path, data=rss_data, headers=headers)
c = validate_api_response(r, 'task_id')
return c['task_id']
def get_task_info(user_id, host_id, task_id):
path = '/user/{user_id}/hosts/{host_id}/turbo/tasks/{task_id}'.format(
user_id=user_id, host_id=host_id, task_id=task_id)
r = SESSION.get(API_URL + path)
c = validate_api_response(r)
return c
def retry_call_until(func, predicate, max_tries=5, initial_delay=60, backoff=2):
current_delay = initial_delay
ret_val = None
for n_try in xrange(0, max_tries + 1):
ret_val = func()
if predicate(ret_val):
break
print 'Will retry. Sleeping for %ds' % current_delay
time.sleep(current_delay)
current_delay *= backoff
return ret_val
user_id = get_user_id()
host_id = url_to_host_id(HOST_ADDRESS)
upload_path = get_rss_upload_path(user_id, host_id)
task_id = upload_rss(upload_path, RSS_STRING)
print 'Waiting for the upload task to complete. This will take a while...'
task_info = retry_call_until(
func=lambda: get_task_info(user_id, host_id, task_id),
predicate=lambda task_info: task_info['load_status'] != 'PROCESSING')
print 'Task status: %s' % task_info['load_status']
task_info = get_task_info(user_id, host_id, task_id)
pp = pprint.PrettyPrinter(indent=4)
pp.pprint(task_info)
Сапорты яндекса молчат. К ним обратился в первую очередь )
