Sunshine GaoKao Project
Project Requirements
Scrape basic information and admission guidelines from various universities (the admission guidelines should be stored in PDF format and entered into the database).
Database Table Design
- id
- task_url
- status: 0 (not scraped), 1 (scraping), 2 (scraping completed), 3 (error), 4 (updating), 5 (data updated successfully), 6 (data not updated, remains the same), 9 (not available), 8 (not available)
- 3: Error, occurs because there are no
ptags under thisdiv, leading to a timeout errorawait page.waitForXPath('//div[@class="content zszc-content UEditor"]//p'), can be handled separately, a total of 4. - Special cases that can be waited for include multiple
ptags, multipledivtags, and tables, all of which have been handled separately. - university_name
- competent_department
- educational_background
- title
- contents
Sunshine Guidelines 1
Due to two page redirects, the previous page object will be destroyed, making it impossible to loop through items for scraping, so it is necessary to first loop through items to scrape task_url and necessary data into the database.
Subsequently, read task_url for scraping.
Source Code:
import asyncio # Coroutine
from pyppeteer import launch
from pyppeteer_stealth import stealth # Eliminate fingerprint
from lxml import etree # XPath data parsing
import pymysql
width, height = 1366, 768 # Set browser width and height
conn = pymysql.connect(user='root', password='123456', db='sunshine')
cursor = conn.cursor()
async def main():
# Set whether to open the browser visibly at startup, eliminate control bar information
browser = await launch(headless=False, args=['--disable-infobars']) # Set browser and add attributes
# Open a page object
page = await browser.newPage()
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
# Eliminate fingerprint
await stealth(page) # <-- Here
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
# Access the first page
await page.goto(
'https://gaokao.chsi.com.cn/zsgs/zhangcheng/listVerifedZszc--method-index,ssdm-,yxls-,xlcc-,zgsx-,yxjbz-,start-0.dhtml')
await page.waitForXPath('//li[@class="ivu-page-item"]/@title') # Wait for a certain node to appear based on XPath
# Get the maximum page
max_page_1 = await page.xpath('//li[@class="ivu-page-item"]/@title')
max_page = await (await max_page_1[-1].getProperty("textContent")).jsonValue()
# print(max_page)
for pp in range(int(max_page)):
print(pp*100)
await page.goto('https://gaokao.chsi.com.cn/zsgs/zhangcheng/listVerifedZszc--method-index,ssdm-,yxls-,xlcc-,zgsx-,yxjbz-,start-{}.dhtml'.format(pp*100))
# Scrape single page data
await asyncio.sleep(2)
# Wait for elements to appear based on CSS selector syntax, similar to pyquery syntax
await page.waitForSelector('div.info-box')
# Scroll the document
# arg1: pixels to scroll right, arg2: pixels to scroll down
await page.evaluate('window.scrollBy(200, document.body.scrollHeight)')
# Wait for the last item to appear
await asyncio.sleep(2)
# Parse single page data
div_list = await page.xpath('//div[@class="info-box"]')
for i in div_list:
university_name_1 = await i.xpath('div[1]/a')
competent_department_1 = await i.xpath('a[1]')
educational_background_1 = await i.xpath('a[2]')
# University name
university_name = await (await university_name_1[0].getProperty("textContent")).jsonValue()
# Competent department
competent_department = await (await competent_department_1[0].getProperty("textContent")).jsonValue()
# Educational background
educational_background = await (await educational_background_1[0].getProperty("textContent")).jsonValue()
educational_background = educational_background.replace('\n', '')
educational_background = educational_background.replace(' ', '')
# Admission guideline URL
# zszc-link text-decoration-none no-info
# zszc-link text-decoration-none
task_url_1 = await i.xpath('a[@class="zszc-link text-decoration-none" and not(contains(@class, "no-info"))]/@href')
if len(task_url_1) > 0:
task_url = await (await task_url_1[0].getProperty("textContent")).jsonValue()
task_url = "https://gaokao.chsi.com.cn{}".format(task_url)
sql = 'insert into tasks(task_url,status,university_name,competent_department,educational_background)' \
' values("{}", "0", "{}", "{}", "{}")'.format(task_url.strip(), university_name.strip(),
competent_department.strip(), educational_background.strip())
else:
task_url = "暂无"
sql = 'insert into tasks(task_url,status,university_name,competent_department,educational_background)' \
' values("{}", "9", "{}", "{}", "{}")'.format(task_url.strip(), university_name.strip(),
competent_department.strip(), educational_background.strip())
# strip to remove spaces: the data obtained by xpath may have spaces on both sides, occupying database space
# print(1, university_name.strip(), 2, competent_department.strip(), 3, educational_background.strip(), 4,
# task_url.strip())
# print(sql)
cursor.execute(sql)
conn.commit()
await asyncio.sleep(100)
if __name__ == '__main__':
asyncio.get_event_loop().run_until_complete(main())
Sunshine Guidelines 2
Source Code:
import asyncio # Coroutine
import multiprocessing
import time
from pyppeteer import launch
from pyppeteer_stealth import stealth # Eliminate fingerprint
from lxml import etree # XPath data parsing
import pymysql
from pymysql.converters import escape_string
import os
width, height = 1366, 768 # Set browser width and height
r = redis.Redis(host="127.0.0.1", port=6379, db=1)
MAX_RETRIES = 3
# Save chapter pages as PDF or Word and store in the database
# Use multiprocessing or multithreading to improve scraping speed
# Resume scraping from the last point
# Incremental web scraping
async def main():
conn = pymysql.connect(user='root', password='123456', db='sunshine')
cursor = conn.cursor()
# Set whether to open the browser visibly at startup, eliminate control bar information
global retries
browser = await launch(headless=True, args=['--disable-infobars']) # Set browser and add attributes
# Open a page object
page = await browser.newPage()
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
# Eliminate fingerprint
await stealth(page) # <-- Here
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
# Access a certain page
allline = get_count()
for i in range(allline):
retries = 0 # Reset retry count
result = get_task()
url = result[1]
id = result[0]
while retries < MAX_RETRIES:
await page.goto(url)
# Smart wait: cannot be 100% sure that this link exists, so error handling
try:
await page.waitForXPath('//a[@class="zszc-zc-title"]/@href')
except:
sql = 'update tasks set status="8" where id = {}'.format(id)
cursor.execute(sql)
conn.commit()
continue
# info_url
info_url_1 = await page.xpath('//a[@class="zszc-zc-title"]/@href')
info_url = await (await info_url_1[0].getProperty("textContent")).jsonValue()
info_url = "https://gaokao.chsi.com.cn{}".format(info_url)
# Access info_url
await page.goto(info_url)
# Smart wait
try:
await page.waitForXPath('//div[@class="content zszc-content UEditor"]//p')
except Exception:
retries += 1
await asyncio.sleep(1)
continue
# title
title = await page.xpath('//h2[@class="zszc-content-title"]')
title = await (await title[0].getProperty("textContent")).jsonValue()
# Screenshot as PDF
# Currently only supports headless mode, head mode does not work
if not os.path.isdir('阳光/pdf'):
os.makedirs('阳光/pdf')
await page.pdf({'path': '阳光/pdf/{}.pdf'.format(title), 'format': 'a4'})
# contents
contents_p = await page.xpath('//div[@class="content zszc-content UEditor"]//p')
contents_list = '\n'.join([await (await x.getProperty("textContent")).jsonValue() for x in contents_p])
# Handle special cases for tables
if not contents_list.strip():
# content_div = await page.xpath('//table')[0].xpath('string(.)')
# content_list = await (await content_div[0].getProperty("textContent")).jsonValue()
# print(55555,content_list)
try:
content = etree.HTML(await page.content()).xpath('//table')[0]
contents = escape_string(etree.tostring(content, encoding='utf-8').decode())
except IndexError:
pass
try:
contents_div = await page.xpath('//div[@class="content zszc-content UEditor"]//div')
contents_list = '\n'.join(
[await (await x.getProperty("textContent")).jsonValue() for x in contents_div])
contents = escape_string(contents_list)
except Exception:
pass
# print(555555, content_list)
else:
# escape_string: escape single and double quotes in text to prevent conflicts
contents = escape_string(contents_list)
# print(title, contents_list)
print(title)
# Store in database
sql = 'update tasks set title="{}", contents="{}", status="2" where id={}'.format(title, contents, id)
cursor.execute(sql)
conn.commit()
break
if retries == 3:
sql = 'update tasks set status="3" where id={}'.format(id)
cursor.execute(sql)
conn.commit()
# Close the browser
await browser.close()
# Get the number of tasks
def get_count():
conn = pymysql.connect(user='root', password='123456', db='sunshine')
cursor = conn.cursor()
sql = 'select count(*) from tasks where status="0"'
cursor.execute(sql)
result = cursor.fetchone()
print(result)
return result[0]
# Get a task
def get_task():
conn = pymysql.connect(user='root', password='123456', db='sunshine')
cursor = conn.cursor()
sql = 'select * from tasks where status="0"'
cursor.execute(sql)
result = cursor.fetchone()
sql1 = 'update tasks set status="1" where id={}'.format(result[0])
cursor.execute(sql1)
conn.commit()
return result
# Only run based on async
def run_async():
asyncio.get_event_loop().run_until_complete(main())
# Run multiple processes
def run_mutiprocess():
pool = multiprocessing.Pool(8)
for _ in range(8):
# multiprocessing.Pool is designed for synchronous functions, if you must use multiprocessing, ensure each process has its own event loop.
pool.apply_async(run_async)
print('Waiting for all subprocesses done...')
pool.close()
pool.join()
print('All subprocesses done.')
async def run_gather():
tasks = [main() for _ in range(8)]
await asyncio.gather(*tasks)
# Run multiple coroutines
def run_coroutine():
asyncio.get_event_loop().run_until_complete(run_gather())
if __name__ == '__main__':
start_time = time.time()
# Only based on async
# run_async()
# Multiple processes
run_mutiprocess()
# Multiple coroutines
# run_coroutine()
end_time = time.time()
print("Total time taken: {}".format(end_time - start_time))
Multiprocessing
33 min
Multicore
Project Highlights
Interview points for the above project
The necessity of the status field being 1
Issues with shared resources in multiprocessing:
If there is no 1, multiple processes may compete for the same resource during data scraping, and setting the status field to 1 while scraping this <code>task_url avoids this situation.
Error Handling
Generally occurs in smart waiting (a series of errors caused by timeout errors), establish a retry mechanism, and upon reaching the maximum retry count, set the status field to 0, which will allow for re-scraping later, preventing program termination due to errors.
Improve the above project
Resume Scraping
If the program is interrupted manually, running the program again will ensure that scraping continues from where it left off.
# Improve project: Resume scraping
async def crawler_resumpt():
conn = pymysql.connect(user='root', password='123456', db='sunshine')
cursor = conn.cursor()
sql = 'select * from tasks where status="1" or status="0" order by id'
cursor.execute(sql)
results = cursor.fetchall()
# Set whether to open the browser visibly at startup, eliminate control bar information
global retries
browser = await launch(headless=True, args=['--disable-infobars']) # Set browser and add attributes
# Open a page object
page = await browser.newPage()
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
# Eliminate fingerprint
await stealth(page) # <-- Here
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
for result in results:
# Access a certain page
retries = 0 # Reset retry count
url = result[1]
id = result[0]
while retries < MAX_RETRIES:
await page.goto(url)
# Smart wait: cannot be 100% sure that this link exists, so error handling
try:
await page.waitForXPath('//a[@class="zszc-zc-title"]/@href')
except:
sql = 'update tasks set status="8" where id = {}'.format(id)
cursor.execute(sql)
conn.commit()
continue
# info_url
info_url_1 = await page.xpath('//a[@class="zszc-zc-title"]/@href')
info_url = await (await info_url_1[0].getProperty("textContent")).jsonValue()
info_url = "https://gaokao.chsi.com.cn{}".format(info_url)
# Access info_url
await page.goto(info_url)
# Smart wait
try:
await page.waitForXPath('//div[@class="content zszc-content UEditor"]//p')
except Exception:
retries += 1
await asyncio.sleep(1)
continue
# title
title = await page.xpath('//h2[@class="zszc-content-title"]')
title = await (await title[0].getProperty("textContent")).jsonValue()
# Screenshot as PDF
# Currently only supports headless mode, head mode does not work
if not os.path.isdir('阳光/pdf'):
os.makedirs('阳光/pdf')
await page.pdf({'path': '阳光/pdf/{}.pdf'.format(title), 'format': 'a4'})
# contents
contents_p = await page.xpath('//div[@class="content zszc-content UEditor"]//p')
contents_list = '\n'.join([await (await x.getProperty("textContent")).jsonValue() for x in contents_p])
# Handle special cases for tables
if not contents_list.strip():
# content_div = await page.xpath('//table')[0].xpath('string(.)')
# content_list = await (await content_div[0].getProperty("textContent")).jsonValue()
# print(55555,content_list)
content = etree.HTML(await page.content()).xpath('//table')[0]
contents = escape_string(etree.tostring(content, encoding='utf-8').decode())
else:
# escape_string: escape single and double quotes in text to prevent conflicts
contents = escape_string(contents_list)
print(title)
# Store in database
sql = 'update tasks set title="{}", contents="{}", status="2" where id={}'.format(title, contents, id)
cursor.execute(sql)
conn.commit()
break
if retries == 3:
sql = 'update tasks set status="0" where id={}'.format(id)
cursor.execute(sql)
conn.commit()
# Close the browser
await browser.close()
Incremental Scraping, Fingerprint Deduplication
Fingerprint: The key string formed by concatenating the scraped data and encrypting it using md5 or sha1 is the fingerprint.
Store the fingerprint and id in an unordered set in the Redis database.
During subsequent data scraping, construct the key string and check for its presence; if it exists, skip data updates; if not, update based on id.
Initial scraping source code:
# Store in database
sql = 'update tasks set title="{}", contents="{}", status="2" where id={}'.format(title, contents, id)
cursor.execute(sql)
conn.commit()
# Fingerprint storage
data = title + contents
r.sadd("sunshine:key", encryption(data))
Incremental scraping source code:
import asyncio # Coroutine
import multiprocessing
import time
from pyppeteer import launch
from pyppeteer_stealth import stealth # Eliminate fingerprint
from lxml import etree # XPath data parsing
import pymysql
from pymysql.converters import escape_string
import os
import redis
import hashlib
width, height = 1366, 768 # Set browser width and height
r = redis.Redis(host="127.0.0.1", port=6379, db=1)
MAX_RETRIES = 3
async def main():
conn = pymysql.connect(user='root', password='123456', db='sunshine2')
cursor = conn.cursor()
# Set whether to open the browser visibly at startup, eliminate control bar information
global retries, contents
browser = await launch(headless=True, args=['--disable-infobars']) # Set browser and add attributes
# Open a page object
page = await browser.newPage()
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
# Eliminate fingerprint
await stealth(page) # <-- Here
# Set browser width and height
await page.setViewport({'width': width, 'height': height})
# Access a certain page
allline = get_count()
for i in range(allline):
retries = 0 # Reset retry count
result = get_finished_task()
url = result[1]
id = result[0]
while retries < MAX_RETRIES:
await page.goto(url)
# Smart wait: cannot be 100% sure that this link exists, so error handling
try:
await page.waitForXPath('//a[@class="zszc-zc-title"]/@href')
except:
sql = 'update tasks set status="8" where id = {}'.format(id)
cursor.execute(sql)
conn.commit()
continue
# info_url
info_url_1 = await page.xpath('//a[@class="zszc-zc-title"]/@href')
info_url = await (await info_url_1[0].getProperty("textContent")).jsonValue()
info_url = "https://gaokao.chsi.com.cn{}".format(info_url)
# Access info_url
await page.goto(info_url)
# Smart wait
try:
await page.waitForXPath('//div[@class="content zszc-content UEditor"]//p')
except Exception:
retries += 1
await asyncio.sleep(1)
continue
# title
title = await page.xpath('//h2[@class="zszc-content-title"]')
title = await (await title[0].getProperty("textContent")).jsonValue()
# Screenshot as PDF
# Currently only supports headless mode, head mode does not work
if not os.path.isdir('阳光/pdf'):
os.makedirs('阳光/pdf')
await page.pdf({'path': '阳光/pdf/{}.pdf'.format(title), 'format': 'a4'})
# contents
contents_p = await page.xpath('//div[@class="content zszc-content UEditor"]//p')
contents_list = '\n'.join([await (await x.getProperty("textContent")).jsonValue() for x in contents_p])
# Handle special cases for tables
if not contents_list.strip():
# content_div = await page.xpath('//table')[0].xpath('string(.)')
# content_list = await (await content_div[0].getProperty("textContent")).jsonValue()
# print(55555,content_list)
try:
content = etree.HTML(await page.content()).xpath('//table')[0]
contents = escape_string(etree.tostring(content, encoding='utf-8').decode())
except IndexError:
pass
try:
contents_div = await page.xpath('//div[@class="content zszc-content UEditor"]//div')
contents_list = '\n'.join(
[await (await x.getProperty("textContent")).jsonValue() for x in contents_div])
contents = escape_string(contents_list)
except Exception:
pass
# print(555555, content_list)
else:
# escape_string: escape single and double quotes in text to prevent conflicts
contents = escape_string(contents_list)
# print(title, contents_list)
print(title)
# Store in database
data = title + contents
if not is_crawlered(data):
print("Data updated...")
sql = 'update tasks set title="{}", contents="{}", status="5" where id={}'.format(title, contents, id)
cursor.execute(sql)
conn.commit()
else:
print("Data has been scraped...")
sql = 'update tasks set status="6" where id={}'.format(id)
cursor.execute(sql)
conn.commit()
break
if retries == 3:
sql = 'update tasks set status="3" where id={}'.format(id)
cursor.execute(sql)
conn.commit()
# Close the browser
await browser.close()
# Get the number of tasks
def get_count():
conn = pymysql.connect(user='root', password='123456', db='sunshine2')
cursor = conn.cursor()
sql = 'select count(*) from tasks where status="2"'
cursor.execute(sql)
result = cursor.fetchone()
print(result)
return result[0]
# md5 encryption
def encryption(data):
md5 = hashlib.md5()
md5.update(data.encode("utf-8"))
return md5.hexdigest()
# Get a finished task
def get_finished_task():
conn = pymysql.connect(user='root', password='123456', db='sunshine2')
cursor = conn.cursor()
sql = 'select * from tasks where status="2"'
cursor.execute(sql)
result = cursor.fetchone()
sql1 = 'update tasks set status="4" where id={}'.format(result[0])
cursor.execute(sql1)
conn.commit()
return result
# Deduplication
def is_crawlered(data):
res = r.sadd("sunshine:key", encryption(data))
return res == 0
# Only run based on async
def run_async():
asyncio.get_event_loop().run_until_complete(main())
# Run multiple processes
def run_mutiprocess():
pool = multiprocessing.Pool(8)
for _ in range(8):
# multiprocessing.Pool is designed for synchronous functions, if you must use multiprocessing, ensure each process has its own event loop.
pool.apply_async(run_async)
print('Waiting for all subprocesses done...')
pool.close()
pool.join()
print('All subprocesses done.')
async def run_gather():
tasks = [main() for _ in range(4)]
await asyncio.gather(*tasks)
# Run multiple coroutines
def run_coroutine():
asyncio.get_event_loop().run_until_complete(run_gather())
if __name__ == '__main__':
start_time = time.time()
# Only based on async
# run_async()
# Multiple processes
run_mutiprocess()
# Multiple coroutines
# run_coroutine()
end_time = time.time()
print("Total time taken: {}".format(end_time - start_time))