-
Notifications
You must be signed in to change notification settings - Fork 58
/
Copy pathwordpress_rest.py
237 lines (185 loc) · 7.94 KB
/
wordpress_rest.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
"""WordPress REST API (including WordPress.com) hosted blog implementation.
To use, go to your WordPress.com blog's admin console, then go to Appearance,
Widgets, add a Text widget, and put this in its text section::
<a href="https://brid.gy/webmention/wordpress" rel="webmention"></a>
Not this, it breaks::
<link rel="webmention" href="https://brid.gy/webmention/wordpress">
https://developer.wordpress.com/docs/api/
Create returns id, can lookup by id.
Test command line::
curl localhost:8080/webmention/wordpress -d 'source=http://localhost/response.html&target=http://ryandc.wordpress.com/2013/03/24/mac-os-x/'
Making an API call with an access token from the command line::
curl -H 'Authorization: Bearer [TOKEN]' URL...
"""
import collections
import logging
import urllib.request, urllib.parse, urllib.error
from flask import render_template, request
from google.cloud import ndb
from oauth_dropins import wordpress_rest as oauth_wordpress
from oauth_dropins.webutil.flask_util import error, flash
from oauth_dropins.webutil.util import json_dumps, json_loads
from flask_app import app
import models
import superfeedr
import util
from util import redirect
logger = logging.getLogger(__name__)
API_CREATE_COMMENT_URL = 'https://public-api.wordpress.com/rest/v1/sites/%s/posts/%d/replies/new?pretty=true'
API_POST_SLUG_URL = 'https://public-api.wordpress.com/rest/v1/sites/%s/posts/slug:%s?pretty=true'
API_SITE_URL = 'https://public-api.wordpress.com/rest/v1/sites/%s?pretty=true'
class WordPress(models.Source):
"""A WordPress blog.
The key name is the blog hostname.
"""
GR_CLASS = collections.namedtuple('FakeGrClass', ('NAME',))(NAME='WordPress.com')
OAUTH_START = oauth_wordpress.Start
SHORT_NAME = 'wordpress'
site_info = ndb.JsonProperty(compressed=True) # from /sites/$site API call
def feed_url(self):
# http://en.support.wordpress.com/feeds/
return urllib.parse.urljoin(self.silo_url(), 'feed/')
def silo_url(self):
return self.domain_urls[0]
def edit_template_url(self):
return urllib.parse.urljoin(self.silo_url(), 'wp-admin/widgets.php')
@staticmethod
def new(auth_entity=None, **kwargs):
"""Creates and returns a WordPress for the logged in user.
Args:
auth_entity (oauth_dropins.wordpress_rest.WordPressAuth):
"""
site_info = WordPress.get_site_info(auth_entity)
if site_info is None:
return
urls = util.dedupe_urls(util.trim_nulls(
[site_info.get('URL'), auth_entity.blog_url]))
domains = [util.domain_from_link(u) for u in urls]
avatar = (json_loads(auth_entity.user_json).get('avatar_URL')
if auth_entity.user_json else None)
return WordPress(id=domains[0],
auth_entity=auth_entity.key,
name=auth_entity.user_display_name(),
picture=avatar,
superfeedr_secret=util.generate_secret(),
url=urls[0],
domain_urls=urls,
domains=domains,
site_info=site_info,
**kwargs)
def urls_and_domains(self, auth_entity):
"""Returns this blog's URL and domain.
Args:
auth_entity: unused
Returns:
([str url], [str domain]) tuple:
"""
return [self.url], [self.key_id()]
def create_comment(self, post_url, author_name, author_url, content):
r"""Creates a new comment in the source silo.
If the last part of the post URL is numeric, e.g.
``http://site/post/123999``\, it's used as the post id. Otherwise, we
extract the last part of the path as the slug, e.g.
``http://site/post/the-slug``\, and look up the post id via the API.
Args:
post_url (str)
author_name (str)
author_url (str)
content (str)
Returns:
dict: JSON response with ``id`` and other fields
"""
auth_entity = self.auth_entity.get()
logger.info(f'Determining WordPress.com post id for {post_url}')
# extract the post's slug and look up its post id
path = urllib.parse.urlparse(post_url).path
if path.endswith('/'):
path = path[:-1]
slug = path.split('/')[-1]
try:
post_id = int(slug)
except ValueError:
logger.info(f'Looking up post id for slug {slug}')
url = API_POST_SLUG_URL % (auth_entity.blog_id, slug)
post_id = self.urlopen(auth_entity, url).get('ID')
if not post_id:
return error('Could not find post id')
logger.info(f'Post id is {post_id}')
# create the comment
url = API_CREATE_COMMENT_URL % (auth_entity.blog_id, post_id)
content = f'<a href="{author_url}">{author_name}</a>: {content}'
data = {'content': content.encode()}
try:
resp = self.urlopen(auth_entity, url, data=urllib.parse.urlencode(data))
except urllib.error.HTTPError as e:
code, body = util.interpret_http_exception(e)
try:
parsed = json_loads(body) if body else {}
if code == '400' and parsed.get('error') == 'invalid_token':
self.status = 'disabled'
self.put()
return error('User is disabled', status=401)
elif ((code == '400' and parsed.get('error') == 'invalid_input') or
(code == '403' and parsed.get('message') == 'Comments on this post are closed')):
return parsed # known error: https://github.com/snarfed/bridgy/issues/161
except ValueError:
pass # fall through
raise e
resp['id'] = resp.pop('ID', None)
return resp
@classmethod
def get_site_info(cls, auth_entity):
"""Fetches the site info from the API.
Args:
auth_entity (oauth_dropins.wordpress_rest.WordPressAuth)
Returns:
dict: site info, or None if API calls are disabled for this blog
"""
try:
return cls.urlopen(auth_entity, API_SITE_URL % auth_entity.blog_id)
except urllib.error.HTTPError as e:
code, body = util.interpret_http_exception(e)
if (code == '403' and '"API calls to this blog have been disabled."' in body):
flash(f'You need to <a href="http://jetpack.me/support/json-api/">enable the Jetpack JSON API</a> in {util.pretty_link(auth_entity.blog_url)}\'s WordPress admin console.')
redirect('/')
return None
raise
@staticmethod
def urlopen(auth_entity, url, **kwargs):
resp = auth_entity.urlopen(url, **kwargs).read()
logger.debug(resp)
return json_loads(resp)
class Add(oauth_wordpress.Callback):
"""This handles both add and delete.
(WordPress.com only allows a single OAuth redirect URL.)
"""
def finish(self, auth_entity, state=None):
if auth_entity:
if int(auth_entity.blog_id) == 0:
flash('Please try again and choose a blog before clicking Authorize.')
return redirect('/')
# Check if this is a self-hosted WordPress blog
site_info = WordPress.get_site_info(auth_entity)
if site_info is None:
return
elif site_info.get('jetpack'):
logger.info(f'This is a self-hosted WordPress blog! {auth_entity.key_id()} {auth_entity.blog_id}')
return render_template('confirm_self_hosted_wordpress.html',
auth_entity_key=auth_entity.key.urlsafe().decode(),
state=state)
util.maybe_add_or_delete_source(WordPress, auth_entity, state)
@app.route('/wordpress/confirm', methods=['POST'])
def confirm_self_hosted():
util.maybe_add_or_delete_source(
WordPress,
ndb.Key(urlsafe=request.form['auth_entity_key']).get(),
request.form['state'])
class SuperfeedrNotify(superfeedr.Notify):
SOURCE_CLS = WordPress
# wordpress.com doesn't seem to use scope
# https://developer.wordpress.com/docs/oauth2/
start = util.oauth_starter(oauth_wordpress.Start).as_view(
'wordpress_start', '/wordpress/add')
app.add_url_rule('/wordpress/start', view_func=start, methods=['POST'])
app.add_url_rule('/wordpress/add', view_func=Add.as_view('wordpress_add', 'unused'))
app.add_url_rule('/wordpress/notify/<id>', view_func=SuperfeedrNotify.as_view('wordpress_notify'), methods=['POST'])