通过python分布式开发 网页数据抓取(三)——代码

作者:课课家教育更新于： 2016-01-08 14:04:31

　　这一节主要是贴代码

      本程序连跑了24小时，然后分布式在10台机器上部署，长时间续航基本没有问题。

　　之后每天将进行10万次网页的爬取。

　　源码如下：

　　内容爬取及工具

   Created on 2010-9-15

   @author: chenggong

   import urllib2

   import re

   import socket

   DEBUG = 0

   工具类

   class Tools():

   #log函数

   @staticmethod

   def writelog(level,info,notify=False):

   if DEBUG == 0:

   try:

   print "["+level+"]"+info.decode('UTF-8').encode('GBK')

  except:

  print "["+level+"]"+info.encode('GBK')

  else:

  print "["+level+"]"+info

  #if notify:

  # print "[notify]报告管理员!!"

  #转unicode

  @staticmethod

  def toUnicode(s,charset):

  if( charset == "" ):

  return s

  else:

  try:

  u = unicode( s, charset )

  except:

  u = ""

  return u

#正则抓取

#@param single 是否只抓取一个

@staticmethod

def getFromPatten(patten,src,single=False):

rst = "";

p = re.compile(patten,re.S)

all = p.findall(src)

for matcher in all:

rst += matcher + " "

if( single ):

break

return rst.strip()

'''

网页内容爬虫

'''

class PageGripper():

URL_OPEN_TIMEOUT = 10 #网页超时时间

MAX_RETRY = 3 #最大重试次数

def __init__(self):

socket.setdefaulttimeout(self.URL_OPEN_TIMEOUT)

#获取字符集

def getCharset(self,s):

rst = Tools.getFromPatten(u'charset=(.*?)"',s,True)

if rst != "":

if rst == "utf8":

rst = "utf-8"

return rst

#尝试获取页面

def downloadUrl(self,url):

charset = ""

page = ""

retry = 0

while True:

try:

fp = urllib2.urlopen(url)

break

except urllib2.HTTPError,e: #状态错误

Tools.writelog('error','HTTP状态错误 code='+e.code)

raise urllib2.HTTPError

except urllib2.URLError,e: #网络错误超时

Tools.writelog('warn','页面访问超时,重试..')

retry+=1

if( retry > self.MAX_RETRY ):

Tools.writelog('warn','超过最大重试次数,放弃')

raise urllib2.URLError

while True:

line = fp.readline()

if charset == "":

charset = self.getCharset(line)

if not line:

break

page += Tools.toUnicode(line,charset)

fp.close()

return page

#获取页面

def getPageInfo(self,url):

Tools.writelog( "info","开始抓取网页,url= "+url)

info = ""

try:

info = self.downloadUrl(url)

except:

raise

Tools.writelog("debug","网页抓取成功")

return info

'''

内容提取类

'''

class InfoGripper():

pageGripper = PageGripper()

def __init__(self):

Tools.writelog('debug',"爬虫启动")

#抓取标题

def griptitle(self,data):

