"Those people who have nothing better to do than post on the Internet all day long are rarely the ones who have the most insights."

- Jakob Nielsen, December 1998

五年前我偶然遇到了 Python, 开始了我的 Python 之旅, 我花费了大量的时间 在 comp.lang.python 新闻组里回答问题. 也许某个人发现一个模块正是他想要的, 但是却不知道如何使用它. 也许某个人为他的任务挑选的不合适的模块. 也许某个人已经厌 倦了发明新轮子. 大多时候, 一个简短的例子要比一份手册文档更有帮助.

本书是超过3,000个新闻组讨论的精华部分, 当然也有很多的新脚本, 为了涵盖标准库的每个角落.

我尽力使得每个脚本都易于理解, 易于重用代码. 我有意缩短注释的长度, 如果你想更深入地 了解背景, 那么你可以参阅每个 Python 发布中的参考手册. 本书的重要之处在于范例代码.

本书使用以下习惯用法:

斜体

用于文件名和命令. 还用于定义术语.

等宽字体 e.g. Python

用于代码以及方法,模块,操作符,函数,语句,属性等的名称.

等宽粗体

用于代码执行结果.

除非提到,所有例子都可以在 Python 1.5.2 和 Python 2.0 下运行. 能不能在 Python 2.4/2.5 下执行.....看参与翻译各位的了.

除了一些平台相关模块的脚本, 所有例子都可以在 Windows, Solaris, 以及 Linux 下正常执行.

所有代码都是有版权的. 当然,你可以自由地使用这些这些模块,别忘记你是从哪得到(?学会)这些的.

大多例子的文件名都包含它所使用的模块名称,后边是 "-example-" 以及一个唯一的"序号". 注意有些例子并不是按顺序出现的, 这是为了匹配本书的较早版本 -(the eff-bot guide to) The Standard Python Library.

Python 江湖 QQ 群: 43680167

Feather (校对) QQ: 85660100

"Since the functions in the C runtime library are not part of the Win32 API, we believe the number of applications that will be affected by this bug to be very limited."

- Microsoft, January 1999

Python 的标准库包括了很多的模块, 从 Python 语言自身特定的类型和声明, 到一些只用于少数程序的不著名的模块.

本章描述了一些基本的标准库模块. 任何大型 Python 程序都有可能直接或间接地使用到这类模块的大部分.

下面的这两个模块比其他模块加在一起还要重要: 定义内建函数(例如 len, int, range ...)的 _ _builtin_ _ 模块, 以及定义所有内建异常的 exceptions 模块.

Python 在启动时导入这两个模块, 使任何程序都能够使用它们.

Python 有许多使用了 POSIX 标准 API 和标准 C 语言库的模块. 它们为底层操作系统提供了平台独立的接口.

这类的模块包括: 提供文件和进程处理功能的 os 模块; 提供平台独立的文件名处理 (分拆目录名, 文件名, 后缀等)的 os.path 模块; 以及时间日期处理相关的time/datetime 模块.

[!Feather注: datetime 为 Py2.3 新增模块, 提供增强的时间处理方法 ]

延伸一点说, 网络和线程模块同样也可以归为这一个类型. 不过 Python 并没有在所有的平台/版本实现这些.

标准库里有许多用于支持内建类型操作的库. string 模块实现了常用的字符串处理. math 模块提供了数学计算操作和常量(pi, e都属于这类常量), cmath 模块为复数提供了和 math 一样的功能.

re 模块为 Python 提供了正则表达式支持. 正则表达式是用于匹配字符串或特定子字符串的 有特定语法的字符串模式.

sys 模块可以让你访问解释器相关参数,比如模块搜索路径,解释器版本号等. operator 模块提供了和内建操作符作用相同的函数. copy 模块允许 你复制对象, Python 2.0 新加入的 gc 模块提供了对垃圾收集的相关控制功能.

这个模块包含 Python 中使用的内建函数. 一般不用手动导入这个模块; Python会帮你做好一切.

File: builtin-apply-example-1.py

def function(a, b):
    print a, b

apply(function, ("whither", "canada?"))
apply(function, (1, 2 + 3))

whither canada?
1 5

File: builtin-apply-example-2.py

def function(a, b):
    print a, b

apply(function, ("crunchy", "frog"))
apply(function, ("crunchy",), {"b": "frog"})
apply(function, (), {"a": "crunchy", "b": "frog"})

crunchy frog
crunchy frog
crunchy frog

File: builtin-apply-example-3.py

class Rectangle:
    def _ _init_ _(self, color="white", width=10, height=10):
        print "create a", color, self, "sized", width, "x", height

class RoundedRectangle(Rectangle):
    def _ _init_ _(self, **kw):
        apply(Rectangle._ _init_ _, (self,), kw)

rect = Rectangle(color="green", height=100, width=100)
rect = RoundedRectangle(color="blue", height=20)

create a green <Rectangle instance at 8c8260> sized 100 x 100
create a blue <RoundedRectangle instance at 8c84c0> sized 10 x 20

Python 2.0 提供了另个方法来做相同的事. 你只需要使用一个传统的函数调用 , 使用 * 来标记元组, ** 来标记字典.

下面两个语句是等价的:

result = function(*args, **kwargs)
result = apply(function, args, kwargs)

如果你写过较庞大的 Python 程序, 那么你就应该知道 import 语句是用来导入外部模块的 (当然也可以使用 from-import 版本). 不过你可能不知道 import 其实是靠调用内建 函数 _ _import_ _ 来工作的.

File: builtin-import-example-1.py

import glob, os

modules = []

for module_file in glob.glob("*-plugin.py"):
    try:
        module_name, ext = os.path.splitext(os.path.basename(module_file))
        module = _ _import_ _(module_name)
        modules.append(module)
    except ImportError:
        pass # ignore broken modules

# say hello to all modules
for module in modules:
    module.hello()

example-plugin says hello

注意这个 plug-in 模块文件名中有个 "-" (hyphens). 这意味着你不能使用普通的 import 命令, 因为 Python 的辨识符不允许有 "-" .

File: example-plugin.py

def hello():
    print "example-plugin says hello"

File: builtin-import-example-2.py

def getfunctionbyname(module_name, function_name):
    module = _ _import_ _(module_name)
    return getattr(module, function_name)

print repr(getfunctionbyname("dumbdbm", "open"))

<function open at 794fa0>

File: builtin-import-example-3.py

class LazyImport:
    def _ _init_ _(self, module_name):
        self.module_name = module_name
        self.module = None
    def _ _getattr_ _(self, name):
        if self.module is None:
            self.module = _ _import_ _(self.module_name)
        return getattr(self.module, name)

string = LazyImport("string")

print string.lowercase

abcdefghijklmnopqrstuvwxyz

