def setUp(self):
self.server = DummyPOP3Server((HOST, 0))
self.server.handler = DummyPOP3_SSLHandler
self.server.start()
self.client = poplib.POP3_SSL(self.server.host, self.server.port)
python类POP3_SSL的实例源码
def test__all__(self):
self.assertIn('POP3_SSL', poplib.__all__)
def authenticate(self, request, **credentials):
"""
Returns user for credentials provided if credentials are valid.
Returns ``None`` otherwise.
:param request: HttpRequest instance
:param credentials: keyword arguments
:return: user object
"""
username = credentials.get('username')
password = credentials.get('password')
login_server = credentials.get('login_server')
port = credentials.get('port')
user_model = get_user_model()
try:
user = user_model.objects.get(username=username)
except user_model.DoesNotExist:
return None
try:
response = poplib.POP3_SSL(host=login_server, port=port)
response.user(user=username)
password_string = response.pass_(pswd=password)
if b'OK' in password_string:
response.quit()
return user
except poplib.error_proto:
return None
except (ValueError, TypeError) as e:
raise e
def popssl(self):
try:
self.pop3 = poplib.POP3_SSL(self.pop3host, self.pop3_port)
# self.pop3.set_debuglevel(1)
except Exception as e:
self.log.error(e)
time.sleep(20)
def setUp(self):
self.server = DummyPOP3Server((HOST, PORT))
self.server.handler = DummyPOP3_SSLHandler
self.server.start()
self.client = poplib.POP3_SSL(self.server.host, self.server.port)
def test__all__(self):
self.assertIn('POP3_SSL', poplib.__all__)
def connectMail():
global pop_conn
pop_conn = poplib.POP3_SSL(mailServer, mailPort)
pop_conn.set_debuglevel(mailDebug)
try:
pop_conn.user(mailLogin)
pop_conn.pass_(mailPass)
print("Logged in %s as %s" % (mailServer, mailLogin))
return True
except:
print("Error connecting "+ mailServer)
return False
def CheckServer(domain,port):
#Check the sever connection
try:
if port==995:
pop_chk=poplib.POP3_SSL(domain,port)
else:
pop_chk=poplib.POP3(domain,port)
pop_chk.quit()
return True
except Exception,e:
return False
05_04_download_google_mail_via_pop3.py 文件源码
项目：011_python_network_programming_cookbook_demo
作者: jerry-0824
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def download_email(username):
mailbox = poplib.POP3_SSL(GOOGLE_POP3_SERVER, '995')
mailbox.user(username)
password = getpass(prompt="Enter your google's password: ")
mailbox.pass_(password)
num_messages = len(mailbox,list()[1])
print "Total emails: %s" %num_messages
print "Getting last message"
for msg in mailbox.retr(num_messages)[1]:
print msg
mailbox.quit()
def run(self):
while 1:
p = None
first = True
pre_number = -1
while service == True:
try:
if first == True:
self.trigger1.emit()
#?????????????POP???110
if popport == '110':
p = poplib.POP3(popserver, popport)
else:
p = poplib.POP3_SSL(popserver, popport)
except:
self.trigger2.emit()
else:
try:
p.user(user)
p.pass_(passwd)
except:
self.trigger3.emit()
else:
try:
resp, mails, octets = p.list()
number = len(mails)
if number != pre_number:
if pre_number != -1:
resp, lines, octets = p.retr(number)
msg_content = b'\r\n'.join(lines)
msg = Parser().parsestr(msg_content)
info = get_info(msg)
subject = info[0].strip()
addr = info[1].strip()
content = info[2].strip()
if addr in whitelist:
thread.start_new_thread(self.processing, (p, subject, content, addr))
except:
self.trigger4.emit()
else:
if pre_number == -1:
self.trigger5.emit()
pre_number = number
finally:
try:
p.quit()
except:
self.trigger4.emit()
first = False
exception_id = -1
time.sleep(2)
time.sleep(float(sleep))
def recv_mail(self):
print 'POP3: recv_mail'
pop = None
try:
pop = poplib.POP3_SSL(POP3_MAIL_IP, POP3_PORT)
except (poplib.error_proto, socket.error) as e:
print 'POP3: ERROR: %s : %s'%(POP3_MAIL_IP,e)
time.sleep(20)
if pop == None:
return
try:
pop.user(POP3_ADDR)
pop.pass_(POP3_PASS)
except poplib.error_proto, e:
print 'POP3: ERROR: %s : %s'%(POP3_ADDR, e)
time.sleep(20)
if pop.stat()[0] == 0:
print 'POP3: no messages'
pop.quit()
time.sleep(1) #long sleep
return
for i in xrange( 1,len(pop.list()[1])+1 ):
print 'NUM: %d'%i
status, lines, octets = pop.retr(i)
msg = email.message_from_string('\n'.join(lines))
#->> P2: validate_subject
if self.isCurrentMsg( msg ) == 0:
self.deleteMsg(pop, i)
#pop.quit()
continue
ip = self.getAgentIP( msg.get_all('Received')[2] )
timestamp = msg.get('Date')
for part in msg.walk():
if part.get_content_type():
if part.get_content_type() == "application/octet-stream":
body = part.get_payload(decode=True)
if body is not None:
agent_id = struct.unpack('<I', body[:4])[0]
data = self.P3Scheme.pack_agent_data(body)
#timestamp = datetime.utcnow().isoformat()
print 'POP3: Datas: %s'%body
meta_data = self.P3Scheme.pack_data(self.P3Scheme.separator.join([ip, timestamp]))
self.LocalStorage.save_data_from_agent(agent_id, meta_data + self.P3Scheme.separator + data)
self.Logger.log_message("POP3: Data saved!")
else:
print 'POP3: empty message'
self.deleteMsg(pop, i)
continue
if self.deleteMsg(pop, i) == 0:
print 'POP3: ERROR: Msg is\'t deleted'
pop.guit()
exit(1)
time.sleep(0.1)
break
pop.quit()
#get data from fs, send to MailServer
def checkemail(email,password,pop3_server,prenum):
server = poplib.POP3_SSL(pop3_server, '995')
# server = poplib.POP3(pop3_server)
# ???????????:
# server.set_debuglevel(1)
# ??:??POP3????????:
# print(time.localtime(time.time()))
# print(server.getwelcome())
# ????:
server.user(email)
server.pass_(password)
# stat()???????????:
# print('Messages: %s. Size: %s' % server.stat())
# list()?????????:
resp, mails, octets = server.list()
# ???????????['1 82923', '2 2184', ...]
# print(mails)
# ????????, ??????1??:
index = len(mails)
if index == prenum:
return index, None
unseennum = index - prenum
print('num of mails:', index)
print('unseennum:', unseennum)
for j in range(unseennum):
ii = index - j
# print(ii)
try:
resp, lines, octets = server.retr(ii)
for i in range(len(lines)):
lines[i] = lines[i].decode()
msg_content = '\r\n'.join(lines)
# ???????:
msg = Parser().parsestr(msg_content)
# subj = print_info(msg, specified_email)
subj = print_info(msg)
if subj != None:
print(subj)
for i in range(index):
server.dele(i+1)
server.quit()
return index,subj
except Exception as e:
# raise('exception:', e)
print('exception:', e)
continue
# ???????????????????:
server.dele(index)
# ????:
server.quit()
return index,None
def checkemail(email,password,pop3_server,prenum):
server = poplib.POP3_SSL(pop3_server, '995')
# server = poplib.POP3(pop3_server)
# ???????????:
# server.set_debuglevel(1)
# ??:??POP3????????:
# print(time.localtime(time.time()))
# print(server.getwelcome())
# ????:
server.user(email)
server.pass_(password)
# stat()???????????:
# print('Messages: %s. Size: %s' % server.stat())
# list()?????????:
resp, mails, octets = server.list()
# ???????????['1 82923', '2 2184', ...]
# print(mails)
# ????????, ??????1??:
index = len(mails)
if index == prenum:
return index, None
unseennum = index - prenum
print('num of mails:', index)
print('unseennum:', unseennum)
for j in range(unseennum):
ii = index - j
# print(ii)
try:
resp, lines, octets = server.retr(ii)
for i in range(len(lines)):
lines[i] = lines[i].decode()
msg_content = '\r\n'.join(lines)
# ???????:
msg = Parser().parsestr(msg_content)
# subj = print_info(msg, specified_email)
subj = print_info(msg)
if subj != None:
print(subj)
for i in range(index):
server.dele(i+1)
server.quit()
return index,subj
except Exception as e:
# raise('exception:', e)
print('exception:', e)
continue
# ???????????????????:
server.dele(index)
# ????:
server.quit()
return index,None
def checkemail(email,password,pop3_server,prenum):
server = poplib.POP3_SSL(pop3_server, '995')
# server = poplib.POP3(pop3_server)
# ???????????:
# server.set_debuglevel(1)
# ??:??POP3????????:
# print(time.localtime(time.time()))
# print(server.getwelcome())
# ????:
server.user(email)
server.pass_(password)
# stat()???????????:
# print('Messages: %s. Size: %s' % server.stat())
# list()?????????:
resp, mails, octets = server.list()
# ???????????['1 82923', '2 2184', ...]
# print(mails)
# ????????, ??????1??:
index = len(mails)
if index == prenum:
return index, None
unseennum = index - prenum
print('num of mails:', index)
print('unseennum:', unseennum)
for j in range(unseennum):
ii = index - j
# print(ii)
try:
resp, lines, octets = server.retr(ii)
for i in range(len(lines)):
lines[i] = lines[i].decode()
msg_content = '\r\n'.join(lines)
# ???????:
msg = Parser().parsestr(msg_content)
# subj = print_info(msg, specified_email)
subj = print_info(msg)
if subj != None:
print(subj)
for i in range(index):
server.dele(i+1)
server.quit()
return index,subj
except Exception as e:
# raise('exception:', e)
print('exception:', e)
continue
# ???????????????????:
server.dele(index)
# ????:
server.quit()
return index,None
