1. 程式人生 > >python爬取電影天堂的下載連結

python爬取電影天堂的下載連結

電影天堂下載連結都是magnet的,搞下來想下就下沒有廣告

# coding=utf-8
import urllib.request
#import requests
import re
import random
import json

#爬取電影天堂電影下載連結
def get_film(url):
    #設定模擬瀏覽器擡頭
    ua_list = [
        "Mozilla/5.0 (Windows NT 6.1; ) Apple.... ",
        "Mozilla/5.0 (X11; CrOS i686 2268.111.0)... ",
        "Mozilla/5.0 (Macintosh; U; PPC Mac OS X.... "
, "Mozilla/5.0 (Macintosh; Intel Mac OS... " ] user_agent = random.choice(ua_list) #構建請求 myRequest = urllib.request.Request(url) #也可以通過呼叫Request.add_header() 新增/修改一個特定的header myRequest.add_header("Connection", "keep-alive") #也可以通過呼叫Request.add_header() 新增/修改一個特定的header
myRequest.add_header("User-Agent", user_agent) #第一個字母大寫,後面的全部小寫 myRequest.get_header("User-agent") #請求響應 response = urllib.request.urlopen(myRequest) #讀取並轉碼為GBK的字串 try: encodeHtml = response.read() #使用"ignore"忽略錯誤 html = encodeHtml.decode("GB18030",
"ignore") detail_list=re.findall('<a href="(.*?)" class="ulink',html) for m in detail_list: b_url = 'http://www.dytt8.net/%s'%m #html_2=requests.get(b_url) #構建請求 request2 = urllib.request.Request(b_url) #也可以通過呼叫Request.add_header() 新增/修改一個特定的header request2.add_header("Connection", "keep-alive") #也可以通過呼叫Request.add_header() 新增/修改一個特定的header request2.add_header("User-Agent", user_agent) #第一個字母大寫,後面的全部小寫 request2.get_header("User-agent") #請求響應 response2 = urllib.request.urlopen(request2) #html_2 = response2.read().decode("GBK") 這個可能報錯 用下面的 html_2 = response2.read().decode("GB18030","ignore") #print(html_2) #新建一個字典 movieInfo = {'name':'','ftp':'','magnet':'','info':'','originalName':'','grade':''} ftp = re.findall('<a href="(.*?)">ftp.*?</a></td>',html_2) if ftp != []: #print (ftp[0]) movieInfo['ftp']=ftp[0] name = re.findall('com.*?.m',ftp[0]) if name != []: #print ('電影名稱:'+name[0].split('.')[1]) movieInfo['name']=name[0].split('.')[1] magnet = re.findall('<a href="magnet:(.*?)"><strong><font style="BACKGROUND-COLOR: #ff9966"><font color="#0000ff"><font size="4">磁力鏈下載點選這裡</font></font></font></strong></a>',html_2) if magnet != []: #print (magnet[0]) movieInfo['magnet']='magnet:'+magnet[0] Info = re.findall(' /> <br /><br />.*?<br /><br /><img',html_2) if Info != []: tmpInfo = Info[0].split('/>',1)[1].split('<img',1)[0] #print (tmpInfo) movieInfo['info']= tmpInfo #原始片名 originalNameList = re.findall('◎片  名.*?<br />',tmpInfo) if originalNameList !=[]: movieInfo['originalName'] = originalNameList[0].split('◎片  名',1)[1].split('<br />',1)[0] #IMDb評分 滿分10分 這裡擷取我寫的差 還是大家自己去匹配把 gradeList = re.findall('◎IMDb評分.*?<br />',tmpInfo) if gradeList !=[]: #print (gradeList[0].split('/',1)[0]) if len(gradeList[0].split('/',1)[0].split(' ',1))>1: #print (gradeList[0].split('/',1)[0].split(' ',1)[1]) movieInfo['grade'] = gradeList[0].split('/',1)[0].split(' ',1)[1] if len(gradeList[0].split('/',1)[0].split(' ',1))>1: #print (gradeList[0].split('/',1)[0].split(' ',1)[1]) movieInfo['grade'] = gradeList[0].split('/',1)[0].split(' ',1)[1] for key in movieInfo: print (key, ' value : ', movieInfo[key]) ''' #將資料儲存到本地 #jsonStr = json.dumps(movieInfo) #print (jsonStr) with open('d:\movieList.txt','a',encoding='utf-8') as f: #寫文字write到本地 f.write(jsonStr+',') ''' #傳送到java伺服器 postDaa = urllib.parse.urlencode(movieInfo).encode('utf-8') header_dict = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Trident/7.0; rv:11.0) like Gecko'} #我的地址就不貼了 url3='https://www.xxx.xxxPython.json' req3 = urllib.request.Request(url=url3,data=postDaa,headers=header_dict,method = 'POST') res3 = urllib.request.urlopen(req3) res3 = res3.read() print(res3) #print(res3.decode(encoding='utf-8')) except Exception as e: print('發生了異常:',e) ''' 這裡是單執行緒的 for n in range(1,110): #goUrl = 'http://www.dytt8.net/html/gndy/dyzz/list_23_'+str(n)+'.html' goUrl = 'http://www.dytt8.net/html/gndy/china/list_4_'+str(n)+'.html' print(goUrl) get_film(goUrl) '''

