[Python 爬虫这样学，一定是大拇指拉！] DAY28 - 实战演练：集大成 - 自动更新每日个股日成交资讯

自动更新每日个股日成交资讯

结合前几篇所学，我们来做一个可以自动更新日成交资讯的程序吧！

程序部分：

import json
import time
import requests
from pathlib import Path
from threading import Thread
from datetime import datetime
from fake_useragent import UserAgent


class DailyPriceSpider:

    def __init__(self):
    
        # 重复使用 TCP 连线
        self.req = requests.Session()
        self.url = "https://www.twse.com.tw/exchangeReport/STOCK_DAY"
        self.headers = self.req.headers
        
        # 伪装 User-Agent
        ua = UserAgent()
        self.headers["User-Agent"] = ua.random
 
    # 供内部使用
    def __get(self, date, stock_no):
        res = self.req.get(self.url,
                           headers = self.headers,
                           params={
                               "response": "csv",  # 这次抓的是 csv 格式
                               "date": date,
                               "stockNo": stock_no
                           })
        return res
    
    # 供内部使用
    def __save_file(self, res_text, path):
        # 去掉 res_text 里多余的空白行
        res_text = '\n'.join(
            filter(None,
                   res_text.splitlines()
                   )
        )

        path = Path(path)
        # parents=True，如果父资料夹不存在则会一并创建
        # exist_ok=True，创建资料夹时，该资料夹已存在则不会 throw exception
        path.parent.mkdir(parents=True, exist_ok=True)
        
        # 这边我比较懒一点，是直接覆盖整个档案，可以依照个人喜好去修改
        with open(path, 'w', encoding="utf-8") as file:
            file.write(res_text)
        return
 
    # 把上面的两个 function 整合成一个 function，供外部使用
    def scrape(self, date, stock_no, save_path=""):
        res = self.__get(date, stock_no)
        if save_path:
            res_text = res.text
            self.__save_file(res_text, save_path)
        # 虽然这边没用到但还是可以先 return response 保留日後扩充弹性
        return res

if __name__ == '__main__':
    SAVE_PATH_ROOT = "./daily_stock_price/"
    SLEEP_TIME = 3
    
    # 之前有实作过抓取股票清单的程序了，我们直接读取清单里的内容来使用
    # 程序实作篇章传送门请至本篇文章最上面
    stock_info_list_file = {}
    with open("./stock_info_list.json", "r", encoding="utf-8") as f:
        stock_info_list_file = json.load(f)

    # 不管日期是几号，他回传都是给我们一整个月的，所以就固定设成1号就好
    # strftime()，把 datetime 输出成我们要的格式
    today_date = "{}01".format(datetime.now().strftime("%Y%m"))

    stock_info_list = stock_info_list_file.get("stock", [])
    dps = DailyPriceSpider()
    req_thread_list = []

    for stock_info in stock_info_list:

        stock_no = stock_info.get("stockNo")
        stock_name = stock_info.get("stockName")
        stock_industry = stock_info.get("stockIndustry")
        file_name = "{}_{}_daily_price.csv".format(
            today_date[:-2],  # 字串只需要用到年跟月
            stock_no+stock_name)

        save_path = "{}/{}/{}/{}".format(SAVE_PATH_ROOT,
                                         stock_industry,
                                         stock_no+stock_name,
                                         file_name)

        if stock_no and stock_name and stock_industry:
            # daemon=True，就是 main 结束时，daemon=True 的执行绪也会跟着结束
            req_thread = Thread(target=dps.scrape,
                                args=(today_date, stock_no, save_path),
                                daemon=True)
            req_thread.start()
            req_thread_list.append(req_thread)
            time.sleep(SLEEP_TIME)

    for req_thread in req_thread_list:
        req_thread.join()

    print("Finished")

输出的资料会长这样：