python中怎么實現(xiàn)分布式抓取網(wǎng)頁,很多新手對此不是很清楚,為了幫助大家解決這個難題,下面小編將為大家詳細講解,有這方面需求的人可以來學習下,希望你能有所收獲。
成都創(chuàng)新互聯(lián)公司專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務,包含不限于成都做網(wǎng)站、網(wǎng)站制作、安吉網(wǎng)絡(luò)推廣、重慶小程序開發(fā)、安吉網(wǎng)絡(luò)營銷、安吉企業(yè)策劃、安吉品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運營等,從售前售中售后,我們都將竭誠為您服務,您的肯定,是我們最大的嘉獎;成都創(chuàng)新互聯(lián)公司為所有大學生創(chuàng)業(yè)者提供安吉建站搭建服務,24小時服務熱線:13518219792,官方網(wǎng)址:www.cdcxhl.com
[python]
view plaincopy
'''''
Created on 2010-9-15
@author: chenggong
'''
import urllib2
import re
import socket
DEBUG = 0
'''''
工具類
'''
class Tools():
#log函數(shù)
@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]報告管理員!!"
#轉(zhuǎn)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()
'''''
網(wǎng)頁內(nèi)容爬蟲
'''
class PageGripper():
URL_OPEN_TIMEOUT = 10 #網(wǎng)頁超時時間
MAX_RETRY = 3 #最大重試次數(shù)
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: #狀態(tài)錯誤
Tools.writelog('error','HTTP狀態(tài)錯誤 code='+e.code)
raise urllib2.HTTPError
except urllib2.URLError,e: #網(wǎng)絡(luò)錯誤超時
Tools.writelog('warn','頁面訪問超時,重試..')
retry+=1
if( retry > self.MAX_RETRY ):
Tools.writelog('warn','超過最大重試次數(shù),放棄')
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","開始抓取網(wǎng)頁,url= "+url)
info = ""
try:
info = self.downloadUrl(url)
except:
raise
Tools.writelog("debug","網(wǎng)頁抓取成功")
return info
'''''
內(nèi)容提取類
'''
class InfoGripper():
pageGripper = PageGripper()
def __init__(self):
Tools.writelog('debug',"爬蟲啟動")
#抓取標題
def griptitle(self,data):
title = Tools.getFromPatten(u'box2t sp">
if title == "":
title = Tools.getFromPatten(u'
return title.strip()
#抓取頻道
def gripchannel(self,data):
zone = Tools.getFromPatten(u'頻道:(.*?)',data,True)
channel = Tools.getFromPatten(u'
return channel
#抓取標簽
def griptag(self,data):
zone = Tools.getFromPatten(u'標簽:(.*?)[^a].*>',data,True);
rst = Tools.getFromPatten(u'>(.*?)',zone,False);
return rst
#抓取觀看次數(shù)
def gripviews(self,data):
rst = Tools.getFromPatten(u'已經(jīng)有(.*?)次觀看',data);
return rst
#抓取發(fā)布時間
def griptime(self,data):
rst = Tools.getFromPatten(u'在(.*?)發(fā)布',data,True)
return rst
#抓取發(fā)布者
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相關(guān)數(shù)據(jù)
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
#獲取站內(nèi)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: #狀態(tài)錯誤
Tools.writelog('error','HTTP狀態(tài)錯誤 code='+"%d"%e.code)
return ""
except urllib2.URLError,e: #網(wǎng)絡(luò)錯誤超時
Tools.writelog('warn','頁面訪問超時,重試..')
retry+=1
if( retry > self.MAX_RETRY ):
Tools.writelog('warn','超過最大重試次數(shù),放棄')
return ""
content = fp.read()
fp.close()
views = Tools.getFromPatten(u'"viewcount":(.*?),',content,True)
views = views.replace('"','')
return views
#從網(wǎng)頁內(nèi)容中爬取點擊量
def gripViewsFromData(self,data):
views = Tools.getFromPatten(u'已經(jīng)有<.*?>(.*?)<.*?>次觀看',data,True)
return views
def gripBaseDir(self,data):
dir = Tools.getFromPatten(u"base_dir = '(.*?)'",data,True)
return dir
#抓取數(shù)據(jù)
def gripinfo(self,url):
try:
data = self.pageGripper.getPageInfo(url)
except:
Tools.writelog ("error", url+" 抓取失敗")
raise
Tools.writelog('info','開始內(nèi)容匹配')
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","獲取觀看次數(shù)失敗")
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服務及任務調(diào)度
[python]
view plaincopy
'''''
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
'''''
數(shù)據(jù)庫適配器
'''
class DBAdapter(object):
def __init__(self):
self.param = {'ip':'',
'port':0,
'user':'',
'pw':'',
'db':''}
self.connect_once = False #是否連接過數(shù)據(jù)庫
'''''
創(chuàng)建/更新數(shù)據(jù)庫連接池
'''
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','更換數(shù)據(jù)庫連接池,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'數(shù)據(jù)庫連接失敗',True)
raise
else:
Tools.writelog('info',u'數(shù)據(jù)庫連接成功')
'''''
執(zhí)行SQL語句
'''
def execute(self,sql):
Tools.writelog('debug',u'執(zhí)行SQL: '+sql)
try:
self.cur.execute(sql)
except:
Tools.writelog('error',u'SQL執(zhí)行錯誤:'+sql)
raise
'''''
查詢數(shù)據(jù)庫
'''
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()
'''''
更新查詢結(jié)果
'''
def update(self,obj,videoId,isUpdateTitle=True):
Tools.writelog('debug','開始更新數(shù)據(jù)庫')
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)