JE API的Python实现

peiyuan1030

　　距上一篇文章10小时，我终于把API的主要功能实现了一遍。但未经测试。
　　废话少说，上代码：
　　首先是__init__.py
　　就是定义API的URL：

'''
初始化jetallker包的配置
Created on 2009-3-15
@author: phy
'''
JE_URL = "http://www.iteye.com/"
JE_API_HOME = JE_URL + "api/"
"""以下是闲聊API"""
JE_AUTH_URL = JE_API_HOME + "auth/verify"   #认证
JE_CHAT_LIST = JE_API_HOME + "twitters/list"   #闲聊列表
JE_CHAT_REPLIES = JE_API_HOME + "twitters/replies"   #@我的
JE_CHAT_ALL = JE_API_HOME + "twitters/all"   #全站闲聊
JE_CHAT_CREATE = JE_API_HOME + "witters/create"   #发布新闲聊
JE_CHAT_DELETE = JE_API_HOME + "twitters/destroy"   #删除闲聊
JE_CHAT_BY_ID = JE_API_HOME + "twitters/show"     #根据ID获取闲聊
#其他……
　　再来，到utils.py，定义一些urllib2的简单工具方法

#!/usr/bin/env python
#coding=UTF-8
'''
Created on 2009-3-15
@author: phy
'''
import urllib2, gzip, base64
from StringIO import StringIO
DEFAULT_USER_AGENT = "PJETallker/0.1"
class SmartRedirectHandler(urllib2.HTTPRedirectHandler):
def http_error_301(self, req, fp, code, msg, headers):
result = urllib2.HTTPRedirectHandler.http_error_301(self, req, fp, code, msg, headers)
result.status = code
return result
def http_error_302(self, req, fp, code, msg, headers):
result = urllib2.HTTPRedirectHandler.http_error_302(self, req, fp, code, msg, headers)
result.status = code
return result
class DefaultErrorHandler(urllib2.HTTPDefaultErrorHandler):
def http_error_default(self, req, fp, code, msg, headers):
result = urllib2.HTTPError(req.get_full_url(), code, msg, headers, fp)
result.status = code
return result
def initHeaders(username, pwd):
return {"Authorization": "Basic " + encodeUser(username, pwd)}
def read(f):
'''读取数据，支持gzip
@param f:
'''
data = f.read()
if(hasattr(f, "headers") and f.headers.get("content-encoding", '') == 'zip'):
data = gzip.GzipFile(fileobj=StringIO(data)).read()
return data
def sendRequest(request, auth=None):
'''发送请求
@param request: urllib2.Request对象
@param auth: 验证字符串，经过encodeUser处理的用户名与密码
@see:jetallker.utils.encodeUser
'''
if(auth):
request.add_header("Authorization", auth)
request.add_header('Accept-encoding', 'gzip')
request.add_header('User-Agent', DEFAULT_USER_AGENT)
request.add_header('Content-Type', "application/x-www-form-urlencoded; charset=UTF-8")
opener = urllib2.build_opener(SmartRedirectHandler(), DefaultErrorHandler())
f = None
try:
f = opener.open(request)
result = {}
result["data"] = read(f)
if(hasattr(f, "headers")):
result["etag"] = f.headers.get("ETag")
result["lastmodified"] = f.headers.get("Last-Modified")
if hasattr(f, "url"):
result["url"] = f.url
result["status"] = 200
if hasattr(f, "status"):
result["status"] = f.status
except e:
raise e
finally:
f.close()
return result
def evalJson(json):
'''将json转为Python对象，处理json时将null替换为None   
@param json:
'''
json = json.replace("null", "None")
return eval(json)
def encodeUser(name, pwd, keyStr="ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="):
'''将用户名密码加密
@param name:用户名
@param pwd:密码
@param keyStr:
'''
output, input = "", name + ":" + pwd;
enc1 = enc2 = enc3 = enc4 = chr1 = chr2 = chr3 = '';
i = 0;
while (i < len(input)):
chr1 = ord(input)
chr2 = ord(input[i + 1]) if i + 1 < len(input) else 0
chr3 = ord(input[i + 2]) if i + 2 < len(input) else 0
enc1 = chr1 >> 2;
enc2 = ((chr1 & 3) << 4) | (chr2 >> 4);
enc3 = ((chr2 & 15) << 2) | (chr3 >> 6);
enc4 = chr3 & 63;
if not chr2:
enc3 = enc4 = 64
elif not chr3:
enc4 = 64;
output = output + keyStr[enc1] + keyStr[enc2] + keyStr[enc3] + keyStr[enc4];
i = i + 3
return output;
　　最后是core.py，定义API的实际访问，包括登录验证，闲聊，收藏等。。

