Python 網路爬蟲工作坊｜金融應用篇 - Cupoy

即時直播注意事項

請事先下載 Zoom 會議軟體，並參考「Zoom 操作手冊」熟悉使用方式，以便獲得最好的教學體驗。 講師授課過程中請學員務必「保持靜音模式」，如聽講過程有問題可使用事先提供的提問表單進行詢問。 實作階段請學員「務必」跟著操作，相信跟著操作後才能真正獲得專業技能。 如學員在實作階段有問題，講師會透過「遠端操作」方式協助您解決問題，到時請您不吝惜地讓講師協助您解決問題。 2022/05/28（六）由於為免費體驗課程，不受限於訂閱制用戶參與，如會議參與人數超過 100 人 則會無法再進入，敬請提早進入會議室。 Zoom 會議室預計皆會在活動開始前 20 分鐘開放參與學員進入。

Zoom操作手冊

本手冊大綱 如何進入 Zoom 會議室？ 如何確認是否靜音、是否開啟或關閉視訊鏡頭 如何確認是否有聲音以及是否可以說話？ 如何進行分享畫面 若需要講師實際協助您的實作，該怎麼做？ 如何進入 Zoom 會議室？ Step1. 點擊指定的會議室連結（直播活動前將以 Email 告知，請務必密切注意信箱） Step2. 點擊「開啟Zoom Meetings」，即可開啟Zoom會議室 Step3. 跳出會議室軟體後，等待主持人允許加入。 注意：若主持人有任何訊息需要給予欲加入的成員，訊息會呈現在畫面右手邊 Step4. 順利加入Zoom會議室。 如何確認是否靜音、是否開啟或關閉視訊鏡頭？ 畫面左下角分別有麥克風跟攝影機的圖示，可以分別控制麥克風與視訊鏡頭的開或關。若為「靜音狀態」以及「關閉攝影機」，圖示會畫出一條斜直線，如想要開啟麥克風或開啟視訊鏡頭，直接點擊圖示即可。 如何確認是否有聲音以及是否可以說話？ Step1. 點擊畫面左下角「麥克風」圖示中右上角的小箭頭 Step2. 確認自身的麥克風與喇叭裝置來源 Step3. 進行聲音與說話的測試 如何進行分享畫面 Step1. 點擊畫面下方綠色的「分享畫面」按鈕，畫面正中間會新跑出一個畫面 Step2. 點擊新畫面中的「螢幕」選項 Step3. 按下新畫面中右下角的「分享」按鈕 若需要講師實際協助您的實作，該怎麼做？ Step1. 學員需要先分享畫面，分享後工具列會出現在畫面最上方。 Step2. 講師端會提出遠端操控需求，學員端會出現「XXX 現正請求遠端控制您的螢幕」，請記得按下「接受」讓講師能夠直接操控您的畫面進行 Debug。

API 串接：公開資訊觀測站 - 董監事持股明細 - 簡報

