Kỹ thuật lập trình Web Scraper cào dữ liệu điểm thi THPTQG năm 2023
Tìm giải pháp
Xây dựng ý tưởng
pip install beautifulsoup4
pip install requests
Đến đây thì mình đã có thể truy cập và lấy được nội dung số báo danh, điểm từ một trang web bất kỳ bằng đoạn code đơn giản sau:
import requests
from bs4 import BeautifulSoup
so_bao_danh = "01000000"
URL = "https://vietnamnet.vn/giao-duc/diem-thi/tra-cuu-diem-thi-tot-nghiep-thpt/2023/{}.html".format(so_bao_danh)
r = requests.get(URL)
soup = BeautifulSoup(r.content, 'html.parser')
target = soup.find('div', attrs={'class': 'resultSearch__right'})
table = target.find('tbody')
rows = table.find_all('tr')
placeHolder = []
for row in rows:
lst = row.find_all('td')
cols = [ele.text.strip() for ele in lst]
placeHolder.append([ele for ele in cols if ele])
content = "{},{}\n".format(so_bao_danh, placeHolder)
Bằng cách đưa vào một vòng lặp đơn giản, mình đã có thể tạo ra một chương trình tự động thu thập điểm số trong kì thi THPTQG như sau:
import requests
from bs4 import BeautifulSoup
for so_bao_danh in range(1000001, 100000000):
so_bao_danh = str(so_bao_danh).rjust(8, '0')
URL = "https://vietnamnet.vn/giao-duc/diem-thi/tra-cuu-diem-thi-tot-nghiep-thpt/2023/{}.html".format(so_bao_danh)
r = requests.get(URL)
soup = BeautifulSoup(r.content, 'html.parser')
target = soup.find('div', attrs={'class': 'resultSearch__right'})
table = target.find('tbody')
rows = table.find_all('tr')
placeHolder = []
for row in rows:
lst = row.find_all('td')
cols = [ele.text.strip() for ele in lst]
placeHolder.append([ele for ele in cols if ele])
content = "{},{}\n".format(so_bao_danh, placeHolder)
Tại đoạn code trên lưu ý khi dùng vòng lặp phải chuyển đôi số báo danh ở dạng int sang dạng string trước khi nạp vào url. Lúc này số báo danh ở dạng string phải được điền đủ 8 chữ số, hàm rjust() trong python cho phép xử lý nhiệm vụ này.
Kết hợp đa luồng
Quá trình xử lý đã hoàn thiện, tuy nhiên nếu xử lý từng trang một thế này mình sẽ phải tốn thời gian rất lâu để thu thập hết đống dữ liệu này, khoảng 99000000 lần. Để rút ngắn thời gian xử lý, ý tưởng ban đầu là viết một chương trình cho phép xử lý song song hàng loạt các trong cùng một lúc. Để thực hiện điều này chúng ta cần biết một vài kiến thức liên quan đến Multi-Thread hoặc Multi-Process. Hiểu đơn giản thì CPU của máy tính bao gồm nhiều nhân và nhiều luồng, điều này cho phép người dùng xử lý nhiều tác vụ cùng một lúc. Lúc này mình cần đến worker, worker được hiểu như một luồng xử lý song song với luồng chính. Một máy nhiều nhân nhiều luồng sẽ có thể sử dụng nhiều worker hơn. Tiếp theo là chunksize, hiểu nôm na là số lượng url được xử lý trong mỗi lần nạp. Mình đã viết đoạn mã để thực hiện Scraper Multi-Thread như sau:
import requests
from bs4 import BeautifulSoup
from multiprocessing.dummy import Pool
from multiprocessing import cpu_count
def Crawl_THPTQG(so_bao_danh):
so_bao_danh = str(so_bao_danh).rjust(8, '0')
URL = "https://vietnamnet.vn/giao-duc/diem-thi/tra-cuu-diem-thi-tot-nghiep-thpt/2023/{}.html".format(so_bao_danh)
r = requests.get(URL)
soup = BeautifulSoup(r.content, 'html.parser')
target = soup.find('div', attrs={'class': 'resultSearch__right'})
table = target.find('tbody')
rows = table.find_all('tr')
placeHolder = []
for row in rows:
lst = row.find_all('td')
cols = [ele.text.strip() for ele in lst]
placeHolder.append([ele for ele in cols if ele])
content = "{},{}\n".format(so_bao_danh, placeHolder)
return content
if __name__ == "__main__":
lst = range(1000001, 1000000000)
NUM_WORKERS = cpu_count() * 2
pool = Pool(NUM_WORKERS)
result_iter = pool.imap(Crawl_THPTQG, lst)
for result in result_iter:
print("Đang xử lý điểm thí sinh số {}".format(result.split(',')[0]))
Khắc phục hiện tượng chặn truy cập
Với kỹ thuật xử lý song song thì chương trình đã chạy nhanh hơn gấp nhiều lần. Tuy nhiên khi chạy thì mình phát hiện một vấn đề. Với lượt truy cập nhanh và liên tục như vậy đôi khi Vietnamnet không kịp phải hồi khiến chương trình hay dừng đột ngột với mã lỗi: Max retries exceeded with URL in requests
Để giải quyết vấn đề này mình viết lại tính năng session thông qua requests. Phương pháp này cho phép requests tạo một phiên mới khi bị chặn và thử truy cập lại với số lượt thử được chỉ định.
session = requests.Session()
retry = Retry(connect=3, backoff_factor=0.5)
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
Tại session.mount, mình khai báo loại kết nối có trong url, trong trường hợp này là https://, kết nối được mã hoá. Lúc này chúng ta đã có thể khắc phục lỗi bị chặn khi truy cập trang web quá nhiều lần trong thời gian ngắn. bước cuối cùng là viết lệnh ghi file cho mỗi lần hoàn tất thu thập dữ liệu. Lưu ý file phải được cài đặt ở chế độ ghi nối tiếp, không được ghi đè. Chương trình hoàn thiện như sau:
import requests
from bs4 import BeautifulSoup
from multiprocessing.dummy import Pool
from multiprocessing import cpu_count
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
retry = Retry(connect=3, backoff_factor=0.5)
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
def Crawl_THPTQG(so_bao_danh):
so_bao_danh = str(so_bao_danh).rjust(8, '0')
URL = "https://vietnamnet.vn/giao-duc/diem-thi/tra-cuu-diem-thi-tot-nghiep-thpt/2023/{}.html".format(so_bao_danh)
r = session.get(URL)
soup = BeautifulSoup(r.content, 'html.parser')
target = soup.find('div', attrs={'class': 'resultSearch__right'})
table = target.find('tbody')
rows = table.find_all('tr')
placeHolder = []
for row in rows:
lst = row.find_all('td')
cols = [ele.text.strip() for ele in lst]
placeHolder.append([ele for ele in cols if ele])
content = "{},{}\n".format(so_bao_danh, placeHolder)
return content
if __name__ == "__main__":
lst = range(1000001, 1000000000)
NUM_WORKERS = cpu_count() * 2
pool = Pool(NUM_WORKERS)
result_iter = pool.imap(Crawl_THPTQG, lst)
with open("Output.csv", "a", encoding='utf-8') as f:
for result in result_iter:
f.write(result)
print("Đang xử lý điểm của thí sinh số {}".format(result.split(',')[0]))
Tối ưu
import requests
from bs4 import BeautifulSoup
from multiprocessing.dummy import Pool
from multiprocessing import cpu_count
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
retry = Retry(connect=3, backoff_factor=0.5)
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
def Crawl_THPTQG(so_bao_danh):
so_bao_danh = str(so_bao_danh).rjust(8, '0')
URL = "https://vietnamnet.vn/giao-duc/diem-thi/tra-cuu-diem-thi-tot-nghiep-thpt/2023/{}.html".format(so_bao_danh)
r = session.get(URL)
if r.status_code != 404:
soup = BeautifulSoup(r.content, 'html.parser')
target = soup.find('div', attrs={'class': 'resultSearch__right'})
table = target.find('tbody')
rows = table.find_all('tr')
placeHolder = []
for row in rows:
lst = row.find_all('td')
cols = [ele.text.strip() for ele in lst]
placeHolder.append([ele for ele in cols if ele])
content = "{},{}\n".format(so_bao_danh, placeHolder)
else:
return None
return content
if __name__ == "__main__":
provinces = range(0, 65)
for province in provinces:
count = 0
start = province*1000000 + 1
lst = range(start, start + 1000000)
NUM_WORKERS = cpu_count() * 8
pool = Pool(NUM_WORKERS)
result_iter = pool.imap(Crawl_THPTQG, lst)
with open("Output2.csv", "a", encoding='utf-8') as f:
for result in result_iter:
if result is not None:
f.write(result)
print("Đang xử lý {}".format(result.split(',')[0]))
else:
count += 1
if count == 10:
break
Vừa rồi mình đã trình bày một vài kỹ thuật xử lý đa luồng khi cào dữ liệu điểm thi THPTQG, hy vọng bài viết cung cấp cho các bạn các kiến thức thú vị.
Đăng nhận xét