mirror of
https://github.com/NaiboWang/EasySpider.git
synced 2026-10-31 02:35:27 +08:00
122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
# 控制流程的暂停和继续
|
|
|
|
import csv
|
|
import os
|
|
import time
|
|
import uuid
|
|
import keyboard
|
|
from openpyxl import Workbook, load_workbook
|
|
import requests
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
def is_valid_url(url):
|
|
try:
|
|
result = urlparse(url)
|
|
return all([result.scheme, result.netloc])
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def check_pause(key, event):
|
|
while True:
|
|
if keyboard.is_pressed(key): # 按下p键，暂停程序
|
|
if event._flag == False:
|
|
print("任务执行中，长按p键暂停执行。")
|
|
print("Task is running, long press 'p' to pause.")
|
|
# 设置Event的值为True，使得线程b可以继续执行
|
|
event.set()
|
|
else:
|
|
# 设置Event的值为False，使得线程b暂停执行
|
|
print("任务已暂停，长按p键继续执行...")
|
|
print("Task paused, press 'p' to continue...")
|
|
event.clear()
|
|
time.sleep(1) # 每秒检查一次
|
|
|
|
|
|
def download_image(url, save_directory):
|
|
# 定义浏览器头信息
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
|
|
}
|
|
if is_valid_url(url):
|
|
# 发送 GET 请求获取图片数据
|
|
response = requests.get(url, headers=headers)
|
|
|
|
# 检查响应状态码是否为成功状态
|
|
if response.status_code == requests.codes.ok:
|
|
# 提取文件名
|
|
file_name = url.split('/')[-1].split("?")[0]
|
|
|
|
# 生成唯一的新文件名
|
|
new_file_name = file_name + '_' + \
|
|
str(uuid.uuid4()) + '_' + file_name
|
|
|
|
# 构建保存路径
|
|
save_path = os.path.join(save_directory, new_file_name)
|
|
|
|
# 保存图片到本地
|
|
with open(save_path, 'wb') as file:
|
|
file.write(response.content)
|
|
|
|
print("图片已成功下载到:", save_path)
|
|
print("The image has been successfully downloaded to:", save_path)
|
|
else:
|
|
print("下载图片失败，请检查此图片链接是否有效:", url)
|
|
print(
|
|
"Failed to download image, please check if this image link is valid:", url)
|
|
else:
|
|
print("下载图片失败，请检查此图片链接是否有效:", url)
|
|
print("Failed to download image, please check if this image link is valid:", url)
|
|
|
|
|
|
def get_output_code(output):
|
|
try:
|
|
if output.find("rue") != -1: # 如果返回值中包含true
|
|
code = 1
|
|
else:
|
|
code = int(output)
|
|
except:
|
|
code = 0
|
|
return code
|
|
|
|
# 判断字段是否为空
|
|
|
|
|
|
def isnull(s):
|
|
return len(s) != 0
|
|
|
|
|
|
def write_to_csv(file_name, data):
|
|
with open(file_name, 'a', encoding='utf-8-sig', newline="") as f:
|
|
f_csv = csv.writer(f)
|
|
for line in data:
|
|
f_csv.writerow(line)
|
|
f.close()
|
|
|
|
|
|
def write_to_excel(file_name, data):
|
|
if os.path.exists(file_name):
|
|
# 加载现有的工作簿
|
|
wb = load_workbook(file_name)
|
|
ws = wb.active
|
|
else:
|
|
# 创建新的工作簿和工作表
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
# 追加数据到工作表
|
|
for line in data:
|
|
ws.append(line)
|
|
# 保存工作簿
|
|
wb.save(file_name)
|
|
|
|
|
|
class Time:
|
|
def __init__(self, type1=""):
|
|
self.t = int(round(time.time() * 1000))
|
|
self.type = type1
|
|
|
|
def end(self):
|
|
at = int(round(time.time() * 1000))
|
|
print("Time used for", self.type, ":", at - self.t, "ms")
|