講師介紹 目錄 爬蟲目標介紹 爬蟲邏輯複習 爬蟲流程講解 爬蟲程式 Demo 與講解（colab） 爬蟲目標介紹 公開資訊觀測站 - 董監事持股明細 學習觀察：以後面對不同的爬蟲任務，多一項找到解法的武器 學習發送 Https Requests：使用 Requests 套件對網站發出相對應型態的請求，獲取資料 學習將爬蟲包裝為函式：透過函式打包讓爬蟲代碼輕鬆套用到多個標的上 學習資料儲存方式：了解如何資料存為 json 格式 Step1. 取得董監事資料的介面 Step2. 有沒有跳過按按鈕的方式？ Step3. 用程式設定選項的方式？ 最新資料/歷史資料 公司代號 年度 月份 爬蟲邏輯複習 網頁結構分析：分析各項途徑，找到正確、合法且最簡單方法取得資料 網頁爬蟲開發：使用各項套件，將分析好的環節寫成程式 資料儲存：選擇合適的資料儲存格式，思考資料更新頻率和維護問題 爬蟲流程講解 網頁結構分析 打開「瀏覽器開發者工具」 切到 Network Panel，執行一遍取得資料的動作 「搜尋」載入我們目標資料的那一項資源 點擊該資源並確認 Response 內容 點擊該資源並確認 Header 內容 點擊該資源並確認 Payload 內容 網頁爬蟲開發 載入套件庫     requests     bs4     pandas     json 設定變數     target_url     headers     payload     is_new     co_id     year     month 程式邏輯 資料儲存 從回傳資料、大小、用途等判斷合適格式 爬蟲程式 demo 與講解 (colab) API串接 - 程式碼爬蟲目標：公開資訊觀測站 - 董監事持股明細 我們希望能取得每間上市櫃公司每月的董監事名單和持股數，人物跨公司的關係和持股數的變化，用來做對股價變化或投資標的選擇的運用。 頁面連結：https://mops.twse.com.tw/mops/web/stapap1 API: https://mops.twse.com.tw/mops/web/ajax_stapap1 資料單位：每月、每間公司，會有一份名單和持股數的表格 載入套件 from bs4 import BeautifulSoup # 網頁解析 import datetime as dt import json import numpy as np import pandas as pd import requests # 發送 requests from time import sleep # 暫停 from tqdm import tqdm # progress 進度條 import warnings warnings.filterwarnings("ignore") 設定變數 headers = { "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.69 Mobile Safari/537.36", "Content-Type": "application/x-www-form-urlencoded" } url = 'https://mops.twse.com.tw/mops/web/ajax_stapap1' stock_id = 1102 year = 110 month = 11 payload = { 'encodeURIComponent': '1', 'step': '1', 'firstin': '1', 'off': '1', 'keyword4': '', 'code1': '', 'TYPEK2': '', 'checkbtn': '', 'queryName': 'co_id', 'inpuType': 'co_id', 'TYPEK': 'all', 'isnew': 'false', 'co_id': str(stock_id), 'year': str(year), 'month': str(month).zfill(2) } 發 POST Requests 測試取得資料 # 發送 Requests res = requests.post(url, data=payload, headers=headers).content # 解析網頁 soup = BeautifulSoup(res, 'html.parser') # 捕捉所需的資料 shareholdings_records = [ [i.text.strip() for i in item.find_all('td')] for item in soup.find('table', {'class': 'hasBorder'}).find_all('tr') ] print(shareholdings_records) 作法1: 處理 list 中的欄位再轉換 DataFrame # 欄位處理 pre = shareholdings_records[0][-1] # 配偶、未成年子女及利用他人名義持有部份 total_cols = shareholdings_records[0][:-1] + [f"{pre}-{col}" for col in shareholdings_records[1]] shareholdings_records[0] = total_cols shareholdings_records.pop(1) # 將資料轉成 DataFrame cols = shareholdings_records[0] data = shareholdings_records[1:] df = pd.DataFrame(data, columns=cols) df = df[(df['職稱']!="職稱") & (df['目前持股'].notnull())].reset_index(drop=True) df 作法2: 以 pd.read_html 轉換為 DataFrame 再處理 table = soup.find('table', {'class': 'hasBorder'}) df = pd.read_html(str(table))[0] merge_headers = (df.iloc[0, -3:] + "-" + df.iloc[1, -3:]).values.tolist() df.columns = df.iloc[0, :-3].values.tolist() + merge_headers df = df.iloc[2:, :].reset_index(drop=True) df = df[df['職稱']!="職稱"].reset_index(drop=True) df 將爬取程式寫為 function 便於呼叫 def getShareHoldings(stock_id, year, month): """ 要送入的參數： - 公司代碼 stock_id - 年份 year - 月份 month """ payload['co_id'] = stock_id payload['year'] = year payload['month'] = month # 發 requests res = requests.post(url, data=payload, headers=headers).content # 網頁解析 soup = BeautifulSoup(res, 'html.parser') # 抓取想要的資料，拼成所需格式 shareholdings_records = [ [i.text.strip() for i in item.find_all('td')] for item in soup.find('table', {'class': 'hasBorder'}).find_all('tr') ] # 欄位處理 pre = shareholdings_records[0][-1] total_cols = shareholdings_records[0][:-1] + [f"{pre}-{col}" for col in shareholdings_records[1]] shareholdings_records[0] = total_cols shareholdings_records.pop(1) return shareholdings_records # 取得 2330 110年 1月資料 shareholdings_records = getShareHoldings( stock_id=2330, year=110, month=1 ) df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.head() 載入台股上市的公司代碼列表 listed_company.csv stock_ids = pd.read_csv('https://raw.githubusercontent.com/A-baoYang/Crawlers/jupyter_gcp_cathayddt/Financial/stock/listed_company.csv', delimiter="　") stock_ids.columns = ["id", "name"] stock_id_list = stock_ids["id"].unique().tolist() print(stock_ids.shape) stock_ids 練習 1 取得「雄獅」公司 109 年度 5 月的董監事持股明細 以 .csv 格式儲存stock_id = 2731 year = 109 month = 5 # shareholdings_records = getShareHoldings( # stock_id=stock_id, year=year, month=month # ) # shareholdings_records[:5] df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.to_csv(f"{stock_id}-sharehold-{year}-{month}.csv", index=False) 練習 2 取得「玉山金」公司 108 年度 7-12 月的董監事持股明細 以 .csv 格式儲存 stock_ids[stock_ids["name"].str.contains("玉山金")] for number in range(10): if number == 5: break # continue here print('Number is ' + str(number)) stock_id = 2884 year = 108 broken_records = [] for month in range(7, 13): try: print(f"Crawling {stock_id} {year} {month}") sleep(3) shareholdings_records = getShareHoldings( stock_id=stock_id, year=year, month=month ) df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.to_csv(f"{stock_id}-{year}{str(month).zfill(2)}.csv", index=False) except Exception as err: print(f"At {stock_id} {year} {month}") print(err) broken_records.append((stock_id, year, month)) 練習 3 取一部分公司列表示範，爬取董監事持股資料，並以 .csv 格式儲存 取列表前10間公司 取得 109-110 年度各個月份的董監事持股明細 (2020-2021)stock_id_list[:10] for stock_id in tqdm(stock_id_list[:10]): for year in range(109, 111): for month in range(1, 13): try: print(f"Crawling {stock_id} {year} {month}") sleep(3) shareholdings_records = getShareHoldings( stock_id=stock_id, year=year, month=month ) df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.to_csv(f"{stock_id}-{year}{str(month).zfill(2)}.csv", index=False) except Exception as err: print(f"At {stock_id} {year} {month}") print(err) pass 注意：「強烈推薦」在 google colab 上執行。

API 串接：公開資訊觀測站 - 董監事持股明細 - 程式碼