一個一個挨著去爬肯定慢啊,建一個多執行緒的去爬 mui.py

這個是參照了 AutomationTesting的部落格內容,很多欄位我也沒改。

# -*-coding:utf-8-*-
import threading;
import main;
mutex_lock = threading.RLock();  # 互斥鎖的宣告
ticket = 180;  # 總頁碼
start = 1; # 起始頁碼
# 用於統計各個執行緒的得票數
ticket_stastics=[];
   
class myThread(threading.Thread):  # 執行緒處理函式
    def __init__(self, name): 
        threading.Thread.__init__(self);  # 執行緒類必須的初始化
        self.thread_name = name;  # 將傳遞過來的name構造到類中的name
    def run(self):
        # 宣告在類中使用全域性變數
        global mutex_lock;
        global ticket;   
        global start;  
        while 1:          
            if ticket > start:                
                start += 1;
                goUrl = 'http://www.dytt8.net/html/gndy/dyzz/list_23_'+str(start)+'.html'
                print (goUrl)
                #呼叫你寫的那個類
                main.get_film(goUrl)          
                ticket_stastics[self.thread_name]+=1;                                     
            else:                
                break;
        mutex_lock.release();  # python線上程死亡的時候,不會清理已存在線上程函式的互斥鎖,必須程式猿自己主動清理
        print ("%s被銷燬了!" % (self.thread_name));  
              
# 初始化執行緒 
threads = [];#存放執行緒的陣列,相當於執行緒池
for i in range(0,50):#執行緒數量,電腦好可以多設點2333
    thread = myThread(i);#指定執行緒i的執行函式為myThread
    threads.append(thread);#先講這個執行緒放到執行緒threads
    ticket_stastics.append(0);# 初始化執行緒的得票數統計陣列
for t in threads:#讓執行緒池中的所有陣列開始
    t.start(); 
for t in threads:
    t.join();#等待所有執行緒執行完畢才執行一下的程式碼
print ("錄入完畢!");
print ("=========統計=========");
for i in range(0,len(ticket_stastics)):
    print ("執行緒%d:%d頁" % (i,ticket_stastics[i]));

這兩個執行一下就可以爬了;平常替換下goUrl就可以了~

多執行緒有可能會發生WinError 10060報錯,使用time.sleep(1)避免

發生了異常: <urlopen error [WinError 10060] 由於連線方在一段時間後沒有正確答覆或連線的主機沒有反應,連線嘗試失敗。>
這是由於連線數太多被對方防火牆ban了… 發生這類錯誤就暫停一下再繼續應該就好了,估計是防DDoS攻擊?