Python 也提供了重新加载已加载模块的基本支持. [Example 1-8 #eg-1-8 会加载 3 次 hello.py 文件.

File: builtin-reload-example-1.py

import hello
reload(hello)
reload(hello)

hello again, and welcome to the show
hello again, and welcome to the show
hello again, and welcome to the show

reload 直接接受模块作为参数.

[!Feather 注:  ^ 原句无法理解, 稍后讨论.]

注意，当你重加载模块时, 它会被重新编译, 新的模块会代替模块字典里的老模块. 但是, 已经用原模块里的类建立的实例仍然使用的是老模块(不会被更新).

同样地, 使用 from-import 直接创建的到模块内容的引用也是不会被更新的.

File: builtin-dir-example-1.py

def dump(value):
    print value, "=>", dir(value)

import sys

dump(0)
dump(1.0)
dump(0.0j) # complex number
dump([]) # list
dump({}) # dictionary
dump("string")
dump(len) # function
dump(sys) # module

0 => []
1.0 => []
0j => ['conjugate', 'imag', 'real']
[] => ['append', 'count', 'extend', 'index', 'insert',
    'pop', 'remove', 'reverse', 'sort']
{} => ['clear', 'copy', 'get', 'has_key', 'items',
    'keys', 'update', 'values']
string => []
<built-in function len> => ['_ _doc_ _', '_ _name_ _', '_ _self_ _']
<module 'sys' (built-in)> => ['_ _doc_ _', '_ _name_ _',
    '_ _stderr_ _', '_ _stdin_ _', '_ _stdout_ _', 'argv',
    'builtin_module_names', 'copyright', 'dllhandle',
    'exc_info', 'exc_type', 'exec_prefix', 'executable',
...

File: builtin-dir-example-2.py

class A:
    def a(self):
        pass
    def b(self):
        pass

class B(A):
    def c(self):
        pass
    def d(self):
        pass

def getmembers(klass, members=None):
    # get a list of all class members, ordered by class
    if members is None:
        members = []
    for k in klass._ _bases_ _:
        getmembers(k, members)
    for m in dir(klass):
        if m not in members:
            members.append(m)
    return members

print getmembers(A)
print getmembers(B)
print getmembers(IOError)

['_ _doc_ _', '_ _module_ _', 'a', 'b']
['_ _doc_ _', '_ _module_ _', 'a', 'b', 'c', 'd']
['_ _doc_ _', '_ _getitem_ _', '_ _init_ _', '_ _module_ _', '_ _str_ _']

getmembers 函数返回了一个有序列表. 成员在列表中名称出现的越早, 它所处的类层次就越高. 如果无所谓顺序的话, 你可以使用字典代替列表.

[!Feather 注: 字典是无序的, 而列表和元组是有序的, 网上有关于有序字典的讨论]

File: builtin-vars-example-1.py

book = "library2"
pages = 250
scripts = 350


print "the %(book)s book contains more than %(scripts)s scripts" % vars()

the library book contains more than 350 scripts

Python 是一种动态类型语言, 这意味着给一个定变量名可以在不同的场合绑定到不同的类型上. 在接下面例子中, 同样的函数分别被整数, 浮点数, 以及一个字符串调用:

def function(value):
    print value
function(1)
function(1.0)
function("one")

File: builtin-type-example-1.py

def dump(value):
    print type(value), value

dump(1)
dump(1.0)
dump("one")

<type 'int'> 1
<type 'float'> 1.0
<type 'string'> one

File: builtin-type-example-2.py

def load(file):
    if isinstance(file, type("")):
        file = open(file, "rb")
    return file.read()

print len(load("samples/sample.jpg")), "bytes"
print len(load(open("samples/sample.jpg", "rb"))), "bytes"


4672 bytes
4672 bytes

File: builtin-callable-example-1.py

def dump(function):
    if callable(function):
        print function, "is callable"
    else:
        print function, "is *not* callable"

class A:
    def method(self, value):
        return value

class B(A):
    def _ _call_ _(self, value):
        return value

a = A()
b = B()

dump(0) # simple objects
dump("string")
dump(callable)
dump(dump) # function

dump(A) # classes
dump(B)
dump(B.method)

dump(a) # instances
dump(b)
dump(b.method)

0 is *not* callable
string is *not* callable
<built-in function callable> is callable
<function dump at 8ca320> is callable
A is callable
B is callable
<unbound method A.method> is callable
<A instance at 8caa10> is *not* callable
<B instance at 8cab00> is callable
<method A.method of B instance at 8cab00> is callable

注意类对象 (A 和 B) 都是可调用的; 如果调用它们, 就产生新的对象(类实例). 但是 A 类的实例不可调用, 因为它的类没有实现 _ _call_ _ 方法.

你可以在 operator 模块中找到检查对象是否为某一内建类型(数字, 序列, 或者字典等) 的函数. 但是, 因为创建一个类很简单(比如实现基本序列方法的类), 所以对这些 类型使用显式的类型判断并不是好主意.

在处理类和实例的时候会复杂些. Python 不会把类作为本质上的类型对待; 相反地, 所有的类都属于一个特殊的类类型(special class type), 所有的类实例属于一个特殊的实例类型(special instance type).

File: builtin-isinstance-example-1.py

class A:
    pass

class B:
    pass

class C(A):
    pass

class D(A, B):
    pass

def dump(object):
    print object, "=>",
    if isinstance(object, A):
        print "A",
    if isinstance(object, B):
        print "B",
    if isinstance(object, C):
        print "C",
    if isinstance(object, D):
        print "D",
    print

a = A()
b = B()
c = C()
d = D()

dump(a)
dump(b)
dump(c)
dump(d)
dump(0)
dump("string")

<A instance at 8ca6d0> => A
<B instance at 8ca750> => B
<C instance at 8ca780> => A C
<D instance at 8ca7b0> => A B D
0 =>
string =>

File: builtin-issubclass-example-1.py

class A:
    pass

class B:
    pass

class C(A):
    pass

class D(A, B):
    pass

def dump(object):
    print object, "=>",
    if issubclass(object, A):
        print "A",
    if issubclass(object, B):
        print "B",
    if issubclass(object, C):
        print "C",
    if issubclass(object, D):
        print "D",
    print

dump(A)
dump(B)
dump(C)
dump(D)
dump(0)
dump("string")

A => A
B => B
C => A C
D => A B D
0 =>
Traceback (innermost last):
  File "builtin-issubclass-example-1.py", line 29, in ?
  File "builtin-issubclass-example-1.py", line 15, in dump
TypeError: arguments must be classes

File: builtin-eval-example-1.py

def dump(expression):
    result = eval(expression)
    print expression, "=>", result, type(result)

dump("1")
dump("1.0")
dump("'string'")
dump("1.0 + 2.0")
dump("'*' * 10")
dump("len('world')")

1 => 1 <type 'int'>
1.0 => 1.0 <type 'float'>
'string' => string <type 'string'>
1.0 + 2.0 => 3.0 <type 'float'>
'*' * 10 => ********** <type 'string'>
len('world') => 5 <type 'int'>

File: builtin-eval-example-2.py

print eval("_ _import_ _('os').getcwd()")
print eval("_ _import_ _('os').remove('file')")

/home/fredrik/librarybook
Traceback (innermost last):
 File "builtin-eval-example-2", line 2, in ?
 File "<string>", line 0, in ?
os.error: (2, 'No such file or directory')

这里我们得到了一个 os.error 异常, 这说明 Python 事实上在尝试删除文件!

幸运地是, 这个问题很容易解决. 你可以给 eval 函数传递第 2 个参数, 一个定义了该表达式求值时名称空间的字典. 我们测试下, 给函数传递个空字典:

>>> print eval("_ _import_ _('os').remove('file')", {})
Traceback (innermost last):
  File "<stdin>", line 1, in ?
  File "<string>", line 0, in ?
os.error: (2, 'No such file or directory')

呃.... 我们还是得到了个 os.error 异常.

这是因为 Python 在求值前会检查这个字典, 如果没有发现名称为 _ _builtins_ _ 的变量(复数形式), 它就会添加一个:

>>> namespace = {}
>>> print eval("_ _import_ _('os').remove('file')", namespace)
Traceback (innermost last):
  File "<stdin>", line 1, in ?
  File "<string>", line 0, in ?
os.error: (2, 'No such file or directory')
>>> namespace.keys()
['_ _builtins_ _']

如果你打印这个 namespace 的内容, 你会发现里边有所有的内建函数.

[!Feather 注: 如果我RP不错的话, 添加的这个_ _builtins_ _就是当前的_ _builtins_ _]

File: builtin-eval-example-3.py

print eval("_ _import_ _('os').getcwd()", {})
print eval("_ _import_ _('os').remove('file')", {"_ _builtins_ _": {}})

/home/fredrik/librarybook
Traceback (innermost last):
  File "builtin-eval-example-3.py", line 2, in ?
  File "<string>", line 0, in ?
NameError: _ _import_ _

即使这样, 你仍然无法避免针对 CPU 和内存资源的攻击. (比如, 形如 eval("'*'*1000000*2*2*2*2*2*2*2*2*2") 的语句在执行后会使你的程序耗尽系统资源).

File: builtin-compile-example-1.py

NAME = "script.py"

BODY = """
prnt 'owl-stretching time'
"""

try:
    compile(BODY, NAME, "exec")
except SyntaxError, v:
    print "syntax error:", v, "in", NAME

# syntax error: invalid syntax in script.py

File: builtin-compile-example-2.py

BODY = """
print 'the ant, an introduction'
"""

code = compile(BODY, "<script>", "exec")

print code

exec code

<code object ? at 8c6be0, file "<script>", line 0>
the ant, an introduction

File: builtin-compile-example-3.py

import sys, string

class CodeGeneratorBackend:
    "Simple code generator for Python"

    def begin(self, tab="/t"):
        self.code = []
        self.tab = tab
        self.level = 0

    def end(self):
        self.code.append("") # make sure there's a newline at the end 
        return compile(string.join(self.code, "/n"), "<code>", "exec")

    def write(self, string):
        self.code.append(self.tab * self.level + string)

    def indent(self):
        self.level = self.level + 1
        # in 2.0 and later, this can be written as: self.level += 1

    def dedent(self):
        if self.level == 0:
            raise SyntaxError, "internal error in code generator"
        self.level = self.level - 1
        # or: self.level -= 1

#
# try it out!

c = CodeGeneratorBackend()
c.begin()
c.write("for i in range(5):")
c.indent()
c.write("print 'code generation made easy!'")
c.dedent()
exec c.end()

code generation made easy!
code generation made easy!
code generation made easy!
code generation made easy!
code generation made easy!

File: builtin-execfile-example-1.py

execfile("hello.py")

def EXECFILE(filename, locals=None, globals=None):
    exec compile(open(filename).read(), filename, "exec") in locals, globals

EXECFILE("hello.py")

hello again, and welcome to the show
hello again, and welcome to the show

File: hello.py

print "hello again, and welcome to the show"

File: builtin-open-example-1.py

def open(filename, mode="rb"):
    import _ _builtin_ _
    file = _ _builtin_ _.open(filename, mode)
    if file.read(5) not in("GIF87", "GIF89"):
        raise IOError, "not a GIF file"
    file.seek(0)
    return file

fp = open("samples/sample.gif")
print len(fp.read()), "bytes"

fp = open("samples/sample.jpg")
print len(fp.read()), "bytes"

3565 bytes
Traceback (innermost last):
  File "builtin-open-example-1.py", line 12, in ?
  File "builtin-open-example-1.py", line 5, in open
IOError: not a GIF file

[!Feather 注: 明白这个open()函数是干什么的么? 检查一个文件是否是 GIF 文件, 
一般如这类的图片格式都在文件开头有默认的格式. 
另外打开文件推荐使用file()而不是open() , 虽然暂时没有区别]

exceptions 模块提供了标准异常的层次结构. Python 启动的时候会自动导入这个模块, 并且将它加入到 _ _builtin_ _ 模块中. 也就是说, 一般不需要手动导入这个模块.

在 1.5.2 版本时它是一个普通模块, 2.0 以及以后版本成为内建模块.

该模块定义了以下标准异常:

• Exception 是所有异常的基类. 强烈建议(但不是必须)自定义的异常异常也继承这个类.
• SystemExit(Exception) 由 sys.exit 函数引发. 如果它在最顶层没有被 try-except 语句捕获, 那么解释器将直接关闭而不会显示任何跟踪返回信息.
• StandardError(Exception) 是所有内建异常的基类(除 SystemExit 外).
• KeyboardInterrupt(StandardError) 在用户按下 Control-C(或其他打断按键)后 被引发. 如果它可能会在你使用 "捕获所有" 的 try-except 语句时导致奇怪的问题.
• ImportError(StandardError) 在 Python 导入模块失败时被引发.
• EnvironmentError 作为所有解释器环境引发异常的基类. (也就是说, 这些异常一般不是由于程序 bug 引起).
• IOError(EnvironmentError) 用于标记 I/O 相关错误.
• OSError(EnvironmentError) 用于标记 os 模块引起的错误.
• WindowsError(OSError) 用于标记 os 模块中 Windows 相关错误.
• NameError(StandardError) 在 Python 查找全局或局部名称失败时被引发.
• UnboundLocalError(NameError) , 当一个局部变量还没有赋值就被使用时, 会引发这个异常. 这个异常只有在2.0及之后的版本有; 早期版本只会引发一个普通的 NameError .
• AttributeError(StandardError) , 当 Python 寻找(或赋值)给一个实例属性, 方法, 模块功能或其它有效的命名失败时, 会引发这个异常.
• SyntaxError(StandardError) , 当解释器在编译时遇到语法错误, 这个异常就被引发.
• (2.0 及以后版本) IndentationError(SyntaxError) 在遇到非法的缩进时被引发. 该异常只用于 2.0 及以后版本, 之前版本会引发一个 SyntaxError 异常.
• (2.0 及以后版本) TabError(IndentationError) , 当使用 -tt 选项检查不一致缩进时有可能被引发. 该异常只用于 2.0 及以后版本, 之前版本会引发一个SyntaxError 异常.
• TypeError(StandardError) , 当给定类型的对象不支持一个操作时被引发.
• AssertionError(StandardError) 在 assert 语句失败时被引发(即表达式为 false 时).
• LookupError(StandardError) 作为序列或字典没有包含给定索引或键时所引发异常的基类.
• IndexError(LookupError) , 当序列对象使用给定索引数索引失败时(不存在索引对应对象)引发该异常.
• KeyError(LookupError) 当字典对象使用给定索引索引失败时(不存在索引对应对象)引发该异常.
• ArithmeticError(StandardError) 作为数学计算相关异常的基类.
• OverflowError(ArithmeticError) 在操作溢出时被引发(例如当一个整数太大, 导致不能符合给定类型).
• ZeroDivisionError(ArithmeticError) , 当你尝试用 0 除某个数时被引发.
• FloatingPointError(ArithmeticError) , 当浮点数操作失败时被引发.
• ValueError(StandardError) , 当一个参数类型正确但值不合法时被引发.
• (2.0 及以后版本) UnicodeError(ValueError) , Unicode 字符串类型相关异常. 只使用在 2.0 及以后版本.
• RuntimeError(StandardError) , 当出现运行时问题时引发, 包括在限制模式下尝试访问外部内容, 未知的硬件问题等等.
• NotImplementedError(RuntimeError) , 用于标记未实现的函数, 或无效的方法.
• SystemError(StandardError) , 解释器内部错误. 该异常值会包含更多的细节 (经常会是一些深层次的东西, 比如 "eval_code2: NULL globals" ). 这本书的作者编了 5 年程序都没见过这个错误. (想必是没有用 raise SystemError).
• MemoryError(StandardError) , 当解释器耗尽内存时会引发该异常. 注意只有在底层内存分配抱怨时这个异常才会发生; 如果是在你的旧机器上, 这个异常发生之前系统会陷入混乱的内存交换中.

File: exceptions-example-1.py

# python imports this module by itself, so the following
# line isn't really needed
# python 会自动导入该模块, 所以以下这行是不必要的
# import exceptions

class HTTPError(Exception):
    # indicates an HTTP protocol error
    def _ _init_ _(self, url, errcode, errmsg):
        self.url = url
        self.errcode = errcode
        self.errmsg = errmsg
    def _ _str_ _(self):
        return (
            "<HTTPError for %s: %s %s>" %
            (self.url, self.errcode, self.errmsg)
            )

try:
    raise HTTPError("http://www.python.org/foo", 200, "Not Found")
except HTTPError, error:
    print "url", "=>", error.url
    print "errcode", "=>", error.errcode
    print "errmsg", "=>", error.errmsg
    raise # reraise exception

url => http://www.python.org/foo
errcode => 200
errmsg => Not Found
Traceback (innermost last):
  File "exceptions-example-1", line 16, in ?
HTTPError: <HTTPError for http://www.python.org/foo: 200 Not Found>

这个模块中的大部分函数通过对应平台相关模块实现, 比如 posix 和 nt. os 模块会在第一次导入的时候自动加载合适的执行模块.

File: os-example-3.py

import os
import string

def replace(file, search_for, replace_with):
    # replace strings in a text file

    back = os.path.splitext(file)[0] + ".bak"
    temp = os.path.splitext(file)[0] + ".tmp"

    try:
        # remove old temp file, if any
        os.remove(temp)
    except os.error:
        pass

    fi = open(file)
    fo = open(temp, "w")

    for s in fi.readlines():
        fo.write(string.replace(s, search_for, replace_with))

    fi.close()
    fo.close()

    try:
        # remove old backup file, if any
        os.remove(back)
    except os.error:
        pass

    # rename original to backup...
    os.rename(file, back)

    # ...and temporary to original
    os.rename(temp, file)

#
# try it out!

file = "samples/sample.txt"

replace(file, "hello", "tjena")
replace(file, "tjena", "hello")

os 模块也包含了一些用于目录处理的函数.

File: os-example-5.py

import os

for file in os.listdir("samples"):
    print file

sample.au
sample.jpg
sample.wav
...

File: os-example-4.py

import os

# where are we?
cwd = os.getcwd()
print "1", cwd

# go down
os.chdir("samples")
print "2", os.getcwd()

# go back up
os.chdir(os.pardir)
print "3", os.getcwd()

1 /ematter/librarybook
2 /ematter/librarybook/samples
3 /ematter/librarybook

File: os-example-6.py

import os

os.makedirs("test/multiple/levels")

fp = open("test/multiple/levels/file", "w")
fp.write("inspector praline")
fp.close()

# remove the file
os.remove("test/multiple/levels/file")

# and all empty directories above it
os.removedirs("test/multiple/levels")

File: os-example-7.py

import os

os.mkdir("test")
os.rmdir("test")

os.rmdir("samples") # this will fail

Traceback (innermost last):
  File "os-example-7", line 6, in ?
OSError: [Errno 41] Directory not empty: 'samples'

如果需要删除非空目录, 你可以使用 shutil 模块中的 rmtree 函数.

File: os-example-1.py

import os
import time

file = "samples/sample.jpg"

def dump(st):
    mode, ino, dev, nlink, uid, gid, size, atime, mtime, ctime = st
    print "- size:", size, "bytes"
    print "- owner:", uid, gid
    print "- created:", time.ctime(ctime)
    print "- last accessed:", time.ctime(atime)
    print "- last modified:", time.ctime(mtime)
    print "- mode:", oct(mode)
    print "- inode/dev:", ino, dev

#
# get stats for a filename

st = os.stat(file)

print "stat", file
dump(st)
print

#
# get stats for an open file

fp = open(file)

st = os.fstat(fp.fileno())

print "fstat", file
dump(st)

stat samples/sample.jpg
- size: 4762 bytes
- owner: 0 0
- created: Tue Sep 07 22:45:58 1999
- last accessed: Sun Sep 19 00:00:00 1999
- last modified: Sun May 19 01:42:16 1996
- mode: 0100666
- inode/dev: 0 2

fstat samples/sample.jpg
- size: 4762 bytes
- owner: 0 0
- created: Tue Sep 07 22:45:58 1999
- last accessed: Sun Sep 19 00:00:00 1999
- last modified: Sun May 19 01:42:16 1996
- mode: 0100666
- inode/dev: 0 0

返回对象中有些属性在非 Unix 平台下是无意义的, 比如 (st_inode , st_dev)为 Unix 下的为每个文件提供了唯一标识, 但在其他平台可能为任意无意义数据 .

stat 模块包含了很多可以处理该返回对象的常量及函数. 下面的代码展示了其中的一些.

File: os-example-2.py

import os
import stat, time

infile = "samples/sample.jpg"
outfile = "out.jpg"

# copy contents
fi = open(infile, "rb")
fo = open(outfile, "wb")

while 1:
    s = fi.read(10000)
    if not s:
        break
    fo.write(s)

fi.close()
fo.close()

# copy mode and timestamp
st = os.stat(infile)
os.chmod(outfile, stat.S_IMODE(st[stat.ST_MODE]))
os.utime(outfile, (st[stat.ST_ATIME], st[stat.ST_MTIME]))

print "original", "=>"
print "mode", oct(stat.S_IMODE(st[stat.ST_MODE]))
print "atime", time.ctime(st[stat.ST_ATIME])
print "mtime", time.ctime(st[stat.ST_MTIME])

print "copy", "=>"
st = os.stat(outfile)
print "mode", oct(stat.S_IMODE(st[stat.ST_MODE]))
print "atime", time.ctime(st[stat.ST_ATIME])
print "mtime", time.ctime(st[stat.ST_MTIME])

original =>
mode 0666
atime Thu Oct 14 15:15:50 1999
mtime Mon Nov 13 15:42:36 1995
copy =>
mode 0666
atime Thu Oct 14 15:15:50 1999
mtime Mon Nov 13 15:42:36 1995

File: os-example-8.py

import os

if os.name == "nt":
    command = "dir"
else:
    command = "ls -l"

os.system(command)

-rwxrw-r--   1 effbot  effbot        76 Oct  9 14:17 README
-rwxrw-r--   1 effbot  effbot      1727 Oct  7 19:00 SimpleAsyncHTTP.py
-rwxrw-r--   1 effbot  effbot       314 Oct  7 20:29 aifc-example-1.py
-rwxrw-r--   1 effbot  effbot       259 Oct  7 20:38 anydbm-example-1.py
...

命令通过操作系统的标准 shell 执行, 并返回 shell 的退出状态. 需要注意的是在 Windows 95/98 下, shell 通常是 command.com , 它的推出状态总是 0.

由于 11os.system11 直接将命令传递给 shell , 所以如果你不检查传入参数的时候会很危险 (比如命令 os.system("viewer %s" % file), 将 file 变量设置为 "sample.jpg; rm -rf $HOME" ....). 如果不确定参数的安全性, 那么最好使用 exec 或 spawn 代替(稍后介绍).

File: os-exec-example-1.py

import os
import sys

program = "python"
arguments = ["hello.py"]

print os.execvp(program, (program,) +  tuple(arguments))
print "goodbye"

hello again, and welcome to the show

File: os-exec-example-2.py

import os
import sys

def run(program, *args):
    pid = os.fork()
    if not pid:
        os.execvp(program, (program,) +  args)
    return os.wait()[0]

run("python", "hello.py")

print "goodbye"

hello again, and welcome to the show
goodbye

fork 函数在子进程返回中返回 0 (这个进程首先从 fork 返回值), 在父进程中返回一个非 0 的进程标识符(子进程的 PID ). 也就是说, 只有当我们处于子进程的时候 "not pid" 才为真.

File: os-spawn-example-1.py

import os
import string

def run(program, *args):
    # find executable
    for path in string.split(os.environ["PATH"], os.pathsep):
        file = os.path.join(path, program) + ".exe"
        try:
            return os.spawnv(os.P_WAIT, file, (file,) + args)
        except os.error:
            pass
    raise os.error, "cannot find executable"

run("python", "hello.py")

print "goodbye"

hello again, and welcome to the show
goodbye

File: os-spawn-example-2.py

import os
import string

def run(program, *args, **kw):
    # find executable
    mode = kw.get("mode", os.P_WAIT)
    for path in string.split(os.environ["PATH"], os.pathsep):
        file = os.path.join(path, program) + ".exe"
        try:
            return os.spawnv(mode, file, (file,) + args)
        except os.error:
            pass
    raise os.error, "cannot find executable"

run("python", "hello.py", mode=os.P_NOWAIT)
print "goodbye"

goodbye
hello again, and welcome to the show

File: os-spawn-example-3.py

import os
import string

if os.name in ("nt", "dos"):
    exefile = ".exe"
else:
    exefile = ""

def spawn(program, *args):
    try:
        # possible 2.0 shortcut!
        return os.spawnvp(program, (program,) + args)
    except AttributeError:
        pass
    try:
        spawnv = os.spawnv
    except AttributeError:

        # assume it's unix
        pid = os.fork()
        if not pid:
            os.execvp(program, (program,) + args)
        return os.wait()[0]
    else:
        # got spawnv but no spawnp: go look for an executable
        for path in string.split(os.environ["PATH"], os.pathsep):
            file = os.path.join(path, program) + exefile
            try:
                return spawnv(os.P_WAIT, file, (file,) + args)
            except os.error:
                pass
        raise IOError, "cannot find executable"

#
# try it out!

spawn("python", "hello.py")

print "goodbye"

hello again, and welcome to the show
goodbye

File: os-example-14.py

import os
import time

pid = os.fork()
if pid:
    os._exit(0) # kill original

print "daemon started"
time.sleep(10)
print "daemon terminated"

需要创建一个真正的后台程序稍微有点复杂, 首先调用 setpgrp 函数创建一个 "进程组首领/process group leader". 否则, 向无关进程组发送的信号(同时)会引起守护进程的问题:

os.setpgrp()

为了确保守护进程创建的文件能够获得程序指定的 mode flags(权限模式标记?), 最好删除 user mode mask:

os.umask(0)

然后, 你应该重定向 stdout/stderr 文件, 而不能只是简单地关闭它们(如果你的程序需要 stdout 或 stderr 写入内容的时候, 可能会出现意想不到的问题).

class NullDevice:
    def write(self, s):
        pass
sys.stdin.close()
sys.stdout = NullDevice()
sys.stderr = NullDevice()

换言之, 由于 Python 的 print 和 C 中的 printf/fprintf 在设备(device) 没有连接后不会关闭你的程序, 此时守护进程中的 sys.stdout.write() 会抛出一个IOError 异常, 而你的程序依然在后台运行的很好....

File: os-example-9.py

import os
import sys

try:
    sys.exit(1)
except SystemExit, value:
    print "caught exit(%s)" % value

try:
    os._exit(2)
except SystemExit, value:
    print "caught exit(%s)" % value

print "bye!"

caught exit(1)

os.path 模块包含了各种处理长文件名(路径名)的函数. 先导入 (import) os 模块, 然后就可以以 os.path 访问该模块.

File: os-path-example-1.py

import os

filename = "my/little/pony"

print "using", os.name, "..."
print "split", "=>", os.path.split(filename)
print "splitext", "=>", os.path.splitext(filename)
print "dirname", "=>", os.path.dirname(filename)
print "basename", "=>", os.path.basename(filename)
print "join", "=>", os.path.join(os.path.dirname(filename),
                                 os.path.basename(filename))

using nt ...
split => ('my/little', 'pony')
splitext => ('my/little/pony', '')
dirname => my/little
basename => pony
join => my/little/pony

注意这里的 split 只分割出最后一项(不带斜杠).

File: os-path-example-2.py

import os

FILES = (
    os.curdir,
    "/",
    "file",
    "/file",
    "samples",
    "samples/sample.jpg",
    "directory/file",
    "../directory/file",
    "/directory/file"
    )

for file in FILES:
    print file, "=>",
    if os.path.exists(file):
        print "EXISTS",
    if os.path.isabs(file):
        print "ISABS",
    if os.path.isdir(file):
        print "ISDIR",
    if os.path.isfile(file):
        print "ISFILE",
    if os.path.islink(file):
        print "ISLINK",
    if os.path.ismount(file):
        print "ISMOUNT",
    print

. => EXISTS ISDIR
/ => EXISTS ISABS ISDIR ISMOUNT
file =>
/file => ISABS
samples => EXISTS ISDIR
samples/sample.jpg => EXISTS ISFILE
directory/file =>
../directory/file =>
/directory/file => ISABS

File: os-path-expanduser-example-1.py

import os

print os.path.expanduser("~/.pythonrc")

# /home/effbot/.pythonrc

File: os-path-expandvars-example-1.py

import os

os.environ["USER"] = "user"

print os.path.expandvars("/home/$USER/config")
print os.path.expandvars("$USER/folders")

/home/user/config
user/folders

File: os-path-walk-example-1.py

import os

def callback(arg, directory, files):
    for file in files:
        print os.path.join(directory, file), repr(arg)

os.path.walk(".", callback, "secret message")

./aifc-example-1.py 'secret message'
./anydbm-example-1.py 'secret message'
./array-example-1.py 'secret message'
...
./samples 'secret message'
./samples/sample.jpg 'secret message'
./samples/sample.txt 'secret message'
./samples/sample.zip 'secret message'
./samples/articles 'secret message'
./samples/articles/article-1.txt 'secret message'
./samples/articles/article-2.txt 'secret message'
...

File: os-path-walk-example-2.py

import os

def index(directory):
    # like os.listdir, but traverses directory trees
    stack = [directory]
    files = []
    while stack:
        directory = stack.pop()
        for file in os.listdir(directory):
            fullname = os.path.join(directory, file)
            files.append(fullname)
            if os.path.isdir(fullname) and not os.path.islink(fullname):
                stack.append(fullname)
    return files

for file in index("."):
    print file

./aifc-example-1.py
./anydbm-example-1.py
./array-example-1.py
...

File: os-path-walk-example-3.py

import os

class DirectoryWalker:
    # a forward iterator that traverses a directory tree

    def _ _init_ _(self, directory):
        self.stack = [directory]
        self.files = []
        self.index = 0

    def _ _getitem_ _(self, index):
        while 1:
            try:
                file = self.files[self.index]
                self.index = self.index + 1
            except IndexError:
                # pop next directory from stack
                self.directory = self.stack.pop()
                self.files = os.listdir(self.directory)
                self.index = 0
            else:
                # got a filename
                fullname = os.path.join(self.directory, file)
                if os.path.isdir(fullname) and not os.path.islink(fullname):
                    self.stack.append(fullname)
                return fullname

for file in DirectoryWalker("."):
    print file

./aifc-example-1.py
./anydbm-example-1.py
./array-example-1.py
...

注意 DirectoryWalker 类并不检查传递给 _ _getitem_ _ 方法的索引值. 这意味着如果你越界访问序列成员(索引数字过大)的话, 这个类将不能正常工作.

File: os-path-walk-example-4.py

import os, stat

class DirectoryStatWalker:
    # a forward iterator that traverses a directory tree, and
    # returns the filename and additional file information

    def _ _init_ _(self, directory):
        self.stack = [directory]
        self.files = []
        self.index = 0

    def _ _getitem_ _(self, index):
        while 1:
            try:
                file = self.files[self.index]
                self.index = self.index + 1
            except IndexError:
                # pop next directory from stack
                self.directory = self.stack.pop()
                self.files = os.listdir(self.directory)
                self.index = 0
            else:
                # got a filename
                fullname = os.path.join(self.directory, file)
                st = os.stat(fullname)
                mode = st[stat.ST_MODE]
                if stat.S_ISDIR(mode) and not stat.S_ISLNK(mode):
                    self.stack.append(fullname)
                return fullname, st

for file, st in DirectoryStatWalker("."):
    print file, st[stat.ST_SIZE]

./aifc-example-1.py 336
./anydbm-example-1.py 244
./array-example-1.py 526

File: stat-example-1.py

import stat
import os, time

st = os.stat("samples/sample.txt")

print "mode", "=>", oct(stat.S_IMODE(st[stat.ST_MODE]))

print "type", "=>",
if stat.S_ISDIR(st[stat.ST_MODE]):
    print "DIRECTORY",
if stat.S_ISREG(st[stat.ST_MODE]):
    print "REGULAR",
if stat.S_ISLNK(st[stat.ST_MODE]):
    print "LINK",
print

print "size", "=>", st[stat.ST_SIZE]

print "last accessed", "=>", time.ctime(st[stat.ST_ATIME])
print "last modified", "=>", time.ctime(st[stat.ST_MTIME])
print "inode changed", "=>", time.ctime(st[stat.ST_CTIME])

mode => 0664
type => REGULAR
size => 305
last accessed => Sun Oct 10 22:12:30 1999
last modified => Sun Oct 10 18:39:37 1999
inode changed => Sun Oct 10 15:26:38 1999

File: string-example-1.py

import string

text = "Monty Python's Flying Circus"

print "upper", "=>", string.upper(text)
print "lower", "=>", string.lower(text)
print "split", "=>", string.split(text)
print "join", "=>", string.join(string.split(text), "+")
print "replace", "=>", string.replace(text, "Python", "Java")
print "find", "=>", string.find(text, "Python"), string.find(text, "Java")
print "count", "=>", string.count(text, "n")

upper => MONTY PYTHON'S FLYING CIRCUS
lower => monty python's flying circus
split => ['Monty', "Python's", 'Flying', 'Circus']
join => Monty+Python's+Flying+Circus
replace => Monty Java's Flying Circus
find => 6 -1
count => 3

在 Python 1.5.2 以及更早版本中, string 使用 strop 中的函数来实现模块功能.

File: string-example-2.py

text = "Monty Python's Flying Circus"

print "upper", "=>", text.upper()
print "lower", "=>", text.lower()
print "split", "=>", text.split()
print "join", "=>", "+".join(text.split())
print "replace", "=>", text.replace("Python", "Perl")
print "find", "=>", text.find("Python"), text.find("Perl")
print "count", "=>", text.count("n")

upper => MONTY PYTHON'S FLYING CIRCUS
lower => monty python's flying circus
split => ['Monty', "Python's", 'Flying', 'Circus']
join => Monty+Python's+Flying+Circus
replace => Monty Perl's Flying Circus
find => 6 -1
count => 3

File: string-example-3.py

import string

print int("4711"),
print string.atoi("4711"),
print string.atoi("11147", 8), # octal 八进制
print string.atoi("1267", 16), # hexadecimal 十六进制
print string.atoi("3mv", 36) # whatever...

print string.atoi("4711", 0),
print string.atoi("04711", 0),
print string.atoi("0x4711", 0)

print float("4711"),
print string.atof("1"),
print string.atof("1.23e5")

4711 4711 4711 4711 4711
4711 2505 18193
4711.0 1.0 123000.0

大多数情况下 (特别是当你使用的是1.6及更高版本时) ，你可以使用 int 和 float 函数代替 string 模块中对应的函数。

atoi 函数可以接受可选的第二个参数, 指定数基(number base). 如果数基为 0, 那么函数将检查字符串的前几个字符来决定使用的数基: 如果为 "0x," 数基将为 16 (十六进制), 如果为 "0," 则数基为 8 (八进制). 默认数基值为 10 (十进制), 当你未传递参数时就使用这个值.

在 1.6 及以后版本中, int 函数和 atoi 一样可以接受第二个参数. 与字符串版本函数不一样的是 , int 和 float 可以接受 Unicode 字符串对象.

"Some people, when confronted with a problem, think 'I know, I'll use regular expressions.' Now they have two problems."

- Jamie Zawinski, on comp.lang.emacs

re 模块提供了一系列功能强大的正则表达式 (regular expression) 工具, 它们允许你快速检查给定字符串是否与给定的模式匹配 (使用 match 函数), 或者包含这个模式 (使用 search 函数). 正则表达式是以紧凑(也很神秘)的语法写出的字符串模式.

File: re-example-1.py

import re

text = "The Attila the Hun Show"

# a single character 单个字符
m = re.match(".", text)
if m: print repr("."), "=>", repr(m.group(0))

# any string of characters 任何字符串
m = re.match(".*", text)
if m: print repr(".*"), "=>", repr(m.group(0))

# a string of letters (at least one) 只包含字母的字符串(至少一个)
m = re.match("/w+", text)
if m: print repr("/w+"), "=>", repr(m.group(0))

# a string of digits 只包含数字的字符串
m = re.match("/d+", text)
if m: print repr("/d+"), "=>", repr(m.group(0))

 '.' => 'T'
'.*' => 'The Attila the Hun Show'
'//w+' => 'The'

File: re-example-2.py

import re

text ="10/15/99"

m = re.match("(/d{2})/(/d{2})/(/d{2,4})", text)
if m:
    print m.group(1, 2, 3)

('10', '15', '99')

File: re-example-3.py

import re

text = "Example 3: There is 1 date 10/25/95 in here!"

m = re.search("(/d{1,2})/(/d{1,2})/(/d{2,4})", text)

print m.group(1), m.group(2), m.group(3)

month, day, year = m.group(1, 2, 3)
print month, day, year

date = m.group(0)
print date

10 25 95
10 25 95
10/25/95

File: re-example-4.py

import re

text = "you're no fun anymore..."

# literal replace (string.replace is faster)
# 文字替换 (string.replace 速度更快)
print re.sub("fun", "entertaining", text)

# collapse all non-letter sequences to a single dash 
# 将所有非字母序列转换为一个"-"(dansh,破折号)
print re.sub("[^/w]+", "-", text)

# convert all words to beeps 
# 将所有单词替换为 BEEP
print re.sub("/S+", "-BEEP-", text)

you're no entertaining anymore...
you-re-no-fun-anymore-
-BEEP- -BEEP- -BEEP- -BEEP-

File: re-example-5.py

import re
import string

text = "a line of text//012another line of text//012etc..."

def octal(match):
    # replace octal code with corresponding ASCII character
    # 使用对应 ASCII 字符替换八进制代码
    return chr(string.atoi(match.group(1), 8))

octal_pattern = re.compile(r"//(/d/d/d)")

print text
print octal_pattern.sub(octal, text)

a line of text/012another line of text/012etc...
a line of text
another line of text
etc...

如果你不编译, re 模块会为你缓存一个编译后版本, 所有的小脚本中, 通常不需要编译正则表达式. Python1.5.2 中, 缓存中可以容纳 20 个匹配模式, 而在 2.0 中, 缓存则可以容纳 100 个匹配模式.

File: re-example-6.py

import re, string

def combined_pattern(patterns):
    p = re.compile(
        string.join(map(lambda x: "("+x+")", patterns), "|")
        )
    def fixup(v, m=p.match, r=range(0,len(patterns))):
        try:
            regs = m(v).regs
        except AttributeError:
            return None # no match, so m.regs will fail
        else:
            for i in r:
                if regs[i+1] != (-1, -1):
                    return i
    return fixup

#
# try it out!

patterns = [
    r"/d+",
    r"abc/d{2,4}",
    r"p/w+"
]

p = combined_pattern(patterns)

print p("129391")
print p("abc800")
print p("abc1600")
print p("python")
print p("perl")
print p("tcl")

0
1
1
2
2
None

File: math-example-1.py

import math

print "e", "=>", math.e
print "pi", "=>", math.pi
print "hypot", "=>", math.hypot(3.0, 4.0)

# and many others...

e => 2.71828182846
pi => 3.14159265359
hypot => 5.0

完整函数列表请参阅 Python Library Reference .

File: cmath-example-1.py

import cmath

print "pi", "=>", cmath.pi
print "sqrt(-1)", "=>", cmath.sqrt(-1)

pi => 3.14159265359
sqrt(-1) => 1j

完整函数列表请参阅 Python Library Reference .

File: operator-example-1.py

import operator

sequence = 1, 2, 4

print "add", "=>", reduce(operator.add, sequence)
print "sub", "=>", reduce(operator.sub, sequence)
print "mul", "=>", reduce(operator.mul, sequence)
print "concat", "=>", operator.concat("spam", "egg")
print "repeat", "=>", operator.repeat("spam", 5)
print "getitem", "=>", operator.getitem(sequence, 2)
print "indexOf", "=>", operator.indexOf(sequence, 2)
print "sequenceIncludes", "=>", operator.sequenceIncludes(sequence, 3)

add => 7
sub => -5
mul => 8
concat => spamegg
repeat => spamspamspamspamspam

getitem => 4
indexOf => 1
sequenceIncludes => 0

File: operator-example-2.py

import operator
import UserList

def dump(data):
    print type(data), "=>",
    if operator.isCallable(data):
        print "CALLABLE",
    if operator.isMappingType(data):
        print "MAPPING",
    if operator.isNumberType(data):
        print "NUMBER",
    if operator.isSequenceType(data):
        print "SEQUENCE",
    print
        
dump(0)
dump("string")
dump("string"[0])
dump([1, 2, 3])
dump((1, 2, 3))
dump({"a": 1})
dump(len) # function 函数
dump(UserList) # module 模块
dump(UserList.UserList) # class 类
dump(UserList.UserList()) # instance 实例

<type 'int'> => NUMBER
<type 'string'> => SEQUENCE
<type 'string'> => SEQUENCE
<type 'list'> => SEQUENCE
<type 'tuple'> => SEQUENCE
<type 'dictionary'> => MAPPING
<type 'builtin_function_or_method'> => CALLABLE
<type 'module'> =>
<type 'class'> => CALLABLE
<type 'instance'> => MAPPING NUMBER SEQUENCE

这里需要注意 operator 模块使用非常规的方法处理对象实例. 所以使用 isNumberType , isMappingType , 以及 isSequenceType 函数的时候要小心, 这很容易降低代码的扩展性.

同样需要注意的是一个字符串序列成员 (单个字符) 也是序列. 所以当在递归函数使用 isSequenceType 来截断对象树的时候, 别把普通字符串作为参数(或者是任何包含字符串的序列对象).

File: copy-example-1.py

import copy

a = [[1],[2],[3]]
b = copy.copy(a)

print "before", "=>"
print a
print b

# modify original
a[0][0] = 0
a[1] = None

print "after", "=>"
print a
print b

before =>
[[1], [2], [3]]
[[1], [2], [3]]
after =>
[[0], None, [3]]
[[0], [2], [3]]

你也可以使用[:]语句 (完整切片) 来对列表进行浅层复制, 也可以使用 copy 方法复制字典.

File: copy-example-2.py

import copy

a = [[1],[2],[3]]
b = copy.deepcopy(a)

print "before", "=>"
print a
print b

# modify original
a[0][0] = 0
a[1] = None

print "after", "=>"
print a
print b

before =>
[[1], [2], [3]]
[[1], [2], [3]]
after =>
[[0], None, [3]]
[[1], [2], [3]]

sys 模块提供了许多函数和变量来处理 Python 运行时环境的不同部分.

File: sys-argv-example-1.py

import sys

print "script name is", sys.argv[0]

if len(sys.argv) > 1:
    print "there are", len(sys.argv)-1, "arguments:"
    for arg in sys.argv[1:]:
        print arg
else:
    print "there are no arguments!"

script name is sys-argv-example-1.py
there are no arguments!

如果是从标准输入读入脚本 (比如 "python < sys-argv-example-1.py"), 脚本的名称将被设置为空串. 如果把脚本作为字符串传递给python (使用 -c 选项), 脚本名会被设置为 "-c".

File: sys-path-example-1.py

import sys

print "path has", len(sys.path), "members"

# add the sample directory to the path
sys.path.insert(0, "samples")
import sample

# nuke the path
sys.path = []
import random # oops!

path has 7 members
this is the sample module!
Traceback (innermost last):
  File "sys-path-example-1.py", line 11, in ?
    import random # oops!
ImportError: No module named random

File: sys-builtin-module-names-example-1.py

import sys

def dump(module):
    print module, "=>",
    if module in sys.builtin_module_names:
        print "<BUILTIN>"
    else:
        module = _ _import_ _(module)
        print module._ _file_ _

dump("os")
dump("sys")
dump("string")
dump("strop")
dump("zlib")

os => C:/python/lib/os.pyc
sys => <BUILTIN>
string => C:/python/lib/string.pyc
strop => <BUILTIN>
zlib => C:/python/zlib.pyd

modules 字典包含所有加载的模块. import 语句在从磁盘导入内容之前会先检查这个字典.

File: sys-modules-example-1.py

import sys

print sys.modules.keys()

['os.path', 'os', 'exceptions', '_ _main_ _', 'ntpath', 'strop', 'nt',
'sys', '_ _builtin_ _', 'site', 'signal', 'UserDict', 'string', 'stat']

File: sys-getrefcount-example-1.py

import sys

variable = 1234

print sys.getrefcount(0)
print sys.getrefcount(variable)
print sys.getrefcount(None)

50
3
192

注意这个值总是比实际的数量大, 因为该函数本身在确定这个值的时候依赖这个对象.

== 检查主机平台===

File: sys-platform-example-1.py

import sys

#
# emulate "import os.path" (sort of)...

if sys.platform == "win32":
    import ntpath
    pathmodule = ntpath
elif sys.platform == "mac":
    import macpath
    pathmodule = macpath
else:
    # assume it's a posix platform
    import posixpath
    pathmodule = posixpath

print pathmodule

典型的平台有Windows 9X/NT(显示为 win32 ), 以及 Macintosh(显示为 mac ) . 对于 Unix 系统而言, platform 通常来自 "uname -r" 命令的输出, 例如 irix6 ,linux2 , 或者 sunos5 (Solaris).

File: sys-setprofiler-example-1.py

import sys

def test(n):
    j = 0
    for i in range(n):
        j = j + i
    return n

def profiler(frame, event, arg):
    print event, frame.f_code.co_name, frame.f_lineno, "->", arg

# profiler is activated on the next call, return, or exception
# 分析函数将在下次函数调用, 返回, 或异常时激活
sys.setprofile(profiler)

# profile this function call
# 分析这次函数调用
test(1)

# disable profiler
# 禁用分析函数
sys.setprofile(None)

# don't profile this call
# 不会分析这次函数调用
test(2)

call test 3 -> None
return test 7 -> 1

基于该函数, profile 模块提供了一个完整的分析器框架.

File: sys-settrace-example-1.py

import sys

def test(n):
    j = 0
    for i in range(n):
        j = j + i
    return n

def tracer(frame, event, arg):
    print event, frame.f_code.co_name, frame.f_lineno, "->", arg
    return tracer

# tracer is activated on the next call, return, or exception
# 跟踪器将在下次函数调用, 返回, 或异常时激活
sys.settrace(tracer)

# trace this function call
# 跟踪这次函数调用
test(1)

# disable tracing
# 禁用跟踪器
sys.settrace(None)

# don't trace this call
# 不会跟踪这次函数调用
test(2)

call test 3 -> None
line test 3 -> None
line test 4 -> None
line test 5 -> None
line test 5 -> None
line test 6 -> None
line test 5 -> None
line test 7 -> None
return test 7 -> 1

基于该函数提供的跟踪功能, pdb 模块提供了完整的调试( debug )框架.

File: sys-stdout-example-1.py

import sys
import string

class Redirect:

    def _ _init_ _(self, stdout):
        self.stdout = stdout

    def write(self, s):
        self.stdout.write(string.lower(s))

# redirect standard output (including the print statement)
# 重定向标准输出(包括print语句)
old_stdout = sys.stdout
sys.stdout = Redirect(sys.stdout)

print "HEJA SVERIGE",
print "FRISKT HUM/303/226R"

# restore standard output
# 恢复标准输出
sys.stdout = old_stdout

print "M/303/205/303/205/303/205/303/205L!"

heja sverige friskt hum/303/266r
M/303/205/303/205/303/205/303/205L!

要重定向输出只要创建一个对象, 并实现它的 write 方法.

(除非 C 类型的实例外：Python 使用一个叫做 softspace 的整数属性来控制输出中的空白. 如果没有这个属性, Python 将把这个属性附加到这个对象上. 你不需要在使用 Python 对象时担心, 但是在重定向到一个 C 类型时, 你应该确保该类型支持 softspace 属性.)

File: sys-exit-example-1.py

import sys

print "hello"

sys.exit(1)

print "there"

hello

File: sys-exit-example-2.py

import sys

print "hello"

try:
    sys.exit(1)
except SystemExit:
    pass

print "there"

hello
there

File: sys-exitfunc-example-1.py

import sys

def exitfunc():
    print "world"

sys.exitfunc = exitfunc

print "hello"
sys.exit(1)
print "there" # never printed # 不会被 print

hello
world

在 Python 2.0 以后, 你可以使用 atexit 模块来注册多个退出处理函数.

(用于2.0版本及以上) atexit 模块允许你注册一个或多个终止函数(暂且这么叫), 这些函数将在解释器终止前被自动调用.

File: atexit-example-1.py

import atexit

def exit(*args):
    print "exit", args

# register two exit handler
atexit.register(exit)
atexit.register(exit, 1)
atexit.register(exit, "hello", "world")

exit ('hello', 'world')
exit (1,)
exit ()

该模块其实是一个对 sys.exitfunc 钩子( hook )的简单封装.

time 模块提供了一些处理日期和一天内时间的函数. 它是建立在 C 运行时库的简单封装.

给定的日期和时间可以被表示为浮点型(从参考时间, 通常是 1970.1.1 到现在经过的秒数. 即 Unix 格式), 或者一个表示时间的 struct (类元组).

File: time-example-1.py

import time

now = time.time()

print now, "seconds since", time.gmtime(0)[:6]
print
print "or in other words:"
print "- local time:", time.localtime(now)
print "- utc:", time.gmtime(now)

937758359.77 seconds since (1970, 1, 1, 0, 0, 0)

or in other words:
- local time: (1999, 9, 19, 18, 25, 59, 6, 262, 1)
- utc: (1999, 9, 19, 16, 25, 59, 6, 262, 0)

localtime 和 gmtime 返回的类元组包括年, 月, 日, 时, 分, 秒, 星期, 一年的第几天, 日光标志. 其中年是一个四位数(在有千年虫问题的平台上另有规定, 但还是四位数), 星期从星期一(数字 0 代表)开始, 1月1日是一年的第一天.

File: time-example-2.py

import time

now = time.localtime(time.time())

print time.asctime(now)
print time.strftime("%y/%m/%d %H:%M", now)
print time.strftime("%a %b %d", now)
print time.strftime("%c", now)
print time.strftime("%I %p", now)
print time.strftime("%Y-%m-%d %H:%M:%S %Z", now)

# do it by hand...
year, month, day, hour, minute, second, weekday, yearday, daylight = now
print "%04d-%02d-%02d" % (year, month, day)
print "%02d:%02d:%02d" % (hour, minute, second)
print ("MON", "TUE", "WED", "THU", "FRI", "SAT", "SUN")[weekday], yearday

Sun Oct 10 21:39:24 1999
99/10/10 21:39
Sun Oct 10
Sun Oct 10 21:39:24 1999
09 PM
1999-10-10 21:39:24 CEST
1999-10-10
21:39:24
SUN 283

File: time-example-6.py

import time

# make sure we have a strptime function!
# 确认有函数 strptime
try:
    strptime = time.strptime
except AttributeError:
    from strptime import strptime

print strptime("31 Nov 00", "%d %b %y")
print strptime("1 Jan 70 1:30pm", "%d %b %y %I:%M%p")

File: strptime.py

import re
import string

MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug",
          "Sep", "Oct", "Nov", "Dec"]

SPEC = {
    # map formatting code to a regular expression fragment
    "%a": "(?P<weekday>[a-z]+)",
    "%A": "(?P<weekday>[a-z]+)",
    "%b": "(?P<month>[a-z]+)",
    "%B": "(?P<month>[a-z]+)",
    "%C": "(?P<century>/d/d?)",
    "%d": "(?P<day>/d/d?)",
    "%D": "(?P<month>/d/d?)/(?P<day>/d/d?)/(?P<year>/d/d)",
    "%e": "(?P<day>/d/d?)",
    "%h": "(?P<month>[a-z]+)",
    "%H": "(?P<hour>/d/d?)",
    "%I": "(?P<hour12>/d/d?)",
    "%j": "(?P<yearday>/d/d?/d?)",
    "%m": "(?P<month>/d/d?)",
    "%M": "(?P<minute>/d/d?)",
    "%p": "(?P<ampm12>am|pm)",
    "%R": "(?P<hour>/d/d?):(?P<minute>/d/d?)",
    "%S": "(?P<second>/d/d?)",
    "%T": "(?P<hour>/d/d?):(?P<minute>/d/d?):(?P<second>/d/d?)",
    "%U": "(?P<week>/d/d)",
    "%w": "(?P<weekday>/d)",
    "%W": "(?P<weekday>/d/d)",
    "%y": "(?P<year>/d/d)",
    "%Y": "(?P<year>/d/d/d/d)",
    "%%": "%"
}

class TimeParser:
    def _ _init_ _(self, format):
        # convert strptime format string to regular expression
        format = string.join(re.split("(?:/s|%t|%n)+", format))
        pattern = []
        try:
            for spec in re.findall("%/w|%%|.", format):
                if spec[0] == "%":
                    spec = SPEC[spec]
                pattern.append(spec)
        except KeyError:
            raise ValueError, "unknown specificer: %s" % spec
        self.pattern = re.compile("(?i)" + string.join(pattern, ""))
    def match(self, daytime):
        # match time string
        match = self.pattern.match(daytime)
        if not match:
            raise ValueError, "format mismatch"
        get = match.groupdict().get
        tm = [0] * 9
        # extract date elements
        y = get("year")
        if y:
            y = int(y)
            if y < 68:
                y = 2000 + y
            elif y < 100:
                y = 1900 + y
            tm[0] = y
        m = get("month")
        if m:
            if m in MONTHS:
                m = MONTHS.index(m) + 1
            tm[1] = int(m)
        d = get("day")
        if d: tm[2] = int(d)
        # extract time elements
        h = get("hour")
        if h:
            tm[3] = int(h)
        else:
            h = get("hour12")
            if h:
                h = int(h)
                if string.lower(get("ampm12", "")) == "pm":
                    h = h + 12
                tm[3] = h
        m = get("minute")
        if m: tm[4] = int(m)
        s = get("second")
        if s: tm[5] = int(s)
        # ignore weekday/yearday for now
        return tuple(tm)

def strptime(string, format="%a %b %d %H:%M:%S %Y"):
    return TimeParser(format).match(string)

if _ _name_ _ == "_ _main_ _":
    # try it out
    import time
    print strptime("2000-12-20 01:02:03", "%Y-%m-%d %H:%M:%S")
    print strptime(time.ctime(time.time()))

(2000, 12, 20, 1, 2, 3, 0, 0, 0)
(2000, 11, 15, 12, 30, 45, 0, 0, 0)

File: time-example-3.py

import time

t0 = time.time()
tm = time.localtime(t0)

print tm

print t0
print time.mktime(tm)

(1999, 9, 9, 0, 11, 8, 3, 252, 1)
936828668.16
936828668.0

File: time-example-4.py

import time

def _d(y, m, d, days=(0,31,59,90,120,151,181,212,243,273,304,334,365)):
    # map a date to the number of days from a reference point
    return (((y - 1901)*1461)/4 + days[m-1] + d +
        ((m > 2 and not y % 4 and (y % 100 or not y % 400)) and 1))

def timegm(tm, epoch=_d(1970,1,1)):
    year, month, day, h, m, s = tm[:6]
    assert year >= 1970
    assert 1 <= month <= 12
    return (_d(year, month, day) - epoch)*86400 + h*3600 + m*60 + s

t0 = time.time()
tm = time.gmtime(t0)

print tm

print t0
print timegm(tm)

(1999, 9, 8, 22, 12, 12, 2, 251, 0)
936828732.48
936828732

从 1.6 版本开始, calendar 模块提供了一个类似的函数 calendar.timegm .

File: time-example-5.py

import time

def procedure():
    time.sleep(2.5)

# measure process time
t0 = time.clock()
procedure()
print time.clock() - t0, "seconds process time"

# measure wall time
t0 = time.time()
procedure()
print time.time() - t0, "seconds wall time"

0.0 seconds process time
2.50903499126 seconds wall time

并不是所有的系统都能测量真实的进程时间. 一些系统中(包括 Windows ), clock 函数通常测量从程序启动到测量时的 wall time.

进程时间的精度受限制. 在一些系统中, 它超过 30 分钟后进程会被清理. (原文: On many systems, it wraps around after just over 30 minutes.)

另参见 timing 模块( Windows 下的朋友不用忙活了,没有地~), 它可以测量两个事件之间的 wall time.

File: types-example-1.py

import types

def check(object):
    print object,

    if type(object) is types.IntType:
        print "INTEGER",
    if type(object) is types.FloatType:
        print "FLOAT",
    if type(object) is types.StringType:
        print "STRING",
    if type(object) is types.ClassType:
        print "CLASS",
    if type(object) is types.InstanceType:
        print "INSTANCE",
    print

check(0)
check(0.0)
check("0")

class A:
    pass

class B:
    pass

check(A)
check(B)

a = A()
b = B()

check(a)
check(b)

0 INTEGER
0.0 FLOAT
0 STRING
A CLASS
B CLASS
<A instance at 796960> INSTANCE
<B instance at 796990> INSTANCE

注意所有的类都具有相同的类型, 所有的实例也是一样. 要测试一个类或者实例所属的类, 可以使用内建的 issubclass 和 isinstance 函数.

types 模块在第一次引入的时候会破坏当前的异常状态. 也就是说, 不要在异常处理语句块中导入该模块 (或其他会导入它的模块) .

(可选, 2.0 及以后版本) gc 模块提供了到内建循环垃圾收集器的接口.

Python 使用引用记数来跟踪什么时候销毁一个对象; 一个对象的最后一个引用一旦消失, 这个对象就会被销毁.

File: gc-example-1.py

import gc

# create a simple object that links to itself
class Node:

    def _ _init_ _(self, name):
        self.name = name
        self.parent = None
        self.children = []

    def addchild(self, node):
        node.parent = self
        self.children.append(node)

    def _ _repr_ _(self):
        return "<Node %s at %x>" % (repr(self.name), id(self))

# set up a self-referencing structure
root = Node("monty")

root.addchild(Node("eric"))
root.addchild(Node("john"))
root.addchild(Node("michael"))

# remove our only reference
del root

print gc.collect(), "unreachable objects"
print gc.collect(), "unreachable objects"

12 unreachable objects
0 unreachable objects

如果你确定你的程序不会创建自引用的数据结构, 你可以使用 gc.disable 函数禁用垃圾收集, 调用这个函数以后, Python 的工作方式将与 1.5.2 或更早的版本相同.

"Now, imagine that your friend kept complaining that she didn't want to visit you since she found it too hard to climb up the drain pipe, and you kept telling her to use the friggin' stairs like everyone else..."

- eff-bot, June 1998

本章叙述了许多在 Python 程序中广泛使用的模块. 当然, 在大型的 Python 程序中不使用这些模块也是可以的, 但如果使用会节省你不少时间.

fileinput 模块可以让你更简单地向不同的文件写入内容. 该模块提供了一个简单的封装类, 一个简单的 for-in 语句就可以循环得到一个或多个文本文件的内容.

StringIO 模块 (以及 cStringIO 模块, 作为一个的变种) 实现了一个工作在内存的文件对象. 你可以在很多地方用 StringIO 对象替换普通的文件对象.

UserDict , UserList , 以及 UserString 是对应内建类型的顶层简单封装. 和内建类型不同的是, 这些封装是可以被继承的. 这在你需要一个和内建类型行为相似但由额外新方法的类的时候很有用.

random 模块提供了一些不同的随机数字生成器. whrandom 模块与此相似, 但允许你创建多个生成器对象.

[!Feather 注: whrandom 在版本 2.1 时声明不支持. 请使用 random 替代.]

md5 和 sha 模块用于计算密写的信息标记( cryptographically strong message signatures , 所谓的 "message digests", 信息摘要).

crypt 模块实现了 DES 样式的单向加密. 该模块只在 Unix 系统下可用.

rotor 模块提供了简单的双向加密. 版本 2.4 以后的朋友可以不用忙活了.

[!Feather 注: 它在版本 2.3 时申明不支持, 因为它的加密运算不安全.]

File: fileinput-example-1.py

import fileinput
import sys

for line in fileinput.input("samples/sample.txt"):
    sys.stdout.write("-> ")
    sys.stdout.write(line)

-> We will perhaps eventually be writing only small
-> modules which are identified by name as they are
-> used to build larger ones, so that devices like
-> indentation, rather than delimiters, might become
-> feasible for expressing local structure in the
-> source language.
->      -- Donald E. Knuth, December 1974

File: fileinput-example-2.py

import fileinput
import glob
import string, sys

for line in fileinput.input(glob.glob("samples/*.txt")):
    if fileinput.isfirstline(): # first in a file?
        sys.stderr.write("-- reading %s --/n" % fileinput.filename())
    sys.stdout.write(str(fileinput.lineno()) + " " + string.upper(line))

-- reading samples/sample.txt --
1 WE WILL PERHAPS EVENTUALLY BE WRITING ONLY SMALL
2 MODULES WHICH ARE IDENTIFIED BY NAME AS THEY ARE
3 USED TO BUILD LARGER ONES, SO THAT DEVICES LIKE
4 INDENTATION, RATHER THAN DELIMITERS, MIGHT BECOME
5 FEASIBLE FOR EXPRESSING LOCAL STRUCTURE IN THE
6 SOURCE LANGUAGE.
7    -- DONALD E. KNUTH, DECEMBER 1974

File: fileinput-example-3.py

import fileinput, sys

for line in fileinput.input(inplace=1):
    # convert Windows/DOS text files to Unix files
    if line[-2:] == "/r/n":
        line = line[:-2] + "/n"
    sys.stdout.write(line)

File: shutil-example-1.py

import shutil
import os

for file in os.listdir("."):
    if os.path.splitext(file)[1] == ".py":
        print file
        shutil.copy(file, os.path.join("backup", file))

aifc-example-1.py
anydbm-example-1.py
array-example-1.py
...

File: shutil-example-2.py

import shutil
import os

SOURCE = "samples"
BACKUP = "samples-bak"

# create a backup directory
shutil.copytree(SOURCE, BACKUP)

print os.listdir(BACKUP)

# remove it
shutil.rmtree(BACKUP)

print os.listdir(BACKUP)

['sample.wav', 'sample.jpg', 'sample.au', 'sample.msg', 'sample.tgz',
...
Traceback (most recent call last):
 File "shutil-example-2.py", line 17, in ?
   print os.listdir(BACKUP)
os.error: No such file or directory

File: tempfile-example-1.py

import tempfile
import os

tempfile = tempfile.mktemp()

print "tempfile", "=>", tempfile

file = open(tempfile, "w+b")
file.write("*" * 1000)
file.seek(0)
print len(file.read()), "bytes"
file.close()

try:
    # must remove file when done
    os.remove(tempfile)
except OSError:
    pass

tempfile => C:/TEMP/~160-1
1000 bytes

File: tempfile-example-2.py

import tempfile

file = tempfile.TemporaryFile()

for i in range(100):
    file.write("*" * 100)

file.close() # removes the file!

File: stringio-example-1.py

import StringIO

MESSAGE = "That man is depriving a village somewhere of a computer scientist."

file = StringIO.StringIO(MESSAGE)

print file.read()

That man is depriving a village somewhere of a computer scientist.

File: stringio-example-2.py

import StringIO

file = StringIO.StringIO()
file.write("This man is no ordinary man. ")
file.write("This is Mr. F. G. Superman.")

print file.getvalue()

This man is no ordinary man. This is Mr. F. G. Superman.

File: stringio-example-3.py

import StringIO
import string, sys

stdout = sys.stdout

sys.stdout = file = StringIO.StringIO()

print """
According to Gbaya folktales, trickery and guile
are the best ways to defeat the python, king of
snakes, which was hatched from a dragon at the
world's start. -- National Geographic, May 1997
"""

sys.stdout = stdout

print string.upper(file.getvalue())

ACCORDING TO GBAYA FOLKTALES, TRICKERY AND GUILE
ARE THE BEST WAYS TO DEFEAT THE PYTHON, KING OF
SNAKES, WHICH WAS HATCHED FROM A DRAGON AT THE
WORLD'S START. -- NATIONAL GEOGRAPHIC, MAY 1997

File: cstringio-example-1.py

import cStringIO

MESSAGE = "That man is depriving a village somewhere of a computer scientist."

file = cStringIO.StringIO(MESSAGE)

print file.read()

That man is depriving a village somewhere of a computer scientist.

File: cstringio-example-2.py

try:
    import cStringIO
    StringIO = cStringIO
except ImportError:
    import StringIO

print StringIO

<module  'StringIO' (built-in)>

File: mmap-example-1.py

import mmap
import os

filename = "samples/sample.txt"

file = open(filename, "r+")
size = os.path.getsize(filename)

data = mmap.mmap(file.fileno(), size)

# basics
print data
print len(data), size

# use slicing to read from the file
# 使用切片操作读取文件
print repr(data[:10]), repr(data[:10])

# or use the standard file interface
# 或使用标准的文件接口
print repr(data.read(10)), repr(data.read(10))

<mmap object at 008A2A10>
302 302
'We will pe' 'We will pe'
'We will pe' 'rhaps even'

在 Windows 下, 这个文件必须以既可读又可写的模式打开( `r+` , `w+` , 或 `a+` ), 否则 mmap 调用会失败.

[!Feather 注: 经本人测试, a+ 模式是完全可以的, 原文只有 r+ 和 w+]

File: mmap-example-2.py

import mmap
import os, string, re

def mapfile(filename):
    file = open(filename, "r+")
    size = os.path.getsize(filename)
    return mmap.mmap(file.fileno(), size)

data = mapfile("samples/sample.txt")

# search
index = data.find("small")
print index, repr(data[index-5:index+15])

# regular expressions work too!
m = re.search("small", data)
print m.start(), m.group()

43 'only small/015/012modules '
43 small

UserDict 模块包含了一个可继承的字典类 (事实上是对内建字典类型的 Python 封装).

File: userdict-example-1.py

import UserDict

class FancyDict(UserDict.UserDict):

    def _ _init_ _(self, data = {}, **kw):
        UserDict.UserDict._ _init_ _(self)
        self.update(data)
        self.update(kw)

    def _ _add_ _(self, other):
        dict = FancyDict(self.data)
        dict.update(b)
        return dict

a = FancyDict(a = 1)
b = FancyDict(b = 2)

print a + b

{'b': 2, 'a': 1}

UserList 模块包含了一个可继承的列表类 (事实上是对内建列表类型的 Python 封装).

File: userlist-example-1.py

import UserList

class AutoList(UserList.UserList):

    def _ _setitem_ _(self, i, item):
        if i == len(self.data):
            self.data.append(item)
        else:
            self.data[i] = item

list = AutoList()

for i in range(10):
    list[i] = i

print list

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

(2.0 新增) UserString 模块包含两个类, UserString 和 MutableString . 前者是对标准字符串类型的封装, 后者是一个变种, 允许你修改特定位置的字符(联想下列表就知道了).

File: userstring-example-1.py

import UserString

class MyString(UserString.MutableString):

    def append(self, s):
        self.data = self.data + s

    def insert(self, index, s):
        self.data = self.data[index:] + s + self.data[index:]

    def remove(self, s):
        self.data = self.data.replace(s, "")

file = open("samples/book.txt")
text = file.read()
file.close()

book = MyString(text)

for bird in ["gannet", "robin", "nuthatch"]:
    book.remove(bird)

print book

...
C: The one without the !
P: The one without the -!!! They've ALL got the !! It's a
Standard British Bird, the , it's in all the books!!!
...

File: traceback-example-1.py

# note! importing the traceback module messes up the
# exception state, so you better do that here and not
# in the exception handler
# 注意! 导入 traceback 会清理掉异常状态, 所以
# 最好别在异常处理代码中导入该模块
import traceback

try:
    raise SyntaxError, "example"
except:
    traceback.print_exc()

Traceback (innermost last):
  File "traceback-example-1.py", line 7, in ?
SyntaxError: example

File: traceback-example-2.py

import traceback
import StringIO

try:
    raise IOError, "an i/o error occurred"
except:
    fp = StringIO.StringIO()
    traceback.print_exc(file=fp)
    message = fp.getvalue()

    print "failure! the error was:", repr(message)

failure! the error was: 'Traceback (innermost last):/012  File
"traceback-example-2.py", line 5, in ?/012IOError: an i/o error
occurred/012'

File: traceback-example-3.py

import traceback
import sys

def function():
    raise IOError, "an i/o error occurred"

try:
    function()
except:
    info = sys.exc_info()
    for file, lineno, function, text in traceback.extract_tb(info[2]):
        print file, "line", lineno, "in", function
        print "=>", repr(text)
    print "** %s: %s" % info[:2]

traceback-example-3.py line 8 in ?
=> 'function()'
traceback-example-3.py line 5 in function
=> 'raise IOError, "an i/o error occurred"'
** exceptions.IOError: an i/o error occurred

File: errno-example-1.py

import errno

try:
    fp = open("no.such.file")
except IOError, (error, message):
    if error == errno.ENOENT:
        print "no such file"
    elif error == errno.EPERM:
        print "permission denied"
    else:
        print message

no such file

File: errno-example-2.py

import errno

try:
    fp = open("no.such.file")
except IOError, (error, message):
    print error, repr(message)
    print errno.errorcode[error]

# 2 'No such file or directory'
# ENOENT

File: getopt-example-1.py

import getopt
import sys

# simulate command-line invocation
# 模仿命令行参数
sys.argv = ["myscript.py", "-l", "-d", "directory", "filename"]

# process options
# 处理选项
opts, args = getopt.getopt(sys.argv[1:], "ld:")

long = 0
directory = None

for o, v in opts:
    if o == "-l":
        long = 1
    elif o == "-d":
        directory = v

print "long", "=", long
print "directory", "=", directory
print "arguments", "=", args

long = 1
directory = directory
arguments = ['filename']

File: getopt-example-2.py

import getopt
import sys

# simulate command-line invocation
# 模仿命令行参数
sys.argv = ["myscript.py", "--echo", "--printer", "lp01", "message"]

opts, args = getopt.getopt(sys.argv[1:], "ep:", ["echo", "printer="])

# process options
# 处理选项
echo = 0
printer = None

for o, v in opts:
    if o in ("-e", "--echo"):
        echo = 1
    elif o in ("-p", "--printer"):
        printer = v

print "echo", "=", echo
print "printer", "=", printer
print "arguments", "=", args

echo = 1
printer = lp01
arguments = ['message']

[!Feather 注: 我不知道大家明白没, 可以自己试下:
myscript.py -e -p lp01 message
myscript.py --echo --printer=lp01 message
]

File: getpass-example-1.py

import getpass

usr = getpass.getuser()

pwd = getpass.getpass("enter password for user %s: " % usr)

print usr, pwd

enter password for user mulder:
mulder trustno1

glob 根据给定模式生成满足该模式的文件名列表, 和 Unix shell 相同.

这里的模式和正则表达式类似, 但更简单. 星号(*) 匹配零个或更多个字符, 问号(?) 匹配单个字符. 你也可以使用方括号来指定字符范围, 例如 [0-9] 代表一个数字. 其他所有字符都代表它们本身.

File: glob-example-1.py

import glob

for file in glob.glob("samples/*.jpg"):
    print file

samples/sample.jpg

注意这里的 glob 返回完整路径名, 这点和 os.listdir 函数不同. glob 事实上使用了 fnmatch 模块来完成模式匹配.

File: fnmatch-example-1.py

import fnmatch
import os

for file in os.listdir("samples"):
    if fnmatch.fnmatch(file, "*.jpg"):
        print file

sample.jpg

File: fnmatch-example-2.py

import fnmatch
import os, re

pattern = fnmatch.translate("*.jpg")

for file in os.listdir("samples"):
    if re.match(pattern, file):
        print file

print "(pattern was %s)" % pattern

sample.jpg
(pattern was .*/.jpg$)

glob 和 find 模块在内部使用 fnmatch 模块来实现.

"Anyone who considers arithmetical methods of producing random digits is, of course, in a state of sin."

- John von Neumann, 1951

random 模块包含许多随机数生成器.

File: random-example-1.py

import random

for i in range(5):

    # random float: 0.0 <= number < 1.0
    print random.random(),

    # random float: 10 <= number < 20
    print random.uniform(10, 20),

    # random integer: 100 <= number <= 1000
    print random.randint(100, 1000),

    # random integer: even numbers in 100 <= number < 1000
    print random.randrange(100, 1000, 2)

0.946842713956 19.5910069381 709 172
0.573613195398 16.2758417025 407 120
0.363241598013 16.8079747714 916 580
0.602115173978 18.386796935 531 774
0.526767588533 18.0783794596 223 344

注意这里的 randint 函数可以返回上界, 而其他函数总是返回小于上界的值. 所有函数都有可能返回下界值.

File: random-example-2.py

import random

# random choice from a list
for i in range(5):
    print random.choice([1, 2, 3, 5, 9])

2
3
1
9
1

File: random-example-4.py

import random

try:
    # available in 2.0 and later
    shuffle = random.shuffle
except AttributeError:
    def shuffle(x):
        for i in xrange(len(x)-1, 0, -1):
            # pick an element in x[:i+1] with which to exchange x[i]
            j = int(random.random() * (i+1))
            x[i], x[j] = x[j], x[i]

cards = range(52)

shuffle(cards)

myhand = cards[:5]

print myhand

[4, 8, 40, 12, 30]

File: random-example-3.py

import random

histogram = [0] * 20

# calculate histogram for gaussian
# noise, using average=5, stddev=1
for i in range(1000):
    i = int(random.gauss(5, 1) * 2)
    histogram[i] = histogram[i] + 1

# print the histogram
m = max(histogram)
for v in histogram:
    print "*" * (v * 50 / m)


****
**********
*************************
***********************************
************************************************
**************************************************
*************************************
***************************
*************
***
*

你可以在 Python Library Reference 找到更多关于非恒定分布随机生成器函数的信息.

标准库中提供的随机数生成器都是伪随机数生成器. 不过这对于很多目的来说已经足够了, 比如模拟, 数值分析, 以及游戏. 可以确定的是它不适合密码学用途.

这个模块早在 2.1 就被声明不赞成, 早废了. 请使用 random 代替. 
- Feather

File: whrandom-example-1.py

import whrandom

# same as random
print whrandom.random()
print whrandom.choice([1, 2, 3, 5, 9])
print whrandom.uniform(10, 20)
print whrandom.randint(100, 1000)

0.113412062346
1
16.8778954689
799

File: whrandom-example-2.py

import whrandom

# initialize all generators with the same seed
rand1 = whrandom.whrandom(4,7,11)
rand2 = whrandom.whrandom(4,7,11)
rand3 = whrandom.whrandom(4,7,11)

for i in range(5):
    print rand1.random(), rand2.random(), rand3.random()

0.123993532536 0.123993532536 0.123993532536
0.180951499518 0.180951499518 0.180951499518
0.291924111809 0.291924111809 0.291924111809
0.952048889363 0.952048889363 0.952048889363
0.969794283643 0.969794283643 0.969794283643

md5 (Message-Digest Algorithm 5)模块用于计算信息密文(信息摘要).

File: md5-example-1.py

import md5

hash = md5.new()
hash.update("spam, spam, and eggs")

print repr(hash.digest())

 'L/005J/243/266/355/243u`/305r/203/267/020F/303'

File: md5-example-2.py

import md5
import string
import base64

hash = md5.new()
hash.update("spam, spam, and eggs")

value = hash.digest()
print hash.hexdigest()
# before 2.0, the above can be written as
# 在 2.0 前, 以上应该写做:
# print string.join(map(lambda v: "%02x" % ord(v), value), "")

print base64.encodestring(value)

4c054aa3b6eda37560c57283b71046c3
TAVKo7bto3VgxXKDtxBGww==

File: md5-example-3.py

import md5
import string, random

def getchallenge():
    # generate a 16-byte long random string.  (note that the built-
    # in pseudo-random generator uses a 24-bit seed, so this is not
    # as good as it may seem...)
    # 生成一个 16 字节长的随机字符串. 注意内建的伪随机生成器
    # 使用的是 24 位的种子(seed), 所以这里这样用并不好..
    challenge = map(lambda i: chr(random.randint(0, 255)), range(16))
    return string.join(challenge, "")

def getresponse(password, challenge):
    # calculate combined digest for password and challenge
    # 计算密码和质询(challenge)的联合密文
    m = md5.new()
    m.update(password)
    m.update(challenge)
    return m.digest()

#
# server/client communication
# 服务器/客户端通讯

# 1. client connects.  server issues challenge.
# 1. 客户端连接, 服务器发布质询(challenge)

print "client:", "connect"

challenge = getchallenge()

print "server:", repr(challenge)

# 2. client combines password and challenge, and calculates
# the response.
# 2. 客户端计算密码和质询(challenge)的组合后的密文

client_response = getresponse("trustno1", challenge)

print "client:", repr(client_response)

# 3. server does the same, and compares the result with the
# client response.  the result is a safe login in which the
# password is never sent across the communication channel.
# 3. 服务器做同样的事, 然后比较结果与客户端的返回, 
# 判断是否允许用户登陆. 这样做密码没有在通讯中明文传输.

server_response = getresponse("trustno1", challenge)

if server_response == client_response:
    print "server:", "login ok"

client: connect
server: '/334/352/227Z#/272/273/212KG/330/265/032>/311o'
client: "l'/305/240-x/245/237/035/225A/254/233/337/225/001"
server: login ok

File: md5-example-4.py

import md5
import array

class HMAC_MD5:
    # keyed md5 message authentication

    def _ _init_ _(self, key):
        if len(key) > 64:
            key = md5.new(key).digest()
        ipad = array.array("B", [0x36] * 64)
        opad = array.array("B", [0x5C] * 64)
        for i in range(len(key)):
            ipad[i] = ipad[i] ^ ord(key[i])
            opad[i] = opad[i] ^ ord(key[i])
        self.ipad = md5.md5(ipad.tostring())
        self.opad = md5.md5(opad.tostring())

    def digest(self, data):
        ipad = self.ipad.copy()
        opad = self.opad.copy()
        ipad.update(data)
        opad.update(ipad.digest())
        return opad.digest()

#
# simulate server end
# 模拟服务器端

key = "this should be a well-kept secret"
message = open("samples/sample.txt").read()

signature = HMAC_MD5(key).digest(message)

# (send message and signature across a public network)
# (经过由网络发送信息和签名)

#
# simulate client end
#模拟客户端

key = "this should be a well-kept secret"

client_signature = HMAC_MD5(key).digest(message)

if client_signature == signature:
    print "this is the original message:"
    print
    print message
else:
    print "someone has modified the message!!!"

File: sha-example-1.py

import sha

hash = sha.new()
hash.update("spam, spam, and eggs")

print repr(hash.digest())
print hash.hexdigest()

'/321/333/003/026I/331/272-j/303/247/240/345/343Tvq/364/346/311'
d1db031649d9ba2d6ac3a7a0e5e3547671f4e6c9

关于 sha 密文的使用, 请参阅 md5 中的例子.

(可选, 只用于 Unix) crypt 模块实现了单向的 DES 加密, Unix 系统使用这个加密算法来储存密码, 这个模块真正也就只在检查这样的密码时有用.

File: crypt-example-1.py

import crypt

import random, string

def getsalt(chars = string.letters + string.digits):
    # generate a random 2-character 'salt'
    # 生成随机的 2 字符 'salt'
    return random.choice(chars) + random.choice(chars)

print crypt.crypt("bananas", getsalt())

'py8UGrijma1j6'

File: crypt-example-2.py

import pwd, crypt

def login(user, password):
    "Check if user would be able to log in using password"
    try:
        pw1 = pwd.getpwnam(user)[1]
        pw2 = crypt.crypt(password, pw1[:2])
        return pw1 == pw2
    except KeyError:
        return 0 # no such user

user = raw_input("username:")
password = raw_input("password:")

if login(user, password):
    print "welcome", user
else:
    print "login failed"

关于其他实现验证的方法请参阅 md5 模块一节.

这个模块在 2.3 时被声明不赞成, 2.4 时废了. 因为它的加密算法不安全. 
- Feather

File: rotor-example-1.py

import rotor

SECRET_KEY = "spam"
MESSAGE = "the holy grail"

r = rotor.newrotor(SECRET_KEY)

encoded_message = r.encrypt(MESSAGE)
decoded_message = r.decrypt(encoded_message)

print "original:", repr(MESSAGE)
print "encoded message:", repr(encoded_message)
print "decoded message:", repr(decoded_message)

original: 'the holy grail'
encoded message: '/227/271/244/015/305sw/3340/337/252/237/340U'
decoded message: 'the holy grail'

(可选) zlib 模块为 "zlib" 压缩提供支持. (这种压缩方法是 "deflate".)

File: zlib-example-1.py

import zlib

MESSAGE = "life of brian"

compressed_message = zlib.compress(MESSAGE)
decompressed_message = zlib.decompress(compressed_message)

print "original:", repr(MESSAGE)
print "compressed message:", repr(compressed_message)
print "decompressed message:", repr(decompressed_message)

original: 'life of brian'
compressed message: 'x/234/313/311LKU/310OSH*/312L/314/003/000!/010/004/302'
decompressed message: 'life of brian'

File: zlib-example-2.py

import zlib
import glob

for file in glob.glob("samples/*"):

    indata = open(file, "rb").read()
    outdata = zlib.compress(indata, zlib.Z_BEST_COMPRESSION)

    print file, len(indata), "=>", len(outdata),
    print "%d%%" % (len(outdata) * 100 / len(indata))

samples/sample.au 1676 => 1109 66%
samples/sample.gz 42 => 51 121%
samples/sample.htm 186 => 135 72%
samples/sample.ini 246 => 190 77%
samples/sample.jpg 4762 => 4632 97%
samples/sample.msg 450 => 275 61%
samples/sample.sgm 430 => 321 74%
samples/sample.tar 10240 => 125 1%
samples/sample.tgz 155 => 159 102%
samples/sample.txt 302 => 220 72%
samples/sample.wav 13260 => 10992 82%

File: zlib-example-3.py

import zlib

encoder = zlib.compressobj()

data = encoder.compress("life")
data = data + encoder.compress(" of ")
data = data + encoder.compress("brian")
data = data + encoder.flush()

print repr(data)
print repr(zlib.decompress(data))

'x/234/313/311LKU/310OSH*/312L/314/003/000!/010/004/302'
'life of brian'

File: zlib-example-4.py

import zlib
import string, StringIO

class ZipInputStream:

    def _ _init_ _(self, file):
        self.file = file
        self._ _rewind()

    def _ _rewind(self):
        self.zip = zlib.decompressobj()
        self.pos = 0 # position in zipped stream
        self.offset = 0 # position in unzipped stream
        self.data = ""

    def _ _fill(self, bytes):
        if self.zip:
            # read until we have enough bytes in the buffer
            while not bytes or len(self.data) < bytes:
                self.file.seek(self.pos)
                data = self.file.read(16384)
                if not data:
                    self.data = self.data + self.zip.flush()
                    self.zip = None # no more data
                    break
                self.pos = self.pos + len(data)
                self.data = self.data + self.zip.decompress(data)

    def seek(self, offset, whence=0):
        if whence == 0:
            position = offset
        elif whence == 1:
            position = self.offset + offset
        else:
            raise IOError, "Illegal argument"
        if position < self.offset:
            raise IOError, "Cannot seek backwards"

        # skip forward, in 16k blocks
        while position > self.offset:
            if not self.read(min(position - self.offset, 16384)):
                break

    def tell(self):
        return self.offset

    def read(self, bytes = 0):
        self._ _fill(bytes)
        if bytes:
            data = self.data[:bytes]
            self.data = self.data[bytes:]
        else:
            data = self.data
            self.data = ""
        self.offset = self.offset + len(data)
        return data

    def readline(self):
        # make sure we have an entire line
        while self.zip and "/n" not in self.data:
            self._ _fill(len(self.data) + 512)
        i = string.find(self.data, "/n") + 1
        if i <= 0:
            return self.read()
        return self.read(i)

    def readlines(self):
        lines = []
        while 1:
            s = self.readline()
            if not s:
                break
            lines.append(s)
        return lines

#
# try it out

data = open("samples/sample.txt").read()
data = zlib.compress(data)

file = ZipInputStream(StringIO.StringIO(data))
for line in file.readlines():
    print line[:-1]

We will perhaps eventually be writing only small
modules which are identified by name as they are
used to build larger ones, so that devices like
indentation, rather than delimiters, might become
feasible for expressing local structure in the
source language.
    -- Donald E. Knuth, December 1974

code 模块提供了一些用于模拟标准交互解释器行为的函数.

compile_command 与内建 compile 函数行为相似, 但它会通过测试来保证你传递的是一个完成的 Python 语句.

File: code-example-1.py

import code
import string

# 
SCRIPT = [
    "a = (",
    "  1,",
    "  2,",
    "  3 ",
    ")",
    "print a"
]

script = ""

for line in SCRIPT:
    script = script + line + "/n"
    co = code.compile_command(script, "<stdin>", "exec")
    if co:
        # got a complete statement.  execute it!
        print "-"*40
        print script,
        print "-"*40
        exec co
        script = ""

----------------------------------------
a = (
  1,
  2,
  3 
)
----------------------------------------
----------------------------------------
print a
----------------------------------------
(1, 2, 3)

InteractiveConsole 类实现了一个交互控制台, 类似你启动的 Python 解释器交互模式.

File: code-example-2.py

import code

console = code.InteractiveConsole()
console.interact()

Python 1.5.2
Copyright 1991-1995 Stichting Mathematisch Centrum, Amsterdam
(InteractiveConsole)
>>> a = (
...     1,
...     2,
...     3
... )
>>> print a
(1, 2, 3)

File: code-example-3.py

def keyboard(banner=None):
    import code, sys

    # use exception trick to pick up the current frame
    try:
        raise None
    except:
        frame = sys.exc_info()[2].tb_frame.f_back

    # evaluate commands in current namespace
    namespace = frame.f_globals.copy()
    namespace.update(frame.f_locals)

    code.interact(banner=banner, local=namespace)

def func():
    print "START"
    a = 10
    keyboard()
    print "END"

func()

START
Python 1.5.2
Copyright 1991-1995 Stichting Mathematisch Centrum, Amsterdam
(InteractiveConsole)
>>> print a
10
>>> print keyboard
<function keyboard at 9032c8>
^Z
END

"Well, since you last asked us to stop, this thread has moved from discussing languages suitable for professional programmers via accidental users to computer-phobic users. A few more iterations can make this thread really interesting..."

- eff-bot, June 1996

本章将介绍标准 Python 解释器中所提供的线程支持模块. 注意线程支持模块是可选的, 有可能在一些 Python 解释器中不可用.

本章还涵盖了一些 Unix 和 Windows 下用于执行外部进程的模块.

执行 Python 程序的时候, 是按照从主模块顶端向下执行的. 循环用于重复执行部分代码, 函数和方法会将控制临时移交到程序的另一部分.

通过线程, 你的程序可以在同时处理多个任务. 每个线程都有它自己的控制流. 所以你可以在一个线程里从文件读取数据, 另个向屏幕输出内容.

为了保证两个线程可以同时访问相同的内部数据, Python 使用了 global interpreter lock (全局解释器锁). 在同一时间只可能有一个线程执行 Python 代码; Python 实际上是自动地在一段很短的时间后切换到下个线程执行, 或者等待 一个线程执行一项需要时间的操作(例如等待通过 socket 传输的数据, 或是从文件中读取数据).

全局锁事实上并不能避免你程序中的问题. 多个线程尝试访问相同的数据会导致异常 状态. 例如以下的代码:

def getitem(key):
    item = cache.get(key)
    if item is None:
        # not in cache; create a new one
        item = create_new_item(key)
	cache[key] = item
    return item

如果不同的线程先后使用相同的 key 调用这里的 getitem 方法, 那么它们很可能会导致相同的参数调用两次 create_new_item . 大多时候这样做没有问题, 但在某些时候会导致严重错误.

不过你可以使用 lock objects 来同步线程. 一个线程只能拥有一个 lock object , 这样就可以确保某个时刻 只有一个线程执行 getitem 函数.

在大多现代操作系统中, 每个程序在它自身的进程( process )内执行. 我们通过在 shell 中键入命令或直接在菜单中选择来执行一个程序/进程. Python 允许你在一个脚本内执行一个新的程序.

File: threading-example-1.py

import threading
import time, random

class Counter:
    def _ _init_ _(self):
        self.lock = threading.Lock()
        self.value = 0

    def increment(self):
        self.lock.acquire() # critical section
        self.value = value = self.value + 1
        self.lock.release()
        return value

counter = Counter()

class Worker(threading.Thread):

    def run(self):
        for i in range(10):
            # pretend we're doing something that takes 10�00 ms
            value = counter.increment() # increment global counter
            time.sleep(random.randint(10, 100) / 1000.0)
            print self.getName(), "-- task", i, "finished", value

#
# try it

for i in range(10):
    Worker().start() # start a worker

Thread-1 -- task 0 finished 1
Thread-3 -- task 0 finished 3
Thread-7 -- task 0 finished 8
Thread-1 -- task 1 finished 7
Thread-4 -- task 0 Thread-5 -- task 0 finished 4
finished 5
Thread-8 -- task 0 Thread-6 -- task 0 finished 9
finished 6
...
Thread-6 -- task 9 finished 98
Thread-4 -- task 9 finished 99
Thread-9 -- task 9 finished 100

File: queue-example-1.py

import threading
import Queue
import time, random

WORKERS = 2

class Worker(threading.Thread):

    def _ _init_ _(self, queue):
        self._ _queue = queue
        threading.Thread._ _init_ _(self)

    def run(self):
        while 1:
            item = self._ _queue.get()
            if item is None:
                break # reached end of queue

            # pretend we're doing something that takes 10�00 ms
            time.sleep(random.randint(10, 100) / 1000.0)

            print "task", item, "finished"

#
# try it

queue = Queue.Queue(0)

for i in range(WORKERS):
    Worker(queue).start() # start a worker

for i in range(10):
    queue.put(i)

for i in range(WORKERS):
    queue.put(None) # add end-of-queue markers

task 1 finished
task 0 finished
task 3 finished
task 2 finished
task 4 finished
task 5 finished
task 7 finished
task 6 finished
task 9 finished
task 8 finished

File: queue-example-2.py

import threading
import Queue

import time, random

WORKERS = 2

class Worker(threading.Thread):

    def _ _init_ _(self, queue):
        self._ _queue = queue
        threading.Thread._ _init_ _(self)

    def run(self):
        while 1:
            item = self._ _queue.get()
            if item is None:
                break # reached end of queue

            # pretend we're doing something that takes 10�00 ms
            time.sleep(random.randint(10, 100) / 1000.0)

            print "task", item, "finished"

#
# run with limited queue

queue = Queue.Queue(3)

for i in range(WORKERS):
    Worker(queue).start() # start a worker

for item in range(10):
    print "push", item
    queue.put(item)

for i in range(WORKERS):
    queue.put(None) # add end-of-queue markers

push 0
push 1
push 2
push 3
push 4
push 5
task 0 finished
push 6
task 1 finished
push 7
task 2 finished
push 8
task 3 finished
push 9
task 4 finished
task 6 finished
task 5 finished
task 7 finished
task 9 finished
task 8 finished

File: queue-example-3.py

import Queue
import bisect

Empty = Queue.Empty

class PriorityQueue(Queue.Queue):
    "Thread-safe priority queue"

    def _put(self, item):
        # insert in order
        bisect.insort(self.queue, item)

#
# try it

queue = PriorityQueue(0)

# add items out of order
queue.put((20, "second"))
queue.put((10, "first"))
queue.put((30, "third"))

# print queue contents
try:
    while 1:
        print queue.get_nowait()
except Empty:
    pass

third
second
first

File: queue-example-4.py

import Queue

Empty = Queue.Empty

class Stack(Queue.Queue):
    "Thread-safe stack"
    
    def _put(self, item):
        # insert at the beginning of queue, not at the end
        self.queue.insert(0, item)

    # method aliases
    push = Queue.Queue.put
    pop = Queue.Queue.get
    pop_nowait = Queue.Queue.get_nowait

#
# try it

stack = Stack(0)

# push items on stack
stack.push("first")
stack.push("second")
stack.push("third")

# print stack contents
try:
    while 1:
        print stack.pop_nowait()
except Empty:
    pass

third
second
first

File: thread-example-1.py

import thread
import time, random

def worker():
    for i in range(50):
        # pretend we're doing something that takes 10�00 ms
        time.sleep(random.randint(10, 100) / 1000.0)
        print thread.get_ident(), "-- task", i, "finished"

#
# try it out!

for i in range(2):
    thread.start_new_thread(worker, ())

time.sleep(1)

print "goodbye!"

311 -- task 0 finished
265 -- task 0 finished
265 -- task 1 finished
311 -- task 1 finished
...
265 -- task 17 finished
311 -- task 13 finished
265 -- task 18 finished
goodbye!

注意当主程序退出的时候, 所有的线程也随着退出. 而 threading 模块不存在这个问题 . (该行为可改变)

File: commands-example-1.py

import commands

stat, output = commands.getstatusoutput("ls -lR")

print "status", "=>", stat
print "output", "=>", len(output), "bytes"

status => 0
output => 171046 bytes

File: pipes-example-1.py

import pipes

t = pipes.Template()

# create a pipeline
# 这里 " - " 代表从标准输入读入内容
t.append("sort", "--")
t.append("uniq", "--")

# filter some text
# 这里空字符串代表标准输出
t.copy("samples/sample.txt", "")

Alan Jones (sensible party)
Kevin Phillips-Bong (slightly silly)
Tarquin Fin-tim-lin-bin-whin-bim-lin-bus-stop-F'tang-F'tang-Olé-Biscuitbarrel

popen2 模块允许你执行外部命令, 并通过流来分别访问它的 stdin 和 stdout ( 可能还有 stderr ).

File: popen2-example-1.py

import popen2, string

fin, fout = popen2.popen2("sort")

fout.write("foo/n")
fout.write("bar/n")
fout.close()

print fin.readline(),
print fin.readline(),
fin.close()

bar
foo

File: popen2-example-2.py

import popen2
import string

class Chess:
    "Interface class for chesstool-compatible programs"

    def _ _init_ _(self, engine = "gnuchessc"):
        self.fin, self.fout = popen2.popen2(engine)
        s = self.fin.readline()
        if s != "Chess/n":
            raise IOError, "incompatible chess program"

    def move(self, move):
        self.fout.write(move + "/n")
        self.fout.flush()
        my = self.fin.readline()
        if my == "Illegal move":
            raise ValueError, "illegal move"
        his = self.fin.readline()
        return string.split(his)[2]

    def quit(self):
        self.fout.write("quit/n")
        self.fout.flush()

#
# play a few moves

g = Chess()

print g.move("a2a4")
print g.move("b2b3")

g.quit()

b8c6
e7e5

File: signal-example-1.py

import signal
import time

def handler(signo, frame):
    print "got signal", signo

signal.signal(signal.SIGALRM, handler)

# wake me up in two seconds
signal.alarm(2)

now = time.time()

time.sleep(200)

print "slept for", time.time() - now, "seconds"

got signal 14
slept for 1.99262607098 seconds

"PALO ALTO, Calif. - Intel says its Pentium Pro and new Pentium II chips have a flaw that can cause computers to sometimes make mistakes but said the problems could be fixed easily with rewritten software."

- Reuters telegram

本章描述了一些用于在 Python 对象和其他数据表示类型间相互转换的模块. 这些模块通常用于读写特定的文件格式或是储存/取出 Python 变量.

Python 提供了一些用于二进制数据解码/编码的模块. struct 模块用于在 二进制数据结构(例如 C 中的 struct )和 Python 元组间转换. array 模块将二进制数据阵列 ( C arrays )封装为 Python 序列对象.

marshal 和 pickle 模块用于在不同的 Python 程序间共享/传递数据.

marshal 模块使用了简单的自描述格式( Self-Describing Formats ), 它支持大多的内建数据类型, 包括 code 对象. Python 自身也使用了这个格式来储存编译后代码( .pyc 文件).

pickle 模块提供了更复杂的格式, 它支持用户定义的类, 自引用数据结构等等. pickle 是用 Python 写的, 相对来说速度较慢, 不过还有一个 cPickle 模块, 使用 C 实现了相同的功能, 速度和 marshal 不相上下.

一些模块提供了增强的格式化输出, 用来补充内建的 repr 函数和 % 字符串格式化操作符.

pprint 模块几乎可以将任何 Python 数据结构很好地打印出来(提高可读性).

repr 模块可以用来替换内建同名函数. 该模块与内建函数不同的是它限制了很多输出形式: 他只会 输出字符串的前 30 个字符, 它只打印嵌套数据结构的几个等级, 等等.

Python 支持大部分常见二进制编码, 例如 base64 , binhex (一种 Macintosh 格式) , quoted printable , 以及 uu 编码.

array 模块实现了一个有效的阵列储存类型. 阵列和列表类似, 但其中所有的项目必须为相同的 类型. 该类型在阵列创建时指定.

File: array-example-1.py

import array

a = array.array("B", range(16)) # unsigned char
b = array.array("h", range(16)) # signed short

print a
print repr(a.tostring())

print b
print repr(b.tostring())

array('B', [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])
'/000/001/002/003/004/005/006/007/010/011/012/013/014/015/016/017'

array('h', [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])
'/000/000/001/000/002/000/003/000/004/000/005/000/006/000/007/000
/010/000/011/000/012/000/013/000/014/000/015/000/016/000/017/000'

File: array-example-2.py

import array

a = array.array("B", [1, 2, 3])

a.append(4)

a = a + a

a = a[2:-2]

print a
print repr(a.tostring())
for i in a:
    print i,

array('B', [3, 4, 1, 2])
'/003/004/001/002'
3 4 1 2

File: array-example-3.py

import array

a = array.array("i", "fish license") # signed integer

print a
print repr(a.tostring())
print a.tolist()

array('i', [1752394086, 1667853344, 1702063717])
'fish license'
[1752394086, 1667853344, 1702063717]

File: array-example-4.py

import array

def little_endian():
    return ord(array.array("i",[1]).tostring()[0])

if little_endian():
    print "little-endian platform (intel, alpha)"
else:
    print "big-endian platform (motorola, sparc)"

big-endian platform (motorola, sparc)

Python 2.0 以及以后版本提供了 sys.byteorder 属性, 可以更简单地判断字节序 (属性值为 "little" 或 "big" ), 如 Example 4-5 所示.

File: sys-byteorder-example-1.py

import sys

# 2.0 and later
if sys.byteorder == "little":
    print "little-endian platform (intel, alpha)"
else:
    print "big-endian platform (motorola, sparc)"

big-endian platform (motorola, sparc)

File: struct-example-1.py

import struct

# native byteorder
buffer = struct.pack("ihb", 1, 2, 3)
print repr(buffer)
print struct.unpack("ihb", buffer)

# data from a sequence, network byteorder
data = [1, 2, 3]
buffer = apply(struct.pack, ("!ihb",) + tuple(data))
print repr(buffer)
print struct.unpack("!ihb", buffer)

# in 2.0, the apply statement can also be written as:
# buffer = struct.pack("!ihb", *data)

'/001/000/000/000/002/000/003'
(1, 2, 3)
'/000/000/000/001/000/002/003'
(1, 2, 3)

File: xdrlib-example-1.py

import xdrlib

#
# create a packer and add some data to it

p = xdrlib.Packer()
p.pack_uint(1)
p.pack_string("spam")

data = p.get_buffer()

print "packed:", repr(data)

#
# create an unpacker and use it to decode the data

u = xdrlib.Unpacker(data)

print "unpacked:", u.unpack_uint(), repr(u.unpack_string())

u.done()

packed: '/000/000/000/001/000/000/000/004spam'
unpacked: 1 'spam'

File: xdrlib-example-2.py

import xdrlib

# some constants (see the RPC specs for details)
RPC_CALL = 1
RPC_VERSION = 2

MY_PROGRAM_ID = 1234 # assigned by Sun
MY_VERSION_ID = 1000
MY_TIME_PROCEDURE_ID = 9999

AUTH_NULL = 0

transaction = 1

p = xdrlib.Packer()

# send a Sun RPC call package
p.pack_uint(transaction)
p.pack_enum(RPC_CALL)
p.pack_uint(RPC_VERSION)
p.pack_uint(MY_PROGRAM_ID)
p.pack_uint(MY_VERSION_ID)
p.pack_uint(MY_TIME_PROCEDURE_ID)
p.pack_enum(AUTH_NULL)
p.pack_uint(0)
p.pack_enum(AUTH_NULL)
p.pack_uint(0)

print repr(p.get_buffer())

'/000/000/000/001/000/000/000/001/000/000/000/002/000/000/004/322
/000/000/003/350/000/000/'/017/000/000/000/000/000/000/000/000/000
/000/000/000/000/000/000/000'

File: marshal-example-1.py

import marshal

value = (
    "this is a string",
    [1, 2, 3, 4],
    ("more tuples", 1.0, 2.3, 4.5),
    "this is yet another string"
    )

data = marshal.dumps(value)

# intermediate format
print type(data), len(data)

print "-"*50
print repr(data)
print "-"*50

print marshal.loads(data)

<type 'string'> 118
--------------------------------------------------
'(/004/000/000/000s/020/000/000/000this is a string
[/004/000/000/000i/001/000/000/000i/002/000/000/000
i/003/000/000/000i/004/000/000/000(/004/000/000/000
s/013/000/000/000more tuplesf/0031.0f/0032.3f/0034.
5s/032/000/000/000this is yet another string'
--------------------------------------------------
('this is a string', [1, 2, 3, 4], ('more tuples',
1.0, 2.3, 4.5), 'this is yet another string')

File: marshal-example-2.py

import marshal

script = """
print 'hello'
"""

code = compile(script, "<script>", "exec")

data = marshal.dumps(code)

# intermediate format
print type(data), len(data)

print "-"*50
print repr(data)
print "-"*50

exec marshal.loads(data)

<type 'string'> 81
--------------------------------------------------
'c/000/000/000/000/001/000/000/000s/017/000/000/00
0/177/000/000/177/002/000d/000/000GHd/001/000S(/00
2/000/000/000s/005/000/000/000helloN(/000/000/000/
000(/000/000/000/000s/010/000/000/000<script>s/001
/000/000/000?/002/000s/000/000/000/000'
--------------------------------------------------
hello

pickle 模块同 marshal 模块相同, 将数据连续化, 便于保存传输. 它比 marshal 要慢一些, 但它可以处理类实例, 共享的元素, 以及递归数据结构等.

File: pickle-example-1.py

import pickle

value = (
    "this is a string",
    [1, 2, 3, 4],
    ("more tuples", 1.0, 2.3, 4.5),
    "this is yet another string"
    )

data = pickle.dumps(value)

# intermediate format
print type(data), len(data)

print "-"*50
print data
print "-"*50

print pickle.loads(data)

<type 'string'> 121
--------------------------------------------------
(S'this is a string'
p0
(lp1
I1
aI2
aI3
aI4
a(S'more tuples'
p2
F1.0
F2.3
F4.5
tp3
S'this is yet another string'
p4
tp5
.
--------------------------------------------------
('this is a string', [1, 2, 3, 4], ('more tuples',
1.0, 2.3, 4.5), 'this is yet another string')

不过另一方面, pickle 不能处理 code 对象(可以参阅 copy_reg 模块来完成这个).

File: pickle-example-2.py

import pickle
import math

value = (
    "this is a long string" * 100,
    [1.2345678, 2.3456789, 3.4567890] * 100
    )

# text mode
data = pickle.dumps(value)
print type(data), len(data), pickle.loads(data) == value

# binary mode
data = pickle.dumps(value, 1)
print type(data), len(data), pickle.loads(data) == value

File: cpickle-example-1.py

try:
    import cPickle
    pickle = cPickle
except ImportError:
    import pickle

你可以使用 copy_reg 模块注册你自己的扩展类型. 这样 pickle 和 copy 模块就会知道 如何处理非标准类型.

例如, 标准的 pickle 实现不能用来处理 Python code 对象, 如下所示:

File: copy-reg-example-1.py

import pickle

CODE = """
print 'good evening'
"""

code = compile(CODE, "<string>", "exec")

exec code
exec pickle.loads(pickle.dumps(code))

good evening
Traceback (innermost last):
...
pickle.PicklingError: can't pickle 'code' objects

File: copy-reg-example-2.py

import copy_reg
import pickle, marshal, types

#
# register a pickle handler for code objects

def code_unpickler(data):
    return marshal.loads(data)

def code_pickler(code):
    return code_unpickler, (marshal.dumps(code),)

copy_reg.pickle(types.CodeType, code_pickler, code_unpickler)

#
# try it out

CODE = """
print "suppose he's got a pointed stick"
"""

code = compile(CODE, "<string>", "exec")

exec code
exec pickle.loads(pickle.dumps(code))

suppose he's got a pointed stick
suppose he's got a pointed stick

如果你是在网络中传输 pickle 后的数据, 那么请确保自定义的 unpickler 在数据接收端也是可用的.

Example 4-15 展示了如何实现 pickle 一个打开的文件对象.

File: copy-reg-example-3.py

import copy_reg
import pickle, types
import StringIO

#
# register a pickle handler for file objects

def file_unpickler(position, data):
    file = StringIO.StringIO(data)
    file.seek(position)
    return file

def file_pickler(code):
    position = file.tell()
    file.seek(0)
    data = file.read()
    file.seek(position)
    return file_unpickler, (position, data)

copy_reg.pickle(types.FileType, file_pickler, file_unpickler)

#
# try it out

file = open("samples/sample.txt", "rb")

print file.read(120),
print "<here>",
print pickle.loads(pickle.dumps(file)).read()

We will perhaps eventually be writing only small
modules, which are identified by name as they are
used to build larger <here> ones, so that devices like
indentation, rather than delimiters, might become
feasible for expressing local structure in the
source language.
     -- Donald E. Knuth, December 1974

pprint 模块( pretty printer )用于打印 Python 数据结构. 当你在命令行下打印 特定数据结构时你会发现它很有用(输出格式比较整齐, 便于阅读).

File: pprint-example-1.py

import pprint

data = (
    "this is a string", [1, 2, 3, 4], ("more tuples",
    1.0, 2.3, 4.5), "this is yet another string"
    )

pprint.pprint(data)

('this is a string',
 [1, 2, 3, 4],
 ('more tuples', 1.0, 2.3, 4.5),
 'this is yet another string')

File: repr-example-1.py

# note: this overrides the built-in 'repr' function
from repr import repr

# an annoyingly recursive data structure
data = (
    "X" * 100000,
    )
data = [data]
data.append(data)

print repr(data)

[('XXXXXXXXXXXX...XXXXXXXXXXXXX',), [('XXXXXXXXXXXX...XXXXXXXXXX
XXX',), [('XXXXXXXXXXXX...XXXXXXXXXXXXX',), [('XXXXXXXXXXXX...XX
XXXXXXXXXXX',), [('XXXXXXXXXXXX...XXXXXXXXXXXXX',), [(...), [...
]]]]]]]

base64 编码体系用于将任意二进制数据转换为纯文本. 它将一个 3 字节的二进制字节组 转换为 4 个文本字符组储存, 而且规定只允许以下集合中的字符出现:

ABCDEFGHIJKLMNOPQRSTUVWXYZ
abcdefghijklmnopqrstuvwxyz
0123456789+/

另外, = 用于填充数据流的末尾.

File: base64-example-1.py

import base64

MESSAGE = "life of brian"

file = open("out.txt", "w")
file.write(MESSAGE)
file.close()

base64.encode(open("out.txt"), open("out.b64", "w"))
base64.decode(open("out.b64"), open("out.txt", "w"))

print "original:", repr(MESSAGE)
print "encoded message:", repr(open("out.b64").read())
print "decoded message:", repr(open("out.txt").read())

original:  'life of brian'
encoded message: 'bGlmZSBvZiBicmlhbg==/012'
decoded message: 'life of brian'

File: base64-example-2.py

import base64

MESSAGE = "life of brian"

data = base64.encodestring(MESSAGE)

original_data = base64.decodestring(data)

print "original:", repr(MESSAGE)
print "encoded data:", repr(data)
print "decoded data:", repr(original_data)

original: 'life of brian'
encoded data: 'bGlmZSBvZiBicmlhbg==/012'
decoded data: 'life of brian'

File: base64-example-3.py

import base64

def getbasic(user, password):
    # basic authentication (according to HTTP)
    return base64.encodestring(user + ":" + password)    

print getbasic("Aladdin", "open sesame")

'QWxhZGRpbjpvcGVuIHNlc2FtZQ=='

File: base64-example-4.py

import base64, sys

if not sys.argv[1:]:
    print "Usage: gif2tk.py giffile >pyfile"
    sys.exit(1)

data = open(sys.argv[1], "rb").read()

if data[:4] != "GIF8":
    print sys.argv[1], "is not a GIF file"
    sys.exit(1)

print '# generated from', sys.argv[1], 'by gif2tk.py'
print
print 'from Tkinter import PhotoImage' 
print
print 'image = PhotoImage(data="""'
print base64.encodestring(data),
print '""")'

# generated from samples/sample.gif by gif2tk.py

from Tkinter import PhotoImage

image = PhotoImage(data="""
R0lGODlhoAB4APcAAAAAAIAAAACAAICAAAAAgIAAgACAgICAgAQEBIwEBIyMBJRUlISE/LRUBAQE
...
AjmQBFmQBnmQCJmQCrmQDNmQDvmQEBmREnkRAQEAOw==
""")

File: binhex-example-1.py

import binhex
import sys

infile = "samples/sample.jpg"

binhex.binhex(infile, sys.stdout)

(This file must be converted with BinHex 4.0)

:#R0KEA"XC5jUF'F!2j!)!*!%%TS!N!4RdrrBrq!!%%T'58B!!3%!!!%!!3!!rpX
!3`!)"JB("J8)"`F(#3N)#J`8$3`,#``C%K-2&"dD(aiG'K`F)#3Z*b!L,#-F(#J
h+5``-63d0"mR16di-M`Z-c3brpX!3`%*#3N-#``B$3dB-L%F)6+3-[r!!"%)!)!
!J!-")J!#%3%$%3(ra!!I!!!""3'3"J#3#!%#!`3&"JF)#3S,rm3!Y4!!!J%$!`)
%!`8&"!3!!!&p!3)$!!34"4)K-8%'%e&K"b*a&$+"ND%))d+a`495dI!N-f*bJJN

该模块有两个函数 binhex 和 hexbin .

quopri 模块基于 MIME 标准实现了引用的可打印编码( quoted printable encoding ).

File: quopri-example-1.py

import quopri
import StringIO

# helpers (the quopri module only supports file-to-file conversion)

def encodestring(instring, tabs=0):
    outfile = StringIO.StringIO()
    quopri.encode(StringIO.StringIO(instring), outfile, tabs)
    return outfile.getvalue()

def decodestring(instring):
    outfile = StringIO.StringIO()
    quopri.decode(StringIO.StringIO(instring), outfile)
    return outfile.getvalue()

#
# try it out

MESSAGE = "å i åa ä e ö!"

encoded_message = encodestring(MESSAGE)
decoded_message = decodestring(encoded_message)

print "original:", MESSAGE
print "encoded message:", repr(encoded_message)
print "decoded message:", decoded_message

original: å i åa ä e ö!
encoded message: '=E5 i =E5a =E4 e =F6!/012'
decoded message: å i åa ä e ö!

uu 编码体系用于将任意二进制数据转换为普通文本格式. 该格式在新闻组中很流行, 但逐渐被 base64 编码取代.

uu 编码将每个 3 字节( 24 位)的数据组转换为 4 个可打印字符(每个字符 6 位), 使用从 chr(32) (空格) 到 chr(95) 的字符. uu 编码通常会使数据大小增加 40% .

一个编码后的数据流以一个新行开始, 它包含文件的权限( Unix 格式)和文件名, 以 end 行结尾:

begin 666 sample.jpg
M_]C_X  02D9)1@ ! 0   0 !  #_VP!#  @&!@<&!0@'!P<)'0@*#!0-# L+
...more lines like this...
end

uu 模块提供了两个函数: encode 和 decode .

File: uu-example-1.py

import uu
import os, sys

infile = "samples/sample.jpg"

uu.encode(infile, sys.stdout, os.path.basename(infile))

begin 666 sample.jpg
M_]C_X  02D9)1@ ! 0   0 !  #_VP!#  @&!@<&!0@'!P<)"0@*#!0-# L+
M#!D2$P/4'1H?'AT:'!P@)"XG("(L(QP<*#<I+# Q-#0T'R<Y/3@R/"XS-#+_
MVP!# 0D)"0P+#!@-#1@R(1PA,C(R,C(R,C(R,C(R,C(R,C(R,C(R,C(R,C(R
M,C(R,C(R,C(R,C(R,C(R,C(R,C(R,C+_P  1" "  ( # 2(  A$! Q$!_/0 
M'P   04! 0$! 0$           $" P0%!@<("0H+_/0 M1   @$# P($ P4%

File: uu-example-2.py

import uu
import StringIO

infile = "samples/sample.uue"
outfile = "samples/sample.jpg"

#
# decode

fi = open(infile)
fo = StringIO.StringIO()

uu.decode(fi, fo)

#
# compare with original data file

data = open(outfile, "rb").read()

if fo.getvalue() == data:
    print len(data), "bytes ok"

File: binascii-example-1.py

import binascii

text = "hello, mrs teal"

data = binascii.b2a_base64(text)
text = binascii.a2b_base64(data)
print text, "<=>", repr(data)

data = binascii.b2a_uu(text)
text = binascii.a2b_uu(data)
print text, "<=>", repr(data)

data = binascii.b2a_hqx(text)
text = binascii.a2b_hqx(data)[0]
print text, "<=>", repr(data)

# 2.0 and newer
data = binascii.b2a_hex(text)
text = binascii.a2b_hex(data)
print text, "<=>", repr(data)

hello, mrs teal <=> 'aGVsbG8sIG1ycyB0ZWFs/012'
hello, mrs teal <=> '/:&5L;&//L(&UR<R!T96%L/012'
hello, mrs teal <=> 'D/'9XE/'mX)/'ebFb"dC@&X'
hello, mrs teal <=> '68656c6c6f2c206d7273207465616c'

本章将描述用于处理不同文件格式的模块.

Python 提供了一些用于处理可扩展标记语言( Extensible Markup Language , XML ) 和超文本标记语言( Hypertext Markup Language , HTML )的扩展. Python 同样提供了对 标准通用标记语言( Standard Generalized Markup Language , SGML )的支持.

所有这些格式都有着相同的结构, 因为 HTML 和 XML 都来自 SGML . 每个文档都是由 起始标签( start tags ), 结束标签( end tags ), 文本(又叫字符数据), 以及实体引用( entity references )构成:

<document name="sample.xml">
    <header>This is a header</header>
    <body>This is the body text.  The text can contain
    plain text ("character data"), tags, and
    entities.
    </body>
</document>

在这个例子中, <document>, <header>, 以及 <body> 是起始标签. 每个起始标签都有一个对应的结束标签, 使用斜线 "/" 标记. 起始标签可以包含多个属性, 比如这里的 name 属性.

起始标签和它对应的结束标签中的任何东西被称为 元素( element ). 这里 document 元素包含 header 和 body 两个元素.

" 是一个字符实体( character entity ). 字符实体用于在文本区域中表示特殊的保留字符, 使用 & 指示. 这里它代表一个引号, 常见字符实体还有 " < ( < )" 和 " > ( > )" .

虽然 XML , HTML , SGML 使用相同的结构块, 但它们还有一些不同点. 在 XML 中, 所有元素必须有起始和结束标签, 所有标签必须正确嵌套( well-formed ). 而且 XML 是区分大小写的, 所以 <document> 和 <Document> 是不同的元素类型.

HTML 有很高灵活性, HTML 语法分析器一般会自动补全缺失标签; 例如, 当遇到一个以 <P> 标签开始的新段落, 却没有对应结束标签, 语法分析器会自动添加一个 </P> 标签. HTML 也是区分大小写的. 另一方面, XML 允许你定义任何元素, 而 HTML 使用一些由 HTML 规范定义的固定元素.

SGML 有着更高的灵活性, 你可以使用自己的声明( declaration ) 定义源文件如何转换到元素结构, DTD ( document type description , 文件类型定义)可以用来 检查结构并补全缺失标签. 技术上来说, HTML 和 XML 都是 SGML 应用, 有各自的 SGML 声明, 而且 HTML 有一个标准 DTD .

Python 提供了多个 makeup 语言分析器. 由于 SGML 是最灵活的格式, Python 的 sgmllib 事实上很简单. 它不会去处理 DTD , 不过你可以继承它来提供更复杂的功能.

Python 的 HTML 支持基于 SGML 分析器. htmllib 将具体的格式输出工作交给 formatter 对象. formatter 模块包含一些标准格式化标志.

Python 的 XML 支持模块很复杂. 先前是只有与 sgmllib 类似的 xmllib , 后来加入了更高级的 expat 模块(可选). 而最新版本中已经准备废弃 xmllib ,启用xml 包作为工具集.

ConfigParser 模块用于读取简单的配置文件, 类似 Windows 下的 INI 文件.

netrc 模块用于读取 .netrc 配置文件, shlex 模块用于读取类似 shell 脚本语法的配置文件.

Python 的标准库提供了对 GZIP 和 ZIP ( 2.0 及以后) 格式的支持. 基于 zlib 模块, gzip 和 zipfile 模块分别用来处理这类文件.

xmllib 已在当前版本中申明不支持.

File: xmllib-example-1.py

import xmllib

class Parser(xmllib.XMLParser):
    # get quotation number

    def _ _init_ _(self, file=None):
        xmllib.XMLParser._ _init_ _(self)
        if file:
            self.load(file)

    def load(self, file):
        while 1:
            s = file.read(512)
            if not s:
                break
            self.feed(s)
        self.close()

    def start_quotation(self, attrs):
        print "id =>", attrs.get("id")
        raise EOFError

try:
    c = Parser()
    c.load(open("samples/sample.xml"))
except EOFError:
    pass

id => 031

File: xmllib-example-2.py

import xmllib
import string, sys

STYLESHEET = {
    # each element can contribute one or more style elements
    "quotation": {"style": "italic"},
    "lang": {"weight": "bold"},
    "name": {"weight": "medium"},
}

class Parser(xmllib.XMLParser):
    # a simple styling engine

    def _ _init_ _(self, renderer):
        xmllib.XMLParser._ _init_ _(self)
        self._ _data = []
        self._ _tags = []
        self._ _renderer = renderer

    def load(self, file):
        while 1:
            s = file.read(8192)
            if not s:
                break
            self.feed(s)
        self.close()

    def handle_data(self, data):
        self._ _data.append(data)

    def unknown_starttag(self, tag, attrs):
        if self._ _data:
            text = string.join(self._ _data, "")
            self._ _renderer.text(self._ _tags, text)
        self._ _tags.append(tag)
        self._ _data = []

    def unknown_endtag(self, tag):
        self._ _tags.pop()
        if self._ _data:
            text = string.join(self._ _data, "")
            self._ _renderer.text(self._ _tags, text)
        self._ _data = []

class DumbRenderer:

    def _ _init_ _(self):
        self.cache = {}

    def text(self, tags, text):
        # render text in the style given by the tag stack
        tags = tuple(tags)
        style = self.cache.get(tags)
        if style is None:
            # figure out a combined style
            style = {}
            for tag in tags:
                s = STYLESHEET.get(tag)
                if s:
                    style.update(s)
            self.cache[tags] = style # update cache
        # write to standard output
        sys.stdout.write("%s =>/n" % style)
        sys.stdout.write("  " + repr(text) + "/n")

#
# try it out

r = DumbRenderer()
c = Parser(r)
c.load(open("samples/sample.xml"))

{'style': 'italic'} =>
  'I/'ve had a lot of developers come up to me and/012say,
  "I haven/'t had this much fun in a long time. It sure
  beats/012writing '
{'style': 'italic', 'weight': 'bold'} =>
  'Cobol'
{'style': 'italic'} =>
  '" -- '
{'style': 'italic', 'weight': 'medium'} =>
  'James Gosling'
{'style': 'italic'} =>
  ', on/012'
{'weight': 'bold'} =>
  'Java'
{'style': 'italic'} =>
  '.'

File: xml-parsers-expat-example-1.py

from xml.parsers import expat

class Parser:

    def _ _init_ _(self):
        self._parser = expat.ParserCreate()
        self._parser.StartElementHandler = self.start
        self._parser.EndElementHandler = self.end
        self._parser.CharacterDataHandler = self.data

    def feed(self, data):
        self._parser.Parse(data, 0)

    def close(self):
        self._parser.Parse("", 1) # end of data
        del self._parser # get rid of circular references

    def start(self, tag, attrs):
        print "START", repr(tag), attrs

    def end(self, tag):
        print "END", repr(tag)

    def data(self, data):
        print "DATA", repr(data)

p = Parser()
p.feed("<tag>data</tag>")
p.close()

START u'tag' {}
DATA u'data'
END u'tag'

File: xml-parsers-expat-example-2.py

from xml.parsers import expat

class Parser:

    def _ _init_ _(self):
        self._parser = expat.ParserCreate()
        self._parser.StartElementHandler = self.start
        self._parser.EndElementHandler = self.end
        self._parser.CharacterDataHandler = self.data

    def feed(self, data):
        self._parser.Parse(data, 0)

    def close(self):
        self._parser.Parse("", 1) # end of data
        del self._parser # get rid of circular references

    def start(self, tag, attrs):
        print "START", repr(tag), attrs

    def end(self, tag):
        print "END", repr(tag)

    def data(self, data):
        print "DATA", repr(data)

p = Parser()
p.feed("""/
<?xml version='1.0' encoding='iso-8859-1'?>
<author>
<name>fredrik lundh</name>
<city>linköping</city>
</author>
"""
)
p.close()

START u'author' {}
DATA u'/012'
START u'name' {}
DATA u'fredrik lundh'
END u'name'
DATA u'/012'
START u'city' {}
DATA u'link/366ping'
END u'city'
DATA u'/012'
END u'author'

File: sgmllib-example-1.py

import sgmllib
import string

class FoundTitle(Exception):
    pass

class ExtractTitle(sgmllib.SGMLParser):

    def _ _init_ _(self, verbose=0):
        sgmllib.SGMLParser._ _init_ _(self, verbose)
        self.title = self.data = None

    def handle_data(self, data):
        if self.data is not None:
            self.data.append(data)

    def start_title(self, attrs):
        self.data = []

    def end_title(self):
        self.title = string.join(self.data, "")
        raise FoundTitle # abort parsing!

def extract(file):
    # extract title from an HTML/SGML stream
    p = ExtractTitle()
    try:
        while 1:
            # read small chunks
            s = file.read(512)
            if not s:
                break
            p.feed(s)
        p.close()
    except FoundTitle:
        return p.title
    return None

#
# try it out

print "html", "=>", extract(open("samples/sample.htm"))
print "sgml", "=>", extract(open("samples/sample.sgm"))

html => A Title.
sgml => Quotations

File: sgmllib-example-2.py

import sgmllib
import cgi, sys

class PrettyPrinter(sgmllib.SGMLParser):
    # A simple SGML pretty printer

    def _ _init_ _(self):
        # initialize base class
        sgmllib.SGMLParser._ _init_ _(self)
        self.flag = 0

    def newline(self):
        # force newline, if necessary
        if self.flag:
            sys.stdout.write("/n")
        self.flag = 0

    def unknown_starttag(self, tag, attrs):
        # called for each start tag

        # the attrs argument is a list of (attr, value)
        # tuples. convert it to a string.
        text = ""
        for attr, value in attrs:
            text = text + " %s='%s'" % (attr, cgi.escape(value))

        self.newline()
        sys.stdout.write("<%s%s>/n" % (tag, text))

    def handle_data(self, text):
        # called for each text section
        sys.stdout.write(text)
        self.flag = (text[-1:] != "/n")

    def handle_entityref(self, text):
        # called for each entity
        sys.stdout.write("&%s;" % text)
        
    def unknown_endtag(self, tag):
        # called for each end tag
        self.newline()
        sys.stdout.write("<%s>" % tag)

#
# try it out

file = open("samples/sample.sgm")

p = PrettyPrinter()
p.feed(file.read())
p.close()

<chapter>
<title>
Quotations
<title>
<epigraph>
<attribution>
eff-bot, June 1997
<attribution>
<para>
<quote>
Nobody expects the Spanish Inquisition! Amongst
our weaponry are such diverse elements as fear, surprise,
ruthless efficiency, and an almost fanatical devotion to
Guido, and nice red uniforms — oh, damn!
<quote>
<para>
<epigraph>
<chapter>

File: sgmllib-example-3.py

import sgmllib

class WellFormednessChecker(sgmllib.SGMLParser):
    # check that an SGML document is 'well-formed'
    # (in the XML sense).

    def _ _init_ _(self, file=None):
        sgmllib.SGMLParser._ _init_ _(self)
        self.tags = []
        if file:
            self.load(file)

    def load(self, file):
        while 1:
            s = file.read(8192)
            if not s:
                break
            self.feed(s)
        self.close()

    def close(self):
        sgmllib.SGMLParser.close(self)
        if self.tags:
            raise SyntaxError, "start tag %s not closed" % self.tags[-1]

    def unknown_starttag(self, start, attrs):
        self.tags.append(start)

    def unknown_endtag(self, end):
        start = self.tags.pop()
        if end != start:
            raise SyntaxError, "end tag %s does't match start tag %s" %/
                  (end, start)

try:
    c = WellFormednessChecker()
    c.load(open("samples/sample.htm"))
except SyntaxError:
    raise # report error
else:
    print "document is well-formed"

Traceback (innermost last):
...
SyntaxError: end tag head does't match start tag meta

File: sgmllib-example-4.py

import sgmllib
import cgi, string, sys

class SGMLFilter(sgmllib.SGMLParser):
    # sgml filter.  override start/end to manipulate
    # document elements

    def _ _init_ _(self, outfile=None, infile=None):
        sgmllib.SGMLParser._ _init_ _(self)
        if not outfile:
            outfile = sys.stdout
        self.write = outfile.write
        if infile:
            self.load(infile)

    def load(self, file):
        while 1:
            s = file.read(8192)
            if not s:
                break
            self.feed(s)
        self.close()

    def handle_entityref(self, name):
        self.write("&%s;" % name)

    def handle_data(self, data):
        self.write(cgi.escape(data))

    def unknown_starttag(self, tag, attrs):
        tag, attrs = self.start(tag, attrs)
        if tag:
            if not attrs:
                self.write("<%s>" % tag)
            else:
                self.write("<%s" % tag)
                for k, v in attrs:
                    self.write(" %s=%s" % (k, repr(v)))
                self.write(">")

    def unknown_endtag(self, tag):
        tag = self.end(tag)
        if tag:
            self.write("</%s>" % tag)

    def start(self, tag, attrs):
        return tag, attrs # override

    def end(self, tag):
        return tag # override

class Filter(SGMLFilter):

    def fixtag(self, tag):
        if tag == "em":
            tag = "i"
        if tag == "string":
            tag = "b"
        return string.upper(tag)

    def start(self, tag, attrs):
        return self.fixtag(tag), attrs

    def end(self, tag):
        return self.fixtag(tag)

c = Filter()
c.load(open("samples/sample.htm"))

File: htmllib-example-1.py

import htmllib
import formatter
import string

class Parser(htmllib.HTMLParser):
    # return a dictionary mapping anchor texts to lists
    # of associated hyperlinks

    def _ _init_ _(self, verbose=0):
        self.anchors = {}
        f = formatter.NullFormatter()
        htmllib.HTMLParser._ _init_ _(self, f, verbose)

    def anchor_bgn(self, href, name, type):
        self.save_bgn()
        self.anchor = href

    def anchor_end(self):
        text = string.strip(self.save_end())
        if self.anchor and text:
            self.anchors[text] = self.anchors.get(text, []) + [self.anchor]

file = open("samples/sample.htm")
html = file.read()
file.close()

p = Parser()
p.feed(html)
p.close()

for k, v in p.anchors.items():
    print k, "=>", v

print

link => ['http://www.python.org']

如果你只是想解析一个 HTML 文件, 而不是将它交给输出设备, 那么 sgmllib 模块会是更好的选择.

File: htmlentitydefs-example-1.py

import htmlentitydefs

entities = htmlentitydefs.entitydefs

for entity in "amp", "quot", "copy", "yen":
    print entity, "=", entities[entity]

amp = &
quot = "
copy = /302/251
yen = /302/245

Example 5-11 展示了如何将正则表达式与这个字典结合起来翻译字符串中的实体 ( cgi.escape 的逆向操作).

File: htmlentitydefs-example-2.py

import htmlentitydefs
import re
import cgi

pattern = re.compile("&(/w+?);")

def descape_entity(m, defs=htmlentitydefs.entitydefs):
    # callback: translate one entity to its ISO Latin value
    try:
        return defs[m.group(1)]
    except KeyError:
        return m.group(0) # use as is

def descape(string):
    return pattern.sub(descape_entity, string)

print descape("<spam&eggs>")
print descape(cgi.escape("<spam&eggs>"))

<spam&eggs>
<spam&eggs>

File: htmlentitydefs-example-3.py

import htmlentitydefs
import re, string

# this pattern matches substrings of reserved and non-ASCII characters
pattern = re.compile(r"[&<>/"/x80-/xff]+")

# create character map
entity_map = {}

for i in range(256):
    entity_map[chr(i)] = "&%d;" % i

for entity, char in htmlentitydefs.entitydefs.items():
    if entity_map.has_key(char):
        entity_map[char] = "&%s;" % entity

def escape_entity(m, get=entity_map.get):
    return string.join(map(get, m.group()), "")

def escape(string):
    return pattern.sub(escape_entity, string)

print escape("<spam&eggs>")
print escape("/303/245 i /303/245a /303/244 e /303/266")

<spam&eggs>
å i åa ä e ö

formatter 模块提供了一些可用于 htmllib 的格式类( formatter classes ).

File: formatter-example-1.py

import formatter
import htmllib

w = formatter.AbstractWriter()
f = formatter.AbstractFormatter(w)

file = open("samples/sample.htm")

p = htmllib.HTMLParser(f)
p.feed(file.read())
p.close()

file.close()

send_paragraph(1)
new_font(('h1', 0, 1, 0))
send_flowing_data('A Chapter.')
send_line_break()
send_paragraph(1)
new_font(None)
send_flowing_data('Some text. Some more text. Some')
send_flowing_data(' ')
new_font((None, 1, None, None))
send_flowing_data('emphasized')
new_font(None)
send_flowing_data(' text. A')
send_flowing_data(' link')
send_flowing_data('[1]')
send_flowing_data('.')

File: formatter-example-2.py

import formatter
import htmllib

w = formatter.DumbWriter() # plain text
f = formatter.AbstractFormatter(w)

file = open("samples/sample.htm")

# print html body as plain text
p = htmllib.HTMLParser(f)
p.feed(file.read())
p.close()

file.close()

# print links
print
print
i = 1
for link in p.anchorlist:
    print i, "=>", link
    i = i + 1

A Chapter.

Some text. Some more text. Some emphasized text. A link[1].

1 => http://www.python.org

File: formatter-example-3.py

import formatter
import htmllib, string

class Writer(formatter.DumbWriter):

    def _ _init_ _(self):
        formatter.DumbWriter._ _init_ _(self)
        self.tag = ""
        self.bold = self.italic = 0
        self.fonts = []

    def new_font(self, font):
        if font is None:
            font = self.fonts.pop()
            self.tag, self.bold, self.italic = font
        else:
            self.fonts.append((self.tag, self.bold, self.italic))
            tag, bold, italic, typewriter = font
            if tag is not None:
                self.tag = tag
            if bold is not None:
                self.bold = bold
            if italic is not None:
                self.italic = italic

    def send_flowing_data(self, data):
        if not data:
            return
        atbreak = self.atbreak or data[0] in string.whitespace
        for word in string.split(data):
            if atbreak:
                self.file.write(" ")
            if self.tag in ("h1", "h2", "h3"):
                word = string.upper(word)
            if self.bold:
                word = "*" + word + "*"
            if self.italic:
                word = "_" + word + "_"
            self.file.write(word)
            atbreak = 1
        self.atbreak = data[-1] in string.whitespace

w = Writer()
f = formatter.AbstractFormatter(w)

file = open("samples/sample.htm")

# print html body as plain text
p = htmllib.HTMLParser(f)
p.feed(file.read())
p.close()

_A_ _CHAPTER._

Some text. Some more text. Some *emphasized* text. A link[1].

ConfigParser 模块用于读取配置文件.

配置文件的格式与 Windows INI 文件类似, 可以包含一个或多个区域( section ), 每个区域可以有多个配置条目.

File: configparser-example-1.py

import ConfigParser
import string

config = ConfigParser.ConfigParser()

config.read("samples/sample.ini")

# print summary
print
print string.upper(config.get("book", "title"))
print "by", config.get("book", "author"),
print  "(" + config.get("book", "email") + ")"
print
print config.get("ematter", "pages"), "pages"
print

# dump entire config file
for section in config.sections():
    print section
    for option in config.options(section):
        print " ", option, "=", config.get(section, option)

THE PYTHON STANDARD LIBRARY
by Fredrik Lundh (fredrik@pythonware.com)

250 pages

book
  title = The Python Standard Library
  email = fredrik@pythonware.com
  author = Fredrik Lundh
  version = 2.0-001115
  _ _name_ _ = book
ematter
  _ _name_ _ = ematter
  pages = 250
hardcopy
  _ _name_ _ = hardcopy
  pages = 350

File: configparser-example-2.py

import ConfigParser
import sys

config = ConfigParser.ConfigParser()

# set a number of parameters
config.add_section("book")
config.set("book", "title", "the python standard library")
config.set("book", "author", "fredrik lundh")

config.add_section("ematter")
config.set("ematter", "pages", 250)

# write to screen
config.write(sys.stdout)

[book]
title = the python standard library
author = fredrik lundh

[ematter]
pages = 250

netrc 模块可以用来解析 .netrc 配置文件, 如 Example 5-18 所示. 该文件用于在用户的 home 目录储存 FTP 用户名和密码. (别忘记设置这个文件的属性为: "chmod 0600 ~/.netrc," 这样只有当前用户能访问).

File: netrc-example-1.py

import netrc


# default is $HOME/.netrc
info = netrc.netrc("samples/sample.netrc")

login, account, password = info.authenticators("secret.fbi")
print "login", "=>", repr(login)
print "account", "=>", repr(account)
print "password", "=>", repr(password)

login => 'mulder'
account => None
password => 'trustno1'

File: shlex-example-1.py

import shlex

lexer = shlex.shlex(open("samples/sample.netrc", "r"))
lexer.wordchars = lexer.wordchars + "._"

while 1:
    token = lexer.get_token()
    if not token:
        break
    print repr(token)

'machine'
'secret.fbi'
'login'
'mulder'
'password'
'trustno1'
'machine'
'non.secret.fbi'
'login'
'scully'
'password'
'noway'

( 2.0 新增) zipfile 模块可以用来读写 ZIP 格式.

File: zipfile-example-1.py

import zipfile

file = zipfile.ZipFile("samples/sample.zip", "r")

# list filenames
for name in file.namelist():
    print name,
print

# list file information
for info in file.infolist():
    print info.filename, info.date_time, info.file_size

sample.txt sample.jpg
sample.txt (1999, 9, 11, 20, 11, 8) 302
sample.jpg (1999, 9, 18, 16, 9, 44) 4762

File: zipfile-example-2.py

import zipfile

file = zipfile.ZipFile("samples/sample.zip", "r")

for name in file.namelist():
    data = file.read(name)
    print name, len(data), repr(data[:10])

sample.txt 302 'We will pe'
sample.jpg 4762 '/377/330/377/340/000/020JFIF'

向压缩档加入文件很简单, 将文件名, 文件在 ZIP 档中的名称传递给 write 方法即可.

File: zipfile-example-3.py

import zipfile
import glob, os

# open the zip file for writing, and write stuff to it

file = zipfile.ZipFile("test.zip", "w")

for name in glob.glob("samples/*"):
    file.write(name, os.path.basename(name), zipfile.ZIP_DEFLATED)

file.close()

# open the file again, to see what's in it

file = zipfile.ZipFile("test.zip", "r")
for info in file.infolist():
    print info.filename, info.date_time, info.file_size, info.compress_size

sample.wav (1999, 8, 15, 21, 26, 46) 13260 10985
sample.jpg (1999, 9, 18, 16, 9, 44) 4762 4626
sample.au (1999, 7, 18, 20, 57, 34) 1676 1103
...

write 方法的第三个可选参数用于控制是否使用压缩. 默认为 zipfile.ZIP_STORED , 意味着只是将数据储存在档案里而不进行任何压缩. 如果安装了 zlib 模块, 那么就可以使用 zipfile.ZIP_DEFLATED 进行压缩.

File: zipfile-example-4.py

import zipfile
import glob, os, time

file = zipfile.ZipFile("test.zip", "w")

now = time.localtime(time.time())[:6]

for name in ("life", "of", "brian"):
    info = zipfile.ZipInfo(name)
    info.date_time = now
    info.compress_type = zipfile.ZIP_DEFLATED
    file.writestr(info, name*1000)

file.close()

# open the file again, to see what's in it

file = zipfile.ZipFile("test.zip", "r")

for info in file.infolist():
    print info.filename, info.date_time, info.file_size, info.compress_size

life (2000, 12, 1, 0, 12, 1) 4000 26
of (2000, 12, 1, 0, 12, 1) 2000 18
brian (2000, 12, 1, 0, 12, 1) 5000 31

File: gzip-example-1.py

import gzip

file = gzip.GzipFile("samples/sample.gz")

print file.read()

Well it certainly looks as though we're in for
a splendid afternoon's sport in this the 127th
Upperclass Twit of the Year Show.

File: gzip-example-2.py

import gzip

class gzipFile(gzip.GzipFile):
    # adds seek/tell support to GzipFile

    offset = 0

    def read(self, size=None):
        data = gzip.GzipFile.read(self, size)
        self.offset = self.offset + len(data)
        return data

    def seek(self, offset, whence=0):
        # figure out new position (we can only seek forwards)
        if whence == 0:
            position = offset
        elif whence == 1:
            position = self.offset + offset
        else:
            raise IOError, "Illegal argument"
        if position < self.offset:
            raise IOError, "Cannot seek backwards"

        # skip forward, in 16k blocks
        while position > self.offset:
            if not self.read(min(position - self.offset, 16384)):
                break

    def tell(self):
        return self.offset

#
# try it

file = gzipFile("samples/sample.gz")
file.seek(80)

print file.read()

this the 127th
Upperclass Twit of the Year Show.

"To be removed from our list of future commercial postings by [SOME] PUBLISHING COMPANY an Annual Charge of Ninety Five dollars is required. Just send $95.00 with your Name, Address and Name of the Newsgroup to be removed from our list."

- Newsgroup spammer, July 1996

"想要退出 '某' 宣传公司的未来商业广告列表吗, 您需要付 95 美元. 只要您支付95美元, 并且告诉我们您的姓名, 地址, 和需要退出的新闻组, 我们就会把您从列表中移除."

- 新闻组垃圾发送者, 1996 年 7 月

Python 有大量用于处理邮件和新闻组的模块, 其中包括了许多常见的邮件格式.

rfc822 模块包括了一个邮件和新闻组的解析器 (也可用于其它符合 RFC 822 标准的消息, 比如 HTTP 头).

通常, RFC 822 格式的消息包含一些标头字段, 后面至少有一个空行, 然后是信息主体.

For example, here's a short mail message. The first five lines make up the message header, and the actual message (a single line, in this case) follows after an empty line:

例如这里的邮件信息. 前五行组成了消息标头, 隔一个空行后是消息主体.

Message-Id: <20001114144603.00abb310@oreilly.com>
Date: Tue, 14 Nov 2000 14:55:07 -0500
To: "Fredrik Lundh" <fredrik@effbot.org>
From: Frank
Subject: Re: python library book!

Where is it?

File: rfc822-example-1.py

import rfc822

file = open("samples/sample.eml")

message = rfc822.Message(file)

for k, v in message.items():
    print k, "=", v

print len(file.read()), "bytes in body"

subject = Re: python library book!
from = "Frank" <your@editor>
message-id = <20001114144603.00abb310@oreilly.com>
to = "Fredrik Lundh" <fredrik@effbot.org>
date = Tue, 14 Nov 2000 14:55:07 -0500
25 bytes in body

File: rfc822-example-2.py

import rfc822

file = open("samples/sample.eml")

message = rfc822.Message(file)

print message.getdate("date")
print message.getaddr("from")
print message.getaddrlist("to")

(2000, 11, 14, 14, 55, 7, 0, 0, 0)
('Frank', 'your@editor')
[('Fredrik Lundh', 'fredrik@effbot.org')]

地址字段被解析为 (实际名称, 邮件地址) 这样的元组. 数据字段被解析为 9 元时间元组, 可以使用 time 模块处理.

多用途因特网邮件扩展 ( Multipurpose Internet Mail Extensions, MIME ) 标准定义了如何在 RFC 822 格式的消息中储存非 ASCII 文本, 图像以及其它数据.

File: mimetools-example-1.py

import mimetools

file = open("samples/sample.msg")

msg = mimetools.Message(file)

print "type", "=>", msg.gettype()
print "encoding", "=>", msg.getencoding()
print "plist", "=>", msg.getplist()

print "header", "=>"
for k, v in msg.items():
    print "  ", k, "=", v

type => text/plain
encoding => 7bit
plist => ['charset="iso-8859-1"']
header =>
   mime-version = 1.0
   content-type = text/plain;
 charset="iso-8859-1"
   to = effbot@spam.egg
   date = Fri, 15 Oct 1999 03:21:15 -0400
   content-transfer-encoding = 7bit
   from = "Fredrik Lundh" <fredrik@pythonware.com>
   subject = By the way...
...

File: mimewriter-example-1.py

import MimeWriter

# data encoders
# 数据编码
import quopri
import base64
import StringIO

import sys

TEXT = """
here comes the image you asked for.  hope
it's what you expected.

</F>"""

FILE = "samples/sample.jpg"

file = sys.stdout

#
# create a mime multipart writer instance

mime = MimeWriter.MimeWriter(file)
mime.addheader("Mime-Version", "1.0")

mime.startmultipartbody("mixed")

# add a text message
# 加入文字信息

part = mime.nextpart()
part.addheader("Content-Transfer-Encoding", "quoted-printable")
part.startbody("text/plain")

quopri.encode(StringIO.StringIO(TEXT), file, 0)

# add an image
# 加入图片

part = mime.nextpart()
part.addheader("Content-Transfer-Encoding", "base64")
part.startbody("image/jpeg")

base64.encode(open(FILE, "rb"), file)

mime.lastpart()

输出结果如下:

Content-Type: multipart/mixed;
    boundary='host.1.-852461.936831373.130.24813'

--host.1.-852461.936831373.130.24813
Content-Type: text/plain
Context-Transfer-Encoding: quoted-printable

here comes the image you asked for.  hope
it's what you expected.

</F>

--host.1.-852461.936831373.130.24813
Content-Type: image/jpeg
Context-Transfer-Encoding: base64

/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRof
HBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/2wBDAQkJCQwLDBgNDRgyIRwhMjIy
...
1e5vLrSYbJnEVpEgjCLx5mPU0qsVK0UaxjdNlS+1U6pfzTR8IzEhj2HrVG6m8m18xc8cIKSC
tCuFyC746j/Cq2pTia4WztfmKjGBXTCmo6IUpt==

--host.1.-852461.936831373.130.24813--

[Example 6-5 #eg-6-5 ] 使用辅助类储存每个子部分.

File: mimewriter-example-2.py

import MimeWriter
import string, StringIO, sys
import re, quopri, base64

# check if string contains non-ascii characters
must_quote = re.compile("[/177-/377]").search


#
# encoders

def encode_quoted_printable(infile, outfile):
    quopri.encode(infile, outfile, 0)

class Writer:

    def _ _init_ _(self, file=None, blurb=None):
        if file is None:
            file = sys.stdout
        self.file = file
        self.mime = MimeWriter.MimeWriter(file)
        self.mime.addheader("Mime-Version", "1.0")

        file = self.mime.startmultipartbody("mixed")
        if blurb:
            file.write(blurb)

    def close(self):
        "End of message"
        self.mime.lastpart()
        self.mime = self.file = None

    def write(self, data, mimetype="text/plain"):
        "Write data from string or file to message"

        # data is either an opened file or a string
        if type(data) is type(""):
            file = StringIO.StringIO(data)
        else:
            file = data
            data = None

        part = self.mime.nextpart()

        typ, subtyp = string.split(mimetype, "/", 1)

        if typ == "text":

            # text data
            encoding = "quoted-printable"
            encoder = lambda i, o: quopri.encode(i, o, 0)

            if data and not must_quote(data):
                # copy, don't encode
                encoding = "7bit"
                encoder = None

        else:

            # binary data (image, audio, application, ...)
            encoding = "base64"
            encoder = base64.encode

        #
        # write part headers

        if encoding:
            part.addheader("Content-Transfer-Encoding", encoding)

        part.startbody(mimetype)

        #
        # write part body

        if encoder:
            encoder(file, self.file)
        elif data:
            self.file.write(data)
        else:
            while 1:
                data = infile.read(16384)
                if not data:
                    break
                outfile.write(data)

#
# try it out

BLURB = "if you can read this, your mailer is not MIME-aware/n"

mime = Writer(sys.stdout, BLURB)

# add a text message
mime.write("""/
here comes the image you asked for.  hope
it's what you expected.
""", "text/plain")

# add an image
mime.write(open("samples/sample.jpg", "rb"), "image/jpeg")

mime.close()

File: mailbox-example-1.py

import mailbox

mb = mailbox.UnixMailbox(open("/var/spool/mail/effbot"))

while 1:
    msg = mb.next()
    if not msg:
        break
    for k, v in msg.items():
        print k, "=", v
    body = msg.fp.read()
    print len(body), "bytes in body"

subject = for he's a ...
message-id = <199910150027.CAA03202@spam.egg>
received = (from fredrik@pythonware.com)
 by spam.egg (8.8.7/8.8.5) id CAA03202
 for effbot; Fri, 15 Oct 1999 02:27:36 +0200
from = Fredrik Lundh <fredrik@pythonware.com>
date = Fri, 15 Oct 1999 12:35:36 +0200
to = effbot@spam.egg
1295 bytes in body

File: mailcap-example-1.py

import mailcap

caps = mailcap.getcaps()

for k, v in caps.items():
    print k, "=", v

image/* = [{'view': 'pilview'}]
application/postscript = [{'view': 'ghostview'}]

File: mailcap-example-2.py

import mailcap

caps = mailcap.getcaps()

command, info = mailcap.findmatch(
    caps, "image/jpeg", "view", "samples/sample.jpg"
    )

print command

pilview samples/sample.jpg

File: mimetypes-example-1.py

import mimetypes
import glob, urllib

for file in glob.glob("samples/*"):
    url = urllib.pathname2url(file)
    print file, mimetypes.guess_type(url)

samples/sample.au ('audio/basic', None)
samples/sample.ini (None, None)
samples/sample.jpg ('image/jpeg', None)
samples/sample.msg (None, None)
samples/sample.tar ('application/x-tar', None)
samples/sample.tgz ('application/x-tar', 'gzip')
samples/sample.txt ('text/plain', None)
samples/sample.wav ('audio/x-wav', None)
samples/sample.zip ('application/zip', None)

File: packmail-example-1.py

import packmail
import sys

packmail.pack(sys.stdout, "samples/sample.txt", "sample.txt")

echo sample.txt
sed "s/^X//" >sample.txt <<"!"
XWe will perhaps eventually be writing only small
Xmodules, which are identified by name as they are
Xused to build larger ones, so that devices like
Xindentation, rather than delimiters, might become
Xfeasible for expressing local structure in the
Xsource language.
X    -- Donald E. Knuth, December 1974
!

====Example 6-11. 使用 packmail 打包整个目录树===[eg-6-11]

File: packmail-example-2.py

import packmail
import sys

packmail.packtree(sys.stdout, "samples")

注意, 这个模块不能处理二进制文件, 例如声音或者图像文件.

mimify 模块用于在 MIME 编码的文本信息和普通文本信息(例如 ISO Latin 1 文本)间相互转换. 它可以用作命令行工具, 或是特定邮件代理的转换过滤器:

$ mimify.py -e raw-message mime-message
$ mimify.py -d mime-message raw-message

File: mimify-example-1.py

import mimify
import sys

mimify.unmimify("samples/sample.msg", sys.stdout, 1)

这里是一个包含两部分的 MIME 信息, 一个是引用的可打印信息, 另个是 base64 编码信息. unmimify 的第三个参数决定是否自动解码 base64 编码的部分:

MIME-Version: 1.0
Content-Type: multipart/mixed; boundary='boundary'

this is a multipart sample file.  the two
parts both contain ISO Latin 1 text, with
different encoding techniques.

--boundary
Content-Type: text/plain
Content-Transfer-Encoding: quoted-printable

sillmj=F6lke! blindstyre! medisterkorv!

--boundary
Content-Type: text/plain
Content-Transfer-Encoding: base64

a29tIG5lciBiYXJhLCBvbSBkdSB09nJzIQ==

--boundary--

解码结果如下 (可读性相对来说更好些):

MIME-Version: 1.0
Content-Type: multipart/mixed; boundary= 'boundary'

this is a multipart sample file.  the two
parts both contain ISO Latin 1 text, with
different encoding techniques.

--boundary
Content-Type: text/plain

sillmjölke! blindstyre! medisterkorv!

--boundary
Content-Type: text/plain

kom ner bara, om du törs!

File: mimify-example-2.py

import mimify
import StringIO, sys

#
# decode message into a string buffer

file = StringIO.StringIO()

mimify.unmimify("samples/sample.msg", file, 1)

#
# encode message from string buffer

file.seek(0) # rewind

mimify.mimify(file, sys.stdout)

File: multifile-example-1.py

import multifile
import cgi, rfc822

infile = open("samples/sample.msg")

message = rfc822.Message(infile)

# print parsed header
for k, v in message.items():
    print k, "=", v

# use cgi support function to parse content-type header
type, params = cgi.parse_header(message["content-type"])

if type[:10] == "multipart/":

    # multipart message
    boundary = params["boundary"]

    file = multifile.MultiFile(infile)

    file.push(boundary)

    while file.next():

        submessage = rfc822.Message(file)

        # print submessage
        print "-" * 68
        for k, v in submessage.items():
            print k, "=", v
        print
        print file.read()

    file.pop()

else:

    # plain message
    print infile.read()

"Increasingly, people seem to misinterpret complexity as sophistication, which is baffling - the incomprehensible should cause suspicion rather than admiration. Possibly this trend results from a mistaken belief that using a somewhat mysterious device confers an aura of power on the user."

- Niklaus Wirth

本章描述了 Python 的 socket 协议支持以及其他建立在 socket 模块上的网络 模块. 这些包含了对大多流行 Internet 协议客户端的支持, 以及一些可用来 实现 Internet 服务器的框架.

对于那些本章中的底层的例子, 我将使用两个协议作为样例: Internet Time Protocol ( Internet 时间协议 ) 以及 Hypertext Transfer Protocol (超文本传输协议, HTTP 协议).

Internet 时间协议 ( RFC 868, Postel 和 Harrenstien, 1983) 可以让 一个网络客户端获得一个服务器的当前时间.

因为这个协议是轻量级的, 许多 Unix 系统(但不是所有)都提供了这个服务. 它可能是最简单的网络协议了. 服务器等待连接请求并在连接后返回当前时间 ( 4 字节整数, 自从 1900 年 1 月 1 日到当前的秒数).

协议很简单, 这里我们提供规格书给大家:

File: rfc868.txt

Network Working Group                                    J. Postel - ISI
Request for Comments: 868                           K. Harrenstien - SRI
                                                                May 1983

                             Time Protocol

This RFC specifies a standard for the ARPA Internet community.  Hosts on
the ARPA Internet that choose to implement a Time Protocol are expected
to adopt and implement this standard.

本 RFC 规范提供了一个 ARPA Internet community 上的标准. 
在 ARPA Internet 上的所有主机应当采用并实现这个标准.

This protocol provides a site-independent, machine readable date and
time.  The Time service sends back to the originating source the time in
seconds since midnight on January first 1900.

此协议提供了一个独立于站点的, 机器可读的日期和时间信息. 
时间服务返回的是从 1900 年 1 月 1 日午夜到现在的秒数.

One motivation arises from the fact that not all systems have a
date/time clock, and all are subject to occasional human or machine
error.  The use of time-servers makes it possible to quickly confirm or
correct a system's idea of the time, by making a brief poll of several
independent sites on the network.

设计这个协议的一个重要目的在于, 网络上的一些主机并没有时钟, 
这有可能导致人工或者机器错误. 我们可以依靠时间服务器快速确认或者修改
一个系统的时间.

This protocol may be used either above the Transmission Control Protocol
(TCP) or above the User Datagram Protocol (UDP).

该协议可以用在 TCP 协议或是 UDP 协议上.

When used via TCP the time service works as follows:

通过 TCP 访问时间服务器的步骤:

 *  S: Listen on port 37 (45 octal).
 *  U: Connect to port 37.
 *  S: Send the time as a 32 bit binary number.
 *  U: Receive the time.
 *  U: Close the connection.
 *  S: Close the connection.
 
 *  S: 监听 37 ( 45 的八进制) 端口.
 *  U: 连接 37 端口.
 *  S: 将时间作为 32 位二进制数字发送.
 *  U: 接收时间.
 *  U: 关闭连接.
 *  S: 关闭连接.  

   The server listens for a connection on port 37.  When the connection
   is established, the server returns a 32-bit time value and closes the
   connection.  If the server is unable to determine the time at its
   site, it should either refuse the connection or close it without
   sending anything.
   
   服务器在 37 端口监听. 当连接建立的时候, 服务器返回一个 32 位的数字值
   并关闭连接. 如果服务器自己无法决定当前时间, 那么它应该拒绝这个连接或者
   不发送任何数据立即关闭连接.

When used via UDP the time service works as follows:

通过 TCP 访问时间服务器的步骤:
   
   S: Listen on port 37 (45 octal).
   U: Send an empty datagram to port 37.
   S: Receive the empty datagram.
   S: Send a datagram containing the time as a 32 bit binary number.
   U: Receive the time datagram.

   S: 监听 37 ( 45 的八进制) 端口.
   U: 发送空数据报文到 37 端口.
   S: 接受空报文.
   S: 发送包含时间( 32 位二进制数字 )的报文.
   U: 接受时间报文.
   
   The server listens for a datagram on port 37.  When a datagram
   arrives, the server returns a datagram containing the 32-bit time
   value.  If the server is unable to determine the time at its site, it
   should discard the arriving datagram and make no reply.
   
   服务器在 37 端口监听报文. 当报文到达时, 服务器返回包含 32 位时间值
   的报文. 如果服务器无法决定当前时间, 那么它应该丢弃到达的报文, 
   不做任何回复.

The Time

时间

The time is the number of seconds since 00:00 (midnight) 1 January 1900
GMT, such that the time 1 is 12:00:01 am on 1 January 1900 GMT; this
base will serve until the year 2036.

时间是自 1900 年 1 月 1 日 0 时到当前的秒数, 
这个协议标准会一直服务到2036年. 到时候数字不够用再说.

For example:

   the time  2,208,988,800 corresponds to 00:00  1 Jan 1970 GMT,
             2,398,291,200 corresponds to 00:00  1 Jan 1976 GMT,
             2,524,521,600 corresponds to 00:00  1 Jan 1980 GMT,
             2,629,584,000 corresponds to 00:00  1 May 1983 GMT,
        and -1,297,728,000 corresponds to 00:00 17 Nov 1858 GMT.
		

例如:

     时间值  2,208,988,800 对应 to 00:00  1 Jan 1970 GMT,
             2,398,291,200 对应 to 00:00  1 Jan 1976 GMT,
             2,524,521,600 对应 to 00:00  1 Jan 1980 GMT,
             2,629,584,000 对应 to 00:00  1 May 1983 GMT,
       最后 -1,297,728,000 对应 to 00:00 17 Nov 1858 GMT.

	   
RFC868.txt Translated By Andelf(gt: andelf@gmail.com )
非商业用途, 转载请保留作者信息. Thx.

超文本传输协议 ( HTTP, RFC 2616 ) 是另个完全不同的东西. 最近的格式说明书( Version 1.1 )超过了 100 页.

从它最简单的格式来看, 这个协议是很简单的. 客户端发送如下的请求到服务器, 请求一个文件:

GET /hello.txt HTTP/1.0
Host: hostname
User-Agent: name

[optional request body , 可选的请求正文]

服务器返回对应的响应:

HTTP/1.0 200 OK
Content-Type: text/plain
Content-Length: 7

Hello

请求和响应的 headers (报头)一般会包含更多的域, 但是请求 header 中的 Host 域/字段是必须提供的.

header 行使用 "/r/n" 分割, 而且 header 后必须有一个空行, 即使没有正文 (请求和响应都必须符合这条规则).

socket 模块实现了到 socket 通讯层的接口. 你可以使用该模块创建 客户端或是服务器的 socket .

File: socket-example-1.py

import socket
import struct, time

# server
HOST = "www.python.org"
PORT = 37

# reference time (in seconds since 1900-01-01 00:00:00)
TIME1970 = 2208988800L # 1970-01-01 00:00:00

# connect to server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))

# read 4 bytes, and convert to time value
t = s.recv(4)
t = struct.unpack("!I", t)[0]
t = int(t - TIME1970)

s.close()

# print results
print "server time is", time.ctime(t)
print "local clock is", int(time.time()) - t, "seconds off"

server time is Sat Oct 09 16:42:36 1999
local clock is 8 seconds off

socket 工厂函数( factory function )根据给定类型(该例子中为 Internet stream socket , 即就是 TCP socket )创建一个新的 socket . connect 方法尝试将这个 socket 连接到指定服务器上. 成功后, 就可以使用 recv 方法读取数据.

创建一个服务器 socket 使用的是相同的方法, 不过这里不是连接到服务器, 而是将 socket bind (绑定)到本机的一个端口上, 告诉它去监听连接请求, 然后尽快处理每个到达的请求.

File: socket-example-2.py

import socket
import struct, time

# user-accessible port
PORT = 8037

# reference time
TIME1970 = 2208988800L

# establish server
service = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
service.bind(("", PORT))
service.listen(1)

print "listening on port", PORT

while 1:
    # serve forever
    channel, info = service.accept()
    print "connection from", info
    t = int(time.time()) + TIME1970
    t = struct.pack("!I", t)
    channel.send(t) # send timestamp
    channel.close() # disconnect

listening on port 8037
connection from ('127.0.0.1', 1469)
connection from ('127.0.0.1', 1470)
...

listen 函数的调用告诉 socket 我们期望接受连接. 参数代表连接 的队列(用于在程序没有处理前保持连接)大小. 最后 accept 循环将当前时间返回 给每个连接的客户端.

注意这里的 accept 函数返回一个新的 socket 对象, 这个对象是直接连接到客户端 的. 而原 socket 只是用来保持连接; 所有后来的数据传输操作都使用新的 socket .

File: timeclient.py

import socket
import struct, sys, time

# default server
host = "localhost"
port = 8037

# reference time (in seconds since 1900-01-01 00:00:00)
TIME1970 = 2208988800L # 1970-01-01 00:00:00

def gettime(host, port):
    # fetch time buffer from stream server
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((host, port))
    t = s.recv(4)
    s.close()
    t = struct.unpack("!I", t)[0]
    return int(t - TIME1970)

if _ _name_ _ == "_ _main_ _":
    # command-line utility
    if sys.argv[1:]:
        host = sys.argv[1]
        if sys.argv[2:]:
            port = int(sys.argv[2])
        else:
            port = 37 # default for public servers

    t = gettime(host, port)
    print "server time is", time.ctime(t)
    print "local clock is", int(time.time()) - t, "seconds off"

server time is Sat Oct 09 16:58:50 1999
local clock is 0 seconds off

File: socket-example-4.py

import socket
import struct, time

# server
HOST = "localhost"
PORT = 8037

# reference time (in seconds since 1900-01-01 00:00:00)
TIME1970 = 2208988800L # 1970-01-01 00:00:00

# connect to server
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# send empty packet
s.sendto("", (HOST, PORT))

# read 4 bytes from server, and convert to time value
t, server = s.recvfrom(4)
t = struct.unpack("!I", t)[0]
t = int(t - TIME1970)

s.close()

print "server time is", time.ctime(t)
print "local clock is", int(time.time()) - t, "seconds off"

server time is Sat Oct 09 16:42:36 1999
local clock is 8 seconds off

这里的 recvfrom 返回两个值: 数据和发送者的地址. 后者用于发送回复数据.

Example 7-5 展示了对应的服务器代码.

Example 7-5. 使用 socket 模块实现一个报文时间服务器

File: socket-example-5.py

import socket
import struct, time

# user-accessible port
PORT = 8037

# reference time
TIME1970 = 2208988800L

# establish server
service = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
service.bind(("", PORT))

print "listening on port", PORT

while 1:
    # serve forever
    data, client = service.recvfrom(0)
    print "connection from", client
    t = int(time.time()) + TIME1970
    t = struct.pack("!I", t)
    service.sendto(t, client) # send timestamp

listening on port 8037
connection from ('127.0.0.1', 1469)
connection from ('127.0.0.1', 1470)
...

最主要的不同在于服务器使用 bind 来分配一个已知端口给 socket , 根据 recvfrom 函数返回的地址向客户端发送数据.

select 模块允许你检查一个或多个 socket , 管道, 以及其他流兼容对象所接受的数据, 如 Example 7-6 所示.

你可以将一个或更多 socket 传递给 select 函数, 然后等待它们状态改变(可读, 可写, 或是发送错误信号):

• 如果某人在调用了 listen 函数后连接, 当远端数据到达时, socket 就成为可读的(这意味着 accept 不会阻塞). 或者是 socket 被关闭或重置时(在此情况下, recv 会返回一个空字符串).
• 当非阻塞调用 connect 方法后建立连接或是数据可以被写入到 socket 时, socket 就成为可写的.
• 当非阻塞调用 connect 方法后连接失败后, socket 会发出一个错误信号.

File: select-example-1.py

import select
import socket
import time

PORT = 8037

TIME1970 = 2208988800L

service = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
service.bind(("", PORT))
service.listen(1)

print "listening on port", PORT

while 1:
    is_readable = [service]
    is_writable = []
    is_error = []
    r, w, e = select.select(is_readable, is_writable, is_error, 1.0)
    if r:
        channel, info = service.accept()
        print "connection from", info
        t = int(time.time()) + TIME1970
        t = chr(t>>24&255) + chr(t>>16&255) + chr(t>>8&255) + chr(t&255)
        channel.send(t) # send timestamp
        channel.close() # disconnect
    else:
        print "still waiting"

listening on port 8037
still waiting
still waiting
connection from ('127.0.0.1', 1469)
still waiting
connection from ('127.0.0.1', 1470)
...

asyncore 模块提供了一个 "反馈性的( reactive )" socket 实现. 该模块允许你定义特定过程完成后所执行的代码, 而不是创建 socket 对象, 调用它们的方法. 你只需要继承 dispatcher 类, 然后重载如下方法 (可以选择重载某一个或多个)就可以实现异步的 socket 处理器.

• handle_connect : 一个连接成功建立后被调用.
• handle_expt : 连接失败后被调用.
• handle_accept : 连接请求建立到一个监听 socket 上时被调用. 回调时( callback )应该使用 accept 方法来获得客户端 socket .
• handle_read : 有来自 socket 的数据等待读取时被调用. 回调时应该使用 recv 方法来获得数据.
• handle_write : socket 可以写入数据的时候被调用. 使用 send 方法写入数据.
• handle_close : 当 socket 被关闭或复位时被调用.
• handle_error(type, value, traceback) 在任何一个回调函数发生 Python 错误时被调用. 默认的实现会打印跟踪返回消息到 sys.stdout .

File: asyncore-example-1.py

import asyncore
import socket, time

# reference time (in seconds since 1900-01-01 00:00:00)
TIME1970 = 2208988800L # 1970-01-01 00:00:00

class TimeRequest(asyncore.dispatcher):
    # time requestor (as defined in RFC 868)

    def _ _init_ _(self, host, port=37):
        asyncore.dispatcher._ _init_ _(self)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connect((host, port))

    def writable(self):
        return 0 # don't have anything to write

    def handle_connect(self):
        pass # connection succeeded

    def handle_expt(self):
        self.close() # connection failed, shutdown

    def handle_read(self):
        # get local time
        here = int(time.time()) + TIME1970

        # get and unpack server time
        s = self.recv(4)
        there = ord(s[3]) + (ord(s[2])<<8) + (ord(s[1])<<16) + (ord(s[0])<<24L)

        self.adjust_time(int(here - there))

        self.handle_close() # we don't expect more data

    def handle_close(self):
        self.close()

    def adjust_time(self, delta):
        # override this method!
        print "time difference is", delta

#
# try it out

request = TimeRequest("www.python.org")

asyncore.loop()

log: adding channel <TimeRequest  at 8cbe90>
time difference is 28
log: closing channel 192:<TimeRequest connected at 8cbe90>

如果你不想记录任何信息, 那么你可以在你的 dispatcher 类里重载 log 方法.

File: asyncore-example-2.py

import asyncore
import socket, time

# reference time
TIME1970 = 2208988800L

class TimeChannel(asyncore.dispatcher):

    def handle_write(self):
        t = int(time.time()) + TIME1970
        t = chr(t>>24&255) + chr(t>>16&255) + chr(t>>8&255) + chr(t&255)
        self.send(t)
        self.close()

class TimeServer(asyncore.dispatcher):

    def _ _init_ _(self, port=37):
        self.port = port
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.bind(("", port))
        self.listen(5)
        print "listening on port", self.port

    def handle_accept(self):
        channel, addr = self.accept()
        TimeChannel(channel)

server = TimeServer(8037)
asyncore.loop()

log: adding channel <TimeServer  at 8cb940>
listening on port 8037
log: adding channel <TimeChannel  at 8b2fd0>
log: closing channel 52:<TimeChannel connected at 8b2fd0>

除了 dispatcher 外, 这个模块还包含一个 dispatcher_with_send 类. 你可以使用这个类发送大量的数据而不会阻塞网络通讯缓冲区.

File: SimpleAsyncHTTP.py

import asyncore
import string, socket
import StringIO
import mimetools, urlparse

class AsyncHTTP(asyncore.dispatcher_with_send):
    # HTTP requester

    def _ _init_ _(self, uri, consumer):
        asyncore.dispatcher_with_send._ _init_ _(self)

        self.uri = uri
        self.consumer = consumer

        # turn the uri into a valid request
        scheme, host, path, params, query, fragment = urlparse.urlparse(uri)
        assert scheme == "http", "only supports HTTP requests"
        try:
            host, port = string.split(host, ":", 1)
            port = int(port)
        except (TypeError, ValueError):
            port = 80 # default port
        if not path:
            path = "/"
        if params:
            path = path + ";" + params
        if query:
            path = path + "?" + query

        self.request = "GET %s HTTP/1.0/r/nHost: %s/r/n/r/n" % (path, host)

        self.host = host
        self.port = port

        self.status = None
        self.header = None

        self.data = ""

        # get things going!
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connect((host, port))

    def handle_connect(self):
        # connection succeeded
        self.send(self.request)

    def handle_expt(self):
        # connection failed; notify consumer (status is None)
        self.close()
        try:
            http_header = self.consumer.http_header
        except AttributeError:
            pass
        else:
            http_header(self)

    def handle_read(self):
        data = self.recv(2048)
        if not self.header:
            self.data = self.data + data
            try:
                i = string.index(self.data, "/r/n/r/n")
            except ValueError:
                return # continue
            else:
                # parse header
                fp = StringIO.StringIO(self.data[:i+4])
                # status line is "HTTP/version status message"
                status = fp.readline()
                self.status = string.split(status, " ", 2)
                # followed by a rfc822-style message header
                self.header = mimetools.Message(fp)
                # followed by a newline, and the payload (if any)
                data = self.data[i+4:]
                self.data = ""
                # notify consumer (status is non-zero)
                try:
                    http_header = self.consumer.http_header
                except AttributeError:
                    pass
                else:
                    http_header(self)
                if not self.connected:
                    return # channel was closed by consumer

        self.consumer.feed(data)

    def handle_close(self):
        self.consumer.close()
        self.close()

File: asyncore-example-3.py

import SimpleAsyncHTTP
import asyncore

class DummyConsumer:
    size = 0

    def http_header(self, request):
        # handle header
        if request.status is None:
            print "connection failed"
        else:
            print "status", "=>", request.status
            for key, value in request.header.items():
                print key, "=", value

    def feed(self, data):
        # handle incoming data
        self.size = self.size + len(data)

    def close(self):
        # end of data
        print self.size, "bytes in body"

#
# try it out

consumer = DummyConsumer()

request = SimpleAsyncHTTP.AsyncHTTP(
    "http://www.pythonware.com",
    consumer
    )

asyncore.loop()

log: adding channel <AsyncHTTP  at 8e2850>
status => ['HTTP/1.1', '200', 'OK/015/012']
server = Apache/Unix (Unix)
content-type = text/html
content-length = 3730
...
3730 bytes in body
log: closing channel 156:<AsyncHTTP connected at 8e2850>

这里的 consumer 接口设计时是为了与 htmllib 和 xmllib 分析器兼容的, 这样你就可以直接方便地解析 HTML 或是 XML 数据. http_header 方法是可选的; 如果没有定义它, 那么它将被忽略.

File: asyncore-example-4.py

import SimpleAsyncHTTP
import asyncore

class DummyConsumer:
    size = 0

    def http_header(self, request):
        # handle header
        if request.status is None:
            print "connection failed"
        else:
            print "status", "=>", request.status
            for key, value in request.header.items():
                print key, "=", value

    def feed(self, data):
        # handle incoming data
        self.size = self.size + len(data)

    def close(self):
        # end of data
        print self.size, "bytes in body"

class RedirectingConsumer:

    def _ _init_ _(self, consumer):
        self.consumer = consumer

    def http_header(self, request):
        # handle header
        if request.status is None or/
           request.status[1] not in ("301", "302"):
            try:
                http_header = self.consumer.http_header
            except AttributeError:
                pass
            else:
                return http_header(request)
        else:
            # redirect!
            uri = request.header["location"]
            print "redirecting to", uri, "..."
            request.close()
            SimpleAsyncHTTP.AsyncHTTP(uri, self)

    def feed(self, data):
        self.consumer.feed(data)

    def close(self):
        self.consumer.close()

#
# try it out

consumer = RedirectingConsumer(DummyConsumer())

request = SimpleAsyncHTTP.AsyncHTTP(
    "http://www.pythonware.com/library",
    consumer
    )

asyncore.loop()

log: adding channel <AsyncHTTP  at 8e64b0>
redirecting to http://www.pythonware.com/library/ ...
log: closing channel 48:<AsyncHTTP connected at 8e64b0>
log: adding channel <AsyncHTTP  at 8ea790>
status => ['HTTP/1.1', '200', 'OK/015/012']
server = Apache/Unix (Unix)
content-type = text/html
content-length = 387
...
387 bytes in body
log: closing channel 236:<AsyncHTTP connected at 8ea790>

如果服务器返回状态 301 (永久重定向) 或者是 302 (临时重定向), 重定向的 consumer 会关闭当前请求并向新地址发出新请求. 所有对 consumer 的其他调用传递给原来的 consumer .

asynchat 模块是对 asyncore 的一个扩展. 它提供对面向行( line-oriented )的协议的额外支持. 它还提供了增强的缓冲区支持(通过 push 方法和 "producer" 机制.

File: asynchat-example-1.py

import asyncore, asynchat
import os, socket, string

PORT = 8000

class HTTPChannel(asynchat.async_chat):

    def _ _init_ _(self, server, sock, addr):
        asynchat.async_chat._ _init_ _(self, sock)
        self.set_terminator("/r/n")
        self.request = None
        self.data = ""
        self.shutdown = 0

    def collect_incoming_data(self, data):
        self.data = self.data + data

    def found_terminator(self):
        if not self.request:
            # got the request line
            self.request = string.split(self.data, None, 2)
            if len(self.request) != 3:
                self.shutdown = 1
            else:
                self.push("HTTP/1.0 200 OK/r/n")
                self.push("Content-type: text/html/r/n")
                self.push("/r/n")
            self.data = self.data + "/r/n"
            self.set_terminator("/r/n/r/n") # look for end of headers
        else:
            # return payload.
            self.push("<html><body><pre>/r/n")
            self.push(self.data)
            self.push("</pre></body></html>/r/n")
            self.close_when_done()

class HTTPServer(asyncore.dispatcher):

    def _ _init_ _(self, port):
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.bind(("", port))
        self.listen(5)

    def handle_accept(self):
        conn, addr = self.accept()
        HTTPChannel(self, conn, addr)

#
# try it out

s = HTTPServer(PORT)
print "serving at port", PORT, "..."
asyncore.loop()

GET / HTTP/1.1
Accept: */*
Accept-Language: en, sv
Accept-Encoding: gzip, deflate
User-Agent: Mozilla/4.0 (compatible; Bruce/1.0)
Host: localhost:8000
Connection: Keep-Alive

producer 接口允许你传入( "push" )太大以至于无法在内存中储存的对象. asyncore 在需要更多数据的时候自动调用 producer 的 more 方法. 另外, 它使用一个空字符串标记文件的末尾.

File: asynchat-example-2.py

import asyncore, asynchat
import os, socket, string, sys
import StringIO, mimetools

ROOT = "."

PORT = 8000

class HTTPChannel(asynchat.async_chat):

    def _ _init_ _(self, server, sock, addr):
        asynchat.async_chat._ _init_ _(self, sock)
        self.server = server
        self.set_terminator("/r/n/r/n")
        self.header = None
        self.data = ""
        self.shutdown = 0

    def collect_incoming_data(self, data):
        self.data = self.data + data
        if len(self.data) > 16384:
            # limit the header size to prevent attacks
            self.shutdown = 1

    def found_terminator(self):
        if not self.header:
            # parse http header
            fp = StringIO.StringIO(self.data)
            request = string.split(fp.readline(), None, 2)
            if len(request) != 3:
                # badly formed request; just shut down
                self.shutdown = 1
            else:
                # parse message header
                self.header = mimetools.Message(fp)
                self.set_terminator("/r/n")
                self.server.handle_request(
                    self, request[0], request[1], self.header
                    )
                self.close_when_done()
            self.data = ""
        else:
            pass # ignore body data, for now

    def pushstatus(self, status, explanation="OK"):
        self.push("HTTP/1.0 %d %s/r/n" % (status, explanation))


class FileProducer:
    # a producer that reads data from a file object

    def _ _init_ _(self, file):
        self.file = file

    def more(self):
        if self.file:
            data = self.file.read(2048)
            if data:
                return data
            self.file = None
        return ""


class HTTPServer(asyncore.dispatcher):

    def _ _init_ _(self, port=None, request=None):
        if not port:
            port = 80
        self.port = port
        if request:
            self.handle_request = request # external request handler
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.bind(("", port))
        self.listen(5)

    def handle_accept(self):
        conn, addr = self.accept()
        HTTPChannel(self, conn, addr)

    def handle_request(self, channel, method, path, header):
        try:
            # this is not safe!
            while path[:1] == "/":
                path = path[1:]
            filename = os.path.join(ROOT, path)
            print path, "=>", filename
            file = open(filename, "r")
        except IOError:
            channel.pushstatus(404, "Not found")
            channel.push("Content-type: text/html/r/n")
            channel.push("/r/n")
            channel.push("<html><body>File not found.</body></html>/r/n")
        else:
            channel.pushstatus(200, "OK")
            channel.push("Content-type: text/html/r/n")
            channel.push("/r/n")
            channel.push_with_producer(FileProducer(file))

#
# try it out

s = HTTPServer(PORT)
print "serving at port", PORT
asyncore.loop()

serving at port 8000
log: adding channel <HTTPServer  at 8e54d0>
log: adding channel <HTTPChannel  at 8e64a0>
samples/sample.htm => ./samples/sample.htm
log: closing channel 96:<HTTPChannel connected at 8e64a0>

urlib 模块为 HTTP , FTP , 以及 gopher 提供了一个统一的客户端接口. 它会自动地根据 URL 选择合适的协议处理器.

File: urllib-example-1.py

import urllib

fp = urllib.urlopen("http://www.python.org")

op = open("out.html", "wb")

n = 0

while 1:
    s = fp.read(8192)
    if not s:
        break
    op.write(s)
    n = n + len(s)

fp.close()
op.close()

for k, v in fp.headers.items():
    print k, "=", v

print "copied", n, "bytes from", fp.url

server = Apache/1.3.6 (Unix)
content-type = text/html
accept-ranges = bytes
date = Mon, 11 Oct 1999 20:11:40 GMT
connection = close
etag = "741e9-7870-37f356bf"
content-length = 30832
last-modified = Thu, 30 Sep 1999 12:25:35 GMT
copied 30832 bytes from http://www.python.org

这个流对象提供了一些非标准的属性. headers 是一个 Message 对象(在 mimetools 模块中定义), url 是实际的 URL . 后者会根据服务器的重定向而更新.

File: urllib-example-3.py

import urllib

class myURLOpener(urllib.FancyURLopener):
    # read an URL, with automatic HTTP authentication

    def setpasswd(self, user, passwd):
        self._ _user = user
        self._ _passwd = passwd

    def prompt_user_passwd(self, host, realm):
        return self._ _user, self._ _passwd

urlopener = myURLOpener()
urlopener.setpasswd("mulder", "trustno1")

fp = urlopener.open("http://www.secretlabs.com")
print fp.read()

File: urlparse-example-1.py

import urlparse

print urlparse.urlparse("http://host/path;params?query#fragment")

('http', 'host', '/path', 'params', 'query', 'fragment')

File: urlparse-example-2.py

import urlparse

scheme, host, path, params, query, fragment =/
        urlparse.urlparse("http://host/path;params?query#fragment")

if scheme == "http":
    print "host", "=>", host
    if params:
        path = path + ";" + params
    if query:
        path = path + "?" + query
    print "path", "=>", path

host => host
path => /path;params?query

File: urlparse-example-3.py

import urlparse

scheme, host, path, params, query, fragment =/
        urlparse.urlparse("http://host/path;params?query#fragment")

if scheme == "http":
    print "host", "=>", host
    print "path", "=>", urlparse.urlunparse(
    (None, None, path, params, query, None)
    )

host => host
path => /path;params?query

File: urlparse-example-4.py

import urlparse

base = "http://spam.egg/my/little/pony"

for path in "/index", "goldfish", "../black/cat":
    print path, "=>", urlparse.urljoin(base, path)

/index => http://spam.egg/index
goldfish => http://spam.egg/my/little/goldfish
../black/cat => http://spam.egg/my/black/cat

File: cookie-example-1.py

import Cookie
import os, time

cookie = Cookie.SimpleCookie()
cookie["user"] = "Mimi"
cookie["timestamp"] = time.time()

print cookie

# simulate CGI roundtrip
os.environ["HTTP_COOKIE"] = str(cookie)

print

cookie = Cookie.SmartCookie()
cookie.load(os.environ["HTTP_COOKIE"])

for key, item in cookie.items():
    # dictionary items are "Morsel" instances
    # use value attribute to get actual value
    print key, repr(item.value)


Set-Cookie: timestamp=736513200;
Set-Cookie: user=Mimi;

user 'Mimi'
timestamp '736513200'

File: robotparser-example-1.py

import robotparser

r = robotparser.RobotFileParser()
r.set_url("http://www.python.org/robots.txt")
r.read()

if r.can_fetch("*", "/index.html"):
    print "may fetch the home page"

if r.can_fetch("*", "/tim_one/index.html"):
    print "may fetch the tim peters archive"

may fetch the home page

ftplib 模块包含了一个 File Transfer Protocol (FTP , 文件传输协议)客户端的实现.

File: ftplib-example-1.py

import ftplib

ftp = ftplib.FTP("www.python.org")
ftp.login("anonymous", "ftplib-example-1")

print ftp.dir()

ftp.quit()

total 34
drwxrwxr-x  11 root     4127         512 Sep 14 14:18 .
drwxrwxr-x  11 root     4127         512 Sep 14 14:18 ..
drwxrwxr-x   2 root     4127         512 Sep 13 15:18 RCS
lrwxrwxrwx   1 root     bin           11 Jun 29 14:34 README -> welcome.msg
drwxr-xr-x   3 root     wheel        512 May 19  1998 bin
drwxr-sr-x   3 root     1400         512 Jun  9  1997 dev
drwxrwxr--   2 root     4127         512 Feb  8  1998 dup
drwxr-xr-x   3 root     wheel        512 May 19  1998 etc
...

File: ftplib-example-2.py

import ftplib
import sys

def gettext(ftp, filename, outfile=None):
    # fetch a text file
    if outfile is None:
        outfile = sys.stdout
    # use a lambda to add newlines to the lines read from the server
    ftp.retrlines("RETR " + filename, lambda s, w=outfile.write: w(s+"/n"))

def getbinary(ftp, filename, outfile=None):
    # fetch a binary file
    if outfile is None:
        outfile = sys.stdout
    ftp.retrbinary("RETR " + filename, outfile.write)

ftp = ftplib.FTP("www.python.org")
ftp.login("anonymous", "ftplib-example-2")

gettext(ftp, "README")
getbinary(ftp, "welcome.msg")

WELCOME to python.org, the Python programming language home site.

You are number %N of %M allowed users.  Ni!

Python Web site: http://www.python.org/

CONFUSED FTP CLIENT?  Try begining your login password with '-' dash.
This turns off continuation messages that may be confusing your client.
...

File: ftplib-example-3.py

import ftplib
import os

def upload(ftp, file):
    ext = os.path.splitext(file)[1]
    if ext in (".txt", ".htm", ".html"):
        ftp.storlines("STOR " + file, open(file))
    else:
        ftp.storbinary("STOR " + file, open(file, "rb"), 1024)

ftp = ftplib.FTP("ftp.fbi.gov")
ftp.login("mulder", "trustno1")

upload(ftp, "trixie.zip")
upload(ftp, "file.txt")
upload(ftp, "sightings.jpg")

File: gopherlib-example-1.py

import gopherlib

host = "gopher.spam.egg"

f = gopherlib.send_selector("1/", host)
for item in gopherlib.get_directory(f):
    print item

['0', "About Spam.Egg's Gopher Server", "0/About's Spam.Egg's
Gopher Server", 'gopher.spam.egg', '70', '+']
['1', 'About Spam.Egg', '1/Spam.Egg', 'gopher.spam.egg', '70', '+']
['1', 'Misc', '1/Misc', 'gopher.spam.egg', '70', '+']
...

File: httplib-example-1.py

import httplib

USER_AGENT = "httplib-example-1.py"

class Error:
    # indicates an HTTP error
    def _ _init_ _(self, url, errcode, errmsg, headers):
        self.url = url
        self.errcode = errcode
        self.errmsg = errmsg
        self.headers = headers
    def _ _repr_ _(self):
        return (
            "<Error for %s: %s %s>" %
            (self.url, self.errcode, self.errmsg)
            )

class Server:

    def _ _init_ _(self, host):
        self.host = host

    def fetch(self, path):
        http = httplib.HTTP(self.host)

        # write header
        http.putrequest("GET", path)
        http.putheader("User-Agent", USER_AGENT)
        http.putheader("Host", self.host)
        http.putheader("Accept", "*/*")
        http.endheaders()

        # get response
        errcode, errmsg, headers = http.getreply()

        if errcode != 200:
            raise Error(errcode, errmsg, headers)

        file = http.getfile()
        return file.read()

if _ _name_ _ == "_ _main_ _":

    server = Server("www.pythonware.com")
    print server.fetch("/index.htm")

注意 httplib 提供的 HTTP 客户端在等待服务器回复的时候会阻塞程序. 异步的解决方法请参阅 asyncore 模块中的例子.

File: httplib-example-2.py

import httplib

USER_AGENT = "httplib-example-2.py"

def post(host, path, data, type=None):

    http = httplib.HTTP(host)

    # write header
    http.putrequest("PUT", path)
    http.putheader("User-Agent", USER_AGENT)
    http.putheader("Host", host)
    if type:
        http.putheader("Content-Type", type)
    http.putheader("Content-Length", str(len(size)))
    http.endheaders()

    # write body
    http.send(data)

    # get response
    errcode, errmsg, headers = http.getreply()

    if errcode != 200:
        raise Error(errcode, errmsg, headers)

    file = http.getfile()
    return file.read()

if _ _name_ _ == "_ _main_ _":

    post("www.spam.egg", "/bacon.htm", "a piece of data", "text/plain")

File: poplib-example-1.py

import poplib
import string, random
import StringIO, rfc822

SERVER = "pop.spam.egg"

USER  = "mulder"
PASSWORD = "trustno1"

# connect to server
server = poplib.POP3(SERVER)

# login
server.user(USER)
server.pass_(PASSWORD)

# list items on server
resp, items, octets = server.list()

# download a random message
id, size = string.split(random.choice(items))
resp, text, octets = server.retr(id)

text = string.join(text, "/n")
file = StringIO.StringIO(text)

message = rfc822.Message(file)

for k, v in message.items():
    print k, "=", v

print message.fp.read()

subject = ANN: (the eff-bot guide to) The Standard Python Library
message-id = <199910120808.KAA09206@spam.egg>
received = (from fredrik@spam.egg)
 by spam.egg (8.8.7/8.8.5) id KAA09206
 for mulder; Tue, 12 Oct 1999 10:08:47 +0200
from = Fredrik Lundh <fredrik@spam.egg>
date = Tue, 12 Oct 1999 10:08:47 +0200
to = mulder@spam.egg

...

File: imaplib-example-1.py

import imaplib
import string, random
import StringIO, rfc822

SERVER = "imap.spam.egg"

USER  = "mulder"
PASSWORD = "trustno1"

# connect to server
server = imaplib.IMAP4(SERVER)

# login
server.login(USER, PASSWORD)
server.select()

# list items on server
resp, items = server.search(None, "ALL")
items = string.split(items[0])

# fetch a random item
id = random.choice(items)
resp, data = server.fetch(id, "(RFC822)")
text = data[0][1]

file = StringIO.StringIO(text)

message = rfc822.Message(file)

for k, v in message.items():
    print k, "=", v

print message.fp.read()

server.logout()

subject = ANN: (the eff-bot guide to) The Standard Python Library
message-id = <199910120816.KAA12177@larch.spam.egg>
to = mulder@spam.egg
date = Tue, 12 Oct 1999 10:16:19 +0200 (MET DST)
from = <effbot@spam.egg>
received = (effbot@spam.egg) by imap.algonet.se (8.8.8+Sun/8.6.12)
id KAA12177 for effbot@spam.egg; Tue, 12 Oct 1999 10:16:19 +0200
(MET DST)

body text for test 5

File: smtplib-example-1.py

import smtplib
import string, sys

HOST = "localhost"

FROM = "effbot@spam.egg"
TO = "fredrik@spam.egg"

SUBJECT = "for your information!"

BODY = "next week: how to fling an otter"

body = string.join((
    "From: %s" % FROM,
    "To: %s" % TO,
    "Subject: %s" % SUBJECT,
    "",
    BODY), "/r/n")

print body

server = smtplib.SMTP(HOST)
server.sendmail(FROM, [TO], body)
server.quit()

From: effbot@spam.egg
To: fredrik@spam.egg
Subject: for your information!

next week: how to fling an otter

telnetlib 模块提供了一个 telnet 客户端实现.

File: telnetlib-example-1.py

import telnetlib
import sys

HOST = "spam.egg"

USER = "mulder"
PASSWORD = "trustno1"

telnet = telnetlib.Telnet(HOST)

telnet.read_until("login: ")
telnet.write(USER + "/n")

telnet.read_until("Password: ")
telnet.write(PASSWORD + "/n")

telnet.write("ls librarybook/n")
telnet.write("exit/n")

print telnet.read_all()

[spam.egg mulder]$ ls
README                                 os-path-isabs-example-1.py
SimpleAsyncHTTP.py                     os-path-isdir-example-1.py
aifc-example-1.py                      os-path-isfile-example-1.py
anydbm-example-1.py                    os-path-islink-example-1.py
array-example-1.py                     os-path-ismount-example-1.py
...

nntplib 模块提供了一个网络新闻传输协议( Network News Transfer Protocol, NNTP )客户端的实现.

File: nntplib-example-1.py

import nntplib
import string

SERVER = "news.spam.egg"
GROUP  = "comp.lang.python" 
AUTHOR = "fredrik@pythonware.com" # eff-bots human alias

# connect to server
server = nntplib.NNTP(SERVER)

# choose a newsgroup
resp, count, first, last, name = server.group(GROUP)
print "count", "=>", count
print "range", "=>", first, last

# list all items on the server
resp, items = server.xover(first, last)

# extract some statistics
authors = {}
subjects = {}
for id, subject, author, date, message_id, references, size, lines in items:
    authors[author] = None
    if subject[:4] == "Re: ":
        subject = subject[4:]
    subjects[subject] = None
    if string.find(author, AUTHOR) >= 0:
        print id, subject
    
print "authors", "=>", len(authors)
print "subjects", "=>", len(subjects)

count => 607
range => 57179 57971
57474 Three decades of Python!
...
57477 More Python books coming...
authors => 257
subjects => 200

File: nntplib-example-2.py

import nntplib
import string

SERVER = "news.spam.egg"
GROUP  = "comp.lang.python" 
KEYWORD = "tkinter"

# connect to server
server = nntplib.NNTP(SERVER)

resp, count, first, last, name = server.group(GROUP)
resp, items = server.xover(first, last)
for id, subject, author, date, message_id, references, size, lines in items:
    if string.find(string.lower(subject), KEYWORD) >= 0:
        resp, id, message_id, text = server.article(id)
        print author
        print subject
        print len(text), "lines in article"

"Fredrik Lundh" <fredrik@pythonware.com>
Re: Programming Tkinter (In Python)
110 lines in article
...

File: nntplib-example-3.py

import nntplib
import string, random
import StringIO, rfc822

SERVER = "news.spam.egg"
GROUP  = "comp.lang.python"

# connect to server
server = nntplib.NNTP(SERVER)

resp, count, first, last, name = server.group(GROUP)
for i in range(10):
    try:
        id = random.randint(int(first), int(last))
        resp, id, message_id, text = server.article(str(id))
    except (nntplib.error_temp, nntplib.error_perm):
        pass # no such message (maybe it was deleted?)
    else:
        break # found a message!
else:
    raise SystemExit

text = string.join(text, "/n")
file = StringIO.StringIO(text)

message = rfc822.Message(file)

for k, v in message.items():
    print k, "=", v

print message.fp.read()

mime-version = 1.0
content-type = text/plain; charset="iso-8859-1"
message-id = <008501bf1417$1cf90b70$f29b12c2@sausage.spam.egg>
lines = 22
...
from = "Fredrik Lundh" <fredrik@pythonware.com>
nntp-posting-host = parrot.python.org
subject = ANN: (the eff-bot guide to) The Standard Python Library
...
</F>

到这一步后, 你可以使用 htmllib , uu , 以及 base64 继续处理这些消息.

SocketServer 为各种基于 socket 的服务器提供了一个框架. 该模块提供了大量的类, 你可以用它们来创建不同的服务器.

File: socketserver-example-1.py

import SocketServer
import time

# user-accessible port
PORT = 8037

# reference time
TIME1970 = 2208988800L

class TimeRequestHandler(SocketServer.StreamRequestHandler):
    def handle(self):
        print "connection from", self.client_address
        t = int(time.time()) + TIME1970
        b = chr(t>>24&255) + chr(t>>16&255) + chr(t>>8&255) + chr(t&255)
        self.wfile.write(b)

server = SocketServer.TCPServer(("", PORT), TimeRequestHandler)
print "listening on port", PORT
server.serve_forever()

connection from ('127.0.0.1', 1488)
connection from ('127.0.0.1', 1489)
...

这是一个建立在 SocketServer 框架上的基本框架, 用于 HTTP 服务器.

File: basehttpserver-example-1.py

import BaseHTTPServer
import cgi, random, sys

MESSAGES = [
    "That's as maybe, it's still a frog.",
    "Albatross! Albatross! Albatross!",
    "It's Wolfgang Amadeus Mozart.",
    "A pink form from Reading.",
    "Hello people, and welcome to 'It's a Tree.'"
    "I simply stare at the brick and it goes to sleep.",
]

class Handler(BaseHTTPServer.BaseHTTPRequestHandler):

    def do_GET(self):
        if self.path != "/":
            self.send_error(404, "File not found")
            return
        self.send_response(200)
        self.send_header("Content-type", "text/html")
        self.end_headers()
        try:
            # redirect stdout to client
            stdout = sys.stdout
            sys.stdout = self.wfile
            self.makepage()
        finally:
            sys.stdout = stdout # restore
        
    def makepage(self):
        # generate a random message
        tagline = random.choice(MESSAGES)
        print "<html>"
        print "<body>"
        print "<p>Today's quote: "
        print "<i>%s</i>" % cgi.escape(tagline)
        print "</body>"
        print "</html>"

PORT = 8000

httpd = BaseHTTPServer.HTTPServer(("", PORT), Handler)
print "serving at port", PORT
httpd.serve_forever()

更有扩展性的 HTTP 框架请参阅 SimpleHTTPServer 和 CGIHTTPServer 模块.

File: simplehttpserver-example-1.py

import SimpleHTTPServer
import SocketServer

# minimal web server.  serves files relative to the
# current directory.

PORT = 8000

Handler = SimpleHTTPServer.SimpleHTTPRequestHandler

httpd = SocketServer.TCPServer(("", PORT), Handler)

print "serving at port", PORT
httpd.serve_forever()

serving at port 8000
localhost - - [11/Oct/1999 15:07:44] code 403, message Directory listing not sup
ported
localhost - - [11/Oct/1999 15:07:44] "GET / HTTP/1.1" 403 -
localhost - - [11/Oct/1999 15:07:56] "GET /samples/sample.htm HTTP/1.1" 200 -

这个服务器会忽略驱动器符号和相对路径名(例如 `..`). 但它并没有任何访问验证处理, 所以请小心使用.

File: simplehttpserver-example-2.py

# a truly minimal HTTP proxy

import SocketServer
import SimpleHTTPServer
import urllib

PORT = 1234

class Proxy(SimpleHTTPServer.SimpleHTTPRequestHandler):
    def do_GET(self):
        self.copyfile(urllib.urlopen(self.path), self.wfile)

httpd = SocketServer.ForkingTCPServer(('', PORT), Proxy)
print "serving at port", PORT
httpd.serve_forever()

File: cgihttpserver-example-1.py

import CGIHTTPServer
import BaseHTTPServer

class Handler(CGIHTTPServer.CGIHTTPRequestHandler):
    cgi_directories = ["/cgi"]

PORT = 8000

httpd = BaseHTTPServer.HTTPServer(("", PORT), Handler)
print "serving at port", PORT
httpd.serve_forever()

cgi 模块为 CGI 脚本提供了函数和类支持. 它还可以处理 CGI 表单数据.

File: cgi-example-1.py

import cgi
import os, urllib

ROOT = "samples"

# header
print "text/html"
print

query = os.environ.get("QUERY_STRING")
if not query:
    query = "."

script = os.environ.get("SCRIPT_NAME", "")
if not script:
    script = "cgi-example-1.py"

print "<html>"
print "<head>"
print "<title>file listing</title>"
print "</head>"
print "</html>"

print "<body>"

try:
    files = os.listdir(os.path.join(ROOT, query))
except os.error:
    files = []

for file in files:
    link = cgi.escape(file)
    if os.path.isdir(os.path.join(ROOT, query, file)):
        href = script + "?" + os.path.join(query, file)
        print "<p><a href= '%s'>%s</a>" % (href, cgi.escape(link))
    else:
        print "<p>%s" % link

print "</body>"
print "</html>"

text/html

<html>
<head>
<title>file listing</title>
</head>
</html>
<body>
<p>sample.gif
<p>sample.gz
<p>sample.netrc
...
<p>sample.txt
<p>sample.xml
<p>sample~
<p><a href='cgi-example-1.py?web'>web</a>
</body>
</html>

File: webbrowser-example-1.py

import webbrowser
import time

webbrowser.open("http://www.pythonware.com")

# wait a while, and then go to another page
time.sleep(5)
webbrowser.open(
    "http://www.pythonware.com/people/fredrik/librarybook.htm"
    )

在 Unix 下, 该模块支持 lynx , Netscape , Mosaic , Konquerer , 和 Grail . 在 Windows 和 Macintosh 下, 它会调用标准浏览器 (在注册表或是 Internet 选项面板中定义).

File: locale-example-2.py

import locale

language, encoding = locale.getdefaultlocale()

print "language", language
print "encoding", encoding

language sv_SE
encoding cp1252

File: unicodedata-example-1.py

import unicodedata

for char in [u"A", u"-", u"1", u"/N{LATIN CAPITAL LETTER O WITH DIAERESIS}"]:
    print repr(char),
    print unicodedata.category(char),
    print repr(unicodedata.decomposition(char)),

    print unicodedata.decimal(char, None),
    print unicodedata.numeric(char, None)

u'A' Lu '' None None
u'-' Pd '' None None
u'1' Nd '' 1 1.0
u'/303/226' Lu '004F 0308' None None

在 Python 2.0 中缺少 CJK 象形文字和韩语音节的属性. 这影响到了 0x3400-0x4DB5 , 0x4E00-0x9FA5 , 以及 0xAC00-D7A3 中的字符, 不过每个区间内的第一个字符属性是正确的, 我们可以把字符映射到起始 实现正常操作:

def remap(char):
    # fix for broken unicode property database in Python 2.0
    c = ord(char)
    if 0x3400 <= c <= 0x4DB5:
        return unichr(0x3400)
    if 0x4E00 <= c <= 0x9FA5:
        return unichr(0x4E00)
    if 0xAC00 <= c <= 0xD7A3:
        return unichr(0xAC00)
    return char

Python 2.1 修复了这个 bug .

File: ucnhash-example-1.py

# Python imports this module automatically, when it sees
# the first /N{} escape
# import ucnhash

print repr(u"/N{FROWN}")
print repr(u"/N{SMILE}")
print repr(u"/N{SKULL AND CROSSBONES}")

u'/u2322'
u'/u2323'
u'/u2620'

"Wot? No quote?"

- Guido van Rossum

Python 提供了一些用于处理图片和音频文件的模块.

File: imghdr-example-1.py

import imghdr

result = imghdr.what("samples/sample.jpg")

if result:
    print "file format:", result
else:
    print "cannot identify file"

file format: jpeg

# 使用 PIL 
import Image

im = Image.open("samples/sample.jpg")
print im.format, im.mode, im.size

File: sndhdr-example-1.py

import sndhdr

result = sndhdr.what("samples/sample.wav")

if result:
    print "file format:", result
else:
    print "cannot identify file"

file format: ('wav', 44100, 1, -1, 16)

File: whatsound-example-1.py

import whatsound # same as sndhdr

result = whatsound.what("samples/sample.wav")

if result:
    print "file format:", result
else:
    print "cannot identify file"

file format: ('wav', 44100, 1, -1, 16)

File: SimpleAsyncHTTP.py

import asyncore
import string, socket
import StringIO
import mimetools, urlparse

class AsyncHTTP(asyncore.dispatcher_with_send):
    # HTTP requestor

    def _ _init_ _(self, uri, consumer):
        asyncore.dispatcher_with_send._ _init_ _(self)

        self.uri = uri
        self.consumer = consumer

        # turn the uri into a valid request
        scheme, host, path, params, query, fragment = urlparse.urlparse(uri)
        assert scheme == "http", "only supports HTTP requests"
        try:
            host, port = string.split(host, ":", 1)
            port = int(port)
        except (TypeError, ValueError):
            port = 80 # default port
        if not path:
            path = "/"
        if params:
            path = path + ";" + params
        if query:
            path = path + "?" + query

        self.request = "GET %s HTTP/1.0/r/nHost: %s/r/n/r/n" % (path, host)

        self.host = host
        self.port = port

        self.status = None
        self.header = None

        self.data = ""

        # get things going!
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connect((host, port))

    def handle_connect(self):
        # connection succeeded
        self.send(self.request)

    def handle_expt(self):
        # connection failed; notify consumer (status is None)
        self.close()
        try:
            http_header = self.consumer.http_header
        except AttributeError:
            pass
        else:
            http_header(self)

    def handle_read(self):
        data = self.recv(2048)
        if not self.header:
            self.data = self.data + data
            try:
                i = string.index(self.data, "/r/n/r/n")
            except ValueError:
                return # continue
            else:
                # parse header
                fp = StringIO.StringIO(self.data[:i+4])
                # status line is "HTTP/version status message"
                status = fp.readline()
                self.status = string.split(status, " ", 2)
                # followed by a rfc822-style message header
                self.header = mimetools.Message(fp)
                # followed by a newline, and the payload (if any)
                data = self.data[i+4:]
                self.data = ""
                # notify consumer (status is non-zero)
                try:
                    http_header = self.consumer.http_header
                except AttributeError:
                    pass
                else:
                    http_header(self)
                if not self.connected:
                    return # channel was closed by consumer

        self.consumer.feed(data)

    def handle_close(self):
        self.consumer.close()
        self.close()

File: sunau-example-1.py

import sunau

w = sunau.open("samples/sample.au", "r")

if w.getnchannels() == 1:
    print "mono,",
else:
    print "stereo,",

print w.getsampwidth()*8, "bits,",
print w.getframerate(), "Hz sampling rate"

mono, 16 bits, 8012 Hz sampling rate

sunaudio 模块用于识别 Sun AU 音频文件, 并提取其基本信息. sunau 模块为 Sun AU 文件提供了更完成的支持. 如 Example 9-6 所示

File: sunaudio-example-1.py

import sunaudio

file = "samples/sample.au"

print sunaudio.gethdr(open(file, "rb"))

(6761, 1, 8012, 1, 'sample.au')

wave 模块用于读写 Microsoft WAV 音频文件, 如 Example 9-7 所示.

File: wave-example-1.py

import wave

w = wave.open("samples/sample.wav", "r")

if w.getnchannels() == 1:
    print "mono,",
else:
    print "stereo,",

print w.getsampwidth()*8, "bits,",
print w.getframerate(), "Hz sampling rate"

mono, 16 bits, 44100 Hz sampling rate

File: audiodev-example-1.py

import audiodev
import aifc

sound = aifc.open("samples/sample.aiff", "r")

player = audiodev.AudioDev()

player.setoutrate(sound.getframerate())
player.setsampwidth(sound.getsampwidth())
player.setnchannels(sound.getnchannels())

bytes_per_frame = sound.getsampwidth() * sound.getnchannels()
bytes_per_second = sound.getframerate() * bytes_per_frame

while 1:
    data = sound.readframes(bytes_per_second)
    if not data:
        break
    player.writeframes(data)

player.wait()

File: winsound-example-1.py

import winsound

file = "samples/sample.wav"

winsound.PlaySound(
    file,
    winsound.SND_FILENAME|winsound.SND_NOWAIT,
    )

flag 变量说明:

• SND_FILENAME - sound 是一个 wav 文件名
• SND_ALIAS - sound 是一个注册表中指定的别名
• SND_LOOP - 重复播放直到下一次 PlaySound ; 必须指定 SND_ASYNC
• SND_MEMORY - sound 是一个 wav 文件的内存映像
• SND_PURGE - 停止指定 sound 的所有实例
• SND_ASYNC - 异步播放声音, 声音开始播放后函数立即返回
• SND_NODEFAULT - 找不到 sound 时不播放默认的 beep 声音
• SND_NOSTOP - 不打断当前播放中的任何 sound
• SND_NOWAIT - sound 驱动忙时立即返回

"Unlike mainstream component programming, scripts usually do not introduce new components but simply 'wire' existing ones. Scripts can be seen as introducing behavior but no new state ... Of course, there is nothing to stop a 'scripting' language from introducing persistent state — it then simply turns into a normal programming language."

- Clemens Szyperski, in Component Software

Python 提供了多种相似数据库管理( database manager )的驱动, 它们的模型都基于 Unix 的 dbm 库. 这些数据库和普通的字典对象类似, 但这里需要注意的是它只能接受字符串作为键和值. ( shelve 模块可以处理任何类型的值)

anydbm 模块为简单数据库驱动提供了统一标准的接口.

当第一次被导入的时候, anydbm 模块会自动寻找一个合适的数据库驱动, 按照 dbhash , gdbm , dbm , 或 dumbdbm 的顺序尝试. 如果没有找到任何模块, 它将引发一个 ImportError 异常.

File: anydbm-example-1.py

import anydbm

db = anydbm.open("database", "c")
db["1"] = "one"
db["2"] = "two"
db["3"] = "three"
db.close()

db = anydbm.open("database", "r")
for key in db.keys():
    print repr(key), repr(db[key])

'2' 'two'
'3' 'three'
'1' 'one'

File: whichdb-example-1.py

import whichdb

filename = "database"

result = whichdb.whichdb(filename)

if result:
    print "file created by", result
    handler = _ _import_ _(result)
    db = handler.open(filename, "r")
    print db.keys()
else:
    # cannot identify data base
    if result is None:
        print "cannot read database file", filename
    else:
        print "cannot identify database file", filename
    db = None

这个例子中使用了 _ _import_ _ 函数来导入对应模块(还记得我们在第一章的例子么?).

File: shelve-example-1.py

import shelve

db = shelve.open("database", "c")
db["one"] = 1
db["two"] = 2
db["three"] = 3
db.close()

db = shelve.open("database", "r")
for key in db.keys():
    print repr(key), repr(db[key])

'one' 1
'three' 3
'two' 2

File: shelve-example-3.py

import shelve
import gdbm

def gdbm_shelve(filename, flag="c"):
    return shelve.Shelf(gdbm.open(filename, flag))

db = gdbm_shelve("dbfile")

File: dbhash-example-1.py

import dbhash

db = dbhash.open("dbhash", "c")
db["one"] = "the foot"
db["two"] = "the shoulder"
db["three"] = "the other foot"
db["four"] = "the bridge of the nose"
db["five"] = "the naughty bits"
db["six"] = "just above the elbow"
db["seven"] = "two inches to the right of a very naughty bit indeed"
db["eight"] = "the kneecap"
db.close()

db = dbhash.open("dbhash", "r")
for key in db.keys():
    print repr(key), repr(db[key])

File: dbm-example-1.py

import dbm

db = dbm.open("dbm", "c")
db["first"] = "bruce"
db["second"] = "bruce"
db["third"] = "bruce"
db["fourth"] = "bruce"
db["fifth"] = "michael"
db["fifth"] = "bruce" # overwrite
db.close()

db = dbm.open("dbm", "r")
for key in db.keys():
    print repr(key), repr(db[key])

'first' 'bruce'
'second' 'bruce'
'fourth' 'bruce'
'third' 'bruce'
'fifth' 'bruce'

dumbdbm 模块是一个简单的数据库实现, 与 dbm 一类相似, 但使用纯 Python 实现. 它使用两个文件: 一个二进制文件 (.dat) 用于储存数据, 一个文本文件 (.dir) 用于数据描述.

File: dumbdbm-example-1.py

import dumbdbm

db = dumbdbm.open("dumbdbm", "c")
db["first"] = "fear"
db["second"] = "surprise"
db["third"] = "ruthless efficiency"
db["fourth"] = "an almost fanatical devotion to the Pope"
db["fifth"] = "nice red uniforms"
db.close()

db = dumbdbm.open("dumbdbm", "r")
for key in db.keys():
    print repr(key), repr(db[key])

'first' 'fear'
'third' 'ruthless efficiency'
'fifth' 'nice red uniforms'
'second' 'surprise'
'fourth' 'an almost fanatical devotion to the Pope'

File: gdbm-example-1.py

import gdbm

db = gdbm.open("gdbm", "c")
db["1"] = "call"
db["2"] = "the"
db["3"] = "next"
db["4"] = "defendant"
db.close()


db = gdbm.open("gdbm", "r")

keys = db.keys()
keys.sort()
for key in keys:
    print db[key],

call the next defendant

标准库中有一些模块既可用作模块又可以作为命令行实用程序.

dis 模块是 Python 的反汇编器. 它可以把字节码转换为更容易让人看懂的格式.

你可以从命令行调用反汇编器. 它会编译给定的脚本并把反汇编后的字节代码输出到终端上:

$ dis.py hello.py

          0 SET_LINENO              0

          3 SET_LINENO              1
          6 LOAD_CONST              0 ('hello again, and welcome to the show')
          9 PRINT_ITEM
         10 PRINT_NEWLINE
         11 LOAD_CONST              1 (None)
         14 RETURN_VALUE

File: dis-example-1.py

import dis

def procedure():
    print 'hello'

dis.dis(procedure)

          0 SET_LINENO          3

          3 SET_LINENO          4
          6 LOAD_CONST          1 ('hello')
          9 PRINT_ITEM
         10 PRINT_NEWLINE
         11 LOAD_CONST          0 (None)
         14 RETURN_VALUE

pdb 模块是标准 Python 调试器( debugger ). 它基于 bdb 调试器框架.

File: pdb-example-1.py

import pdb

def test(n):
    j = 0
    for i in range(n):
        j = j + i
    return n

db = pdb.Pdb()
db.runcall(test, 1)

> pdb-example-1.py(3)test()
-> def test(n):
(Pdb) s
> pdb-example-1.py(4)test()
-> j = 0
(Pdb) s
> pdb-example-1.py(5)test()
-> for i in range(n):
...

File: bdb-example-1.py

import bdb
import time

def spam(n):
    j = 0
    for i in range(n):
        j = j + i
    return n

def egg(n):
    spam(n)
    spam(n)
    spam(n)
    spam(n)

def test(n):
    egg(n)

class myDebugger(bdb.Bdb):

    run = 0

    def user_call(self, frame, args):
        name = frame.f_code.co_name or "<unknown>"
        print "call", name, args
        self.set_continue() # continue

    def user_line(self, frame):
        if self.run:
            self.run = 0
            self.set_trace() # start tracing
        else:
            # arrived at breakpoint
            name = frame.f_code.co_name or "<unknown>"
            filename = self.canonic(frame.f_code.co_filename)
            print "break at", filename, frame.f_lineno, "in", name
        print "continue..."
        self.set_continue() # continue to next breakpoint

    def user_return(self, frame, value):
        name = frame.f_code.co_name or "<unknown>"
        print "return from", name, value
        print "continue..."
        self.set_continue() # continue

    def user_exception(self, frame, exception):
        name = frame.f_code.co_name or "<unknown>"
        print "exception in", name, exception
        print "continue..."
        self.set_continue() # continue

db = myDebugger()
db.run = 1
db.set_break("bdb-example-1.py", 7)
db.runcall(test, 1)

continue...
call egg None
call spam None
break at C:/ematter/librarybook/bdb-example-1.py 7 in spam
continue...
call spam None
break at C:/ematter/librarybook/bdb-example-1.py 7 in spam
continue...
call spam None
break at C:/ematter/librarybook/bdb-example-1.py 7 in spam
continue...
call spam None
break at C:/ematter/librarybook/bdb-example-1.py 7 in spam
continue...

profile 模块是标准 Python 分析器.

和反汇编器, 调试器相同, 你可以从命令行调用分析器:

$ profile.py hello.py

hello again, and welcome to the show

         3 function calls in 0.785 CPU seconds

  Ordered by: standard name

  ncalls  tottime  percall  cumtime  percall filename:lineno(function)
       1  0.001    0.001    0.002    0.002 <string>:1(?)
       1  0.001    0.001    0.001    0.001 hello.py:1(?)
       1  0.783    0.783    0.785    0.785 profile:0(execfile('hello.py'))
       0  0.000             0.000          profile:0(profiler)

File: profile-example-1.py

import profile

def func1():
    for i in range(1000):
        pass

def func2():
    for i in range(1000):
        func1()

profile.run("func2()")

        1003 function calls in 2.380 CPU seconds

  Ordered by: standard name

  ncalls  tottime  percall  cumtime  percall filename:lineno(function)
       1  0.000    0.000    2.040    2.040 <string>:1(?)
    1000  1.950    0.002    1.950    0.002 profile-example-1.py:3(func1)
       1  0.090    0.090    2.040    2.040 profile-example-1.py:7(func2)
       1  0.340    0.340    2.380    2.380 profile:0(func2())
       0  0.000             0.000          profile:0(profiler)

你可以使用 pstats 模块来修改结果报告的形式.

File: pstats-example-1.py

import pstats
import profile

def func1():
    for i in range(1000):
        pass

def func2():
    for i in range(1000):
        func1()

p = profile.Profile()
p.run("func2()")

s = pstats.Stats(p)
s.sort_stats("time", "name").print_stats()

         1003 function calls in 1.574 CPU seconds

   Ordered by: internal time, function name

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     1000    1.522    0.002    1.522    0.002 pstats-example-1.py:4(func1)
        1    0.051    0.051    1.573    1.573 pstats-example-1.py:8(func2)
        1    0.001    0.001    1.574    1.574 profile:0(func2())
        1    0.000    0.000    1.573    1.573 <string>:1(?)
        0    0.000             0.000          profile:0(profiler)

(2.0 新增) tabnanny 模块用于检查 Python 源文件中的含糊的缩进. 当文件混合了 tab 和空格两种缩进时候, nanny (保姆)会立即给出提示.

在下边使用的 badtabs.py 文件中, if 语句后的第一行使用 4 个空格和 1 个 tab . 第二行只使用了空格.

$ tabnanny.py -v samples/badtabs.py
';samples/badtabs.py': *** Line 3: trouble in tab city! ***
offending line:         print "world"

indent not equal e.g. at tab sizes 1, 2, 3, 5, 6, 7, 9

因为 Python 解释器把 tab 作为 8 个空格来处理, 所以这个脚本可以正常运行. 在所有符合代码标准(一个 tab 为 8 个空格)的编辑器中它也会正常显示. 当然, 这些都骗不过 nanny .

File: tabnanny-example-1.py

import tabnanny

FILE = "samples/badtabs.py"

file = open(FILE)
for line in file.readlines():
    print repr(line)

# let tabnanny look at it
tabnanny.check(FILE)

'if 1:/012'
'    /011print "hello"/012'
'        print "world"/012'
samples/badtabs.py 3 '        print "world"'/012'

将 sys.stdout 重定向到一个 StringIO 对象就可以捕获输出.

本章介绍了一些平台相关的模块. 重点放在了适用于整个平台家族的模块上. (比如 Unix , Windows 家族)

File: fcntl-example-1.py

import fcntl, FCNTL
import os, time

FILE = "counter.txt"

if not os.path.exists(FILE):
    # create the counter file if it doesn't exist
	# 创建 counter 文件
    file = open(FILE, "w")
    file.write("0")
    file.close()

for i in range(20):
    # increment the counter
    file = open(FILE, "r+")
    fcntl.flock(file.fileno(), FCNTL.LOCK_EX)
    counter = int(file.readline()) + 1
    file.seek(0)
    file.write(str(counter))
    file.close() # unlocks the file
    print os.getpid(), "=>", counter
    time.sleep(0.1)

30940 => 1
30942 => 2
30941 => 3
30940 => 4
30941 => 5
30942 => 6

File: pwd-example-1.py

import pwd
import os

print pwd.getpwuid(os.getgid())
print pwd.getpwnam("root")

('effbot', 'dsWjk8', 4711, 4711, 'eff-bot', '/home/effbot', '/bin/bosh')
('root', 'hs2giiw', 0, 0, 'root', '/root', '/bin/bash')

getpwall 函数返回一个包含所有可用用户数据库入口的列表. 你可以使用它搜索一个用户.

File: pwd-example-2.py

import pwd
import os

# preload password dictionary
_pwd = {}
for info in pwd.getpwall():
    _pwd[info[0]] = _pwd[info[2]] = info

def userinfo(uid):
    # name or uid integer
    return _pwd[uid]

print userinfo(os.getuid())
print userinfo("root")

('effbot', 'dsWjk8', 4711, 4711, 'eff-bot', '/home/effbot', '/bin/bosh')
('root', 'hs2giiw', 0, 0, 'root', '/root', '/bin/bash')

File: grp-example-1.py

import grp
import os

print grp.getgrgid(os.getgid())
print grp.getgrnam("wheel")

('effbot', '', 4711, ['effbot'])
('wheel', '', 10, ['root', 'effbot', 'gorbot', 'timbot'])

getgrall 函数返回包含所有可用用户组数据库入口的列表.

File: grp-example-2.py

import grp
import os

# preload password dictionary
_grp = {}
for info in grp.getgrall():
    _grp[info[0]] = _grp[info[2]] = info

def groupinfo(gid):
    # name or gid integer
    return _grp[gid]

print groupinfo(os.getgid())
print groupinfo("wheel")

('effbot', '', 4711, ['effbot'])
('wheel', '', 10, ['root', 'effbot', 'gorbot', 'timbot'])

File: nis-example-1.py

import nis
import string

print nis.cat("ypservers")
print string.split(nis.match("bacon", "hosts.byname"))

{'bacon.spam.egg': 'bacon.spam.egg'}
['194.18.155.250', 'bacon.spam.egg', 'bacon', 'spam-010']

File: curses-example-1.py

import curses

text = [
    "a very simple curses demo",
    "",
    "(press any key to exit)"
]

# connect to the screen
# ��ӵ���Ļ
screen = curses.initscr()

# setup keyboard
# ���ü���
curses.noecho() # no keyboard echo
curses.cbreak() # don't wait for newline

# screen size
# ��Ļ�ߴ�
rows, columns = screen.getmaxyx()

# draw a border around the screen
# ��һ��߿�
screen.border()

# display centered text
# ��ʾ����
y = (rows - len(text)) / 2

for line in text:
    screen.addstr(y, (columns-len(line))/2, line)
    y = y + 1

screen.getch()

curses.endwin()

(只用于 Unix , 可选) termios 为 Unix 的终端控制设备提供了一个接口. 它可用于控制终端通讯端口的大多方面.

File: termios-example-1.py

import termios, TERMIOS
import sys

fileno = sys.stdin.fileno()

attr = termios.tcgetattr(fileno)
orig = attr[:]

print "attr =>", attr[:4] # flags

# disable echo flag
attr[3] = attr[3] & ~TERMIOS.ECHO

try:
    termios.tcsetattr(fileno, TERMIOS.TCSADRAIN, attr)
    message = raw_input("enter secret message: ")
    print
finally:
    # restore terminal settings
    termios.tcsetattr(fileno, TERMIOS.TCSADRAIN, orig)

print "secret =>", repr(message)

attr => [1280, 5, 189, 35387]
enter secret message:
secret => 'and now for something completely different'

File: tty-example-1.py

import tty
import os, sys

fileno = sys.stdin.fileno()

tty.setraw(fileno)
print raw_input("raw input: ")

tty.setcbreak(fileno)
print raw_input("cbreak input: ")

os.system("stty sane") # ...

raw input: this is raw input
cbreak input: this is cbreak input

File: resource-example-1.py

import resource

print "usage stats", "=>", resource.getrusage(resource.RUSAGE_SELF)
print "max cpu", "=>", resource.getrlimit(resource.RLIMIT_CPU)
print "max data", "=>", resource.getrlimit(resource.RLIMIT_DATA)
print "max processes", "=>", resource.getrlimit(resource.RLIMIT_NPROC)
print "page size", "=>", resource.getpagesize()

usage stats => (0.03, 0.02, 0, 0, 0, 0, 75, 168, 0, 0, 0, 0, 0, 0, 0, 0)
max cpu => (2147483647, 2147483647)
max data => (2147483647, 2147483647)
max processes => (256, 256)
page size => 4096

File: resource-example-2.py

import resource

resource.setrlimit(resource.RLIMIT_CPU, (0, 1))

# pretend we're busy
for i in range(1000):
    for j in range(1000):
        for k in range(1000):
            pass

CPU time limit exceeded

File: syslog-example-1.py

import syslog
import sys

syslog.openlog(sys.argv[0])

syslog.syslog(syslog.LOG_NOTICE, "a log notice")
syslog.syslog(syslog.LOG_NOTICE, "another log notice: %s" % "watch out!")

syslog.closelog()

(只用于 Windows/DOS ) msvcrt 模块用于访问 Microsoft Visual C/C++ Runtime Library (MSVCRT) 中函数的方法.

File: msvcrt-example-1.py

import msvcrt

print "press 'escape' to quit..."

while 1:
    char = msvcrt.getch()
    if char == chr(27):
        break
    print char,
    if char == chr(13):
        print

press 'escape' to quit...
h e l l o

File: msvcrt-example-2.py

import msvcrt
import time

print "press SPACE to enter the serial number"

while not msvcrt.kbhit() or msvcrt.getch() != " ":
    # do something else
    print ".",
    time.sleep(0.1)

print

# clear the keyboard buffer
# 清除键盘缓冲区
while msvcrt.kbhit():
    msvcrt.getch()

serial = raw_input("enter your serial number: ")

print "serial number is", serial

press SPACE to enter the serial number
. . . . . . . . . . . . . . . . . . . . . . . .
enter your serial number: 10
serial number is 10

译注: 某翻译在这里评注道: 我能在 cmd 下运行. 用别的 IDLE 要不然卡住, 要不然接受不了键盘输入. 原因未知. 这是因为 IDLE 启动两个 python 线程, 使用 socket 发送数据, 获得程序返回的.

File: msvcrt-example-3.py

import msvcrt
import os

LK_UNLCK = 0 # unlock the file region 解锁区域
LK_LOCK = 1 # lock the file region 锁定文件区域
LK_NBLCK = 2 # non-blocking lock 非阻塞文件锁
LK_RLCK = 3 # lock for writing 为写入文件提供锁定
LK_NBRLCK = 4 # non-blocking lock for writing 为写入文件提供的非阻塞锁定

FILE = "counter.txt"

if not os.path.exists(FILE):
    file = open(FILE, "w")
    file.write("0")
    file.close()

for i in range(20):
    file = open(FILE, "r+")
    # look from current position (0) to end of file
    msvcrt.locking(file.fileno(), LK_LOCK, os.path.getsize(FILE))
    counter = int(file.readline()) + 1
    file.seek(0)
    file.write(str(counter))
    file.close() # unlocks the file
    print os.getpid(), "=>", counter
    time.sleep(0.1)

208 => 21
208 => 22
208 => 23
208 => 24
208 => 25
208 => 26

File: nt-example-1.py

import nt

# in real life, use os.listdir and os.stat instead!
for file in nt.listdir("."):
    print file, nt.stat(file)[6]

aifc-example-1.py 314
anydbm-example-1.py 259
array-example-1.py 48

File: winreg-example-1.py

import _winreg

explorer = _winreg.OpenKey(
    _winreg.HKEY_CURRENT_USER,
    "Software//Microsoft//Windows/CurrentVersion//Explorer"
    )

# list values owned by this registry key 
# 列出该注册表键下的所有值
try:
    i = 0
    while 1:
      name, value, type= _winreg.EnumValue(explorer, i)
      print repr(name),
      i += 1
except WindowsError:
    print

value, type = _winreg.QueryValueEx(explorer, "Logon User Name")

print
print "user is", repr(value)


'Logon User Name' 'CleanShutdown' 'ShellState' 'Shutdown Setting'
'Reason Setting' 'FaultCount' 'FaultTime' 'IconUnderline'...

user is u'Effbot'

File: posix-example-1.py

import posix

for file in posix.listdir("."):
    print file, posix.stat(file)[6]

aifc-example-1.py 314
anydbm-example-1.py 259
array-example-1.py 48

就是其他模块中用到的模块.

File: dospath-example-1.py

import dospath 

file = "/my/little/pony"

print "isabs", "=>", dospath.isabs(file)
print "dirname", "=>", dospath.dirname(file)
print "basename", "=>", dospath.basename(file)
print "normpath", "=>", dospath.normpath(file)
print "split", "=>", dospath.split(file)
print "join", "=>", dospath.join(file, "zorba")

isabs => 1
dirname => /my/little
basename => pony
normpath => /my/little/pony
split => ('/my/little', 'pony')
join => /my/little/pony/zorba

注意 Python 的 DOS 支持可以使用斜杠和反斜杠作为目录分隔符.

File: macpath-example-1.py

import macpath

file = "my:little:pony"

print "isabs", "=>", macpath.isabs(file)
print "dirname", "=>", macpath.dirname(file)
print "basename", "=>", macpath.basename(file)
print "normpath", "=>", macpath.normpath(file)
print "split", "=>", macpath.split(file)
print "join", "=>", macpath.join(file, "zorba")

isabs => 1
dirname => my:little
basename => pony
normpath => my:little:pony
split => ('my:little', 'pony')
join => my:little:pony:zorba

File: ntpath-example-1.py

import ntpath

file = "/my/little/pony"

print "isabs", "=>", ntpath.isabs(file)
print "dirname", "=>", ntpath.dirname(file)
print "basename", "=>", ntpath.basename(file)
print "normpath", "=>", ntpath.normpath(file)
print "split", "=>", ntpath.split(file)
print "join", "=>", ntpath.join(file, "zorba")

isabs => 1
dirname => /my/little
basename => pony

normpath => /my/little/pony
split => ('/my/little', 'pony')
join => /my/little/pony/zorba

注意该模块可以同时使用斜杠和反斜杠作为目录分隔符.

File: posixpath-example-1.py

import posixpath

file = "/my/little/pony"

print "isabs", "=>", posixpath.isabs(file)
print "dirname", "=>", posixpath.dirname(file)
print "basename", "=>", posixpath.basename(file)
print "normpath", "=>", posixpath.normpath(file)
print "split", "=>", posixpath.split(file)
print "join", "=>", posixpath.join(file, "zorba")

isabs => 1
dirname => /my/little
basename => pony
normpath => /my/little/pony
split => ('/my/little', 'pony')
join => /my/little/pony/zorba

(已废弃) strop 为 string 模块中的大多函数提供了底层 C 语言实现. string 模块会自动调用它, 所以一般你不需要直接使用它.

File: strop-example-1.py

import strop
import sys

# assuming we have an executable named ".../executable", add a
# directory named ".../executable-extra" to the path

if strop.lower(sys.executable)[-4:] == ".exe":
    extra = sys.executable[:-4] # windows
else:
    extra = sys.executable

sys.path.insert(0, extra + "-extra")

import mymodule

在 Python 2.0 及以后版本中, 你应该使用字符串方法代替 strop , 例如在上边的代码中. 使用 "sys.executable.lower()" 替换 "strop.lower(sys.executable)" .

File: imp-example-1.py

import imp
import sys

def my_import(name, globals=None, locals=None, fromlist=None):
    try:
        module = sys.modules[name] # already imported?
    except KeyError:
        file, pathname, description = imp.find_module(name)
        print "import", name, "from", pathname, description
        module = imp.load_module(name, file, pathname, description)
    return module

import _ _builtin_ _
_ _builtin_ _._ _import_ _ = my_import

import xmllib

import xmllib from /python/lib/xmllib.py ('.py', 'r', 1)
import re from /python/lib/re.py ('.py', 'r', 1)

import sre from /python/lib/sre.py ('.py', 'r', 1)
import sre_compile from /python/lib/sre_compile.py ('.py', 'r', 1)
import _sre from /python/_sre.pyd ('.pyd', 'rb', 3)

注意这里的导入功能不支持包. 具体实现请参阅 knee 模块的源代码.

File: new-example-1.py

import new

class Sample:

    a = "default"

    def _ _init_ _(self):
        self.a = "initialised"

    def _ _repr_ _(self):
        return self.a

#
# create instances

a = Sample()
print "normal", "=>", a

b = new.instance(Sample, {})
print "new.instance", "=>", b

b._ _init_ _()
print "after _ _init_ _", "=>", b

c = new.instance(Sample, {"a": "assigned"})
print "new.instance w. dictionary", "=>", c

normal => initialised
new.instance => default
after _ _init_ _ => initialised
new.instance w. dictionary => assigned

File: pre-example-1.py

import pre

p = pre.compile("[Python]+")

print p.findall("Python is not that bad")

['Python', 'not', 'th', 't']

File: sre-example-1.py

import sre

text = "The Bookshop Sketch"

# a single character
m = sre.match(".", text)
if m: print repr("."), "=>", repr(m.group(0))

# and so on, for all 're' examples...

'.' => 'T'

File: py-compile-example-1.py

import py_compile

# explicitly compile this module
py_compile.compile("py-compile-example-1.py")

compileall 模块可以把一个目录树下的所有 Python 文件编译为字节代码.

File: compileall-example-1.py

import compileall

print "This may take a while!"

compileall.compile_dir(".", force=1)

This may take a while!
Listing . ...
Compiling ./SimpleAsyncHTTP.py ...
Compiling ./aifc-example-1.py ...
Compiling ./anydbm-example-1.py ...
...

File: ihooks-example-1.py

import ihooks, imp, os

def import_from(filename):
    "Import module from a named file"

    loader = ihooks.BasicModuleLoader()
    path, file = os.path.split(filename)
    name, ext  = os.path.splitext(file)
    m = loader.find_module_in_dir(name, path)
    if not m:
        raise ImportError, name
    m = loader.load_module(name, m)
    return m

colorsys = import_from("/python/lib/colorsys.py")

print colorsys

<module 'colorsys' from '/python/lib/colorsys.py'>

File: linecache-example-1.py

import linecache

print linecache.getline("linecache-example-1.py", 5)

print linecache.getline("linecache-example-1.py", 5)

traceback 模块使用这个模块实现了对导入操作的跟踪.

File: macurl2path-example-1.py

import macurl2path

file = ":my:little:pony"

print macurl2path.pathname2url(file)
print macurl2path.url2pathname(macurl2path.pathname2url(file))

my/little/pony
:my:little:pony

File: nturl2path-example-1.py

import nturl2path

file = r"c:/my/little/pony"

print nturl2path.pathname2url(file)
print nturl2path.url2pathname(nturl2path.pathname2url(file))

///C|/my/little/pony
C:/my/little/pony

File: nturl2path-example-2.py

import urllib

file = r"c:/my/little/pony"

print urllib.pathname2url(file)
print urllib.url2pathname(urllib.pathname2url(file))

///C|/my/little/pony
C:/my/little/pony

tokenize 模块将一段 Python 源文件分割成不同的 token . 你可以在代码高亮工具中使用它.

File: tokenize-example-1.py

import tokenize

file = open("tokenize-example-1.py")

def handle_token(type, token, (srow, scol), (erow, ecol), line):
    print "%d,%d-%d,%d:/t%s/t%s" % /
        (srow, scol, erow, ecol, tokenize.tok_name[type], repr(token))

tokenize.tokenize(
    file.readline,
    handle_token
    )

1,0-1,6:     NAME    'import'
1,7-1,15:    NAME    'tokenize'
1,15-1,16:   NEWLINE '/012'
2,0-2,1:     NL      '/012'
3,0-3,4:     NAME    'file'
3,5-3,6:     OP      '='
3,7-3,11:    NAME    'open'
3,11-3,12:   OP      '('
3,12-3,35:   STRING  '"tokenize-example-1.py"'
3,35-3,36:   OP      ')'
3,36-3,37:   NEWLINE '/012'
...

注意这里的 tokenize 函数接受两个可调用对象作为参数: 前一个用于获取新的代码行, 第二个用于在获得每个 token 时调用.

File: keyword-example-1.py

import keyword

name = raw_input("Enter module name: ")

if keyword.iskeyword(name):
    print name, "is a reserved word."
    print "here's a complete list of reserved words:"
    print keyword.kwlist

Enter module name: assert
assert is a reserved word.
here's a complete list of reserved words:
['and', 'assert', 'break', 'class', 'continue', 'def', 'del',
'elif', 'else', 'except', 'exec', 'finally', 'for', 'from',
'global', 'if', 'import', 'in', 'is', 'lambda', 'not', 'or',
'pass', 'print', 'raise', 'return', 'try', 'while']

(可选) parser 模块提供了一个到 Python 内建语法分析器和编译器的接口.

File: parser-example-1.py

import parser
import symbol, token

def dump_and_modify(node):
    name = symbol.sym_name.get(node[0])
    if name is None:
        name = token.tok_name.get(node[0])
    print name,
    for i in range(1, len(node)):
        item = node[i]
        if type(item) is type([]):
            dump_and_modify(item)
        else:
            print repr(item)
            if name == "NUMBER":
                # increment all numbers!
                node[i] = repr(int(item)+1)

ast = parser.expr("1 + 3")

list = ast.tolist()

dump_and_modify(list)

ast = parser.sequence2ast(list)

print eval(parser.compileast(ast))

eval_input testlist test and_test not_test comparison
expr xor_expr and_expr shift_expr arith_expr term factor
power atom NUMBER '1'
PLUS '+'
term factor power atom NUMBER '3'
NEWLINE ''
ENDMARKER ''
6

File: symbol-example-1.py

import symbol

print "print", symbol.print_stmt
print "return", symbol.return_stmt

print 268
return 274

File: token-example-1.py

import token

print "NUMBER", token.NUMBER
print "PLUS", token.STAR
print "STRING", token.STRING

NUMBER 2
PLUS 16
STRING 3

本章描述的是一些并不怎么常见的模块. 一些是很实用的, 另些是已经废弃的模块.

File: pyclbr-example-1.py

import pyclbr

mod = pyclbr.readmodule("cgi")

for k, v in mod.items():
    print k, v

MiniFieldStorage <pyclbr.Class instance at 7873b0>
InterpFormContentDict <pyclbr.Class instance at 79bd00>
FieldStorage <pyclbr.Class instance at 790e20>
SvFormContentDict <pyclbr.Class instance at 79b5e0>
StringIO <pyclbr.Class instance at 77dd90>
FormContent <pyclbr.Class instance at 79bd60>
FormContentDict <pyclbr.Class instance at 79a9c0>

File: pyclbr-example-3.py

import pyclbr

# 2.0 and later
mod = pyclbr.readmodule_ex("cgi")

for k, v in mod.items():
    print k, v

MiniFieldStorage <pyclbr.Class instance at 00905D2C>
parse_header <pyclbr.Function instance at 00905BD4>
test <pyclbr.Function instance at 00906FBC>
print_environ_usage <pyclbr.Function instance at 00907C94>
parse_multipart <pyclbr.Function instance at 00905294>
FormContentDict <pyclbr.Class instance at 008D3494>
initlog <pyclbr.Function instance at 00904AAC>
parse <pyclbr.Function instance at 00904EFC>
StringIO <pyclbr.Class instance at 00903EAC>
SvFormContentDict <pyclbr.Class instance at 00906824>
...

File: pyclbr-example-2.py

import pyclbr
import string

mod = pyclbr.readmodule("cgi")

def dump(c):
    # print class header
    s = "class " + c.name
    if c.super:
        s = s +  "(" + string.join(map(lambda v: v.name, c.super), ", ") + ")"
    print s + ":"
    # print method names, sorted by line number
    methods = c.methods.items()
    methods.sort(lambda a, b: cmp(a[1], b[1]))
    for method, lineno in methods:
        print "  def " + method
    print

for k, v in mod.items():
    dump(v)

class MiniFieldStorage:
  def _ _init_ _
  def _ _repr_ _

class InterpFormContentDict(SvFormContentDict):
  def _ _getitem_ _
  def values
  def items

...

File: filecmp-example-1.py

import filecmp

if filecmp.cmp("samples/sample.au", "samples/sample.wav"):
    print "files are identical"
else:
    print "files differ!"

# files differ!

1.5.2 以及先前版本中, 你可以使用 cmp 和 dircmp 模块代替.

File: cmd-example-1.py

import cmd
import string, sys

class CLI(cmd.Cmd):

    def _ _init_ _(self):
        cmd.Cmd._ _init_ _(self)
        self.prompt = '> '

    def do_hello(self, arg):
        print "hello again", arg, "!"

    def help_hello(self):
        print "syntax: hello [message]",
        print "-- prints a hello message"

    def do_quit(self, arg):
        sys.exit(1)

    def help_quit(self):
        print "syntax: quit",
        print "-- terminates the application"

    # shortcuts
    do_q = do_quit

#
# try it out

cli = CLI()
cli.cmdloop()

> help

Documented commands (type help <topic>):
========================================
hello           quit

Undocumented commands:
======================
help            q

> hello world
hello again world !
> q

File: rexec-example-1.py

import rexec

r = rexec.RExec()
print r.r_eval("1+2+3")
print r.r_eval("_ _import_ _('os').remove('file')")

6
Traceback (innermost last):
  File "rexec-example-1.py", line 5, in ?
    print r.r_eval("_ _import_ _('os').remove('file')")
  File "/usr/local/lib/python1.5/rexec.py", line 257, in r_eval
    return eval(code, m._ _dict_ _)
  File "<string>", line 0, in ?
AttributeError: remove

File: bastion-example-1.py

import Bastion

class Sample:
    value = 0

    def _set(self, value):
        self.value = value

    def setvalue(self, value):
        if 10 < value <= 20:
            self._set(value)
        else:
            raise ValueError, "illegal value"

    def getvalue(self):
        return self.value

#
# try it

s = Sample()
s._set(100) # cheat
print s.getvalue()

s = Bastion.Bastion(Sample())
s._set(100) # attempt to cheat
print s.getvalue()

100
Traceback (innermost last):
...
AttributeError: _set

File: bastion-example-2.py

import Bastion

class Sample:
    value = 0

    def _set(self, value):
        self.value = value

    def setvalue(self, value):
        if 10 < value <= 20:
            self._set(value)
        else:
            raise ValueError, "illegal value"

    def getvalue(self):
        return self.value

#
# try it

def is_public(name):
    return name[:3] != "get"

s = Bastion.Bastion(Sample(), is_public)
s._set(100) # this works
print s.getvalue() # but not this

100
Traceback (innermost last):
...
AttributeError: getvalue

File: readline-example-1.py

import readline # activate readline editing

(可选, 只用于 Unix ) rlcompleter 模块为 readline 模块提供了单词自动完成功能.

导入该模块就可以启动自动完成功能. 默认情况下完成函数被绑定在了 Esc 键上. 按两次 Esc 键就可以自动完成当前单词. 你可以使用下面的代码修改所绑定的键:

import readline
readline.parse_and_bind("tab: complete")

File: rlcompleter-example-1.py

import rlcompleter
import sys

completer = rlcompleter.Completer()

for phrase in "co", "sys.p", "is":
    print phrase, "=>",
    # emulate readline completion handler
    try:
        for index in xrange(sys.maxint):
            term = completer.complete(phrase, index)
            if term is None:
                break
            print term,
    except:
        pass
    print

co => continue compile complex coerce completer
sys.p => sys.path sys.platform sys.prefix
is => is isinstance issubclass

statvfs 模块包含一些与 os.statvfs (可选)函数配合使用的常量和函数, 该函数会返回文件系统的相关信息. 如 Example 14-11 所示.

File: statvfs-example-1.py

import statvfs
import os

st = os.statvfs(".")

print "preferred block size", "=>", st[statvfs.F_BSIZE]
print "fundamental block size", "=>", st[statvfs.F_FRSIZE]
print "total blocks", "=>", st[statvfs.F_BLOCKS]
print "total free blocks", "=>", st[statvfs.F_BFREE]
print "available blocks", "=>", st[statvfs.F_BAVAIL]
print "total file nodes", "=>", st[statvfs.F_FILES]
print "total free nodes", "=>", st[statvfs.F_FFREE]
print "available nodes", "=>", st[statvfs.F_FAVAIL]
print "max file name length", "=>", st[statvfs.F_NAMEMAX]

preferred block size => 8192
fundamental block size => 1024
total blocks => 749443
total free blocks => 110442
available blocks => 35497
total file nodes => 92158
total free nodes => 68164
available nodes => 68164
max file name length => 255

calendar 模块是 Unix cal 命令的 Python 实现. 它可以将给定年份/月份的日历输出到标准输出设备上.

File: calendar-example-1.py

import calendar
calendar.prmonth(1999, 12)

    December 1999
Mo Tu We Th Fr Sa Su
       1  2  3  4  5
 6  7  8  9 10 11 12
13 14 15 16 17 18 19
20 21 22 23 24 25 26
27 28 29 30 31

File: calendar-example-2.py

import calendar
calendar.prcal(2000)

                                  2000

       January                  February                    March
Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su
                1  2          1  2  3  4  5  6             1  2  3  4  5
 3  4  5  6  7  8  9       7  8  9 10 11 12 13       6  7  8  9 10 11 12
10 11 12 13 14 15 16      14 15 16 17 18 19 20      13 14 15 16 17 18 19
17 18 19 20 21 22 23      21 22 23 24 25 26 27      20 21 22 23 24 25 26
24 25 26 27 28 29 30      28 29                     27 28 29 30 31
31

        April                      May                      June
Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su
                1  2       1  2  3  4  5  6  7                1  2  3  4
 3  4  5  6  7  8  9       8  9 10 11 12 13 14       5  6  7  8  9 10 11
10 11 12 13 14 15 16      15 16 17 18 19 20 21      12 13 14 15 16 17 18
17 18 19 20 21 22 23      22 23 24 25 26 27 28      19 20 21 22 23 24 25
24 25 26 27 28 29 30      29 30 31                  26 27 28 29 30

        July                     August                   September
Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su
                1  2          1  2  3  4  5  6                   1  2  3
 3  4  5  6  7  8  9       7  8  9 10 11 12 13       4  5  6  7  8  9 10
10 11 12 13 14 15 16      14 15 16 17 18 19 20      11 12 13 14 15 16 17
17 18 19 20 21 22 23      21 22 23 24 25 26 27      18 19 20 21 22 23 24
24 25 26 27 28 29 30      28 29 30 31               25 26 27 28 29 30
31

       October                  November                  December
Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su      Mo Tu We Th Fr Sa Su
                   1             1  2  3  4  5                   1  2  3
 2  3  4  5  6  7  8       6  7  8  9 10 11 12       4  5  6  7  8  9 10
 9 10 11 12 13 14 15      13 14 15 16 17 18 19      11 12 13 14 15 16 17
16 17 18 19 20 21 22      20 21 22 23 24 25 26      18 19 20 21 22 23 24
23 24 25 26 27 28 29      27 28 29 30               25 26 27 28 29 30 31
30 31

注意这里的日历是按照欧洲习惯打印的, 也就是说星期一是一个星期的第一天, 其他情况需要请参考模块中的几个类. (和咱们一样, 不用管了)

该模块中的其他类或函数可以帮助你输出自己需要的格式.

sched 模块为非线程环境提供了一个简单的计划任务模式. 如 Example 14-14 所示.

File: sched-example-1.py

import sched
import time, sys

scheduler = sched.scheduler(time.time, time.sleep)

# add a few operations to the queue
scheduler.enter(0.5, 100, sys.stdout.write, ("one/n",))
scheduler.enter(1.0, 300, sys.stdout.write, ("three/n",))
scheduler.enter(1.0, 200, sys.stdout.write, ("two/n",))

scheduler.run()

one
two
three

File: statcache-example-1.py

import statcache
import os, stat, time

now = time.time()
for i in range(1000):
    st = os.stat("samples/sample.txt")
print "os.stat", "=>", time.time() - now

now = time.time()
for i in range(1000):
    st = statcache.stat("samples/sample.txt")
print "statcache.stat", "=>", time.time() - now

print "mode", "=>", oct(stat.S_IMODE(st[stat.ST_MODE]))
print "size", "=>", st[stat.ST_SIZE]
print "last modified", "=>", time.ctime(st[stat.ST_MTIME])

os.stat => 0.371000051498
statcache.stat => 0.0199999809265
mode => 0666
size => 305
last modified => Sun Oct 10 18:39:37 1999

grep 模块提供了在文本文件中搜索字符串的另种方法, 如 Example 14-16 所示.

版本 2.1 时被声明不支持, 及就是说, 当前版本已经无法使用该模块.

File: grep-example-1.py

import grep
import glob

grep.grep("/<rather/>", glob.glob("samples/*.txt"))

# 4: indentation, rather than delimiters, might become

File: dircache-example-1.py

import dircache

import os, time

# 
# test cached version

t0 = time.clock()

for i in range(100):
    dircache.listdir(os.sep)

print "cached", time.clock() - t0

# 
# test standard version

t0 = time.clock()

for i in range(100):
    os.listdir(os.sep)

print "standard", time.clock() - t0

cached 0.0664509964968
standard 0.5560845807

File: dircmp-example-1.py

import dircmp

d = dircmp.dircmp()
d.new("samples", "oldsamples")
d.run()
d.report()

diff samples oldsamples
Only in samples : ['sample.aiff', 'sample.au', 'sample.wav']
Identical files : ['sample.gif', 'sample.gz', 'sample.jpg', ...]

Python 2.0 后, 该模块被 filecmp 替换.

File: cmp-example-1.py

import cmp

if cmp.cmp("samples/sample.au", "samples/sample.wav"):
    print "files are identical"
else:
    print "files differ!"

files differ!

Python 2.0 后, 该模块被 filecmp 替换.

File: cmpcache-example-1.py

import cmpcache

if cmpcache.cmp("samples/sample.au", "samples/sample.wav"):
    print "files are identical"
else:
    print "files differ!"

files differ!

Python 2.0 后, 该模块被 filecmp 替换.

但 filecmp 已经不提供缓存支持.

(已废弃, 只用于 1.5.2) util 模块提供了常见操作的封装函数. 新代码可以使用如 Examples 14-21 到 14-23 的实现方法.

File: util-example-1.py

def remove(sequence, item):
    if item in sequence:
        sequence.remove(item)

File: util-example-2.py

def readfile(filename):
    file = open(filename, "r")
    return file.read()

File: util-example-3.py

def readopenfile(file):
    return file.read()

(已废弃, 只用于 1.5.2) soundex 实现了一个简单的 hash 算法, 基于英文发音将单词转换为 6 个字符的字符串.

版本 2.0 后, 该模块已从标准库中删除.

File: soundex-example-1.py

import soundex

a = "fredrik"
b = "friedrich"

print soundex.get_soundex(a), soundex.get_soundex(b)

print soundex.sound_similar(a, b)

F63620 F63620
1

File: timing-example-1.py

import timing
import time

def procedure():
    time.sleep(1.234)

timing.start()
procedure()
timing.finish()

print "seconds:", timing.seconds()
print "milliseconds:", timing.milli()
print "microseconds:", timing.micro()

seconds: 1
milliseconds: 1239
microseconds: 1239999

File: timing-example-2.py

import time

t0 = t1 = 0

def start():
    global t0
    t0 = time.time()


def finish():
    global t1
    t1 = time.time()

def seconds():
    return int(t1 - t0)

def milli():
    return int((t1 - t0) * 1000)

def micro():
    return int((t1 - t0) * 1000000)

time.clock() 可以替换 time.time() 获得 CPU 时间.

File: posixfile-example-1.py

import posixfile
import string

filename = "counter.txt"

try:
    # open for update
    file = posixfile.open(filename, "r+")
    counter = int(file.read(6)) + 1
except IOError:
    # create it
    file = posixfile.open(filename, "w")
    counter = 0

file.lock("w|", 6)

file.seek(0) # rewind
file.write("%06d" % counter)

file.close() # releases lock

bisect 模块用于向排序后的序列插入对象.

File: bisect-example-1.py

import bisect

list = [10, 20, 30]

bisect.insort(list, 25)
bisect.insort(list, 15)

print list

[10, 15, 20, 25, 30]

File: bisect-example-2.py

import bisect

list = [10, 20, 30]

print list
print bisect.bisect(list, 25)
print bisect.bisect(list, 15)

[10, 20, 30]
2
1

knee 模块用于 Python 1.5 中导入包( package import )的实现. 当然 Python 解释器已经支持了这个, 所以这个模块几乎没有什么作用, 不过你可以看看它的代码, 明白这一切是怎么完成的.

代码请参见 Python-X.tgz/Python-2.4.4/Demo/imputil/knee.py

File: knee-example-1.py

import knee

# that's all, folks!

File: tzparse-example-1.py

import os
if not os.environ.has_key("TZ"):
    # set it to something...
    os.environ["TZ"] = "EST+5EDT;100/2,300/2"

# importing this module will parse the TZ variable
import tzparse

print "tzparams", "=>", tzparse.tzparams
print "timezone", "=>", tzparse.timezone
print "altzone", "=>", tzparse.altzone
print "daylight", "=>", tzparse.daylight
print "tzname", "=>", tzparse.tzname

tzparams => ('EST', 5, 'EDT', 100, 2, 300, 2)
timezone => 18000
altzone => 14400
daylight => 1
tzname => ('EST', 'EDT')

除了这些变量之外, 该模块还提供了一些用于时间计算的函数.

File: regex-example-1.py

import regex

text = "Man's crisis of identity in the latter half of the 20th century"

p = regex.compile("latter") # literal
print p.match(text)
print p.search(text), repr(p.group(0))

p = regex.compile("[0-9]+") # number
print p.search(text), repr(p.group(0))

p = regex.compile("/</w/w/>") # two-letter word
print p.search(text), repr(p.group(0))

p = regex.compile("/w+$") # word at the end
print p.search(text), repr(p.group(0))

-1
32 'latter'
51 '20'
13 'of'
56 'century'

File: regsub-example-1.py

import regsub

text = "Well, there's spam, egg, sausage, and spam."

print regsub.sub("spam", "ham", text) # just the first
print regsub.gsub("spam", "bacon", text) # all of them

Well, there's ham, egg, sausage, and spam.
Well, there's bacon, egg, sausage, and bacon.

File: reconvert-example-1.py

import reconvert

for pattern in "abcd", "a/(b*c/)d", "/</w+/>":
    print pattern, "=>", reconvert.convert(pattern)

abcd => abcd
a/(b*c/)d => a(b*c)d
/</w+/> => /b/w+/b

File: regex-syntax-example-1.py

import regex_syntax
import regex

def compile(pattern, syntax):
    syntax = regex.set_syntax(syntax)
    try:
        pattern = regex.compile(pattern)
    finally:
        # restore original syntax
        regex.set_syntax(syntax)
    return pattern

def compile_awk(pattern):
    return compile(pattern, regex_syntax.RE_SYNTAX_AWK)

def compile_grep(pattern):
    return compile(pattern, regex_syntax.RE_SYNTAX_GREP)

def compile_emacs(pattern):
    return compile(pattern, regex_syntax.RE_SYNTAX_EMACS)

File: find-example-1.py

import find

# find all JPEG files in or beneath the current directory
for file in find.find("*.jpg", "."):
    print file

./samples/sample.jpg

本章将在以后的时间里慢慢完成, 更新.