爬蟲目標：公開資訊觀測站 - 董監事持股明細 我們希望能取得每間上市櫃公司每月的董監事名單和持股數，人物跨公司的關係和持股數的變化，用來做對股價變化或投資標的選擇的運用。 頁面連結：https://mops.twse.com.tw/mops/web/stapap1 API: https://mops.twse.com.tw/mops/web/ajax_stapap1 資料單位：每月、每間公司，會有一份名單和持股數的表格 載入套件 from bs4 import BeautifulSoup # 網頁解析 import datetime as dt import json import numpy as np import pandas as pd import requests # 發送 requests from time import sleep # 暫停 from tqdm import tqdm # progress 進度條 import warnings warnings.filterwarnings("ignore") 設定變數 headers = { "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.69 Mobile Safari/537.36", "Content-Type": "application/x-www-form-urlencoded" } url = 'https://mops.twse.com.tw/mops/web/ajax_stapap1' stock_id = 1102 year = 110 month = 11 payload = { 'encodeURIComponent': '1', 'step': '1', 'firstin': '1', 'off': '1', 'keyword4': '', 'code1': '', 'TYPEK2': '', 'checkbtn': '', 'queryName': 'co_id', 'inpuType': 'co_id', 'TYPEK': 'all', 'isnew': 'false', 'co_id': str(stock_id), 'year': str(year), 'month': str(month).zfill(2) } 發 POST Requests 測試取得資料 # 發送 Requests res = requests.post(url, data=payload, headers=headers).content # 解析網頁 soup = BeautifulSoup(res, 'html.parser') # 捕捉所需的資料 shareholdings_records = [ [i.text.strip() for i in item.find_all('td')] for item in soup.find('table', {'class': 'hasBorder'}).find_all('tr') ] print(shareholdings_records) 作法1: 處理 list 中的欄位再轉換 DataFrame # 欄位處理 pre = shareholdings_records[0][-1] # 配偶、未成年子女及利用他人名義持有部份 total_cols = shareholdings_records[0][:-1] + [f"{pre}-{col}" for col in shareholdings_records[1]] shareholdings_records[0] = total_cols shareholdings_records.pop(1) # 將資料轉成 DataFrame cols = shareholdings_records[0] data = shareholdings_records[1:] df = pd.DataFrame(data, columns=cols) df = df[(df['職稱']!="職稱") & (df['目前持股'].notnull())].reset_index(drop=True) df 作法2: 以 pd.read_html 轉換為 DataFrame 再處理 table = soup.find('table', {'class': 'hasBorder'}) df = pd.read_html(str(table))[0] merge_headers = (df.iloc[0, -3:] + "-" + df.iloc[1, -3:]).values.tolist() df.columns = df.iloc[0, :-3].values.tolist() + merge_headers df = df.iloc[2:, :].reset_index(drop=True) df = df[df['職稱']!="職稱"].reset_index(drop=True) df 將爬取程式寫為 function 便於呼叫 def getShareHoldings(stock_id, year, month): """ 要送入的參數： - 公司代碼 stock_id - 年份 year - 月份 month """ payload['co_id'] = stock_id payload['year'] = year payload['month'] = month # 發 requests res = requests.post(url, data=payload, headers=headers).content # 網頁解析 soup = BeautifulSoup(res, 'html.parser') # 抓取想要的資料，拼成所需格式 shareholdings_records = [ [i.text.strip() for i in item.find_all('td')] for item in soup.find('table', {'class': 'hasBorder'}).find_all('tr') ] # 欄位處理 pre = shareholdings_records[0][-1] total_cols = shareholdings_records[0][:-1] + [f"{pre}-{col}" for col in shareholdings_records[1]] shareholdings_records[0] = total_cols shareholdings_records.pop(1) return shareholdings_records # 取得 2330 110年 1月資料 shareholdings_records = getShareHoldings( stock_id=2330, year=110, month=1 ) df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.head() 載入台股上市的公司代碼列表 listed_company.csv stock_ids = pd.read_csv('https://raw.githubusercontent.com/A-baoYang/Crawlers/jupyter_gcp_cathayddt/Financial/stock/listed_company.csv', delimiter="　") stock_ids.columns = ["id", "name"] stock_id_list = stock_ids["id"].unique().tolist() print(stock_ids.shape) stock_ids 練習 1 取得「雄獅」公司 109 年度 5 月的董監事持股明細 以 .csv 格式儲存stock_id = 2731 year = 109 month = 5 # shareholdings_records = getShareHoldings( # stock_id=stock_id, year=year, month=month # ) # shareholdings_records[:5] df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.to_csv(f"{stock_id}-sharehold-{year}-{month}.csv", index=False) 練習 2 取得「玉山金」公司 108 年度 7-12 月的董監事持股明細 以 .csv 格式儲存 stock_ids[stock_ids["name"].str.contains("玉山金")] for number in range(10): if number == 5: break # continue here print('Number is ' + str(number)) stock_id = 2884 year = 108 broken_records = [] for month in range(7, 13): try: print(f"Crawling {stock_id} {year} {month}") sleep(3) shareholdings_records = getShareHoldings( stock_id=stock_id, year=year, month=month ) df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.to_csv(f"{stock_id}-{year}{str(month).zfill(2)}.csv", index=False) except Exception as err: print(f"At {stock_id} {year} {month}") print(err) broken_records.append((stock_id, year, month)) 練習 3 取一部分公司列表示範，爬取董監事持股資料，並以 .csv 格式儲存 取列表前10間公司 取得 109-110 年度各個月份的董監事持股明細 (2020-2021)stock_id_list[:10] for stock_id in tqdm(stock_id_list[:10]): for year in range(109, 111): for month in range(1, 13): try: print(f"Crawling {stock_id} {year} {month}") sleep(3) shareholdings_records = getShareHoldings( stock_id=stock_id, year=year, month=month ) df = pd.DataFrame(shareholdings_records[1:], columns=shareholdings_records[0]) df.to_csv(f"{stock_id}-{year}{str(month).zfill(2)}.csv", index=False) except Exception as err: print(f"At {stock_id} {year} {month}") print(err) pass

【學員提問補充】(1) 爬蟲套件結合 pandas.read_html 使用

若遇到要爬取的網頁內容是 <table></table> 節點內的資料 可使用 pandas.read_html 解析 html 字串直接輸出表格。 實例如下： # 載入套件 import pandas as pd import requests # 變數設置 headers = { "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.69 Mobile Safari/537.36", "Content-Type": "application/x-www-form-urlencoded" } url = 'https://mops.twse.com.tw/mops/web/ajax_stapap1' stock_id = 1102 year = 110 month = 11 payload = { 'encodeURIComponent': '1', 'step': '1', 'firstin': '1', 'off': '1', 'keyword4': '', 'code1': '', 'TYPEK2': '', 'checkbtn': '', 'queryName': 'co_id', 'inpuType': 'co_id', 'TYPEK': 'all', 'isnew': 'false', 'co_id': str(stock_id), 'year': str(year), 'month': str(month).zfill(2) } # 發送 Requests res = requests.post(url, data=payload, headers=headers).content # 解析網頁 soup = BeautifulSoup(res, 'html.parser') # 解析 html 物件轉為 DataFrame table = soup.find('table', {'class': 'hasBorder'}) df = pd.read_html(str(table))[0] # read_html 會將 html 物件內所有 ###draft_code_symbol_lessthen###table> 都轉成 DataFrame，因此輸出會是 list 形式 df.head(10) 輸出結果： 再經過一些 DataFrame 的處理則可以變成我們預期的樣子 # 將第一列、第二列欄位合併 merge_headers = (df.iloc[0, -3:] + "-" + df.iloc[1, -3:]).values.tolist() df.columns = df.iloc[0, :-3].values.tolist() + merge_headers # 排除無意義列、重新 index （避免後面的處理發生問題） df = df.iloc[2:, :].reset_index(drop=True) df = df[df['職稱']!="職稱"].reset_index(drop=True) df.head(10) 輸出結果：

【學員提問補充】(2) Python 迴圈控制語法比較：break, continue, pass

