--- title: "03-spider.py" created: 2026-04-01 tags: - 项目 aliases: - spider.py --- # spider.py ```python # -*- 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()) ``` --- **项目分区导航**：[[02-getProxy-setProxyRecord|getProxy-setProxyRecord]] ⬅️ | 03-spider.py | ➡️ [[00-项目导航|项目导航]]