python版webpower的edm的api接口

webpower似乎只提供php,java,.net接口，，正好要用到python的接口,于是就参照php的接口写了个webpower的接口

python版：

# -*- coding: utf-8 -*-import os,sys,traceback,timereload(sys)sys.setdefaultencoding(‘utf-8’)from suds.client import Clientclass webpower(object):"""docstring for webpower"""def __init__(self, arg):super(webpower, self).__init__()self.arg = argself.client = Noneself.url = self.arg.get(‘url’,”)self.login = self.arg.get(‘login’,{})def connection(self):”’创建连接”’self.client = Client(self.url)self.client.set_options(headers={"Content-Type":"text/xml; charset=utf-8"})def createSenderAddress(self,fromEmail = ”):”’创建发送的邮件地址@param fromEmail@return Dict”’resultDict = {}try:if not self.client : self.connection()tmpResult = self.client.service.createSenderAddress(self.login,fromEmail)if hasattr(tmpResult, ‘statusMsg’):resultDict[‘statusMsg’] = tmpResult.statusMsgif hasattr(tmpResult, ‘id’):resultDict[‘id’] = tmpResult.idexcept:resultDict[‘error’] = traceback.format_exc()return resultDictdef createCampaign(self,campaignName = ”,defaultSenderName=”,defaultSenderAddrID=”):”’创建活动@param campaignName活动名称@param defaultSenderName 发送者名称@param defaultSenderAddrID 发送的ID地址return Dict”’resultDict = {}try:if not self.client : self.connection()tmpResult = self.client.service.createCampaign(self.login,0,#brandIDcampaignName,#The name of the new campaigndefaultSenderName, #default Sender NamedefaultSenderAddrID, #default Sender Address Id0,#default Reply Address’cn’,#langFalse,#is overall"","","")if hasattr(tmpResult, ‘status’):resultDict[‘status’] = tmpResult.statusif hasattr(tmpResult, ‘id’):resultDict[‘id’] = tmpResult.idexcept:resultDict[‘error’] = traceback.format_exc()return resultDictdef createMailing(self,campaignID=”,title = ”,fromName =”,html =”):”’创建活动对应的邮件内容@param campaignID 活动ID@param title活动标题/活动唯一标题@param fromName 来源名称@param html发送的html内容@return Dict”’resultDict = {}try:if not self.client : self.connection()mailingName = ‘%s_%s’ % (title, int(time.time()))tmpResult = self.client.service.createMailing(self.login,campaignID,#campaign IDmailingName,#mailing Name’cn’,#langtitle,#subjectfromName,#fromName0,#sender IDhtml#html)if hasattr(tmpResult, ‘status’):resultDict[‘status’] = tmpResult.statusif hasattr(tmpResult, ‘id’):resultDict[‘id’] = tmpResult.idexcept:resultDict[‘error’] = traceback.format_exc()return resultDictdef getMailingStatsSummary(self,campaignID=”,mailingID=”):”’通过邮件ID来获取该邮件的发送情况@param campaignID 品牌ID@param mailingID 邮件ID@return Dict”’resultDict = {}try:if not self.client : self.connection()tmpResult = self.client.service.getMailingStatsSummary(self.login,campaignID,#campaign IDmailingID,#mailing ID)resultDict[‘result’] = dict(tmpResult)except:resultDict[‘error’] = traceback.format_exc()return resultDictdef addRecipient(self,campaignID =”,groupIDs =[],recipientData = []):”’增加一条数据到组@param campaignID 品牌ID@param groupIDs组ID@param recipientData 添加的数据集例: [{‘name’:’email’,’value’:’xx@shopex.cn’}]return Dict”’resultDict = {}try:if not self.client : self.connection()#数组结构arrayofint = self.client.factory.create(‘ArrayOfIntType’)arrayofint.int = groupIDs#NewRecipient 结构NewRecipientDataType = self.client.factory.create(‘NewRecipientType’)NewRecipientDataType.fields = recipientDatatmpResult = self.client.service.addRecipient(self.login,campaignID,#campaign IDarrayofint,#groupIDs arrayofint1NewRecipientDataType, #recipient Data recipientDataTrue,True)resultDict[‘result’] = dict(tmpResult)except:resultDict[‘error’] = traceback.format_exc()return resultDictdef addGroup(self,campaignID =”,name = ”):”’新增组@param campaignID 品牌ID@param name组名称@return Dict”’resultDict = {}try:if not self.client : self.connection()#NewGroupTypeNewGroupType = self.client.factory.create(‘NewGroupType’)NewGroupType.name = nameNewGroupType.is_test = TrueNewGroupType.remarks = ”tmpResult = self.client.service.addGroup(self.login,campaignID,#campaign IDNewGroupType,#New Group Type)resultDict[‘result’] = dict(tmpResult)except:resultDict[‘error’] = traceback.format_exc()return resultDictdef sendMailing(self,campaignID =”,mailingID=”,resultsEmail=”,groupIDs=[],callbackUrl=”):”’发送邮件@param campaignID 品牌ID@param mailingID 邮件ID@param isTest是否是测试@param resultsEmail 结果email@param groupIDs组ID@param callbackUrl 返回的URL@return Dict”’resultDict = {}try:if not self.client : self.connection()#groupIDsArrayOfIntType = self.client.factory.create(‘ArrayOfIntType’)ArrayOfIntType.int = groupIDs#langsArrayOfStringType = self.client.factory.create(‘ArrayOfStringType’)ArrayOfStringType.string = [‘cn’]tmpResult = self.client.service.sendMailing(self.login,campaignID,#campaign IDmailingID,#mailing IDFalse,#is testresultsEmail,#results EmailArrayOfIntType, #group ids0,ArrayOfStringType,#langs”,callbackUrl)resultDict[‘result’] = dict(tmpResult)except:resultDict[‘error’] = traceback.format_exc()return resultDictdef sendSingleMailing(self,campaignID=”,mailingID=”,recipientID=”):”’发送单个EMAIL@param campaignID 品牌ID@param mailingID 发送的EMAIL地址@param recipientID recipient id@return Dict”’resultDict = {}try:if not self.client : self.connection()tmpResult = self.client.service.sendSingleMailing(self.login,campaignID,#campaign IDmailingID,#mailing IDrecipientID ,#is test)resultDict[‘result’] = {‘status’:tmpResult}except:resultDict[‘error’] = traceback.format_exc()return resultDictif __name__ == ‘__main__’:arg = {‘login’ : {‘username’:’******’,’password’:’******’},’url’ : ‘http://******.dmdelivery.com/x/soap-v4/wsdl.php’}#print webpower(arg).createSenderAddress(‘wangzhaoming@shopex.cn’)#print webpower(arg).createCampaign(‘1sssss’)#print webpower(arg).createMailing(237,’subjecttest22′,’67272740′,'<html><body>test</body></html>’)#print webpower(arg).getMailingStatsSummary(237,1745)#print webpower(arg).addRecipient(237,[81],[{‘name’:’email’,’value’:’test1@shopex.cn’}])#print webpower(arg).addGroup(237,’create003′)#print webpower(arg).sendMailing(237,1745,’wangzhaoming@shopex.cn’,[81],callbackUrl=’http://callback.webpower-inc.com/callback.php’)#print webpower(arg).sendSingleMailing(237,1745,3)pass

『不可能』只存在於蠢人的字典里

相关文章：

你感兴趣的文章：

标签云：