本次工作坊程式在迴圈中使用到 try... except... 例外處理及迴圈控制語法 pass 這邊補充三種迴圈控制語法  break：當條件符合時，強制跳出整個迴圈 continue：當條件符合時，跳過本次迴圈、進到下一次 pass：加與不加對輸出無差異，通常用於空函式暫時的填塞值 P.S. 程式中有部分函式還沒完成，只是先將名稱命名好時，想要測試整份程式是否可以運行，避免空函式的部分報錯 def if_exist(item, target_list): if [i for i in target_list if i == item] return True else: return False def function_to_be_done(x): pass 這邊用同一段程式，放入上面三種不同控制語法，比較他們的輸出，這樣就更一目瞭然他們的用途了 for number in range(10): print('Number is ' + str(number)) for number in range(10): if number == 5: pass # pass here print('Number is ' + str(number)) for number in range(10): if number == 5: continue # continue here print('Number is ' + str(number)) for number in range(10): if number == 5: break # break here print('Number is ' + str(number))

【學員提問補充】(3) 爬蟲防擋的作法統整

由於課堂上大家對於如何防網站擋爬非常感興趣 這邊幫大家課後統整爬蟲防擋的小技巧 會統一在第二堂課教到如何使用 proxy 時，一併和學員示範如何實作 同時這些內容也會更新到第六屆爬蟲馬拉松，搭配不同網站的作業練習，加深學習效果 以下方法由簡單到複雜排序： (I)  發送 Request 時帶上 Headers 並切換關鍵項目 Request Headers (請求標頭)是客戶端用來告知伺服器端，客戶端這邊的瀏覽器資訊是什麼 因此模仿你在瀏覽網頁時，Network Panel 所看到的 Request Headers，也是降低網站把你判斷成爬蟲程序的機率。 import requests from fake_useragent import UserAgent # 可隨機產生 User-Agent headers = { "Accept": "*/*", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-TW,zh;q=0.9,en-US;q=0.8,en;q=0.7", "Host": "https://mops.twse.com.tw", "Connection": "keep-alive", "Content-Type": "application/x-www-form-urlencoded", "Sec-GPC": 1, "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.69 Mobile Safari/537.36", } user_agent = UserAgent() headers["User-Agent"] = user_agent.random r = requests.get(url, headers=headers) Accept-Language: 語系代碼及各個語系的權重值（詳細說明） Content-Type: 網站上的資料型態、網站會從伺服器端接收到的資料編碼格式（型態種類列表） User-Agent: 告知網站發出請求的瀏覽器的資訊 切換 User-Agent 扮演不同瀏覽器發出請求，也可以降低被伺服器判定為機器人的機率 Referer: 指定參照位址，也就是紀錄是從哪邊進到當前網頁 可以用 SEO 工具來獲取反向連結（哪個網址有連到當前頁面）當作切換池，例如下圖使用Ahrefs 獲取反向連結排名列表： (II)  設定隨機延遲時間 就是讓 sleep 的秒數從隨機整數產生，讓網站不易發現，否則頻率一致的固定模式就比較容易被偵測到是爬蟲程序在發送請求了。 import numpy as np import requests from time import sleep def fetch(url, header): sleep(np.random.randint(5, 20)) r = requests.get(url, headers=headers) return r (III)  輪流切換 IP 位址 如果同一組 IP 位址短時間內大量發送請求到目標網站，會有很大機率被網站判定為程序，因而被擋。 因此輪流切換 IP 位址是一種降低被擋機率的方式。 import requests proxies = { "http": "http://username:password@proxy_provider_host.com:20001", "https": "http://username:password@proxy_provider_host.com:20001", } r = requests.get(url, headers=headers, proxies=proxies) 其中，HTTP 及 HTTPS Proxy 間的差異如下： HTTP Proxy會從客戶端收到純文字的 Request，隨後再發送另一 Request 到目標伺服器端，最後伺服器端將資訊回傳至客戶端。 HTTPS Proxy一個中繼站的概念，會先收到客戶端一個特殊的 Request 後，與目標伺服器端建立一個不透明的通道，隨後客戶端會發送 SSL/TLS Request 到伺服器端、繼續進行 SSL 交握。 如果不確定目標網站是否需要透過 HTTPS 連線才能回傳資料，則可以在設定 requests proxies 時如上設定一個字典檔分別傳入 http, https proxy (IV)  使用 headless browser  有些網站對爬蟲的偵測比較嚴格，會檢查像是瀏覽器 Cookie 和 JavaScript 執行等，此時更建議使用 Selenium；但是預設用 Selenium 模擬瀏覽器行為時，速度會變慢很多，這時就可以用 headless 的設置，來取消網頁介面彈出，讓效能較好一些。 以下用需要 JavaScript 載入的 Pressplay 網站示範： from fake_useragent import UserAgent import numpy as np from selenium import webdriver import time # 設定 webdriver 啟動檔位址 driver_path = ( # 練習時記得換成自己的 driver path "/Users/jiunyiyang/.wdm/drivers/chromedriver/mac64/102.0.5005.61/chromedriver" ) url = "https://www.pressplay.cc/project?type=507CF74F93AD23B6584EF2C5D5D4430E" # 設定 Options opt = webdriver.ChromeOptions() user_agent = UserAgent() opt.add_argument("--user-agent=%s" % user_agent) opt.add_argument("--window-size=1920,1080") opt.add_argument("--headless") # 啟用 headless 模式 opt.add_argument("--disable-gpu") # 測試關閉 GPU 避免某些錯誤出現 # 載入目標網址 driver = webdriver.Chrome(driver_path, options=opt) driver.get(url) print(driver.title) # 模擬用戶等待網站載入，每次都用隨機時間 time.sleep(np.random.uniform(3, 5)) try: ele = driver.find_elements_by_xpath('//div[@class="project-list-item"]')[1] course = ele.find_elements_by_xpath('.//div[@id="project-card"]')[1] print(course.text) except Exception as e: print(e) finally: driver.close() print("driver closed.") 運行結果： 看到他可以成功印出那堂課程資訊卡的文字內容 P.S. 如果只用 requests 爬取就會抓不到課程區塊內的資訊卡 延伸閱讀： Python 自動化工具 – Headless Selenium 隱藏的網頁瀏覽器 (V)  其他 讓爬蟲步驟增加一些不可預測性 如果爬蟲步驟越固定，越容易被網站偵測出是程序；在使用 Selenium 時可以增加一些隨機的滾動或點擊行為，讓網站覺得你的程序越像真人的行為。 在網站流量的離峰時段爬取 如果在尖峰時段爬取，通常爬蟲程序的換頁速度比正常人瀏覽快很多，如此一來也會更明顯的拖累網站本身的效能和用戶體驗；在離峰時間爬取、並配合前面提過的隨機延時，可以避免讓伺服器超過負荷。

