分类： Python/Ruby
2009-09-13 10:28:29
磁针石：xurongzhong#gmail.com
# python
Python 2.6.2 (r262:71600, Apr 29 2009, 14:54:06)
[GCC 3.4.6 20060404 (Red Hat 3.4.6-9)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import socket
>>> s = socket.socket()
>>> s.connect(('10.50.20.100', 80))
>>> s.send("GET / HTTP/1.0\n\n")
16
>>> s.recv(200)
'HTTP/1.1 200 OK\r\nDate: Tue, 26 May 2009 07:49:43 GMT\r\nServer: Apache/2.0.52 (Red Hat)\r\nX-Powered-By: PHP/4.3.9\r\nContent-Length: 7789\r\nConnection: close\r\nContent-Type: text/html; charset=UTF-8\r\n\r\n
>>> s.close()
以下程序检查指定的机器的制定端口是否打开，即TCP端口检测：
# cat port_checker_tcp.py
#!/usr/bin/env python
import socket
import re
import sys
def check_server(address, port):
#create a TCP socket
s = socket.socket()
print "Attempting to connect to %s on port %s" % (address, port)
try:
s.connect((address, port))
print "Connected to %s on port %s" % (address, port)
return True
except socket.error, e:
print "Connection to %s on port %s failed: %s" % (address, port, e)
return False
if __name__ == '__main__':
from optparse import OptionParser
parser = OptionParser()
parser.add_option("-a", "--address", dest="address", default='localhost',
help="ADDRESS for server", metavar="ADDRESS")
parser.add_option("-p", "--port", dest="port", type="int", default=80,
help="PORT for server", metavar="PORT")
(options, args) = parser.parse_args()
print 'options: %s, args: %s' % (options, args)
check = check_server(options.address, options.port)
print 'check_server returned %s' % check
sys.exit(not check)
# ./port_checker_tcp.py -a 10.50.20.100 -p 80
options: {'port': 80, 'address': '10.50.20.100'}, args: []
Attempting to connect to 10.50.20.100 on port 80
Connected to 10.50.20.100 on port 80
check_server returned True
# ./port_checker_tcp.py -a 10.56.3.110 -p 3302
options: {'port': 3302, 'address': '10.56.3.110'}, args: []
Attempting to connect to 10.56.3.110 on port 3302
Connected to 10.56.3.110 on port 3302
check_server returned True
sys.exit(not check)的实质是把返回值取反，以适合shell的调用。
以下程序检测socket检测web服务，检查指定网站的指定网页是否存在。
# cat web_server_checker_tcp.py
#!/usr/bin/env python
import socket
import re
import sys
def check_webserver(address, port, resource):
#build up HTTP request string
if not resource.startswith('/'):
resource = '/' + resource
request_string = "GET %s HTTP/1.1\r\nHost: %s\r\n\r\n" % (resource, address)
print 'HTTP request:'
print '|||%s|||' % request_string
#create a TCP socket
s = socket.socket()
print "Attempting to connect to %s on port %s" % (address, port)
try:
s.connect((address, port))
print "Connected to %s on port %s" % (address, port)
s.send(request_string)
#we should only need the first 100 bytes or so
rsp = s.recv(100)
print 'Received 100 bytes of HTTP response'
print '|||%s|||' % rsp
except socket.error, e:
print "Connection to %s on port %s failed: %s" % (address, port, e)
return False
finally:
#be a good citizen and close your connection
print "Closing the connection"
s.close()
lines = rsp.splitlines()
print 'First line of HTTP response: %s' % lines[0]
try:
version, status, message = re.split(r'\s+', lines[0], 2)
print 'Version: %s, Status: %s, Message: %s' % (version, status, message)
except ValueError:
print 'Failed to split status line'
return False
if status in ['200', '301']:
print 'Success - status was %s' % status
return True
else:
print 'Status was %s' % status
return False
if __name__ == '__main__':
from optparse import OptionParser
parser = OptionParser()
parser.add_option("-a", "--address", dest="address", default='localhost',
help="ADDRESS for webserver", metavar="ADDRESS")
parser.add_option("-p", "--port", dest="port", type="int", default=80,
help="PORT for webserver", metavar="PORT")
parser.add_option("-r", "--resource", dest="resource", default='index.html',
help="RESOURCE to check", metavar="RESOURCE")
(options, args) = parser.parse_args()
print 'options: %s, args: %s' % (options, args)
check = check_webserver(options.address, options.port, options.resource)
print 'check_webserver returned %s' % check
sys.exit(not check)
如果有高层次的库，建议不要直接调用socket，这样耗时耗力。
这里是把上边的例子用httplib实现。
# cat web_server_checker_httplib.py
#!/usr/bin/env python
import httplib
import sys
def check_webserver(address, port, resource):
#create connection
if not resource.startswith('/'):
resource = '/' + resource
try:
conn = httplib.HTTPConnection(address, port)
print 'HTTP connection created successfully'
#make request
req = conn.request('GET', resource)
print 'request for %s successful' % resource
#get response
response = conn.getresponse()
print 'response status: %s' % response.status
except sock.error, e:
print 'HTTP connection failed: %s' % e
return False
finally:
conn.close()
print 'HTTP connection closed successfully'
if response.status in [200, 301]:
return True
else:
return False
if __name__ == '__main__':
from optparse import OptionParser
parser = OptionParser()
parser.add_option("-a", "--address", dest="address", default='localhost',
help="ADDRESS for webserver", metavar="ADDRESS")
parser.add_option("-p", "--port", dest="port", type="int", default=80,
help="PORT for webserver", metavar="PORT")
parser.add_option("-r", "--resource", dest="resource", default='index.html',
help="RESOURCE to check", metavar="RESOURCE")
(options, args) = parser.parse_args()
print 'options: %s, args: %s' % (options, args)
check = check_webserver(options.address, options.port, options.resource)
print 'check_webserver returned %s' % check
sys.exit(not check)
从ftp服务器获取一个文件：
# cat url_retrieve_ftp_ftplib.py
#!/usr/bin/env python
from ftplib import FTP
import ftplib
import sys
from optparse import OptionParser
parser = OptionParser()
parser.add_option("-a", "--remote_host_address", dest="remote_host_address",
help="REMOTE FTP HOST.",
metavar="REMOTE FTP HOST")
parser.add_option("-r", "--remote_file", dest="remote_file",
help="REMOTE FILE NAME to download.",
metavar="REMOTE FILE NAME")
parser.add_option("-l", "--local_file", dest="local_file",
help="LOCAL FILE NAME to save remote file to", metavar="LOCAL FILE NAME")
parser.add_option("-u", "--username", dest="username",default="Anonymous",
help="USERNAME for ftp server", metavar="USERNAME")
parser.add_option("-p", "--password", dest="password",default="Anonymous",
help="PASSWORD for ftp server", metavar="PASSWORD")
(options, args) = parser.parse_args()
if not (options.remote_file and
options.local_file and
options.remote_host_address):
parser.error('REMOTE HOST, LOCAL FILE NAME, ' \
'and REMOTE FILE NAME are mandatory')
if options.username and not options.password:
parser.error('PASSWORD is mandatory if USERNAME is present')
ftp = FTP(options.remote_host_address)
if options.username:
try:
ftp.login(options.username, options.password)
except ftplib.error_perm, e:
print "Login failed: %s" % e
sys.exit(1)
else:
try:
ftp.login()
except ftplib.error_perm, e:
print "Anonymous login failed: %s" % e
sys.exit(1)
try:
local_file = open(options.local_file, 'wb')
ftp.retrbinary('RETR %s' % options.remote_file, local_file.write)
finally:
local_file.close()
ftp.close()
运行示例：
# ./url_retrieve_ftp_ftplib.py -a ftp.pku.edu.cn -r welcome.msg -l /root/welcome.msg
本例子把上述例子用更高级别的urllib实现。
# cat url_retrieve_urllib.py
#!/usr/bin/env python
"""
url retriever
Usage:
url_retrieve_urllib.py URL FILENAME
URL:
If the URL is an FTP URL the format should be:
ftp://[username[:password]@]hostname/filename
If you want to use absolute paths to the file to download,
you should make the URL look something like this:
ftp://user:password@host/%2Fpath/to/myfile.txt
Notice the '%2F' at the beginning of the path to the file.
FILENAME:
absolute or relative path to the filename to save downloaded file as
"""
import urllib
import sys
if '-h' in sys.argv or '--help' in sys.argv:
print __doc__
sys.exit(1)
if not len(sys.argv) == 3:
print 'URL and FILENAME are mandatory'
print __doc__
sys.exit(1)
url = sys.argv[1]
filename = sys.argv[2]
urllib.urlretrieve(url, filename)
运行示例：
# ./url_retrieve_urllib.py ftp://ftp.pku.edu.cn/welcome.msg /root/wel.msg
urllib2做了扩展，支持更好的认证和cookies。
