开启辅助访问 切换到宽版

精易论坛

 找回密码
 注册

QQ登录

只需一步,快速开始

用微信号发送消息登录论坛

新人指南 邀请好友注册 - 我关注人的新帖 教你赚取精币 - 每日签到


求职/招聘- 论坛接单- 开发者大厅

论坛版规 总版规 - 建议/投诉 - 应聘版主 - 精华帖总集 积分说明 - 禁言标准 - 有奖举报

查看: 1045|回复: 44
收起左侧

[分享] 爬取资源网的附件

[复制链接]
结帖率:60% (6/10)
发表于 2025-5-13 08:27:43 | 显示全部楼层 |阅读模式   浙江省杭州市
精易的资源网,超过数量需要登陆扣分。所有找了一个免费的。
目标:e5a5x的资源网。
运行环境:python
QQ截图20250513082208.png
-----
QQ截图20250513082535.png
不做分类处理,只下载整站附件,异步下载
----

[Python] 纯文本查看 复制代码
import asyncio
import aiohttp
import os
import re
import logging
from urllib.parse import urljoin
from bs4 import BeautifulSoup

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('downloader.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)

# 全局配置
BASE_URL = "https://www.e5a5x.com"
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"
}
DOWNLOAD_DIR = "loads"
os.makedirs(DOWNLOAD_DIR, exist_ok=True)

# 分类配置(已根据您的需求设置完整分页)
CATEGORIES = {
"web3": {
"base_url": f"{BASE_URL}/html/web3/",
"sub_categories": ["B1", "B2", "B3", "B4"],
"list_pattern": "list_45_{page}.html",
"max_pages": 182 # 共182页
},
"web2": {
"base_url": f"{BASE_URL}/html/web2/",
"sub_categories": ["B5", "B6", "B7", "B8"],
"list_pattern": "list_46_{page}.html",
"max_pages": 176 # 共176页
},
"web4": {
"base_url": f"{BASE_URL}/html/web4/",
"sub_categories": ["A1", "A2", "A3", "A4"],
"list_pattern": "list_44_{page}.html",
"max_pages": 225 # 共225页
},
"lib": {
"base_url": f"{BASE_URL}/html/lib/",
"sub_categories": [],
"list_pattern": "list_42_{page}.html",
"max_pages": 20 # 共20页
},
"soft": {
"base_url": f"{BASE_URL}/html/soft/",
"sub_categories": [],
"list_pattern": None,
"max_pages": 0
}
}

async def fetch(session, url, retries=3):
for i in range(retries):
try:
async with session.get(url, headers=HEADERS) as response:
response.raise_for_status()
return await response.text()
except Exception as e:
if i == retries - 1:
logger.error(f"请求失败 {url}: {str(e)}")
return None
await asyncio.sleep(2 ** i)

async def download_file(session, url, title):
try:
# 生成安全的文件名
safe_title = re.sub(r'[\\/*?:"<>|]', '_', title)[:100]
file_ext = os.path.splitext(url)[1] or '.zip'
filename = f"{safe_title}{file_ext}"
save_path = os.path.join(DOWNLOAD_DIR, filename)

# 检查文件是否已存在
if os.path.exists(save_path):
logger.info(f"文件已存在,跳过下载: {filename}")
return

logger.info(f"开始下载: {filename}")
async with session.get(url, headers=HEADERS) as response:
response.raise_for_status()
total_size = int(response.headers.get('content-length', 0))
downloaded = 0

with open(save_path, 'wb') as f:
async for chunk in response.content.iter_chunked(1024 * 1024): # 1MB chunks
f.write(chunk)
downloaded += len(chunk)
if total_size > 0:
logger.info(f"进度: {downloaded}/{total_size} bytes ({downloaded/total_size:.1%})")

logger.info(f"下载完成: {filename} ({os.path.getsize(save_path)/1024:.2f} KB)")
except Exception as e:
logger.error(f"下载失败 {url}: {str(e)}")

def parse_download_url(html):
soup = BeautifulSoup(html, 'html.parser')

# 方法1:检查所有a标签的href属性
for a in soup.find_all('a', href=True):
href = a['href']
if href.lower().endswith(('.zip', '.rar', '.7z', '.exe', '.apk')):
return urljoin(BASE_URL, href)

# 方法2:检查JavaScript中的下载链接
patterns = [
r"(?:window\.location\.href|window\.open|var\s+url)\s*=\s*['\"](.+?\.(?:zip|rar|7z|exe))['\"]",
r"downloadG1\(\)\s*{[^}]+['\"](.+?\.(?:zip|rar|7z|exe))['\"]",
r"Click_down\(\)\s*{[^}]+['\"](.+?\.(?:zip|rar|7z|exe))['\"]"
]
for pattern in patterns:
matches = re.finditer(pattern, html, re.DOTALL)
for match in matches:
url = match.group(1).strip()
return urljoin(BASE_URL, url)

return None

async def process_detail_page(session, url):
html = await fetch(session, url)
if not html:
return

soup = BeautifulSoup(html, 'html.parser')
title = soup.title.text.strip() if soup.title else "无标题"
download_url = parse_download_url(html)

if download_url:
await download_file(session, download_url, title)
else:
logger.warning(f"未找到下载地址: {url}")

async def crawl_list_page(session, list_url):
try:
logger.info(f"正在爬取列表页: {list_url}")
html = await fetch(session, list_url)
if not html:
return

soup = BeautifulSoup(html, 'html.parser')
detail_urls = []

# 查找所有详情页链接
for a in soup.select('a[href^="/html/"]'):
href = a['href']
if not href.startswith(('/html/article_', '/html/feedback_')):
full_url = urljoin(BASE_URL, href)
detail_urls.append(full_url)