第一堂 - 精選影片1 - 爬蟲目標

爬蟲目標介紹 董監事持股明細

第一堂 - 精選影片2 - 爬蟲邏輯複習

爬蟲邏輯複習 網頁結構分析：分析各項途徑，找到正確、合法且最簡單方法取得資料 網頁爬蟲開發：使用各項套件，將分析好的環節寫成程式 資料儲存：選擇合適的資料儲存格式，思考資料更新頻率和維護問題

第一堂 - 精選影片3 - 爬蟲流程講解

第一堂 - 精選影片4 - 爬蟲程式Demo與練習

爬蟲程式 demo google colab

Goodinfo! 台灣股市資訊網 - 類股分類表 - 簡報

Context 爬蟲目標介 爬蟲流程講解 應對反爬蟲機制的方法 爬蟲程式 demo 與講解 (colab) Q & A Goodinfo! 台灣股市資訊網 - 類股分類表要學到什麼？ 學習觀察頁面結構熟悉搜尋物件所在的網頁節點，將抓取邏輯轉為程式 熟悉物件定位語法熟悉使用 BeautifulSoup 語法進行物件定位 學習使用 proxy 發送請求使用 Requests 套件送出經過 proxy 的請求 學習應對網站反爬機制了解不同方式避開網站的反爬蟲機制 爬蟲目標介紹從 Goodinfo! 獲取所有類別及其下方所有股票代號及名稱 目標頁面觀察詳細確認目標資料的所在位置和規律（以頁面講解） 爬蟲遇到反爬蟲機制多測試幾次後，發現 Goodinfo! 會限制同一 IP 請求次數 應對反爬蟲機制的方法 發送 Requests 帶上 Headers 隨機替換 Headers 中的 User-Agent 隨機替換 Headers 中的 Referer 經由 Proxy 發送 Requests 添加隨機延時 在網站流量的離峰時間爬蟲 增加隨機的瀏覽動作 (selenium) 使用 headless browser (selenium) 發送 Requests 帶上 Headers 隨機替換 Headers 中的 User-Agent 隨機替換 Headers 中的 Referer 經由 Proxy 發送 Requests多測試幾次後，發現 Goodinfo! 會限制同一 IP 請求次數 添加隨機延時 在網站流量的離峰時間爬蟲如果在尖峰時段爬取，通常爬蟲程序的換頁速度比正常人瀏覽快很多，如此一來也會更明顯的拖累網站本身的效能和用戶體驗； 在離峰時間爬取、並配合前面提過的隨機延時，可以避免讓伺服器超過負荷。 增加隨機的瀏覽動作 (selenium)如果爬蟲步驟越固定，越容易被網站偵測出是程序；在使用 Selenium 時可以增加一些隨機的滾動或點擊行為，讓網站覺得你的程序越像真人的行為。 使用 headless browser (selenium) 範例程式碼https://www.cupoy.com/collection/00000180B6E4E37F000000026375706F795F72656C656173654355/00000181F35962630000002C6375706F795F72656C656173654349

Goodinfo! 台灣股市資訊網 - 類股分類表 - 程式碼

爬蟲目標說明：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 被網站擋爬了 印出網頁上顯示的文字： <meta http-equiv="Content-Type" content="text/html; charset=utf-8">您的瀏覽量異常, 已影響網站速度, 目前暫時關閉服務, 請稍後再重新使用<br>若您是使用程式大量下載本網站資料, 請適當調降程式查詢頻率, 以維護其他使用者的權益。 防擋爬方法 #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

【學員提問補充】公開資訊觀測站 - 衍生性商品交易資訊爬蟲

當如果遇到彈出式視窗、沒有網址、沒辦法看 Network Panel 的時候，我們還可以嘗試觀察 html 結構。 由於是在按下「詳細資料」之後跳出目標資料，我們觀察下 <button> 節點的 onclick 屬性，其按鈕動作是 document.t15sf_fm.ppp.value='0';document.t15sf_fm.co_id.value='2330';openWindow(this.form ,''); 而 document.t15sf_fm 其實指的是隱藏在 html 結構中的第二個 <form> 元件，這個沒有顯示在網頁上的表單，包覆了整個子母公司的資訊表格。 這個就是 name="t15sf_fm" 的 <form> 元件，我們就發現這個 form 的目標網址是 "https://mops.twse.com.tw/mops/web/ajax_t15sf" ，發送表單使用的是 POST method。 ###draft_code_symbol_lessthen###form action='/mops/web/ajax_t15sf' name='t15sf_fm' id='t15sf_fm' method='post' onsubmit='return false;'> 說到 POST method，是不是也會有 payload 傳遞呢？再細看 <form> 裡面的 <input>： ###draft_code_symbol_lessthen###input type=hidden name=ppp >###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=statef value='110'>###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=TYPEK value='sii'>###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=year value=111>###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=month value=02>###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=co_id >###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=id value=''>###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=key value=''>###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type=hidden name=co_name >###draft_code_symbol_lessthen###/input> ###draft_code_symbol_lessthen###input type='hidden' name='firstin' value='ture'/>###draft_code_symbol_lessthen###/input> 其實這每個 <input> 的 name, value 屬性就是 POST request payload 中的 key-value pairs； 其中 ppp, co_id, co_name 被預設為空值。 再回到前面的按鈕 onclick document.t15sf_fm.ppp.value='0';document.t15sf_fm.co_id.value='2330';openWindow(this.form ,''); 這段 javascript 做的事是，當我按下按鈕： 在 name=ppp 的 input value 填入 "0" 在 name=co_id 的 input value 填入 "2330" 就是在將前面 <form> 中的 <input> 空值處填好並送出表單，將回傳的資料另開視窗呈現。 觀察完畢就很簡單了，我們來寫代碼： import pandas as pd import requests url = "https://mops.twse.com.tw/mops/web/ajax_t15sf" child_co_id = 2330 # 從頁面上的表格獲取的子/母公司代號 payload = { "ppp": "0", "statef": "110", "TYPEK": "sii", "year": "111", "month": "01", "co_id": child_co_id, "id": "", "key": "", "co_name": "", "firstin": "ture" } res = requests.post(url, data=payload, headers=default_headers).content soup = BeautifulSoup(res, "html.parser") df_list = pd.read_html(str(soup)) # 將字串中的 ###draft_code_symbol_lessthen###table> 解析成 DataFrame 格式 （DataFrame 資料需要再額外進行一些清洗） 這樣我們就在不使用 Network Panel 的情況下、以 Requests 方式取得資料啦！ 以上是本週工作坊的精選問答更新！ 如果妳有爬蟲相關的問題或建議，都歡迎留言/詢問我！