#!/usr/bin/env python
#coding=UTF-8
'''
Created on 2009-3-15
@author: phy
'''
from __init__ import *
from utils import *
from urllib2 import Request
class JECore(object):
'''JavaEye Python API
@author: phyeas
'''
def __init__(self, username=None, pwd=None):
self.isLogin = False
self.__authKey = None
self.user = None
if(username and pwd):
login(username, pwd)
def login(self, username, pwd):
'''登录验证
@param username: 用户名
@param pwd:密码
@return: 返回一个元组(登录是否成功,消息)
'''
if not (username and pwd):
return False, "未输入用户名和密码"
request = Request(JE_AUTH_URL, headers=initHeaders(username, pwd))
result = sendRequest(request)
if result["status"] == 401 and result["data"] == "error.auth.fail":
return False, "用户名或密码错误"
elif result["status"] == 401  and result["data"] == "error.auth.over.limit":
return False, "已连续6此验证失败，系统自动禁止登录，请一个小时后重试"
elif result["status"] == 400 and result["data"] == "error.api.over.limit":
return False, "您登录的次数太多了"
elif result["status"] == 200:
self.isLogin = True
self.__authKey = encodeUser(username, pwd)
self.user = evalJson(result["data"])
#print self.__user
return True, "验证成功"
return False, "未知错误，请通知管理员"
def request(self, url, data=None):
'''通用 web请求方法
@param url:请求的url
@param data:要提交的数据
@raise NoLoginError: 如果用户未登录调用该方法将引发NoLoginError
@raise JEAccessError: 如果返回状态不等于200则引发此异常(跳转也不行)
'''
if not(self.isLogin):
raise NoLoginError()
request = Request(url, data=data)
result = sendRequest(request, self.__authKey)
if result["status"] == 200:
try:
return evalJson(result["data"])
except:
pass
raise JEAccessError(result["data"], result["status"])
def request2(self, url, data=None):
if not(self.isLogin):
raise NoLoginError()
request = Request(url, data=data)
return sendRequest(request, self.__authKey)
def getList(self, url, last_id=None, page=None):
'''通用获取闲聊的列表的方法
@param last_id:最后获取到的闲聊ID
@param url: URL
@param page:抓取第N页记录
@raise NoLoginError: 如果用户未登录调用该方法将引发NoLoginError
@raise JEAccessError: 返回结果不正常时引发此异常
'''
data = {"last_id":last_id, "page":page} if last_id and page else None
return self.request(url, data)
def listTalk(self, last_id=None, page=None):
'''获取"闲聊一下 "的列表
@param last_id:最后获取到的闲聊ID
@param page:抓取第N页记录
@raise NoLoginError: 如果用户未登录调用该方法将引发NoLoginError
@raise JEAccessError: 返回结果不正常时引发此异常
'''
return self.getList(JE_CHAT_LIST, last_id, page)
def repliesTalk(self, last_id=None, page=None):
'''获取我的回复列表  --针对闲聊
@param last_id:最后获取到的闲聊ID
@param page:抓取第N页记录
@raise NoLoginError: 如果用户未登录调用该方法将引发NoLoginError
@raise JEAccessError: 返回结果不正常时引发此异常
'''
return self.getList(JE_CHAT_REPLIES, last_id, page)
def allTalk(self, last_id=None, page=None):
'''获取"全站闲聊 "列表
@param last_id:最后获取到的闲聊ID
@param page:抓取第N页记录
@raise NoLoginError: 如果用户未登录调用该方法将引发NoLoginError
@raise JEAccessError: 返回结果不正常时引发此异常
'''
return self.getList(JE_CHAT_ALL, last_id, page)
def createTalk(self, body, reply_to_id=None, via="PJETalker"):
'''创建一个闲聊
@param body:闲聊内容
@param reply_to_id:回复闲聊的ID
'''
data = {"body":body, "reply_to_id":reply_to_id, "via":via}
return self.request(JE_CHAT_CREATE, data)
def deleteTalk(self, id):
'''删除闲聊      
@param id:
'''
return self.request2(JE_CHAT_DELETE, {"id":id})
def showTalk(self, id):
'''根据id获取闲聊，id可以为逗号分割值，如"1,2,3"   
@param id:
'''
data = {"id":id}
return self.request(JE_CHAT_BY_ID, data)
def listFavorites(self):
'''获取用户收藏列表 '''
return self.request(JE_FA_LIST)
def saveFavorites(self, **data):
'''保存用户收藏（添加或更新）'''
return self.request(JE_FA_ADD, data)
def deleteFavorites(self, id):
'''删除用户收藏
@param id:
'''
return self.request2(JE_FA_DELETE, {"id":id})
def listInbox(self, last_id=None, page=None):
'''获取收件箱列表
@param last_id:最后一条信息的id
@param page:抓取第N页记录
@raise NoLoginError: 如果用户未登录调用该方法将引发NoLoginError
@raise JEAccessError: 返回结果不正常时引发此异常
'''
return self.getList(JE_MSG_INBOX, last_id, page)
def sendMsg(self, title, body, receiver_name=None, reply_id=None):
'''发送(或回复)站内消息
@param title:标题
@param body:内容
@param receiver_name:接收人用户名，如为回复则不填
@param reply_id:回复某条消息的ID，如消息为发送则不填
'''
if receiver_name is not None and reply_id is None:
return self.request(JE_MSG_CREATE, data={"title":title, "body":body, "receiver_name":receiver_name})
elif reply_id is not None and receiver_name is None:
return self.request(JE_MSG_REPLY, data={"title":title, "body":body, "id":reply_id})
raise JEAccessError("receiver_name or reply_id must be not null!",0)
def deleteMsg(self,id):
'''删除站内短信
@param id:
'''
return self.request2(JE_MSG_DELETE, {"id":id})
def logOut(self):
'''退出登录'''
self.isLogin, self.__authKey, self.user = False, None, None
class NoLoginError(Exception):
'''用户未登录异常 '''
def __init__(self, message):
self.message = message
def __init__(self):
self.message = "error.auth.fail"
def __str__(self):
return self.message
class JEAccessError(Exception):
'''一般异常：如error.api.over.limit '''
def __init__(self, message, status):
self.message = message
self.status = status
def __str__(self):
return ""
if __name__ == "__main__":
talker = JECore()
success, message = talker.login("username", "pwd")

　　如有需要可下载附件，欢迎各位拍砖。。
　　夜深了，睡觉去咯。