7

我正在编写一个 Python 脚本pycurl来使用Twitter 的 Sreaming API。这是一个简短的片段，正是这样做的（只需输入您的 Twitter 登录名/密码来测试它）：

import pycurl

user = 'USER'
password = 'PWD'

def handleData(data):
    print(data)

conn = pycurl.Curl()  
conn.setopt(pycurl.USERPWD, "%s:%s" % (user, password))  
conn.setopt(pycurl.URL, 'https://stream.twitter.com/1/statuses/sample.json')  
conn.setopt(pycurl.WRITEFUNCTION, handleData)
conn.perform()

问题是因为脚本消耗了一个流，所以conn.perform()永远不会返回（或很少）。因此，有时我需要中断脚本，并且KeyboardInterrupt被perform()方法捕获。

但是，它不能很好地处理它，打印一个丑陋的错误，并引发一个不同的异常。

^CTraceback (most recent call last):
  File "test.py", line 6, in handleData
    def handleData(data):
KeyboardInterrupt
Traceback (most recent call last):
  File "test.py", line 12, in <module>
    conn.perform()
pycurl.error: (23, 'Failed writing body (0 != 2203)')

cURL FAQ说要中断正在进行的传输，回调函数之一（在我的例子中）handleData应该返回一个特殊值。这很好，但KeyboardInterrupt没有被任何回调函数捕获！

我怎样才能巧妙地做到这一点？

编辑：我知道你可以捕获异常，但 pycurl 仍然会做一些有趣的事情：

如果我做：

try:
    conn.perform()
except BaseException as e:
    print('We caught the exception')
    print(type(e))

我得到：

^CTraceback (most recent call last):
  File "test.py", line 6, in handleData
    def handleData(data):
KeyboardInterrupt
We caught the exception
<class 'pycurl.error'>

这意味着在内部，pycurl会进行某种捕获，打印出丑陋的错误消息，然后引发pycurl.error.

4

2 回答 2

2

您需要捕获 CTRL+C 并处理该信号
原始：示例 1
原始：示例 2

示例 1 

#!/usr/bin/env python
import signal
import sys
def signal_handler(signal, frame):
        print 'You pressed Ctrl+C!'
        sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
print 'Press Ctrl+C'
signal.pause()

示例 2

import signal, os

def handler(signum, frame):
    print 'Signal handler called with signal', signum
    raise IOError("Couldn't open device!")

# Set the signal handler and a 5-second alarm
signal.signal(signal.SIGALRM, handler)
signal.alarm(5)

# This open() may hang indefinitely
fd = os.open('/dev/ttyS0', os.O_RDWR)

signal.alarm(0)          # Disable the alarm

至少该推特链接上有些东西不起作用，请参见此处

  • 不要忘记将conn.close()放在conn.perform()之后

并且在测试时启用调试模式很有帮助。

import pycurl

username = 'your_user_name'
password = 'your_password'

def body(buf):
    for item in buf.strip().split('\n'):
        if item.strip():
            print item

def test(debug_type, debug_msg):
    if len(debug_msg) < 300:
        print "debug(%d): %s" % (debug_type, debug_msg.strip())

conn = pycurl.Curl()  
conn.setopt(pycurl.USERNAME, username)
conn.setopt(pycurl.PASSWORD, password)
#conn.setopt(pycurl.SSL_VERIFYPEER, False)
conn.setopt(pycurl.FOLLOWLOCATION, True)
conn.setopt(pycurl.VERBOSE, True)
conn.setopt(pycurl.URL, 'https://stream.twitter.com/1.1/statuses/sample.json')  
conn.setopt(pycurl.DEBUGFUNCTION, test)
conn.setopt(pycurl.WRITEFUNCTION, body)
conn.perform()
conn.close()

只需复制/粘贴工作测试示例

➜  ~  hcat twitter.py 
import pycurl
import signal
import sys
from time import sleep

username = 'bubudee'
password = 'deebubu'

def body(buf):
    for item in buf.strip().split('\n'):
        if item.strip():
            print item

def test(debug_type, debug_msg):
    if len(debug_msg) < 300:
        print "debug(%d): %s" % (debug_type, debug_msg.strip())

def handle_ctrl_c(signal, frame):
    print "Got ctrl+c, going down!"
    sys.exit(0)
signal.signal(signal.SIGINT, handle_ctrl_c)

conn = pycurl.Curl()  
conn.setopt(pycurl.USERNAME, username)
conn.setopt(pycurl.PASSWORD, password)
#conn.setopt(pycurl.SSL_VERIFYPEER, False)
conn.setopt(pycurl.FOLLOWLOCATION, True)
conn.setopt(pycurl.VERBOSE, True)
conn.setopt(pycurl.URL, 'https://stream.twitter.com/1.1/statuses/sample.json')  
conn.setopt(pycurl.DEBUGFUNCTION, test)
conn.setopt(pycurl.WRITEFUNCTION, body)

conn.perform()

print "Who let the dogs out?:p"
sleep(10)

conn.close()

➜  ~  python twitter.py 
debug(0): About to connect() to stream.twitter.com port 443 (#0)
debug(0): Trying 199.16.156.110...
debug(0): Connected to stream.twitter.com (199.16.156.110) port 443 (#0)
debug(0): Initializing NSS with certpath: sql:/etc/pki/nssdb
debug(0): CAfile: /etc/pki/tls/certs/ca-bundle.crt
  CApath: none
debug(0): SSL connection using SSL_RSA_WITH_RC4_128_SHA
debug(0): Server certificate:
debug(0): subject: CN=stream.twitter.com,OU=Twitter Security,O="Twitter, Inc.",L=San Francisco,ST=California,C=US
debug(0): start date: Oct 09 00:00:00 2013 GMT
debug(0): expire date: Dec 30 23:59:59 2016 GMT
debug(0): common name: stream.twitter.com
debug(0): issuer: CN=VeriSign Class 3 Secure Server CA - G3,OU=Terms of use at https://www.verisign.com/rpa (c)10,OU=VeriSign Trust Network,O="VeriSign, Inc.",C=US
debug(0): Server auth using Basic with user 'bubudee'
debug(2): GET /1.1/statuses/sample.json HTTP/1.1
Authorization: Basic YnVidWRlZTpkZWVidWJ1
User-Agent: PycURL/7.29.0
Host: stream.twitter.com
Accept: */*
debug(1): HTTP/1.1 401 Unauthorized
debug(0): Authentication problem. Ignoring this.
debug(1): WWW-Authenticate: Basic realm="Firehose"
debug(1): Content-Type: text/html
debug(1): Cache-Control: must-revalidate,no-cache,no-store
debug(1): Content-Length: 1243
debug(1): Connection: close
debug(1): 
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>
<title>Error 401 Unauthorized</title>
</head>
<body>
<h2>HTTP ERROR: 401</h2>
<p>Problem accessing '/1.1/statuses/sample.json'. Reason:
<pre>    Unauthorized</pre>
</body>
</html>
debug(0): Closing connection 0
Who let the dogs out?:p
^CGot ctrl+c, going down!
于 2013-10-30T21:24:51.823 回答
1

您可以通过捕获 pycurl.error 类型来做到这一点。前任：

try:
    conn.perform()
except pycurl.error, e:
    errorCode, errorText = e.args
    print 'We got an error. Code: %s, Text:%s'%(errorCode, errorText)
于 2012-05-31T21:05:09.907 回答