第二堂 - 精選影片1 - 爬蟲目標

第二堂 - 精選影片2 - 爬蟲流程講解

第二堂 - 精選影片3 - 爬蟲觀念練習與實作

動態爬蟲：MoneyDJ 基金基本資料 - 簡報

Context 爬蟲目標介紹 複習 Selenium WebDriver 複習 XPath 爬蟲程式 demo 與講解 (colab) Q & A 爬蟲目標 #1從 MoneyDJ 獲取所有基金資訊和績效 爬蟲目標 #2從 京東電商 透過下滑載入、點擊換頁獲取商品資訊 Selenium Webdriver 架構 Selenium command 會根據 JSON Wire Protocol 生成 HTTP request HTTP request 透過 browser driver 發送到 real browser 就可以看到啟動的 real browser 在根據我們的 scripts 進行操作 操作執行的訊息會回傳到 browser driver、再回傳到我們程式端的執行中，因此會看到是否成功的 log 、從網頁上抓取回來的資訊等 Selenium Options爬蟲常用到的 arguments: 設定 user-agent 設定編碼 設定螢幕大小 設定 headless …. Selenium 模擬瀏覽動作 等待一定秒數，若物件未能正確定位才噴錯 點擊網頁物件 執行 JavaScript 事件（例如滑動頁面） 按下特定鍵盤按鍵 輸入文字 XPath 語法結構網頁節點定位 節點 語法 h1 //h1 ul > li //ul/li div h4 //div//h4 節點屬性定位 屬性 語法 id //*[@id="id_value"] class > li //*[@class="class_value"] class 包含 course h4 //*[contains(@class, "course")] href //a/@href href 包含 .xml //a[end-with(@href, ".xml")] text //a/text() 依據順序定位 節點 語法 ul 下第二個 li //ul/li[2] 第二個 class 等於 tl 的 div //div[2][@class="tl"] 其他條件定位 屬性 語法 節點文字等於 search //*[text() = "search"] 節點文字包含 search //*[contains(text(), "search")] 有子節點的 ul 節點 //ul[*] 有 li 子傑點的 ul 節點 //ul[li] href 包含 .xml //a[ends - with(@href, ".xml")] 程式碼https://www.cupoy.com/collection/00000180B6E4E37F000000026375706F795F72656C656173654355/000001817B7CB12D000000086375706F795F72656C656173654349

動態爬蟲：MoneyDJ 基金基本資料 - 程式碼

