spider.py
# -*- coding: utf-8 -*-
"""
@Desc : 基类
@Date : 2024/7/3 19:31
@Author : gaofenfei
"""
import time,os,requests,random,re,shutil,sys,traceback,json
from datetime import date,datetime,timedelta
from urllib.parse import urlparse, parse_qs
import configparser
class Spider:
def __init__(self,spiderFolder,comFrom):
config = configparser.ConfigParser() # 创建对象
config.read("spider.ini", encoding="utf-8")
self.chromePath = config.get("Common", "chromePath") # 谷歌浏览器的位置
self.proxies = config.get("Common", 'proxies').split(',') # 代理
self.savePath = config.get('Common', 'savePath') # 保存路径
self.python8084 = config.get("Common", 'python8084')
self.timeInterval = int(config.get("Common", 'timeInterval')) # 时间间隔
self.maxTryTime = int(config.get("Common", 'maxTryTime')) # 最大入库尝试次数
self.comUrl = config.get("Common", 'comUrl') # 入公司接口
self.jobUrl = config.get("Common", 'jobUrl') # 入职位接口
self.getProxyUrl = config.get("Common", 'getProxy') # 获取代理
self.setProxyRecordUrl = config.get("Common", 'setProxyRecord') # 设置代理不可用
self.conditionsApi = config.get('Common','conditionsApi') # 条件接口
self.spiderUrlApi = config.get('Common','spiderUrlApi') # 要抓取的链接接口
self.jobLinksApi = config.get('Common','jobLinksApi') # 职位链接接口
self.jobToExpiredApi = config.get('Common','jobToExpiredApi') # 职位过期接口
self.comFrom = comFrom
self.channel = spiderFolder
self.tryTime = 0 # 尝试次数
self.limitSpiderCount = 0 # 还可用抓取的个数
self.realProxy = '' # 真是代理（外网IP）
self.lastId = '' # 上次抓取的职位ID
self.hasNext = True # 是否还有下一页
# 创建文件夹
if not os.path.exists(self.savePath):
os.mkdir(self.savePath)
self.savePath = os.path.join(self.savePath, spiderFolder)
if not os.path.exists(self.savePath):
os.mkdir(self.savePath)
self.comPath = os.path.join(self.savePath, 'com')
if not os.path.exists(self.comPath):
os.mkdir(self.comPath)
self.jobPath = os.path.join(self.savePath, 'job')
if not os.path.exists(self.jobPath):
os.mkdir(self.jobPath)
self.successPath = os.path.join(self.savePath, 'success') # 成功挪入的文件夹
if not os.path.exists(self.successPath):
os.mkdir(self.successPath)
self.failedPath = os.path.join(self.savePath, 'failed') # 失败挪入的文件夹
if not os.path.exists(self.failedPath):
os.mkdir(self.failedPath)
self.progressPath = os.path.join(self.savePath, 'progress') # 进度文件夹
if not os.path.exists(self.progressPath):
os.mkdir(self.progressPath)
self.companyPath = os.path.join(self.savePath, 'company') # 存储大公司数据的文件夹
if not os.path.exists(self.companyPath):
os.mkdir(self.companyPath)
self.cleanPath = os.path.join(self.savePath, 'clean') # 清理文件夹
if not os.path.exists(self.cleanPath):
os.mkdir(self.cleanPath)
def getProxy(self,tryTimes = 0):
'''获取代理'''
print('trytimes',tryTimes)
if tryTimes >= 3:
return ''
params = {"channel": self.channel, "env": 1}
proxy = ''
try:
ret = requests.post(self.getProxyUrl, data=params)
if ret.status_code == 200:
rescontent = json.loads(ret.content.decode())
if rescontent.get("code") == 200:
proxy = rescontent.get('data').get('proxy')
if rescontent.get('data').get('limitCount'):
self.limitSpiderCount = int(rescontent.get('data').get('limitCount'))
if rescontent.get('data').get('realProxy'):
self.realProxy = rescontent.get('data').get('realProxy')
else:
proxy = ''
if proxy == '':
tryTimes = tryTimes + 1
time.sleep(1)
return self.getProxy(tryTimes)
ret.close()
return proxy
except:
traceback.print_exc()
time.sleep(1)
tryTimes = tryTimes + 1
return self.getProxy(tryTimes)
def setProxyRecord(self,proxy,limitedType,spiderCount = 0,tryTimes = 0):
'''设置代理使用'''
try:
params = {"channel": self.channel, "proxy": proxy, "limitedType": limitedType, "spiderCount": spiderCount}
ret = requests.post(self.setProxyRecordUrl,params=params)
return ret.content.decode()
except:
traceback.print_exc()
tryTimes = tryTimes + 1
if tryTimes < 3:
return self.setProxyRecord(proxy,limitedType,spiderCount,tryTimes)
def jobLinks(self,suffix,tryTimes = 0):
'''获取职位链接'''
try:
# 处理suffix中的英文逗号，替换为减号
processed_suffix = suffix.replace(',', '-')
last_id_file = os.path.join(self.cleanPath, f'last_{processed_suffix}.txt')
# 读取lastId文件
if os.path.exists(last_id_file):
with open(last_id_file, 'r', encoding='utf-8') as f:
self.lastId = f.read().strip()
params = {"channel": self.channel, "suffix": suffix, 'curId':self.lastId}
ret = requests.post(self.jobLinksApi,params=params)
retJson = ret.content.decode('utf-8')
retJson = json.loads(retJson)
self.lastId = retJson.get('lastId')
print('lastId',self.lastId)
self.hasNext = retJson.get('hasNext')
# 保存lastId到文件或清空文件
if self.hasNext:
# 保存lastId到文件
with open(last_id_file, 'w', encoding='utf-8') as f:
f.write(self.lastId)
else:
# 清空lastId文件
if os.path.exists(last_id_file):
with open(last_id_file, 'w', encoding='utf-8') as f:
f.write('')
return retJson.get('links')
except:
traceback.print_exc()
tryTimes = tryTimes + 1
if tryTimes < 3:
return self.jobLinks(suffix,tryTimes)
def jobToExpired(self,jobId,url,suffix,tryTimes = 0):
'''更新职位过期'''
try:
params = {"jobid": jobId}
ret = requests.post(self.jobToExpiredApi,params=params)
self.writeCleanId(self.cleanPath, 'clean_'+suffix + '.txt', jobId, url)
print(f'职位 {jobId} 已招满/404，链接：{url}')
return ret.content.decode()
except:
traceback.print_exc()
tryTimes = tryTimes + 1
if tryTimes < 3:
return self.jobToExpired(jobId,url,suffix,tryTimes)
def writeCleanId(self, writeFolder, writeFile, jobId, url):
'''写结果'''
filePath = writeFolder + '/' + writeFile
with open(filePath, 'a+', encoding='utf-8') as file:
cur = {'time': int(time.time()), 'jobId': jobId, 'url': url}
file.write(json.dumps(cur))
file.write('\n')
def comExist(self, comname):
'''判断公司是否存在'''
url = self.python8084 + "/Com"
params = {"comname": comname, "funname": "comexist"}
ret = requests.post(url, params=params)
if ret.content.decode("utf-8") == "1":
return 1
elif ret.content.decode("utf-8") == "-1":
return -1
else:
return 0
def jobExist(self, comName, jobTitle, city):
'''职位是否存在'''
params = {"comname": comName, "jobtitle": jobTitle, "funname": "jobexist", "jobfrom": self.comFrom}
if city:
params["city"] = city
url = self.python8084 + "/Job"
ret = requests.post(url, params=params)
if ret.content.decode("utf-8") == "1":
return 1
else:
return 0
def insert(self, content, fileName, isCom):
ret = 1
rs = ''
self.tryTime = self.tryTime + 1
try:
data = {
"comFrom": self.comFrom,
"fileName": fileName,
"content": json.dumps(content)
}
url = self.comUrl if isCom else self.jobUrl
res = requests.post(url, data=data, timeout=5)
rs = res.content.decode()
print(rs)
if res.status_code == 200:
rescontent = json.loads(res.content.decode())
if rescontent.get("code") == 200:
ret = 1
elif rescontent.get("code") == 501:
if self.tryTime <= self.maxTryTime:
time.sleep(1)
return self.insert(content, fileName, isCom)
else:
ret = 0
else: # 其他状态码直接抛错误
ret = 0
else: # 状态码不是200，重试
if self.tryTime <= self.maxTryTime:
time.sleep(1)
return self.insert(content, fileName, isCom)
else:
ret = 0
except:
ret = 0
self.tryTime = 0
traceback.print_exc()
self.tryTime = 0
return ret, rs
def createFolder(self, curPath):
'''创建今天的文件夹'''
from datetime import datetime
folder = os.path.join(curPath, datetime.today().strftime("%Y%m%d"))
if not os.path.exists(folder):
os.mkdir(folder)
comFolder = os.path.join(folder, 'com')
if not os.path.exists(comFolder):
os.mkdir(comFolder)
jobFolder = os.path.join(folder, 'job')
if not os.path.exists(jobFolder):
os.mkdir(jobFolder)
return folder
def createTodayFolder(self):
# 创建成功文件夹
successFolder = self.createFolder(self.successPath)
# 创建失败文件夹
failedFolder = self.createFolder(self.failedPath)
return successFolder,failedFolder
def createCompanyFolder(self,comFileId):
'''创建公司文件夹 - 使用comFileId作为文件夹名，并创建job、success、failed子文件夹'''
folder = os.path.join(self.companyPath, comFileId)
if not os.path.exists(folder):
os.mkdir(folder)
# 创建三个子文件夹
job_folder = os.path.join(folder, 'job')
success_folder = os.path.join(folder, 'success')
failed_folder = os.path.join(folder, 'failed')
if not os.path.exists(job_folder):
os.mkdir(job_folder)
if not os.path.exists(success_folder):
os.mkdir(success_folder)
if not os.path.exists(failed_folder):
os.mkdir(failed_folder)
return job_folder,success_folder,failed_folder
def isSpiderCompanyJob(self,comFileId,jobFileId):
'''判断公司下的职位是否已经抓过'''
# 构建公司文件夹路径
companyFolder = os.path.join(self.companyPath, comFileId)
# 如果公司文件夹不存在，返回False
if not os.path.exists(companyFolder):
return False
# 检查job文件夹是否存在且有文件
jobFolder = os.path.join(companyFolder, 'success')
if not os.path.exists(jobFolder):
return False
# 检查具体的职位文件是否存在
jobFilePath = os.path.join(jobFolder, jobFileId)
return os.path.exists(jobFilePath)
def isSpiderToday(self,folder,fileId,type):
'''判断职位或公司今天是否抓过 type: com/job'''
files = os.listdir(folder)
for file in files:
filePath = os.path.join(folder, file)
fileName = os.path.basename(filePath)
if os.path.isdir(filePath):
if fileName == type:
return self.isSpiderToday(filePath,fileId,type)
else:
if fileName == fileId:
return True
return False
def traverseFolder(self, curPath, compareFileName):
# 获取文件夹中的所有文件和子文件夹
files = os.listdir(curPath)
# 遍历文件和子文件夹
for file in files:
# 获取文件或子文件夹的完整路径
filePath = os.path.join(curPath, file)
fileName = os.path.basename(filePath)
# 如果文件是文件夹，则递归遍历它
if os.path.isdir(filePath):
if fileName == 'job':
continue
elif fileName == 'com':
return self.traverseFolder(filePath, compareFileName)
else:
date_obj = datetime.strptime(fileName, "%Y%m%d")
timestamp = int(date_obj.timestamp())
# 小于6个月的才判断
print(int(time.time()) - timestamp < self.timeInterval)
if int(time.time()) - timestamp < self.timeInterval:
return self.traverseFolder(filePath, compareFileName)
else:
continue
else:
# 如果是文件且与传入文件名字相同
print(fileName, compareFileName)
if fileName == compareFileName:
print('公司文件已存在')
return False
return True
def writeLog(self, writeFolder, writeFile, fileName, rs):
'''写结果'''
filePath = writeFolder + '/' + writeFile
with open(filePath, 'a+', encoding='utf-8') as file:
cur = {'time': int(time.time()), 'fileName': fileName, 'error': rs}
file.write(json.dumps(cur))
file.write('\n')
def enterDatabase(self,jsonData,fileId,filePath,type):
'''入库'''
# 入公司
if type == 'com':
ret, rs = self.insert(jsonData, fileId, 1)
else:
ret, rs = self.insert(jsonData, fileId, 0)
if ret:
successFolder = os.path.join(self.successPath, date.today().strftime("%Y%m%d"), type)
shutil.move(filePath, successFolder + '/' + fileId)
# 正确的写入文件
self.writeLog(successFolder, type+'success.txt', fileId, rs)
if type == 'job':
# 判断公司文件夹是否存在，若存在则移动到success文件夹下
if jsonData.get('comFileId'):
comFolder = os.path.join(self.companyPath, jsonData['comFileId'])
if os.path.exists(comFolder):
shutil.move(comFolder+'/job/'+fileId, comFolder+'/success/'+fileId)
self.writeLog(comFolder+'/success', 'jobsuccess.txt', fileId, rs)
else:
failedFolder = os.path.join(self.failedPath, date.today().strftime("%Y%m%d"), type)
shutil.move(filePath, failedFolder + '/' + fileId)
# 错误的记录错误
self.writeLog(failedFolder, type+'error.txt', fileId, rs)
if type == 'job':
# 判断公司文件夹是否存在，若存在则移动到failed文件夹下
if jsonData.get('comFileId'):
comFolder = os.path.join(self.companyPath, jsonData['comFileId'])
if os.path.exists(comFolder):
shutil.move(comFolder+'/job/'+fileId, comFolder+'/failed/'+fileId)
self.writeLog(comFolder+'/failed', 'joberror.txt', fileId, rs)
def getEndDate(self,string):
'''得到节点时间'''
# 如果是日期格式，取日期格式当天零点的时间戳
try:
curDate = datetime.strptime(string, '%Y-%m-%d %H:%M')
timestamp = time.mktime(curDate.timetuple())
return int(timestamp)
except ValueError:
try:
curDate = datetime.strptime(string, '%Y-%m-%d')
timestamp = time.mktime(curDate.timetuple())
return int(timestamp)
except ValueError:
try:
nDay = abs(int(string))
# 获取当天/几天前的时间戳
if nDay >= 0:
# 获取当前日期
today = datetime.today()
# 计算n天前凌晨零点的日期
nDayAgo = today - timedelta(days=nDay)
nDayAgo = nDayAgo.replace(hour=0, minute=0, second=0, microsecond=0)
print('截止', nDayAgo)
# 将日期对象转换为时间戳
timestamp = time.mktime(nDayAgo.timetuple())
return int(timestamp)
except:
today = datetime.today()
# 计算今天凌晨零点的日期
midnight = today.replace(hour=0, minute=0, second=0, microsecond=0)
print('截止 except,注意配置项格式', midnight)
# 将日期对象转换为时间戳
timestamp = time.mktime(midnight.timetuple())
return int(timestamp)
def getContions(self,type,tryTimes = 0):
"""
获取抓取的条件
:param type:
:param tryTimes:
:return:
"""
try:
params = {'site':self.channel,'type':type}
ret = requests.post(self.conditionsApi, params=params)
if ret.status_code == 200:
jsonData = ret.json()
if jsonData.get('rs') == 'success':
return jsonData.get('data')
else:
return []
except:
traceback.print_exc()
tryTimes = tryTimes + 1
if tryTimes < 3:
return self.getContions(type,tryTimes)
else:
return []
def getSpiderUrl(self,index,tryTimes = 0):
"""
获取要抓取的链接
:param index:
:param tryTimes:
:return:
"""
try:
params = {'channel':self.channel,'num':self.num,'index':index}
ret = requests.post(self.spiderUrlApi, data=params)
if ret.status_code == 200:
jsonData = ret.json()
if jsonData.get('msg') == 'ok':
return jsonData.get('data')
else:
return []
except:
traceback.print_exc()
tryTimes = tryTimes + 1
if tryTimes < 3:
return self.getContions(type,tryTimes)
else:
return []
def randomSleep(self,min = 1,max = 3,msg = ''):
'''随机sleep'''
sleepTime = random.randint(min,max)
print(msg,'休眠'+str(sleepTime)+'s')
time.sleep(sleepTime)
if __name__ == '__main__':
spider = Spider('liepin',40001)
# folder = os.path.join(spider.successPath, datetime.today().strftime("%Y%m%d"))
# rs = spider.isSpiderToday(folder,'180996822','job')
# print(rs)
print(spider.getProxy())
项目分区导航：getProxy-setProxyRecord ⬅️ | 03-spider.py | ➡️ 项目导航
💬 评论