logger.info(f"找到 {len(detail_urls)} 个详情页链接")

# 限制并发量
semaphore = asyncio.Semaphore(5)
async def limited_task(url):
async with semaphore:
await process_detail_page(session, url)
await asyncio.sleep(1) # 请求间隔

await asyncio.gather(*[limited_task(url) for url in detail_urls])
except Exception as e:
logger.error(f"处理列表页出错 {list_url}: {str(e)}")

async def crawl_category(session, category_name, category_config):
try:
logger.info(f"开始爬取分类: {category_name} (共{category_config['max_pages']}页)")

# 处理主分类页面
await crawl_list_page(session, category_config["base_url"])

# 处理子分类
for sub in category_config["sub_categories"]:
sub_url = urljoin(category_config["base_url"], sub + "/")
await crawl_list_page(session, sub_url)

# 处理分页(如果有分页模式)
if category_config["list_pattern"] and category_config["max_pages"] > 0:
for page in range(1, category_config["max_pages"] + 1):
list_url = urljoin(
category_config["base_url"],
category_config["list_pattern"].format(page=page)
)
try:
await crawl_list_page(session, list_url)
except Exception as e:
logger.error(f"爬取分页失败 {list_url}: {str(e)}")
continue

# 每10页显示一次进度
if page % 10 == 0:
logger.info(f"{category_name} 进度: {page}/{category_config['max_pages']}页")

logger.info(f"完成爬取分类: {category_name}")
except Exception as e:
logger.error(f"爬取分类 {category_name} 出错: {str(e)}")

async def main():
connector = aiohttp.TCPConnector(limit=10)
timeout = aiohttp.ClientTimeout(total=3600) # 1小时超时

async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
tasks = []
for category_name, config in CATEGORIES.items():
task = asyncio.create_task(crawl_category(session, category_name, config))
tasks.append(task)

await asyncio.gather(*tasks)

if __name__ == "__main__":
logger.info("启动爬虫...")
logger.info(f"下载文件将保存到: {os.path.abspath(DOWNLOAD_DIR)}")

try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("用户中断,停止爬取")
except Exception as e:
logger.error(f"爬虫运行出错: {str(e)}")
finally:
logger.info("爬取任务结束")


结帖率:60% (6/10)
 楼主| 发表于 2025-5-13 08:28:47 | 显示全部楼层   浙江省杭州市
手动置顶:
安装依赖环境:pip install aiohttp beautifulsoup4
回复 支持 反对

使用道具 举报

签到天数: 3 天

发表于 2025-5-19 14:21:36 高大上手机用户 | 显示全部楼层   重庆市重庆市
感谢分享!来看看
回复 支持 反对

使用道具 举报

结帖率:96% (369/383)
发表于 2025-5-18 22:15:37 | 显示全部楼层   陕西省西安市
爬取资源网的附件
回复 支持 反对

使用道具 举报

签到天数: 4 天

发表于 2025-5-17 17:04:29 | 显示全部楼层   黑龙江省哈尔滨市
好厉害作者牛牛牛
回复 支持 反对

使用道具 举报

结帖率:100% (21/21)

签到天数: 1 天

发表于 2025-5-16 19:43:13 | 显示全部楼层   安徽省滁州市
卧槽牛逼
回复 支持 反对

使用道具 举报

结帖率:70% (7/10)

签到天数: 8 天

发表于 2025-5-15 14:41:34 | 显示全部楼层   河南省南阳市
顶一个!!!!!!!!
回复 支持 反对

使用道具 举报

结帖率:50% (7/14)

签到天数: 6 天

发表于 2025-5-15 11:03:52 | 显示全部楼层   安徽省淮北市
感谢我分享啊
回复 支持 反对

使用道具 举报

结帖率:55% (6/11)

签到天数: 1 天

发表于 2025-5-15 00:27:19 | 显示全部楼层   广东省中山市
感谢分享,下载了

回复 支持 反对

使用道具 举报

签到天数: 7 天

发表于 2025-5-13 19:52:18 | 显示全部楼层   贵州省黔东南苗族侗族自治州
下载来试试
回复 支持 反对

使用道具 举报

结帖率:100% (74/74)

签到天数: 2 天

发表于 2025-5-13 18:50:55 | 显示全部楼层   重庆市重庆市
这..........................
回复 支持 反对

使用道具 举报

结帖率:100% (2/2)

签到天数: 10 天

发表于 2025-5-13 18:12:17 | 显示全部楼层   广东省惠州市

挂123分享下,谢谢
回复 支持 反对

使用道具 举报

您需要登录后才可以回帖 登录 | 注册

本版积分规则 致发广告者

发布主题 收藏帖子 返回列表

sitemap| 易语言源码| 易语言教程| 易语言论坛| 易语言模块| 手机版| 广告投放| 精易论坛
拒绝任何人以任何形式在本论坛发表与中华人民共和国法律相抵触的言论,本站内容均为会员发表,并不代表精易立场!
论坛帖子内容仅用于技术交流学习和研究的目的,严禁用于非法目的,否则造成一切后果自负!如帖子内容侵害到你的权益,请联系我们!
防范网络诈骗,远离网络犯罪 违法和不良信息举报QQ: 793400750,邮箱:wp@125.la
网站简介:精易论坛成立于2009年,是一个程序设计学习交流技术论坛,隶属于揭阳市揭东区精易科技有限公司所有。
Powered by Discuz! X3.4 揭阳市揭东区精易科技有限公司 ( 粤ICP备2025452707号) 粤公网安备 44522102000125 增值电信业务经营许可证 粤B2-20192173

快速回复 返回顶部 返回列表