192 lines
8.3 KiB
Python
Executable File
192 lines
8.3 KiB
Python
Executable File
import requests
|
|
import time
|
|
from datetime import datetime
|
|
import random
|
|
from fake_useragent import UserAgent
|
|
import logging
|
|
import write_to_excel as wte
|
|
import utils as Utils
|
|
|
|
# === 获取筛选条件 ===
|
|
KEYWORD =Utils.KEYWORD #关键词筛选
|
|
START_DATE =Utils.START_DATE
|
|
END_DATE =Utils.END_DATE
|
|
|
|
#获取公告来源链接
|
|
SOURCE_URL = Utils.SOURCE_URL
|
|
|
|
# === 反扒配置 ===
|
|
|
|
#获取代理列表
|
|
PROXIES = Utils.PROXIES
|
|
USER_AGENTS = Utils.USER_AGENTS
|
|
|
|
MAX_RETRIES = 3 # 最大重连次数
|
|
DELAY_MIN, DELAY_MAX = 1, 3 # 随机延迟范围（秒）
|
|
|
|
# === 日志配置 ===
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
|
|
|
|
def get_random_user_agent():
|
|
"""随机选择 User-Agent"""
|
|
return random.choice(USER_AGENTS)
|
|
|
|
def get_random_proxy():
|
|
"""随机选择代理"""
|
|
if not PROXIES:
|
|
return None
|
|
proxy = random.choice(PROXIES)
|
|
return {"http": proxy, "https": proxy}
|
|
|
|
def check_sensitive_words(user_agent,payload,category_code,childrencode):
|
|
"""敏感词检查请求"""
|
|
# === 接口地址 ===
|
|
url = "https://zfcg.gxzf.gov.cn/portal/sensitiveWords/check"
|
|
headers = {
|
|
"User-Agent": user_agent,
|
|
"Content-Type": "application/json;charset=UTF-8",
|
|
"Origin": "https://zfcg.gxzf.gov.cn",
|
|
"Referer": f"https://zfcg.gxzf.gov.cn/site/category?parentId={category_code}&childrenCode={childrencode}",
|
|
"Cookie": "_zcy_log_client_uuid=71e283e0-23d2-11f0-844a-eb67dfa7ab64"
|
|
}
|
|
response = requests.post(url, json=payload, headers=headers)
|
|
# print("这是敏感词检查请求返回的响应预览",response.json())
|
|
return response.json()
|
|
|
|
def get_announcements(category_code, childrencode,page_no=1, session=None):
|
|
"""请求接口获取公告列表函数(带反扒机制)"""
|
|
if session is None:
|
|
session = requests.Session()
|
|
payload = {
|
|
# 接口的请求参数
|
|
"keyword": KEYWORD, #关键词筛选，此参数即为搜索框中的关键词输入值
|
|
"publishDateBegin": START_DATE, #最早日期筛选
|
|
"publishDateEnd": END_DATE, #最晚日期筛选
|
|
"pageNo": page_no, #页码数
|
|
"pageSize": 15, #页容量
|
|
"categoryCode": category_code, # 该参数指定当前查找的公告栏目
|
|
"_t": int(time.time() * 1000) # 动态时间戳
|
|
}
|
|
|
|
# 先执行敏感词检查
|
|
user_agent=get_random_user_agent()
|
|
check_response = check_sensitive_words(user_agent,payload, category_code, childrencode)
|
|
if not check_response.get("success", False):
|
|
logging.warning("敏感词检查失败")
|
|
return None
|
|
|
|
# 再执行公告数据请求
|
|
# === 接口地址 ===
|
|
api_url = "https://zfcg.gxzf.gov.cn/portal/category"
|
|
headers = {
|
|
"User-Agent": user_agent,
|
|
"Content-Type": "application/json;charset=UTF-8",
|
|
"Origin": "https://zfcg.gxzf.gov.cn",
|
|
"Referer": f"https://zfcg.gxzf.gov.cn/site/category?parentId={category_code}&childrenCode={childrencode}",
|
|
"Cookie": "_zcy_log_client_uuid=71e283e0-23d2-11f0-844a-eb67dfa7ab64" # 如果需要登录
|
|
}
|
|
#若添加了至少1条代理，则尝试使用代理
|
|
if PROXIES:
|
|
proxy = 'Default_value'
|
|
else:
|
|
proxy=None
|
|
for retry in range(MAX_RETRIES):
|
|
try:
|
|
# 尝试使用代理，失败则切换为无代理
|
|
if proxy :
|
|
proxy = get_random_proxy()
|
|
|
|
if proxy:
|
|
logging.info(f"使用代理: {proxy['http']}")
|
|
else:
|
|
logging.info("未使用代理")
|
|
|
|
response = session.post(
|
|
api_url,
|
|
json=payload,
|
|
headers=headers,
|
|
proxies=proxy,
|
|
timeout=10
|
|
)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get("success",True):
|
|
return data
|
|
else:
|
|
logging.warning(f"接口返回失败: {data.get('error', '未知错误')}")
|
|
else:
|
|
logging.warning(f"请求失败，状态码: {response.status_code}")
|
|
except requests.exceptions.ProxyError as pe:
|
|
# 代理错误时记录并跳过，继续无代理请求
|
|
logging.error(f"代理错误: {pe}. 切换为无代理模式")
|
|
proxy = None # 下次请求不使用代理
|
|
except Exception as e:
|
|
logging.error(f"请求异常: {e}")
|
|
|
|
# 重试前等待
|
|
delay = random.uniform(DELAY_MIN, DELAY_MAX)
|
|
logging.info(f"第 {retry + 1}/{MAX_RETRIES} 次重试，等待 {delay:.1f} 秒...")
|
|
time.sleep(delay)
|
|
|
|
return None # 所有重试失败
|
|
|
|
def parse_data(data,category_code,source_name):
|
|
"""解析公告数据函数"""
|
|
results = []
|
|
for item in data["result"]["data"]["data"]:
|
|
results.append({
|
|
"标题": item["title"],
|
|
"发布时间": datetime.fromtimestamp(int(item["publishDate"]) / 1000).strftime("%Y-%m-%d"),
|
|
"发布单位": item["purchaseName"],
|
|
"内容链接": f"https://zfcg.gxzf.gov.cn/site/detail?parentId={category_code}&articleId={item['articleId']}",
|
|
"来源栏目":source_name
|
|
})
|
|
return results
|
|
|
|
#爬虫主函数
|
|
def web_crawler():
|
|
session = requests.Session()
|
|
# print("这是本地会话存储：",session)
|
|
# 初始化引用值为信息公告,先爬取信息公告栏目的公告数据
|
|
childrencode = "ZcyAnnouncement"
|
|
# 爬取代码主体，外循环为来源栏目的循环，即遍历需要爬取的所有来源栏目
|
|
for key in SOURCE_URL:
|
|
page = 1
|
|
all_results = []
|
|
# print("这是正在爬取的栏目", SOURCE_URL[key][1])
|
|
# print("这是正在爬取的栏目的目录码", str(key))
|
|
# 判断正确的引用值
|
|
# 内循环主体即为在符合筛选条件的公告列表中遍历所有公告，爬取每个公告需要的指定数据。
|
|
while True:
|
|
# 这是分割线
|
|
# logging.info("爬虫日志分割线————————————————————————————————————————————————爬虫日志分割线")
|
|
# logging.info("爬虫日志分割线————————————————————————————————————————————————爬虫日志分割线")
|
|
# logging.info("爬虫日志分割线————————————————————————————————————————————————爬虫日志分割线")
|
|
# logging.info(f"正在爬取第 {page} 页...")
|
|
response = get_announcements(str(key), childrencode, page,session)
|
|
if response["result"]["data"]["empty"] or not response["result"]["data"]["data"]:
|
|
break
|
|
# logging.info(f"请求后共返回 {response["result"]["data"]["total"]} 条公告")
|
|
current_data = parse_data(response, SOURCE_URL[key][0], SOURCE_URL[key][1])
|
|
if not current_data:
|
|
break
|
|
# logging.info(f"解析后共爬取 {len(current_data)} 条公告")
|
|
all_results.extend(current_data)
|
|
page += 1
|
|
# 随机延迟
|
|
delay = random.uniform(DELAY_MIN, DELAY_MAX)
|
|
time.sleep(delay)
|
|
|
|
# print(f"从 {SOURCE_URL[key][1]} 共爬取 {len(all_results)} 条公告")
|
|
# 调用excel读写函数，将公告数据筛选后保存到Excel中
|
|
wte.write_to_excel(all_results,SOURCE_URL[key][1])
|
|
# 这是分割线
|
|
print("公告栏目分割线————————————————————————————————————————————————公告栏目分割线")
|
|
print("公告栏目分割线————————————————————————————————————————————————公告栏目分割线")
|
|
print("公告栏目分割线————————————————————————————————————————————————公告栏目分割线")
|
|
|
|
|
|
|
|
|
|
|