爬蟲目標說明：Goodinfo! 台灣股市資訊網 - 類股分類表 今天的工作坊以取得類股一覽表為例，有兩個目標要達成： 取得六個大類下的類別清單（上市、上櫃、興櫃、電子產業、概念股、集團股） 取得各個類別中包含的公司清單 公司代號 公司名稱 載入套件 from bs4 import BeautifulSoup # 解析網頁結構 import json # 讀寫 json 檔案 import numpy as np # 產生整數亂數 import requests # 發送 HTTP 請求 import re # 用來 regex 規則篩選頁面上的 ip import random # 用來隨機選取列表中的 ip from tqdm import tqdm # 顯示迴圈進度條 from time import sleep # 設定延時時長 設定變數 # 設定 headers default_headers = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-TW,zh;q=0.9,en-US;q=0.8,en;q=0.7", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Sec-Gpc": "1", "Referer": "https://www.google.com/", "Dnt": "1", "Upgrade-Insecure-Requests": "1", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.5005.61 Safari/537.36" } # 使用 proxy 時的 timeout 上限 timeout = 20 爬蟲規則式 頁面觀察與測試 測試取得各類別下的子類別清單 url = 'https://goodinfo.tw/StockInfo/StockList.asp' res = requests.get(url, headers=default_headers).content soup = BeautifulSoup(res, 'html.parser') # 利用節點特徵選取有子類別名稱的區塊 list_stock_cate = soup.find_all('td', {'colspan': '4'}) list_stock_cate = [str(item).split('集團股_')[-1].split('@*=')[0] for item in list_stock_cate] list_stock_cate = [item for item in list_stock_cate if "<" not in item] list_stock_cate 測試取得該子類別下的所有個股清單 import pandas as pd table = soup.find('table', {'class': 'r10_0_0_10 b1 p4_1'}) df = pd.read_html(str(table))[0] df market_cat = "上市" ind = "航運業" url = f'https://goodinfo.tw/StockInfo/StockList.asp?MARKET_CAT={market_cat}&INDUSTRY_CAT={ind}' res = requests.get(url, headers=default_headers).content soup = BeautifulSoup(res, 'html.parser') # 利用節點特徵選取有個股表格的區塊 soup.find('table', {'class': 'r10_0_0_10 b1 p4_1'}).find_all('td') # 排除雜訊只保留公司代號和名稱 list_stock = [item.text for item in soup.find('table', {'class': 'r10_0_0_10 b1 p4_1'}).find_all('td')] list_stock = [(x, y) for x, y in zip(list_stock[::2], list_stock[1::2]) if x != '代號'] list_stock 打包成爬蟲函式 將剛才嘗試成功的爬蟲規則，整理成如下函式：def getStockCates(): """ 取得各類別下的子類別清單 """ market_categories = ['上市','上櫃','興櫃','電子產業','概念股','集團股'] subcates = {} url = f'https://goodinfo.tw/StockInfo/StockList.asp' res = requests.get(url, headers=default_headers).content soup = BeautifulSoup(res, 'html.parser') list_stock_cate = soup.find_all('td', {'colspan': '4'}) for market_cat in tqdm(market_categories): _subcates = [str(item).split(f'{market_cat}_')[-1].split('@*=')[0] for item in list_stock_cate] _subcates = [ item.replace('="" ','').replace('colspan="4"','') for item in _subcates if ("<" not in item) and ("全部" not in item) ] subcates.update({market_cat: _subcates}) return subcates def getCateStockList(market_cat, ind): """ 取得該子類別下的所有個股 """ url = f'https://goodinfo.tw/StockInfo/StockList.asp?MARKET_CAT={market_cat}&INDUSTRY_CAT={ind}' res = requests.get(url, headers=default_headers).content soup = BeautifulSoup(res, 'html.parser') list_stock = [item.text for item in soup.find('table', {'class': 'r10_0_0_10 b1 p4_1'}).find_all('td')] list_stock = [(x, y) for x, y in zip(list_stock[::2], list_stock[1::2]) if x != '代號'] return list_stock 爬蟲流程 1) 取得每個類別下的產業名稱 # 取得六大類別下的子類清單 industry_dict = getStockCates() industry_dict 2) 遍歷產業字典中的項目，取得每一項內的公司清單 ind_stocks = {} # 遍歷 6 個大類 for ind, sub_inds in industry_dict.items(): ind_stocks.update({ind: {}}) # 遍歷該大類下的子類別清單 for sub_ind in tqdm(sub_inds): # try 正常情況下的執行內容 try: url = f'https://goodinfo.tw/StockInfo/StockList.asp?MARKET_CAT={ind}&INDUSTRY_CAT={sub_ind}' list_stock = getCateStockList(market_cat=ind, ind=sub_ind) # except 若取得資料的過程發生錯誤 except Exception as e: print("Error at: ", url, "\n", e) list_stock = [] # 印出當前頁面的文字確認情況 res = requests.get(url, headers=default_headers) res.encoding = "utf-8" print(res.text) # finally 不論是否出現 Exception 都會執行 finally: ind_stocks[ind].update({sub_ind: list_stock}) print(sub_ind, ind_stocks[ind][sub_ind], '\n') sleep(np.random.randint(5, 10)) 起初雖然爬取速度很快，經過數次 request 發送後，自己的 IP 被網站擋爬了 印出網頁上顯示的文字： 您的瀏覽量異常, 已影響網站速度, 目前暫時關閉服務, 請稍後再重新使用
若您是使用程式大量下載本網站資料, 請適當調降程式查詢頻率, 以維護其他使用者的權益。 防擋爬方法 #1 - 變換 User Agent 安裝 User Agent 套件 fake-useragent: !pip install fake-useragent 爬蟲流程 1) 取得每個類別下的產業名稱 （前一次已經完成 直接使用 industry_dict 資料） from fake_useragent import UserAgent ua = UserAgent() # User Agent 產生器 for i in range(5): print(ua.random) default_headers = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-TW,zh;q=0.9,en-US;q=0.8,en;q=0.7", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Sec-Gpc": "1", "Referer": "https://www.google.com/", "Dnt": "1", "Upgrade-Insecure-Requests": "1", "User-Agent": ua.random } default_headers["User-Agent"] 2) 遍歷產業字典中的項目，取得每一項內的公司清單 ind_stocks = {} # 遍歷 6 個大類 for ind, sub_inds in industry_dict.items(): ind_stocks.update({ind: {}}) # 遍歷該大類下的子類別清單 for sub_ind in tqdm(sub_inds): # try 正常情況下的執行內容 try: url = f'https://goodinfo.tw/StockInfo/StockList.asp?MARKET_CAT={ind}&INDUSTRY_CAT={sub_ind}' list_stock = getCateStockList(market_cat=ind, ind=sub_ind) # except 若取得資料的過程發生錯誤 except Exception as e: print("Error at: ", url, "\n", e) list_stock = [] # 印出當前頁面的文字確認情況 res = requests.get(url, headers=default_headers) res.encoding = "utf-8" print(res.text) # finally 不論是否出現 Exception 都會執行 finally: ind_stocks[ind].update({sub_ind: list_stock}) print(sub_ind, ind_stocks[ind][sub_ind], '\n') sleep(np.random.randint(5, 10)) 可以間隔數秒發送的 request 次數增加了；不過還是在爬集團股時達到網站限制爬蟲的上限 with open("industry_companylist_dict.json", "w", encoding="utf-8") as f: json.dump(ind_stocks, f, ensure_ascii=False, indent=4) 防擋爬方法 #2 - 使用 Proxy 借用一些 free proxy list 網站提供的 IP 列表 先利用 ipify 服務檢測該 Proxy 是否可用，減少待會爬蟲的失敗率 def getFreeIpProxy(): """ 取得 ip 列表 """ free_proxy_urls = [ f"https://proxylist.geonode.com/api/proxy-list?limit=200&sort_by=lastChecked&sort_type=desc&google=true&protocols=https", "https://free-proxy-list.net/" ] ip_list = [] for url in tqdm(free_proxy_urls): if "geonode" in url: res = requests.get(url).json() ip_list += [item["ip"]+":"+item["port"] for item in res["data"]] elif "free-proxy-list" in url: res = requests.get(url) ip_list += re.findall('\d+\.\d+\.\d+\.\d+:\d+', res.text) # ip 的 regex 規則: __.__.__.__:__ (\d+ 是指數字格式) return ip_list def getValidIpProxy(num_use): """ 透過 ipify 檢驗是否為可用 IP - num_use: 要檢驗累積到幾個可用 ip 後停止 """ valid_ips = [] check_ip_url = "https://api.ipify.org/?format=json" # 如果 ip 可用則會回傳當前位址 ip_list = getFreeIpProxy() for ip in tqdm(ip_list): try: res = requests.get(check_ip_url, proxies={"http": ip, "https": ip}, timeout=timeout) valid_ips.append(ip) print(res.json()) except: print("Invalid: ", ip) # 如果累積到 num_use 以上個可用 proxy 就結束迴圈 if len(valid_ips) >= num_use: break return valid_ips valid_ips = getValidIpProxy(num_use=10) 更換 Proxy 的函式 同樣的 ip 還是會在多次 requests 後因為達到上限而被擋，因此我們需要一個隨著條件更換 proxy 的判斷函式 def proxySwitch(maxtimes_changeIp, maxtimes_retry, valid_ips, url, method, payload=None): """ 若超過 maxtimes_retry 則換一個 ip 做為 proxy 、若超過 maxtimes_changeIp 次數限制則該資料爬取結果回傳 None - maxtimes_changeIp: `int` 跳過當前 Request URL 的上限次數 - maxtimes_retry: `int` 跳過當前 ip 的上限次數 - valid_ips: `list` 檢驗後可用的 ip 列表 - url: `str` 發送請求的目標網站 - method: `str` 請求類型 {"get", "post"} - payload: `dict` 參數字典 (if method="get", then None) """ changedIp = 0 # 已更換幾次 ip success = 0 # 是否成功正常回傳網頁 res = None # 若更換 ip 次數達上限或資料擷取成功才跳出，否則持續運行 while (changedIp < maxtimes_changeIp) and (success == 0): retry = 0 # 同樣 ip 已嘗試幾次 timeout = 20 # proxy 跳板的 timeout 時長 ip_proxy = random.choice(valid_ips) # 從可用 ip 列表中隨機選取一個 # 若相同 ip 嘗試次數達上限或資料擷取成功才跳出，否則持續運行 while (retry < maxtimes_retry) and (success == 0): try: print(ip_proxy, ' Fetching ', url, ' timeout: ', timeout) if method == 'get': res = requests.get( url=url, headers=default_headers, proxies={'http': ip_proxy, 'https': ip_proxy}, timeout=timeout).content elif method == "post": print(payload) res = requests.post( url=url, data=payload, headers=default_headers, proxies={'http': ip_proxy, 'https': ip_proxy}, timeout=timeout).content else: print("method hasn't defined in function yet.") break soup = BeautifulSoup(res, 'html.parser') test = soup.find('table') # 依照頁面不同判斷成功條件不同 success = 1 # 使跳出 while 迴圈 print('Success') except Exception as e: retry += 1 # 當 retry 次數滿會符合更換 proxy 的條件 timeout += 5 # 每當開始連線失敗就增加 5 秒 timeout 放寬標準 print('retrying: ', retry) changedIp += 1 return res """ 提升使用 proxy 的速率: 1. 成功過的 proxy 先沿用 直到被擋 減少時間浪費 2. 統計 proxy 反應時間是否超過 timeout -> count > 3 -> ip 從 valid_ips 列表當中移除，下一次不會選到這個跳板時間較長的 ip 3. 一次選 5 個不同 ip，安排到不同 thread 執行 多執行緒 假設不得已要長期從 free proxy 網站取得 ip list，是否有方法比較快累積可用 ip 1. ip 檢驗頻率，前一個時段取得的 ip 是否會過期 或是不通過 validation 2. 是不是要定期對 valid_ips 做更新 """ 將 proxySwitch 引用到上面的爬蟲函式中 def getStockCates(maxtimes_changeIp, maxtimes_retry): """ 取得各類別下的子類別清單 - maxtimes_changeIp: `int` 跳過當前 Request URL 的上限次數 - maxtimes_retry: `int` 跳過當前 ip 的上限次數 """ market_categories = ['上市','上櫃','興櫃','電子產業','概念股','集團股'] subcates = {} url = f'https://goodinfo.tw/StockInfo/StockList.asp' # 都和上面寫法一樣，只是將 requests 部分換成 proxySwitch 函式，是一個可以自動更換 proxy 的 requests function res = proxySwitch( maxtimes_changeIp=maxtimes_changeIp, maxtimes_retry=maxtimes_retry, valid_ips=valid_ips, url=url, method="get", payload=None ) soup = BeautifulSoup(res, 'html.parser') list_stock_cate = soup.find_all('td', {'colspan': '4'}) for market_cat in tqdm(market_categories): _subcates = [str(item).split(f'{market_cat}_')[-1].split('@*=')[0] for item in list_stock_cate] _subcates = [ item.replace('="" ','').replace('colspan="4"','') for item in _subcates if ("<" not in item) and ("全部" not in item) ] subcates.update({market_cat: _subcates}) return subcates def getCateStockList(market_cat, ind, maxtimes_changeIp, maxtimes_retry): """ 取得該子類別下的所有個股 """ url = f'https://goodinfo.tw/StockInfo/StockList.asp?MARKET_CAT={market_cat}&INDUSTRY_CAT={ind}' # 都和上面寫法一樣，只是將 requests 部分換成 proxySwitch 函式，是一個可以自動更換 proxy 的 requests function res = proxySwitch( maxtimes_changeIp=maxtimes_changeIp, maxtimes_retry=maxtimes_retry, valid_ips=valid_ips, url=url, method="get", payload=None ) soup = BeautifulSoup(res, 'html.parser') list_stock = [item.text for item in soup.find('table', {'class': 'r10_0_0_10 b1 p4_1'}).find_all('td')] list_stock = [(x, y) for x, y in zip(list_stock[::2], list_stock[1::2]) if x != '代號'] return list_stock 爬蟲流程 1) 取得每個類別下的產業名稱 （前一次已經完成 直接使用 industry_dict 資料） maxtimes_changeIp = 5 maxtimes_retry = 3 test_industry_dict = getStockCates( maxtimes_changeIp=maxtimes_changeIp, maxtimes_retry=maxtimes_retry) print(test_industry_dict) with open("industry_dict.json", "w", encoding="utf-8") as f: json.dump(industry_dict, f, ensure_ascii=False, indent=4) 2) 遍歷產業字典中的項目，取得每一項內的公司清單 P.S. 因為免費 proxy 效用不能保證 本練習不一定能在時間內成功將所有子類別的公司清單都爬下來 最重要目的是希望學員能練習 proxy 的使用和如何設定規則更換 ind_stocks = {} # 遍歷 6 個大類 for ind, sub_inds in industry_dict.items(): ind_stocks.update({ind: {}}) # 遍歷該大類下的子類別清單 for sub_ind in tqdm(sub_inds): # try 正常情況下的執行內容 try: url = f'https://goodinfo.tw/StockInfo/StockList.asp?MARKET_CAT={ind}&INDUSTRY_CAT={sub_ind}' list_stock = getCateStockList(market_cat=ind, ind=sub_ind, maxtimes_changeIp=maxtimes_changeIp, maxtimes_retry=maxtimes_retry) # except 若取得資料的過程發生錯誤 except Exception as e: print("Error at: ", url, "\n", e) list_stock = [] # 印出當前頁面的文字確認情況 res = requests.get(url, headers=default_headers) res.encoding = "utf-8" print(res.text) # finally 不論是否出現 Exception 都會執行 finally: ind_stocks[ind].update({sub_ind: list_stock}) print(sub_ind, ind_stocks[ind][sub_ind], '\n') sleep(np.random.randint(5, 10)) with open("industry_companylist_dict.json", "w", encoding="utf-8") as f: json.dump(ind_stocks, f, ensure_ascii=False, indent=4) 預計爬取結果 📎industry_dict.json📎industry_company_dict.json