Workshop #3 MoneyDJ 基金基本資料 爬蟲目標：將 MoneyDJ 基金頁面上的每支基金資本基料及持股明細爬取下來 # 套件安裝 !pip install fake-useragent selenium webdriver-manager # 載入所需套件 from fake_useragent import UserAgent import json import numpy as np import pandas as pd from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait from time import sleep from tqdm import tqdm from webdriver_manager.chrome import ChromeDriverManager driver = webdriver.Chrome(ChromeDriverManager().install()) 運行下載 chromedriver 的指令後會看到以下訊息，告訴我們最新版本存在什麼地方 這就是 webdriver 的啟動位址 ====== WebDriver manager ====== Current google-chrome version is 102.0.5005 Get LATEST chromedriver version for 102.0.5005 google-chrome Trying to download new driver from https://chromedriver.storage.googleapis.com/102.0.5005.61/chromedriver_mac64.zip Driver has been saved in cache [/Users/jiunyiyang/.wdm/drivers/chromedriver/mac64/102.0.5005.61] # 設定 webdriver 的啟動位址 chrome_path = "/Users/jiunyiyang/.wdm/drivers/chromedriver/mac64/102.0.5005.61/chromedriver" # 設定 User Agent ua = UserAgent() opt = webdriver.ChromeOptions() opt.add_argument("--user-agent=%s" % ua.random) 取得基金列表 載入網頁後，從各個基金分類列表獲取所有基金的名稱和網址列表 url = "https://www.moneydj.com/funddj/yb/YP301000.djhtm" # 初始化 webdrvier driver = webdriver.Chrome(executable_path=chrome_path, options=opt) driver.set_window_size(1024, 850) driver.get(url) # 蒐集所有基金分類 fund_types = [(item.text, item.get_attribute('href')) for item in driver.find_elements_by_xpath('//div[@class="InternalSearch"]//a')] print(fund_types[:5]) # 進入第一個基金分類選單頁(指數型) driver.get(fund_types[0][1]) # 獲取基金列表 fund_company = [(item.text, item.get_attribute('href')) for item in driver.find_elements_by_xpath('//table[@id="oMainTable"]//td/a') if len(item.text.strip())] print(fund_company[0::2][:5]) # 依照上面步驟，遍歷各個基金種類 fund_company_list = list() for item in tqdm(fund_types): driver.get(item[1]) fund_company = [(item.text, item.get_attribute('href')) for item in driver.find_elements_by_xpath('//table[@id="oMainTable"]//td/a') if len(item.text.strip())] fund_list = fund_company[0::2] company_list = [item[0] for item in fund_company[1::2]] for i in range(len(fund_list)): fund_company_list += [(company_list[i], fund_list[i][0], fund_list[i][1])] sleep(np.random.randint(3, 7)) driver.quit() pd.DataFrame( fund_company_list, columns=["company", "fund_name", "fund_url"] ).head() # store to json file fund_company_list = {"list": fund_company_list} with open("fund_company_dict.json", "w", encoding="utf-8") as f: json.dump(fund_company_list, f, ensure_ascii=False, indent=4) fund_company_list = fund_company_list["list"] 取得基金資訊 根據剛才蒐集到的每一支基金的網址，蒐集其基本資訊 fund_id = fund_company_list[888][2].split("a=")[-1] print(fund_id) info_url = "https://www.moneydj.com/funddj/yp/yp011000.djhtm?a=%s" % (fund_id) perf_url = "https://www.moneydj.com/funddj/yp/yp012000.djhtm?a=%s" % (fund_id) 使用「等待物件再抓取」的判斷邏輯，避免因為資料還沒載入造成爬蟲錯誤 driver = webdriver.Chrome(executable_path=chrome_path, options=opt) driver.set_window_size(1024, 850) driver.get(info_url) WAIT_SECONDS = 3 # 等待物件出現，未出現超過 3 秒才會出錯 info_col = WebDriverWait(driver, WAIT_SECONDS).until( # 檢測資訊表格物件是否出現在頁面上 EC.presence_of_element_located( ( # 使用 XPath 定位物件 By.XPATH, '//table[@class="t04"]//td[@class="t2c1"]', ) ) ) # 取得基本資料表格的欄位 info_col = [ item.text for item in driver.find_elements_by_xpath( '//table[@class="t04"]//td[@class="t2" or @class="t2c1"]' ) ] # 取得基本資料表格的值 info_val = [ item.text for item in driver.find_elements_by_xpath( '//table[@class="t04"]//td[@class="t3t2"]' ) ] print(info_col[:5]) print(info_val[:5]) # 將邏輯寫成 function 用來遍歷所有基金 def get_info_table(info_url): driver.get(info_url) # 等待物件出現，未出現超過 3 秒才會出錯 info_col = WebDriverWait(driver, WAIT_SECONDS).until( # 檢測資訊表格物件是否出現在頁面上 EC.presence_of_element_located( ( # 使用 XPath 定位物件 By.XPATH, '//table[@class="t04"]//td[@class="t2c1"]', ) ) ) info_col = [ item.text for item in driver.find_elements_by_xpath( '//table[@class="t04"]//td[@class="t2" or @class="t2c1"]' ) ] info_val = [ item.text for item in driver.find_elements_by_xpath( '//table[@class="t04"]//td[@class="t3t2"]' ) ] return info_col, info_val 取得基金績效表 根據剛才蒐集到的每一支基金的網址，蒐集其績效 driver.get(perf_url) # 等待物件出現，未出現超過 3 秒才會出錯 perf_table = WebDriverWait(driver, WAIT_SECONDS).until( # 等待物件出現，未出現超過 3 秒才會出錯 EC.presence_of_element_located( # 檢測績效表格物件是否出現在頁面上 (By.XPATH, '//table[@class="t01"]') ) ) # 抓取績效表欄位 perf_col = [ item.text.replace("\n","") for item in driver.find_elements_by_xpath('//table[@class="t01"]')[0] .find_elements_by_xpath('.//td[contains(@class, "t2")]') ] # 抓取績效表的值 perf_val = [ item.text for item in driver.find_elements_by_xpath('//table[@class="t01"]')[0] .find_elements_by_xpath('.//td[contains(@class, "t3")]') ] pd.DataFrame([perf_val], columns=perf_col) # 將邏輯寫成 function 用來遍歷所有基金 def get_perf_table(perf_url): driver.get(perf_url) # 等待物件出現，未出現超過 3 秒才會出錯 perf_table = WebDriverWait(driver, WAIT_SECONDS).until( # 等待物件出現，未出現超過 3 秒才會出錯 EC.presence_of_element_located( # 檢測績效表格物件是否出現在頁面上 (By.XPATH, '//table[@class="t01"]') ) ) # 抓取績效表欄位 perf_col = [ item.text.replace("\n","") for item in driver.find_elements_by_xpath('//table[@class="t01"]')[0] .find_elements_by_xpath('.//td[contains(@class, "t2")]') ] # 抓取績效表的值 perf_val = [ item.text for item in driver.find_elements_by_xpath('//table[@class="t01"]')[0] .find_elements_by_xpath('.//td[contains(@class, "t3")]') ] return perf_col, perf_val 遍歷所有基金 # 載入預存好的基金列表 (id 及 url) # load fund item list with open("fund_company_dict.json", "r", encoding="utf-8") as f: fund_company_dict = json.load(f) fund_company_list = fund_company_dict["list"] len(fund_company_list) #driver = webdriver.Chrome(executable_path=chrome_path, options=opt) #driver.set_window_size(1024, 850) WAIT_SECONDS = 3 fund_info_dict = dict() for fund in tqdm(fund_company_list): fund_name = fund[1] fund_id = fund[2].split("a=")[-1] fund_info_dict.update({fund_id: dict()}) fund_info_dict[fund_id].update({"基金名稱": fund_name}) info_url = "https://www.moneydj.com/funddj/yp/yp011000.djhtm?a=%s" % (fund_id) perf_url = "https://www.moneydj.com/funddj/yp/yp012000.djhtm?a=%s" % (fund_id) try: # 基本資料 info_col, info_val = get_info_table(info_url=info_url) # 績效表 perf_col, perf_val = get_perf_table(perf_url=perf_url) except Exception as e: info_col, info_val = [], [] perf_col, perf_val = [], [] finally: fund_info_dict[fund_id].update({ "基本資料": dict(zip(info_col, info_val)), "績效": dict(zip(perf_col, perf_val)), }) # temp store to json file with open("fund_info_dict.json", "w", encoding="utf-8") as f: json.dump(fund_info_dict, f, ensure_ascii=False, indent=4) sleep(np.random.randint(3, 7)) driver.quit() # store to json file with open("fund_info_dict.json", "w", encoding="utf-8") as f: json.dump(fund_info_dict, f, ensure_ascii=False, indent=4) 📎fund_info_dict.json📎fund_company_dict.json

第三堂 - 精選影片1 - 爬蟲目標

第三堂 - 精選影片2 - 複習Xpath

第三堂 - 精選影片3 - 爬蟲程式範例

爬蟲資料結合視覺化：CNYES 鉅亨網財經新聞 - 講義與程式碼