title = Tools.getFromPatten(u'box2t sp">

(.*?)

', data, True)

if title == "":

title = Tools.getFromPatten(u'(.*?)[-<]',data,True)

return title.strip()

#抓取频道

def gripchannel(self,data):

zone = Tools.getFromPatten(u'频道：(.*?)',data,True)

channel = Tools.getFromPatten(u'(.*?)',zone,True)

return channel

#抓取标签

def griptag(self,data):

zone = Tools.getFromPatten(u'标签：(.*?)',data,True);

rst = Tools.getFromPatten(u'>(.*?)',zone,False);

return rst

#抓取观看次数

def gripviews(self,data):

rst = Tools.getFromPatten(u'已经有(.*?)次观看',data);

return rst

#抓取发布时间

def griptime(self,data):

rst = Tools.getFromPatten(u'在(.*?)发布',data,True)

return rst

#抓取发布者

def gripuser(self,data):

rst = Tools.getFromPatten(u'title="点击进入(.*?)的用户空间"',data,True)

return rst

#获取页面字符集

def getPageCharset(self,data):

charset = Tools.getFromPatten(u'charset=(.*?)"',data,True)

if( charset == "utf8" ):

charset = "utf-8"

return charset

#获取CC相关数据

def getCCData(self,data):

zone = Tools.getFromPatten(u'SWFObject(.*?)',data,True)

#判断是否使用bokecc播放

isFromBokeCC = re.match('.*bokecc.com.*', zone)

if( not isFromBokeCC ):

return "",""

ccSiteId = Tools.getFromPatten(u'siteid=(.*?)[&,"]',zone,True)

ccVid = Tools.getFromPatten(u'vid=(.*?)[&,"]',zone,True)

return ccSiteId,ccVid

#获取站内vid

def gripVideoId(self,data):

vid = Tools.getFromPatten(u'var vid = "(.*?)"',data,True)

return vid

#获取点击量

def gripViewsajax(self,vid,url,basedir):

host = Tools.getFromPatten(u'http://(.*?)/',url,True)

ajaxAddr = "http://" + host + basedir + "/index.php/ajax/video_statistic/" + vid

'''

try:

content = self.pageGripper.getPageInfo(ajaxAddr)

except Exception,e:

print e

Tools.writelog ("error", ajaxAddr+u"抓取失败")

return "error"

'''

Tools.writelog('debug', u"开始获取点击量,url="+ajaxAddr)

while True:

try:

fp = urllib2.urlopen(ajaxAddr)

break

except urllib2.HTTPError,e: #状态错误

Tools.writelog('error','HTTP状态错误 code='+"%d"%e.code)

return ""

except urllib2.URLError,e: #网络错误超时

Tools.writelog('warn','页面访问超时,重试..')

retry+=1

if( retry > self.MAX_RETRY ):

Tools.writelog('warn','超过最大重试次数,放弃')

return ""

content = fp.read()

fp.close()

views = Tools.getFromPatten(u'"viewcount":(.*?),',content,True)

views = views.replace('"','')

return views

#从网页内容中爬取点击量

def gripViewsFromData(self,data):

views = Tools.getFromPatten(u'已经有<.*?>(.*?)<.*?>次观看',data,True)

return views

def gripBaseDir(self,data):

dir = Tools.getFromPatten(u"base_dir = '(.*?)'",data,True)

return dir

#抓取数据

def gripinfo(self,url):

try:

data = self.pageGripper.getPageInfo(url)

except:

Tools.writelog ("error", url+" 抓取失败")

raise

Tools.writelog('info','开始内容匹配')

rst = {}

rst['title'] = self.griptitle(data)

rst['channel'] = self.gripchannel(data)

rst['tag'] = self.griptag(data)

rst['release'] = self.griptime(data)

rst['user'] = self.gripuser(data)

ccdata = self.getCCData(data)

rst['ccsiteId'] = ccdata[0]

rst['ccVid'] = ccdata[1]

views = self.gripViewsFromData(data)

if views =="" or not views:

vid = self.gripVideoId(data)

basedir = self.gripBaseDir(data)

views = self.gripViewsAjax(vid,url,basedir)

if( views == "" ):

views = "error"

if( views == "error"):

Tools.writelog("error","获取观看次数失败")

Tools.writelog("debug","点击量:"+views)

rst['views'] = views

Tools.writelog('debug','title=%s,channel=%s,tag=%s'%(rst['title'],rst['channel'],rst['tag']))

return rst

'''

单元测试

'''

if __name__ == '__main__':

list = [

'http://008yx.com/xbsp/index.php/video/index/3138',

'http://vblog.xwhb.com/index.php/video/index/4067',

'http://demo.ccvms.bokecc.com/index.php/video/index/3968',

'http://vlog.cnhubei.com/wuhan/20100912_56145.html',

'http://vlog.cnhubei.com/html/js/30271.html',

'http://www.ddvtv.com/index.php/video/index/15',

'http://boke.2500sz.com/index.php/video/index/60605',

'http://video.zgkqw.com/index.php/video/index/334',

'http://yule.hitmv.com/html/joke/27041.html',

'http://www.ddvtv.com/index.php/video/index/11',

'http://www.zgnyyy.com/index.php/video/index/700',

'http://www.kdianshi.com/index.php/video/index/5330',

'http://www.aoyatv.com/index.php/video/index/127',

'http://v.ourracing.com/html/channel2/64.html',

'http://v.zheye.net/index.php/video/index/93',

'http://vblog.thmz.com/index.php/video/index/7616',

'http://kdianshi.com/index.php/video/index/5330',

'http://tv.seeyoueveryday.com/index.php/video/index/95146',

'http://sp.zgyangzhi.com/html/ji/2.html',

'http://www.xjapan.cc/index.php/video/index/146',

'http://www.jojy.cn/vod/index.php/video/index/399',

'http://v.cyzone.cn/index.php/video/index/99',

]

list1 = ['http://192.168.25.7:8079/vinfoant/versionasdfdf']

infoGripper = InfoGripper()

for url in list:

infoGripper.gripinfo(url)

del infoGripper

WEB服务及任务调度

'''

Created on 2010-9-15

@author: chenggong

'''

# -*- coding: utf-8 -*-

import string,cgi,time

from os import curdir,sep

from BaseHTTPServer import BaseHTTPRequestHandler,HTTPServer

from InfoGripper import *

import re

import MySQLdb

import time

import threading

import urllib

import urllib2

PORT = 8079

VERSION = 0.1

DBCHARSET = "utf8"

PARAMS = [

'callback',

'sessionId',

'retry',

'retryInterval',

'dbhost',

'dbport',

'db',

'dbuser',

'dbpass',

'videoId'

]

DBMAP = ['video_id',

'ccsiteid',

'ccvid',

'desc_url',

'site_id',

'title',

'post_time',

'author',

'elapse',

'channel',

'tags',

'create_time',

'check_time',

'status']

'''

ERROR CODE定义

'''

ERR_OK = 0

ERR_PARAM = 1

ERR_HTTP_TIMEOUT = 5

ERR_HTTP_STATUS = 6

ERR_DB_CONNECT_FAIL = 8

ERR_DB_SQL_FAIL = 9

ERR_GRIPVIEW = 11

ERR_UNKNOW = 12

'''

数据库适配器

'''

class DBAdapter(object):

def __init__(self):

self.param = {'ip':'',

'port':0,

'user':'',

'pw':'',

'db':''}

self.connect_once = False #是否连接过数据库

'''

创建/更新数据库连接池

'''

def connect(self,ip,port,user,pw,db):

if( ip != self.param['ip'] or

port != self.param['port'] or

user != self.param['user'] or

pw != self.param['pw'] or

db != self.param['db']):

Tools.writelog('info','更换数据库连接池,ip='+ip+',port='+port+',user='+user+',pw='+pw+',db='+db)

try:

if self.connect_once == True: #释放上次连接

self.cur.close()

self.conn.close()

self.conn=MySQLdb.connect(user=user,passwd=pw,db=db,host=ip,port=int(port))

self.conn.set_character_set(DBCHARSET)

self.connect_once = True

self.cur=self.conn.cursor(MySQLdb.cursors.Cursor)

self.param['ip'] = ip

self.param['port'] = port

self.param['user'] = user

self.param['pw'] = pw

self.param['db'] = db

except:

Tools.writelog('error',u'数据库连接失败',True)

raise

else:

Tools.writelog('info',u'数据库连接成功')

'''

执行SQL语句

'''

def execute(self,sql):

Tools.writelog('debug',u'执行SQL: '+sql)

try:

self.cur.execute(sql)

except:

Tools.writelog('error',u'SQL执行错误:'+sql)

raise

'''

查询数据库

'''

def query(self,sql):

row = {}

self.execute(sql)

row=self.cur.fetchall()

return row

'''

视频错误

'''

def updateErr(self,videoId):

nowtime = time.strftime('%Y-%m-%d-%H-%M-%S',time.localtime(time.time()))

sql = "UPDATE videos SET "

sql += "check_time='" + nowtime +"',"

sql += "status=-1 "

sql += "WHERE video_id="+videoId

self.execute(sql)

self.conn.commit()

'''

更新查询结果

'''

def update(self,obj,videoId,isUpdateTitle=True):

Tools.writelog('debug','开始更新数据库')

try:

#更新video表

sql = "UPDATE videos SET "

if(obj['ccsiteId'] !="" ):

sql += "ccsiteid='" + obj['ccsiteId'] + "',"

if(obj['ccVid'] != "" ):

sql += "ccvid='" + obj['ccVid'] + "',"

if isUpdateTitle:

sql += "title='" + obj['title'] + "',"

sql += "post_time='" + obj['release'] + "',"

sql += "author='" + obj['user'] + "',"

sql += "channel='" + obj['channel'] + "',"

sql += "tags='" + obj['tag'] + "',"

nowtime = time.strftime('%Y-%m-%d-%H-%M-%S',time.localtime(time.time()))

sql += "check_time='" + nowtime +"',"

sql += "status=0 "

sql += "WHERE video_id="+videoId

self.execute(sql)

#更新count表

if( obj['views'] != 'error' ):

nowdate = time.strftime('%Y-%m-%d',time.localtime(time.time()))

sql = "SELECT * FROM counts WHERE "

sql += "date = '" + nowdate + "' and video_id=" + videoId

rst = self.query(sql)

if len(rst) > 0:#如果当天已有记录，则更新

sql = "UPDATE counts SET count="+obj['views']

sql +=" WHERE video_id=" + videoId + " AND date='" + nowdate+ "'"

else:#否则插入

sql = "INSERT INTO counts VALUES"

sql += "(null," +videoId+",'"+nowdate+"',"+obj['views'] + ")"

self.execute(sql)

self.conn.commit()

Tools.writelog('debug', "db commit ok")

return ERR_OK

except Exception,e:

print e

return ERR_DB_SQL_FAIL

'''

任务线程类

'''

class TaskThread(threading.Thread):

def setTaskTool(self,dbAdapter,gripper):

self.dbAdapter = dbAdapter

self.gripper = gripper

def setParam(self,param):

self.param = param

self.videoId = param['videoId']

assert self.videoId != ""

def init(self):

self.views = "0"

self.errcode = ERR_OK

def run(self):

Tools.writelog('debug','开始爬虫任务,sessionId='+self.param['sessionId'])

self.init()

try:

#更新数据库连接

self.dbAdapter.connect(self.param['dbhost'],self.param['dbport'],self.param['dbuser'],self.param['dbpass'],self.param['db'])

except:

self.errcode = ERR_DB_CONNECT_FAIL #数据库连接失败

callback(self.errcode)

return

#查询该vid的视频

sql = "SELECT "

for column in DBMAP:

sql += column

if column != DBMAP[len(DBMAP)-1]:

sql += ","

sql += " FROM videos"

sql += " WHERE video_id="+self.videoId

video = self.dbAdapter.query(sql)

assert not (len(video)>1 or len(video)==0) #有且仅有一条记录

url = video[0][3]

assert url != ""

try:

rst = self.gripper.gripinfo(url)

except urllib2.HTTPError,e:

self.errcode = ERR_HTTP_STATUS #HTTP状态错误

self.dbAdapter.updateErr(self.videoId)

except urllib2.URLError,e:

self.errcode = ERR_HTTP_TIMEOUT #HTTP连接超时

self.dbAdapter.updateErr(self.videoId)

except:

self.errcode = ERR_UNKNOW #未知错误

self.dbAdapter.updateErr(self.videoId)

else:

self.views = rst['views']

if self.views == "error":

self.views = "-1"

self.errcode = ERR_GRIPVIEW #数据抓取成功，点击量抓取失败

#更新数据库(特殊处理，如果原title中有 "-" 则不更新title字段)

title = video[0][5]

assert title != ""

if re.match('.*-.*', title):

self.errocde = self.dbAdapter.update(rst,self.videoId,True)

else:

self.errcode = self.dbAdapter.update(rst,self.videoId)

self.callback(self.errcode)

Tools.writelog('info','任务结束,sessionId='+self.param['sessionId'])

return

def callback(self,errcode):

results = {'errorcode':errcode,'count':int(self.views)}

results = urllib.urlencode(results)

results = results.replace('&', '%26')

url = self.param['callback']

url += "?"

url += "sessionId=" + self.param['sessionId']

url += "&results=" + results

retry = 0

while True:

try:

Tools.writelog('debug',"回调主控,url="+url)

urllib2.urlopen(url)

Tools.writelog('debug','回调成功')

break

except urllib2.URLError, e: #超时、错误

Tools.writelog('debug','回调主控超时,%s秒后重试'%self.param['retryInterval'])

retry+=1

time.sleep(int(self.param['retryInterval']))

if( retry > int(self.param['retry'])):

Tools.writelog('error','回调主控失败')

return

'''

WEB服务类

'''

class MyHandler(BaseHTTPRequestHandler):

dbAdapter = DBAdapter()

gripper = InfoGripper()

def pageSuccess(self):

self.send_response(200)

self.send_header('Content-type', 'text/html')

self.end_headers()

def pageFail(self):

self.send_error(404, "not found")

def getValue(self,param):

src = self.path + '&'

reg = param + '=' + '(.*?)&'

value = Tools.getFromPatten(reg,src,True)

return value

def do_GET(self):

isGetVersion = re.match('.*vinfoant/version.*', self.path)

isTask = re.match('.*vinfoant/run.*', self.path)

if( isGetVersion ):

self.pageSuccess()

self.wfile.write(VERSION)

elif( isTask ):

self.pageSuccess()

param = {}

for p in PARAMS:

param[p] = self.getValue(p) #获取各项参数

taskThread = TaskThread()

taskThread.setTaskTool(self.dbAdapter, self.gripper)

taskThread.setParam(param)

taskThread.start()#启动任务线程

self.wfile.write("ok")

else:

self.pageFail()

return

'''

启动WEB服务，全局入口

'''

def startHttpd():

try:

Tools.writelog('debug','httpd start..listen on '+str(PORT))

httpd = HTTPServer(('',PORT), MyHandler )

Tools.writelog('debug','success')

httpd.serve_forever()

except KeyboardInterrupt:

Tools.writelog('debug','httpd close..')

httpd.socket.close()

if __name__ == '__main__':

startHttpd()

转载自：http://blog.csdn.net/rcfalcon/article/details/5891781  (课课家）