# -*- coding: utf-8 -*- """ Created on 2025-05-08 --------- @summary: 云南省政府采购网 采购信息 --------- @author: lzz """ import json import execjs import base64 import feapder from items.spider_item import MgpListItem from get_yn_cookies import get_ck from collections import namedtuple from untils.get_imgcode import get_code_det import time from untils.tools import get_proxy from feapder.utils.tools import get_today_of_day def pass_code(session): js_script = ''' uuid = function(){ var s = []; var hexDigits = "0123456789abcdef"; for (var i = 0; i < 36; i++) { s[i] = hexDigits.substr(Math.floor(Math.random() * 0x10), 1); } s[14] = "4"; s[19] = hexDigits.substr((s[19] & 0x3) | 0x8, 1); s[8] = s[13] = s[18] = s[23] = "-"; return 'point' + '-' + s.join(""); } ''' ctx = execjs.compile(js_script) client_uid = ctx.call('uuid') url = "http://www.yngp.com/api/captcha/captcha.get.svc" data = { "captchaType": "clickWord", "clientUid": client_uid, "ts": round(time.time() * 1000) } data = json.dumps(data) headers = { "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Content-Type": "application/json;charset=UTF-8", "Origin": "http://www.yngp.com", "Pragma": "no-cache", "Referer": "http://www.yngp.com/page/procurement/procurementList.html", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" } res = session.post(url, headers=headers, data=data,timeout=60) image_content = res.json().get("data").get("repData").get("originalImageBase64") code_list = res.json().get("data").get("repData").get("wordList") keyWord = res.json().get("data").get("repData").get("secretKey") token = res.json().get("data").get("repData").get("token") image_code = get_code_det(base64.b64decode(image_content)) image_code = image_code.get("r").get("code") word = [] for i in code_list: if image_code.get(i): word.append(image_code.get(i)) else: word.append(image_code.get("")) try: word = [{"x": (i[0] + i[2]) // 2, "y": (i[1] + i[3]) // 2} for i in word] except: word = [{"x": (i[0] + i[2]) // 2, "y": (i[1] + i[3]) // 2} for i in list(image_code.values())[:3]] js_func = ''' var CryptoJS = require("crypto-js") function aesEncrypt(word,keyWord){ word = JSON.stringify(word) var key = CryptoJS.enc.Utf8.parse(keyWord); var srcs = CryptoJS.enc.Utf8.parse(word); var encrypted = CryptoJS.AES.encrypt(srcs, key, {mode:CryptoJS.mode.ECB,padding: CryptoJS.pad.Pkcs7}); return encrypted.toString(); } ''' ctx = execjs.compile(js_func) pointJson = ctx.call("aesEncrypt", word, keyWord) data = { "captchaType": "clickWord", "pointJson": pointJson, "token": token, "clientUid": client_uid, "ts": round(time.time() * 1000) } # with open('ynszfcgw.png', 'wb+') as f: # f.write(base64.b64decode(image_content)) url = "http://www.yngp.com/api/captcha/captcha.check.svc" data = json.dumps(data) session.post(url, headers=headers, data=data, timeout=60) time.sleep(5) js_func = ''' var CryptoJS = require("crypto-js") function aesEncrypt(token,word,keyWord){ word = token + "---" +JSON.stringify(word) var key = CryptoJS.enc.Utf8.parse(keyWord); var srcs = CryptoJS.enc.Utf8.parse(word); var encrypted = CryptoJS.AES.encrypt(srcs, key, {mode:CryptoJS.mode.ECB,padding: CryptoJS.pad.Pkcs7}); return encrypted.toString(); } ''' ctx = execjs.compile(js_func) captchaVerification = ctx.call("aesEncrypt", token, word, keyWord) return captchaVerification class Ynszfcgw_New(feapder.BiddingListSpider): def start_callback(self): Menu = namedtuple('Menu', ['channel', 'code', 'noticeType', 'crawl_page']) self.site = "云南省政府采购网" self.menus = [ Menu('ppp合作伙伴采购信息', 'yn_ynszfcgw_new_ppphzgbcgxx', '4', 1), ] self.headers = { "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Cache-Control": "no-cache", "Connection": "keep-alive", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "Origin": "http://www.yngp.com", "Pragma": "no-cache", "Referer": "http://www.yngp.com/page/procurement/procurementList.html", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36", "X-Requested-With": "XMLHttpRequest" } self.retry = 0 self.proxy = get_proxy() self.cookies = {} self.vcode = "" def start_requests(self): for menu in self.menus: start_url = "http://www.yngp.com/api/procurement/Procurement.gghtMoreList.svc" yield feapder.Request(url=start_url,item=menu._asdict(),proxies=False, page=1) def download_midware(self, request): for _ in range(5): if not self.cookies: self.cookies = get_ck(proxies=self.proxy) if self.cookies: break else: self.proxy = get_proxy() else: break menu = request.item page = request.page if page != 1: url = f"http://www.yngp.com/api/procurement/Procurement.gghtMoreList.svc?captchaCheckFlag={self.vcode}&p={page}" else: url = f"http://www.yngp.com/api/procurement/Procurement.gghtMoreList.svc?captchaCheckFlag=0&p=1" if menu.get('code') == "yn_ynszfcgw_new_zfcghtgg": tm = get_today_of_day() else: tm = "" data = { "current": f"{page}", "rowCount": "10", "searchPhrase": "", "query_bulletintitle": "", "query_startTime": tm, "query_endTime": tm, "query_type": menu.get('noticeType'), "query_code": "", "query_gglxdm": "", "query_purchaser": "", "query_projectid": "" } request.url = url request.data = data request.cookies = self.cookies request.proxies = self.proxy request.headers = self.headers def exception_request(self, request, response): self.proxy = get_proxy() self.cookies = {} yield request def parse(self, request, response): if self.retry > 5: return if "系统异常，请稍后再试" in response.text or "磐云" in response.text: self.retry += 1 self.cookies = {} self.proxy = get_proxy() yield request else: menu = request.item response=response.json info_list = response.get('data').get('rows') for info in info_list: title = info.get('bulletintitle').split("：")[-1].strip() href_id = info.get('bulletin_id') bulletinclassname = info.get('bulletinclassname') href = f'http://www.yngp.com/showBulletinInfo.html?bulletin_id={href_id}' if bulletinclassname == "采购合同公告": if "单位采购合同公告" in title: continue href = f'http://www.yngp.com/ggmxinfo.html?bulletinid={href_id}' elif "公共服务项目验收结果公告" in bulletinclassname: href = f'http://www.yngp.com/showAcceptanceResultsNoticeInfo.html?bulletinid={href_id}' elif bulletinclassname in "单一来源审核前公示": href = f'http://www.yngp.com/dylyggInfo.html?type=3&bulletin_id={href_id}' elif "成交公告" in bulletinclassname: href = f'http://www.yngp.com/showZCYBulletinInfo.html?bulletin_id={href_id}' create_time = info.get("finishday") area = "云南" # 省份 city = "" if info.get('districtname') !="省级": city = info.get('districtname') # 城市 list_item = MgpListItem() # 存储数据的管道 list_item.href = href # 标书链接 list_item.channel = menu.get("channel") # 最上方定义的抓取栏目 （编辑器定的） list_item.spidercode = menu.get("code") # 最上方定义的爬虫code（编辑器定的） list_item.title = title # 标题 list_item.publishtime = create_time # 标书发布时间 list_item.site = self.site list_item.area = area # 城市默认:全国 list_item.city = city # 城市 默认为空 list_item.unique_key = ('href','title',create_time) list_item.parse = "self.detail_get" list_item.render_time = 1 list_item.proxies = False list_item.parse_url = href yield list_item request = self.infinite_pages(request, response) yield request if __name__ == "__main__": Ynszfcgw_New(redis_key="lzz:Ynszfcgw_Jdgl").start()