Web APIに柔軟に対応するために
MetaGatewayは、
さらにこれらのサービスだけでなく、
MetaGatewayでは新しいサービス等が増えても、
たとえば、
実際はファイルを置いただけで機能を追加できるようにでもできますが、
SERVICE_SUPPORTS = [
{
'key':'twitter',
'api':API_TYPE_TWITTER,
'auth':AUTH_TYPE_BASIC,
'type':SERVICE_TYPE_NO_REQUIRE,
},
{
'key':'hatena_haiku',
'api':API_TYPE_HATENA_HAIKU,
'auth':AUTH_TYPE_BASIC,
'type':SERVICE_TYPE_NO_REQUIRE,
},
...
]
MetaGatewayのプラグイン実装
まずはMetaGatewayのコードを実際に出しながら、
まず、
class ApiPlugin():
DEFAULT_CHANNEL_ID = '1'
def __init__(self, end_point, auth):
self.end_point = end_point
self.auth = auth
self.error_msg = None
def isAuthed(self):
return False
def getUserInfo(self, params):
return False
def getUserChannels(self, params):
return False
def getPost(self, params):
assert( params.has_key('post_id') )
post_id = params[ 'post_id']
documents = self.getRecentPosts(params)
for document in documents:
if( document['post_id'] == post_id ):
return document
raise ApplicationException()
def getRecentPosts(self, params):
return []
def createPost(self, params):
raise ApplicationException()
def isCreatePost(self):
return True
def deletePost(self, params):
raise ApplicationException()
def editPost(self, params):
raise ApplicationException()
def createUploadFile(self, params):
raise ApplicationException()
def getUploadFiles(self, params):
raise ApplicationException()
def deleteUploadFile(self, params):
raise ApplicationException()
def getCategories(self, params):
return []
def getPostCategories(self, params):
return []
def setPostCategories(self, params):
return True
def getTrackbackPings(self, params):
return []
def canTrackback(self, params={}):
return False
def rebuildSite(self, params):
return True
def formatTitle(self, title = ''):
return title
def formatBody(self, body = ''):
return body
boundary = None
def makeBoundary(self):
if( self.boundary == None ):
self.boundary = '----------------------------' + str(int(time.time()))
return self.boundary
MULTIPART_CONTENT_TYPE = 'multipart/form-data'
def makeMultipartHeader(self):
content_type = ("%s; boundary=%s" % (self.MULTIPART_CONTENT_TYPE, self.makeBoundary()))
return {'Content-Type':content_type}
def makeMultipartBody(self, params, exinfo={}):
boundary = self.makeBoundary()
orderd = sorted(params.keys())
s = ''
for k in orderd:
s += ( '--' + boundary + "\r\n" )
if( exinfo.has_key(k) ):
s += ('Content-Disposition: form-data; name=\"%s\"; filename="%s"\r\n' % (k, exinfo[k]['filename']) )
s += ('Content-Type: %s' % (exinfo[k]['content-type']) + "\r\n")
s += ('Content-Length: %s' % len(params[k]) + "\r\n\r\n")
else:
s += ('Content-Disposition: form-data; name=\"%s\"\r\n\r\n' % (k) )
s += (params[k] + "\r\n")
s += ( '--' + boundary + '--\r\n\r\n' )
return s
リスト2のコードは、
isAuthedは、
isCreatePostは、
rebuildSiteは、
プラグイン作成の実際
実際のTwitterプラグインはリスト3のようになります。これに関しても、
class ApiPlugin_twitter(ApiPlugin):
DATE_FORMAT = '%a %b %d %H:%M:%S +0000 %Y'
MAX_POST_LEN = 140
RETURN_ID_NODE = 'id'
_capable = CHANNEL_CAPABLE_ALL & ~CHANNEL_CAPABLE_EDIT \
& ~CHANNEL_CAPABLE_HTML \
& ~CHANNEL_CAPABLE_UPLOAD \
& ~CHANNEL_CAPABLE_SEARCH
def _getTitle(self, text, keyword):
return text
def _getText(self, text, keyword):
return text
def _getPubDate(self, text):
return datetime.datetime.strptime(text, self.DATE_FORMAT)
def _getFeedUrl(self, params = {}):
return self.end_point + 'statuses/user_timeline/' + self.auth.username + '.xml'
def _getPostUrl(self, params = {}):
return self.end_point + 'statuses/update.xml'
def _getDeleteUrl(self, params = {}):
id = params['id']
return self.end_point + ( 'statuses/destroy/%(id)s.xml' % { 'id': id } )
def getUserChannels(self, params = {}):
channels = [
{ 'channel_id' : self.DEFAULT_CHANNEL_ID,
'name' :self.auth.username,
'url' : self.end_point + self.auth.username,
'capable': self._capable, },
]
return channels
def getRecentPosts(self, params = {}):
headers = self.auth.getHeaders()
uri = self._getFeedUrl()
result = urlfetch.fetch(uri, None, urlfetch.GET, headers, False)
if( result.status_code >= 400 ):
raise ApplicationException()
root = cElementTree.fromstring(result.content)
entries = []
tag = './status'
for status in root.findall(tag):
pub_date = self._getPubDate(status.find('created_at').text)
id = status.find('id').text
text = status.find('text').text
keyword = status.find('keyword')
if( keyword != None ):
keyword = keyword.text
entry = {
'post_id': id,
'link': self.end_point + self.auth.username + '/status/' + id,
'title': self._getTitle(text, keyword),
'text': self._getText(text, keyword),
'pub_date' : pub_date,
}
entries.append(entry)
return entries
def createPost(self, params = {}):
headers = self.auth.getHeaders()
uri = self._getPostUrl()
text = params['text']
check_text = text.replace('\n', '')
l = len(check_text)
if(len(check_text) = 400 ):
raise ApplicationException()
root = cElementTree.fromstring(result.content)
id = root.find(self.RETURN_ID_NODE).text
return {
'complete':True,
'post_id':id,
'link': self.end_point + self.auth.username + '/status/' + id,
'title': text,
'original_text': text,
'pub_date': params['pub_date'],
}
def isCreatePost(self, params = {}):
# TODO database compare
return True
def deletePost(self, params = {}):
headers = self.auth.getHeaders()
id = params['post_id']
uri = self._getDeleteUrl( {'id':id} )
result = urlfetch.fetch(uri, None, urlfetch.POST, headers, False)
if( result.status_code >= 400 ):
raise ApplicationException()
root = cElementTree.fromstring(result.content)
deleted_id = root.find(self.RETURN_ID_NODE).text
if( id == deleted_id ):
return True
raise ApplicationException()
class ApiPlugin_mogomogo(twitter.ApiPlugin_twitter):
DATE_FORMAT = '%a %b %d %H:%M:%S +0900 %Y'
HOME_URL = 'http://mogo2.jp/home'
RETURN_ID_NODE = './status/id'
_capable = CHANNEL_CAPABLE_ALL & ~CHANNEL_CAPABLE_EDIT \
& ~CHANNEL_CAPABLE_DELETE \
& ~CHANNEL_CAPABLE_HTML \
& ~CHANNEL_CAPABLE_UPLOAD \
& ~CHANNEL_CAPABLE_SEARCH
def _getFeedUrl(self, params = {}):
return self.end_point + 'statuses/user_timeline.xml'
def _getPostUrl(self, params = {}):
return self.end_point + 'statuses/update.xml'
def _getPubDate(self, text):
dt = datetime.datetime.strptime(text, self.DATE_FORMAT)
offset = 60*60*9
of = datetime.timedelta(seconds=offset)
return dt - of
def getUserChannels(self, params = {}):
channels = [
{ 'channel_id' : self.DEFAULT_CHANNEL_ID,
'name' : self.auth.username,
'url' : self.HOME_URL,
'capable': self._capable,
}
]
return channels
def isCreatePost(self, params = {}):
return True
def deletePost(self, params = {}):
raise ApplicationException()
日付だけを見てもマイクロブログごとに好きなフォーマットを使っているので、
'%a %b %d %H:%M:%S +0000 %Y' | |
mogomogo | '%a %b %d %H:%M:%S +0900 %Y' |
nowa | '%Y-%m-%d %H:%M:%S' |
timelog | '%Y/%m/%d %H:%M:%S' |
はてなハイク | '%Y-%m-%dT%H:%M:%SZ' |
これらも含めて各マイクロブログごとに微妙な差異があるため、
ちなみに、
ユーザの「入り口」を広げたい
最近のWebアプリケーションは、
よく言われるのですが、