1. python網路爬蟲-APP端爬蟲
一、環境安裝
1.1 模擬器安裝
藉助模擬器進行APP端調試,通過下載安裝可實現。推薦使用夜神模擬器(yeshen.com/)或網易MuMu模擬器(mumu.163.com/)。
1.2 SDK安裝
提供多種下載渠道,首選官網下載(developer.android.com/s...)或第三方下載平台(androiddevtools.cn/)。使用SDK Manager.exe安裝工具,選擇需要的工具,如Build-tools和特定Android版本,同時勾選Extras中的選項,最後點擊【Install】安裝。注意,安裝過程可能持續數小時。配置環境變數,設置ANDROID_HOME為sdk安裝目錄,並將平台工具和工具路徑添加到Path環境變數中。
1.3 Fiddler安裝
直接從官網下載安裝(telerik.com/download/fi...)以獲取http協議調試代理工具Fiddler,用於記錄和檢查所有電腦與互聯網之間的http通信。
1.4 Appium安裝
通過官網下載(appium.io/)安裝appium,一個跨平台的自動化測試工具,支持iOS、Android應用及web應用測試。
1.5 mitmproxy庫安裝
安裝mitmproxy庫時,通過pip方式可能遇到錯誤,建議下載whl文件安裝(pypi.org/project/mitmpr...)。確保設置好證書,以便在監聽HTTPS請求時工作。
二、抓包工具的使用
2.1 Fiddler配置
確保Fiddler的Capture Traffic功能開啟,然後用瀏覽器訪問網頁,Fiddler中會顯示抓取的數據包。對於HTTPS配置,通過Tools->Options進行設置,並確保Allow remote computers connect選項開啟。
2.2 模擬器配置
在模擬器設置中,手動更改代理設置,輸入本機IP和Fiddler埠,完成與Fiddler的代理連接。
三、移動端自動化控制
3.1 APK包名獲取
通過adb命令獲取apk包名,確保adb服務啟動並連接模擬器,然後在模擬器中獲取所需app的包名和Activity。
3.2 Appium使用
使用appium的python包,啟動appium服務,編寫示例代碼操作模擬器,並使用uiautomatorviewer獲取元素的Xpath路徑。
四、利用mitmproxy抓取存儲數據
4.1 基本原理
mitmproxy提供命令行介面mitmmp,用於處理抓取的數據,並將其存儲到資料庫中,同時支持Python腳本處理請求和響應。
4.2 抓取步驟
使用fiddler分析請求,然後通過mitmmp攔截並保存數據至MySQL資料庫。
五、APK脫殼反編譯
5.1 脫殼
使用Xposed框架安裝FDex2工具,通過Hook ClassLoader方法脫殼APK。推薦從網路下載並安裝FDex2工具。
5.2 APK反編譯
使用apktool反編譯apk文件以獲取靜態資源,而dex2jar則將.dex文件轉換為Java源代碼。此過程需謹慎處理多個.dex文件。
5.3 JAD-反編譯class文件
藉助GitHub上的JAD工具將.class文件反編譯為Java源代碼,便於閱讀和理解。
2. 如何使用python爬蟲jfinal
一、gzip/deflate支持
現在的網頁普遍支持gzip壓縮,這往往可以解決大量傳輸時間,以VeryCD的主頁為例,未壓縮版本247K,壓縮了以後45K,為原來的1/5。這就意味著抓取速度會快5倍。
然而python的urllib/urllib2默認都不支持壓縮,要返回壓縮格式,必須在request的header裡面寫明』accept-
encoding』,然後讀取response後更要檢查header查看是否有』content-encoding』一項來判斷是否需要解碼,很繁瑣瑣
碎。如何讓urllib2自動支持gzip, defalte呢?其實可以繼承BaseHanlder類,然後build_opener的方式來處理:
import urllib2
from gzip import GzipFile
from StringIO import StringIO
class ContentEncodingProcessor(urllib2.BaseHandler):
"""A handler to add gzip capabilities to urllib2 requests """
# add headers to requests
def http_request(self, req):
req.add_header("Accept-Encoding", "gzip, deflate")
return req
# decode
def http_response(self, req, resp):
old_resp = resp
# gzip
if resp.headers.get("content-encoding") == "gzip":
gz = GzipFile(
fileobj=StringIO(resp.read()),
mode="r"
)
resp = urllib2.addinfourl(gz, old_resp.headers, old_resp.url, old_resp.code)
resp.msg = old_resp.msg
# deflate
if resp.headers.get("content-encoding") == "deflate":
gz = StringIO( deflate(resp.read()) )
resp = urllib2.addinfourl(gz, old_resp.headers, old_resp.url, old_resp.code) # 'class to add info() and
resp.msg = old_resp.msg
return resp
# deflate support
import zlib
def deflate(data): # zlib only provides the zlib compress format,not the deflate format;
try: # so on top of all there's this workaround:
return zlib.decompress(data, -zlib.MAX_WBITS)
except zlib.error:
return zlib.decompress(data)然後就簡單了,
encoding_support = ContentEncodingProcessor
opener = urllib2.build_opener( encoding_support, urllib2.HTTPHandler )
#直接用opener打開網頁,如果伺服器支持gzip/defalte則自動解壓縮
content = opener.open(url).read()二、更方便地多線程
總結一文的確提及了一個簡單的多線程模板,但是那個東東真正應用到程序裡面去只會讓程序變得支離破碎,不堪入目。在怎麼更方便地進行多線程方面我也動了一番腦筋。先想想怎麼進行多線程調用最方便呢?
1、用twisted進行非同步I/O抓取
事實上更高效的抓取並非一定要用多線程,也可以使用非同步I/O法:直接用twisted的getPage方法,然後分別加上非同步I/O結束時的callback和errback方法即可。例如可以這么干:
from twisted.web.client import getPage
from twisted.internet import reactor
links = [ 'http://www.verycd.com/topics/%d/'%i for i in range(5420,5430) ]
def parse_page(data,url):
print len(data),url
def fetch_error(error,url):
print error.getErrorMessage(),url
# 批量抓取鏈接
for url in links:
getPage(url,timeout=5)
.addCallback(parse_page,url) #成功則調用parse_page方法
.addErrback(fetch_error,url) #失敗則調用fetch_error方法
reactor.callLater(5, reactor.stop) #5秒鍾後通知reactor結束程序
reactor.run()twisted人如其名,寫的代碼實在是太扭曲了,非正常人所能接受,雖然這個簡單的例子看上去還好;每次寫twisted的程序整個人都扭曲了,累得不得了,文檔等於沒有,必須得看源碼才知道怎麼整,唉不提了。
如果要支持gzip/deflate,甚至做一些登陸的擴展,就得為twisted寫個新的HTTPClientFactory類諸如此類,我這眉頭真是大皺,遂放棄。有毅力者請自行嘗試。
這篇講怎麼用twisted來進行批量網址處理的文章不錯,由淺入深,深入淺出,可以一看。
2、設計一個簡單的多線程抓取類
還是覺得在urllib之類python「本土」的東東裡面折騰起來更舒服。試想一下,如果有個Fetcher類,你可以這么調用
f = Fetcher(threads=10) #設定下載線程數為10
for url in urls:
f.push(url) #把所有url推入下載隊列
while f.taskleft(): #若還有未完成下載的線程
content = f.pop() #從下載完成隊列中取出結果
do_with(content) # 處理content內容這么個多線程調用簡單明了,那麼就這么設計吧,首先要有兩個隊列,用Queue搞定,多線程的基本架構也和「技巧總結」一文類似,push方法和
pop方法都比較好處理,都是直接用Queue的方法,taskleft則是如果有「正在運行的任務」或者」隊列中的任務」則為是,也好辦,於是代碼如
下:import urllib2
from threading import Thread,Lock
from Queue import Queue
import time
class Fetcher:
def __init__(self,threads):
self.opener = urllib2.build_opener(urllib2.HTTPHandler)
self.lock = Lock() #線程鎖
self.q_req = Queue() #任務隊列
self.q_ans = Queue() #完成隊列
self.threads = threads
for i in range(threads):
t = Thread(target=self.threadget)
t.setDaemon(True)
t.start()
self.running = 0
def __del__(self): #解構時需等待兩個隊列完成
time.sleep(0.5)
self.q_req.join()
self.q_ans.join()
def taskleft(self):
return self.q_req.qsize()+self.q_ans.qsize()+self.running
def push(self,req):
self.q_req.put(req)
def pop(self):
return self.q_ans.get()
def threadget(self):
while True:
req = self.q_req.get()
with self.lock: #要保證該操作的原子性,進入critical area
self.running += 1
try:
ans = self.opener.open(req).read()
except Exception, what:
ans = ''
print what
self.q_ans.put((req,ans))
with self.lock:
self.running -= 1
self.q_req.task_done()
time.sleep(0.1) # don't spam
if __name__ == "__main__":
links = [ 'http://www.verycd.com/topics/%d/'%i for i in range(5420,5430) ]
f = Fetcher(threads=10)
for url in links:
f.push(url)
while f.taskleft():
url,content = f.pop()
print url,len(content)