WorkShop#4 - 綜合案例講解 示範案例：鉅亨網新聞爬蟲與資料分析 今日流程 爬蟲：鉅亨網新聞 複習從網頁上尋找 API 複習 API 串接 文章清洗與儲存 複習 Python Regex 語法 文章資料分析 結合 Python Pandas 操作 結合 基礎 NLP 入門學習 (CKIP, Word2Vec, TFIDF, Sentiment Analysis) 結合 Plotly Express 操作 (1) 爬蟲：鉅亨網新聞 我們首先要爬取鉅亨網上的新聞。 複習第一堂工作坊：先找資料載入來源來確定要用哪種爬蟲方式 Step1. 複習從網頁上尋找 API 一開始搜尋文章標題，發現只有靜態頁面；仔細查看靜態頁面中也塞有文章資料的 dictionary 再進一步用文章 id 去搜尋，就發現了 API 所以我們找到了鉅亨網 API：https://api.cnyes.com/media/api/v1/newslist/category/tw_stock 以及其參數 startAt endAt limit Step2. 複習 API 串接 使用 requests 庫串接鉅亨網台股新聞 API import datetime as dt import numpy as np import requests from time import sleep from tqdm import tqdm 觀察最多一次會輸出幾篇文章、下一頁要怎麼取得？int(dt.datetime.today().timestamp()) # datetime object -> unixtimestamp (1655827200) cnyes_api = "https://api.cnyes.com/media/api/v1/newslist/category/tw_stock" params = { "startAt": int((dt.datetime.today()-dt.timedelta(days=30)).timestamp()), # 取最近一個月的新聞 "endAt": int(dt.datetime.today().timestamp()), "limit": 30 } data = requests.get(url=cnyes_api, params=params).json() data["items"]["data"] 觀察獲取了哪些欄位、有哪些可能用得到？# len(data["items"]["data"]) # page_num = data["items"]["last_page"] params["page"] = 2 data = requests.get(url=cnyes_api, params=params).json() data["items"]["data"][0] 最後就來取得所有文章article_list = [] data_cols = ["newsId","title","content","publishAt","market","categoryId","categoryName"] page_num = requests.get(url=cnyes_api, params=params).json()["items"]["last_page"] for i in tqdm(range(1, page_num + 1)): params["page"] = i data = requests.get(url=cnyes_api, params=params).json()["items"]["data"] article_list += [{k: v for k, v in article.items() if k in data_cols} for article in data] sleep(np.random.randint(3, 5)) ```python len(article_list) (2) 文章清洗與儲存 順利抓下文章後，將換行符、與內容無關的段落清洗 再以 .csv 和 .json 儲存import html import json import pandas as pd import re Step3. 複習 Python Regex 語法 觀察數篇文章，他們共同有哪些與內容無關的字符？article_list[0]["content"], article_list[1]["content"] # &lt; p&gt; &amp; nbsp; article_list[7]["content"] _text _text = html.unescape(article_list[2]["content"]) pattern = "<.*?>|\n|&[a-z0-9]+;|http\S+" re.sub(pattern, "", _text) 像是 &lt; &gt; &amp; &nbsp; 其實是 html 標籤的 < > & \xa0 使用 Regex 語法篩選這些字符並排除 排除 html 標籤_html_text = html.unescape(article_list[2]["content"]) # pattern = "<.*?>|&([a-z0-9]+|#[0-9]{1,6}|#x[0-9a-f]{1,6});|\n" pattern = "<.*?>|\n|&[a-z0-9]+;|http\S+" re.sub(pattern, "", _html_text).strip() df = pd.DataFrame(article_list) # 套用 html 標籤清除 regex 語法 df["content"] = df["content"].apply(lambda x: re.sub(pattern, "", html.unescape(x)).strip()) df["content"][0] 儲存檔案df.to_csv("articles_cnyes.csv", index=False) with open("articles_cnyes.json", "w", encoding="utf-8") as f: json.dump(article_list, f, ensure_ascii=False, indent=4) 比較不同檔案格式的大小!ls -l (3) 文章資料分析 結合 Python Pandas 操作 結合 基礎 NLP (CKIP / Word2Vec, TFIDF, Sentiment Analysis) 結合 Plotly Express 操作 資料分析入門題 在這過去一個月中，統整所有文章來看： 熱詞排名情形？ 哪些公司最常一起出現？最可能產生連帶影響？ 文章情緒分數隨時間變化 公司情緒分數隨時間變化基本分析 import datetime as dt from dateutil import tz # 調整時區 import pandas as pd df = pd.read_csv("articles_cnyes.csv") df.info() # 載入爬蟲預存好的資料 df = pd.read_csv("articles_cnyes.csv") for col in ["content","title","market","categoryName"]: if col == "market": df[col] = df[col].fillna("[]") df[col] = df[col].apply(lambda x: eval(x)) else: df[col] = df[col].astype(str) df.info() tz.gettz("Asia/Taipei") dt.datetime.fromtimestamp(1656657673, tz=tz.gettz("Asia/Taipei")) # 為 dataframe 添加可用的時間篩選欄位 df["date"] = df["publishAt"].apply(lambda x: int(dt.datetime.fromtimestamp(x, tz=tz.gettz("Asia/Taipei")).strftime("%Y%m%d"))) df["hour"] = df["publishAt"].apply(lambda x: int(dt.datetime.fromtimestamp(x, tz=tz.gettz("Asia/Taipei")).strftime("%H"))) df["weekday"] = df["publishAt"].apply(lambda x: int(dt.datetime.fromtimestamp(x, tz=tz.gettz("Asia/Taipei")).weekday() + 1)) df["stock_num"] = df["market"].apply(lambda x: len(x)) df.head(1) 時區名稱查表 工作日和週末的發文頻率差異df_agg = df.groupby(["weekday"]).agg({"newsId": "count"}) df_agg 每小時發文的頻率差異!pip install plotly pyyaml==5.4.1 df_agg = df.groupby(["hour"]).agg({"newsId": "count"}).reset_index() df_agg import plotly.express as px fig = px.bar(df_agg, x="hour", y="newsId", width=800, height=400, title="每小時發文的頻率差異") fig.show() 週末和工作日的每小時圖比較df_agg_weekday = df[df["weekday"].isin([1,2,3,4,5])].groupby(["hour"]).agg({"newsId": "count"}) df_agg_weekday["type"] = "weekday" df_agg_weekend = df[df["weekday"].isin([6,7])].groupby(["hour"]).agg({"newsId": "count"}) df_agg_weekend["type"] = "weekend" df_agg_weekend df_agg = pd.concat([df_agg_weekday, df_agg_weekend]) df_agg # 分組長條圖 fig = px.bar(df_agg.reset_index(), x="hour", y="newsId", color="type", text_auto=True, barmode="group") fig.show() 每篇文提及公司數量分佈fig = px.histogram(df, x="stock_num") fig.show() 📎articles_cnyes.csv📎articles_cnyes.json

第四堂 - 精選影片1 - 爬蟲：鉅亨新聞網

第四堂 - 精選影片2 - 資料清洗與儲存

第四堂 - 精選影片3 - 文章資料分析：基本分析

第四堂 - 精選影片4 - 文章資料分析：熱詞分析

第四堂 - 精選影片5 - 文章資料分析：哪些公司會一起出現