JE API的Python实现

距上一篇文章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[i])
        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")
    

 如有需要可下载附件，欢迎各位拍砖。。

夜深了，睡觉去咯。

评论

2 楼 phyeas 2009-03-19  

还测试的哦，这个版本有很多问题，准备再测一下，修改些东西再上传新的

1 楼 woods 2009-03-19  

下载了 楼主动作很快...刚有这想法...