aws boto3 下载文件

起因:有下载 aws s3 需求,但只有web 登录账号,有 id 用户名 密码,没有 boto3 的 key ID

经过分析,发现网页版有个地址会返回临时 keyID,playwright 模拟登录,用 page.on 监测返回数据,获取 keyID 后再使用 boto3 抓取相关文件,比构造网页请求方便快捷

import os, json, urllib, base64
import time, re
from datetime import datetime
from playwright.sync_api import Playwright, sync_playwright, expect
from bs4 import BeautifulSoup
from functools import wrapsproxy = 'http://username:password@192.192.14.32:3128'
proxies = {'http': proxy,'https': proxy
}# 缓存目录
CACHE_DIR = (r'D:\code\aws_s3\cache')# 确保缓存目录存在
os.makedirs(CACHE_DIR, exist_ok=True)def timethis(func):'''Decorator that reports the execution time:param func::return:'''@wraps(func)def wrapper(*args, **kwargs):start = time.time()s1 = datetime.now()result = func(*args, **kwargs)end = time.time()s2 = datetime.now()func_name = func.__name__consume = end - startconsume2 = s2 - s1print(f'{func_name} consume time is ---> {consume}')print(f'{func_name} consume minutes is ---> {consume2}')return resultreturn wrapperdef handle_route(route):# 获取请求的 URLurl = route.request.urlresource_type = route.request.resource_typeurl = route.request.urlresource_type = route.request.resource_typeblock_list = [# 'telemetry', "browserCreds", 'module-utils.js',#             'svg', 'gif', 'image',#           'module', 'panoramaroute', 'log', 'tele', 'index', 'util', 'css']if any(x in url for x in block_list):# print(f"---: {url} (包含 'dist')")route.abort()  # 中止该请求return# print(f"处理请求: {url} ({resource_type})")# 生成对应的缓存文件名# 使用安全的 URL 名称file_name = url.replace("https://", "").replace("http://", "").replace("/", "_").replace(":", "_") + ".json"cache_file = os.path.join(CACHE_DIR, file_name)# 检查缓存文件是否存在if os.path.exists(cache_file):# print(f"从缓存加载: {url}")# 从缓存文件加载数据try:with open(cache_file, 'r') as f:cached_response = json.load(f)# 模拟返回缓存的响应route.fulfill(status=cached_response['status'],headers=cached_response['headers'],body=base64.b64decode(cached_response['body'])  # 解码 body)except:passelse:# 继续请求并缓存响应route.continue_()def log_response(response):url = response.urlresource_type = response.request.resource_type# 仅缓存 CSS、JS 和图片文件if resource_type in ['script', 'stylesheet', 'image']:file_name = url.replace("https://", "").replace("http://", "").replace("/", "_").replace(":", "_") + ".json"cache_file = os.path.join(CACHE_DIR, file_name)# 只有在成功状态时才缓存响应if response.status == 200:try:response_body = {'status': response.status,'headers': dict(response.headers),'body': base64.b64encode(response.body()).decode('utf-8')  # 确保调用 body() 方法获取字节}# 将响应写入缓存文件with open(cache_file, 'w') as f:json.dump(response_body, f)# print(f"缓存资源: {url}")except Exception as e:# print('cache error', url)pass
requests_info = {}def log_request(request):# 记录请求的开始时间requests_info[request.url] = {'start_time': time.time()  # 记录当前时间(开始时间)}def on_response(response, response_data):# 检查响应的 URLif 's3/tb/creds' in response.url and response.status == 200:# 解析响应数据并存储到 response_data 中boto3 = response.json()print('boto3', boto3)response_data.append(response.json())# 使用已保存的状态文件跳过登录状态直接访问系统
@timethis
def get_boto3_token():with sync_playwright() as playwright:browser = playwright.chromium.launch(headless=True,proxy={# 'server': 'http://username:password@192.192.13.193:3128','server': 'http://username:password@192.192.14.32:3128',# 'server': 'http://username:password@10.67.9.200:3128',# 'server': 'http://192.192.163.177:5003',"username": "username","password": "password"})# 创建浏览器上下文时加载状态文件context = browser.new_context()page = context.new_page()should_abort = False# 定义一个列表来存储响应数据response_data = []def handle_route(route):nonlocal should_abort# 检查当前页面是否包含 "open"if should_abort or response_data:print("检测到 'open',停止加载其他内容。")route.abort()  # 中止该请求else:route.continue_()  # 继续请求# 注册请求拦截事件# page.on("route", handle_route)# 直接访问登录后的URLurl = 'https://us-west-2.console.aws.amazon.com/s3/buckets/bs?prefix=RESPONSE/'# 注册请求和响应事件page.on("response", log_response)# page.on("route", handle_route)page.route("*", handle_route)page.goto(url, timeout=30000 * 3)# 屏蔽这一段就正常了# if page.locator("input[id=\"root_user_radio_button\"]"):#     print('find')#     page.locator("input[id=\"iam_user_radio_button\"]").click()#     page.locator("input[id=\"resolving_input\"]").fill("1111111")#     page.locator("button[id=\"next_button\"]").click()if page.locator("input[id=\"account\"]"):print('find')page.locator("input[id=\"account\"]").click()page.locator("input[id=\"account\"]").fill("1111111")# page.locator("button[id=\"next_button\"]").click()print('input username')while True:try:page.locator("input[name=\"username\"]").fill("username")page.locator("input[name=\"password\"]").fill("password")page.locator("#signin_button").click()print('break-->')breakexcept:print(datetime.now(), 'error-->')time.sleep(2)print('wait 6 senconds')time.sleep(2)cookies = page.context.cookies()print('cookie', cookies)url = 'https://us-west-2.console.aws.amazon.com/s3/buckets/bs-tai?region=us-west-2&bucketType=general&prefix=RESPONSE/2023/&showversions=false'# 注册请求和响应事件# 注册响应事件处理函数page.on("response", lambda response: on_response(response, response_data))page.goto(url, timeout=30000 * 3)print('page on response')while True:try:cookies = page.context.cookies()breakexcept:time.sleep(2)print('sleep 2 seconds')soup = BeautifulSoup(page.content(), 'lxml')meta_tag = soup.find('meta', {'name': 'tb-data'})# 提取 content 属性的值tb_data = meta_tag.get('content')# 将 JSON 字符串转换为 Python 字典tb_data_dict = json.loads(tb_data)# 提取 CSRF 令牌xsrf_token = tb_data_dict['csrfToken']print('xsrf token', xsrf_token)print('response_data',response_data)# if not response_data:#     get_boto3_token()# else:#     print('return boto3 token')# page.close()# browser.close()# playwright.stop()return response_data[0]if __name__ == '__main__':get_boto3_token()pass
boto3_token = get_boto3_token()info = boto3_tokenprint(arrow.now())print('boto3_token-->', type(boto3_token), boto3_token)id = info.get("accessKeyId")key = info.get("secretAccessKey")aws_session_token = info.get("sessionToken")session = Session(aws_access_key_id=id, aws_secret_access_key=key, aws_session_token=aws_session_token)# session = Session(aws_access_key_id=id, aws_secret_access_key=key,aws_session_token=aws_session_token)# 获取s3连接的session##bucket = 'bs-tai'client_s3 = session.client('s3', config=Config(proxies=proxies))s3 = session.resource('s3', config=Config(proxies=proxies)).Bucket('bs-tai')def get_prefix_for_months(months_shift=0):arrow_month = arrow.now().shift(months=months_shift)year = arrow_month.format('YYYY')month = arrow_month.format('MM')return f'conn/RESPONSE/{year}/{month}/'# 获取上一个月和当前月的前缀prefix_last_month = get_prefix_for_months(months_shift=-1)prefix_this_month = get_prefix_for_months(months_shift=0)# 组合前缀到列表prefix_list = [prefix_last_month, prefix_this_month]for prefix in prefix_list:for obj in s3.objects.filter(Prefix=prefix):# print(obj.key)if obj.key.endswith('.csv'):file_path = obj.key# 使用字符串分割来提取年月日parts = file_path.split('/')year = parts[2]  # 第四部分是年份month = parts[3]  # 第五部分是月份day = parts[4]  # 第六部分是日期# print(year, month, day)key = obj.keylocal_filename = key.split('/')[-1]local_file_path = os.path.join(public_share_path, f'{year}{month}{day}', local_filename)if not os.path.exists(local_file_path):local_file_dir = os.path.dirname(local_file_path)os.makedirs(local_file_dir, exist_ok=True)client_s3.download_file(bucket, key, local_file_path)print(f'Downloaded {local_file_path}')read_csv(local_file_path, day=f'{year}{month}{day}')export_result_source(day=f'{year}{month}{day}')

参考
https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html

https://cuiqingcai.com/36045.html

https://www.cnblogs.com/neozheng/p/13563841.html

https://stackoverflow.com/questions/35803027/retrieving-subfolders-names-in-s3-bucket-from-b-boto3

https://stackoverflow.com/questions/35803027/retrieving-subfolders-names-in-s3-bucket-from-b-boto3

https://stackoverflow.com/questions/29378763/how-to-save-s3-object-to-a-file-using-boto3

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/884441.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

# Ubuntu 达人九步养成记(1)

Ubuntu 达人九步养成记(1) 目录: 一、ubuntu基本安装 二、设置语言环境 三、设置服务器镜像源 四、在启动栏添加终端图标 五、使用apt更新和升级系统软件 六、使用apt安装软件 七、使用apt删除软件以及apt-get 八、deb格式及谷歌浏览…

QT——TCP网络调试助手

目录 一.项目展示 ​编辑 二.开发流程 三.QTcpServer、QTcpSocket、QUdpSocket类的学习 1.QTcpServer服务端 2.QTcpSocket客户端 3.Udp通信 四.网络调试助手 1.首先我们实现当用户选择不同协议类型时不同的UI组件如何切换 2.实现打开/关闭按键图片的切换 方式一&…

导航栏渐变色iOS

- (void)viewDidLoad {[super viewDidLoad];// 设置导航栏属性self.navigationBar.translucent NO;[self.navigationBar setTitleTextAttributes:{NSForegroundColorAttributeName : [UIColor whiteColor], NSFontAttributeName:[UIFont boldSystemFontOfSize:28]}];// 修复iO…

《Web性能权威指南》-浏览器API与协议-读书笔记

本文是《Web性能权威指南》第四部分——浏览器API与协议的读书笔记。 第一部分——网络技术概览,请参考网络技术概览; 第二部分——无线网络性能,请参考无线网络性能; 第三部分——HTTP,请参考HTTP。 浏览器网络概述 …

使用TypeORM进行数据库操作

💓 博客主页:瑕疵的CSDN主页 📝 Gitee主页:瑕疵的gitee主页 ⏩ 文章专栏:《热点资讯》 使用TypeORM进行数据库操作 引言 TypeORM 简介 安装 TypeORM 配置 TypeORM 定义实体 连接数据库 运行项目 高级功能 事务管理 关…

ESP-HaloPanel:用 ESP32-C2 打造超低成本智能家居面板

项目简介 在生活品质日益提升的今天,智能家居系统已经走进了千家万户,并逐渐成为现代生活的一部份。与此同时,一款设计精致、体积轻盈、操作简便的全屋智能家居控制面板,已经成为众多家庭的新宠。这种高效、直观的智能化的解决方…

Hadoop生态圈框架部署(四)- Hadoop完全分布式部署

文章目录 前言一、Hadoop完全分布式部署(手动部署)1. 下载hadoop2. 上传安装包2. 解压hadoop安装包3. 配置hadoop配置文件3.1 虚拟机hadoop1修改hadoop配置文件3.1.1 修改 hadoop-env.sh 配置文件3.3.2 修改 core-site.xml 配置文件3.3.3 修改 hdfs-site…

数据建模圣经|数据模型资源手册卷3,数据建模最佳实践

简介 本书采用了类设计模式的方式对数据模型进行高度抽象总结,展现了常见的数据模型构建模型等模型的作用、层次、分类、地位、沟通方式,和业务规则。使用一个强大的数据模型模式的数据建模,评估特定与广义模型的优缺点,有助于你改…

【力扣】Go语言回溯算法详细实现与方法论提炼

文章目录 一、引言二、回溯算法的核心概念三、组合问题1. LeetCode 77. 组合2. LeetCode 216. 组合总和III3. LeetCode 17. 电话号码的字母组合4. LeetCode 39. 组合总和5. LeetCode 40. 组合总和 II小结 四、分割问题6. LeetCode 131. 分割回文串7. LeetCode 93. 复原IP地址小…

#渗透测试#SRC漏洞挖掘# 信息收集-Shodan进阶之Mongodb未授权访问

免责声明 本教程仅为合法的教学目的而准备,严禁用于任何形式的违法犯罪活动及其他商业行为,在使用本教程前,您应确保该行为符合当地的法律法规,继续阅读即表示您需自行承担所有操作的后果,如有异议,请立即停…

Golang--流程控制

1、分支结构 1.1 if分支 单分支 语法:if 条件表达式 { 逻辑代码 } 当条件表达式为true时,就会执行代码块的代码。条件表达式左右的()可以不写,也建议不写 if和表达式中间,一定要有空格在Golang中,{}是必须有的,就算你…

【补补漏洞吧 | 02】等保测评ZooKeeperElasticsearch未授权访问漏洞补漏方法

一、项目背景 客户新系统上线,因为行业网络安全要求,需要做等保测评, 通过第三方漏扫工具扫描系统,漏扫报告显示ZooKeeper和 Elasticsearch 服务各拥有一个漏洞,具体结果如下: 1、ZooKeeper 未授权访问【…

Serverless + AI 让应用开发更简单

本文整理自 2024 云栖大会,阿里云智能高级技术专家,史明伟演讲议题《Serverless AI 让应用开发更简单》 随着云计算和人工智能(AI)技术的飞速发展,企业对于高效、灵活且成本效益高的解决方案的需求日益增长。本文旨在…

从0开始学PHP面向对象内容之(类,对象,构造/析构函数)

上期我们讲了面向对象的一些基本信息&#xff0c;这期让我们详细的了解一下 一、面向对象—类 1、PHP类的定义语法&#xff1a; <?php class className {var $var1;var $var2 "constant string";function classfunc ($arg1, $arg2) {[..]}[..] } ?>2、解…

(八)JavaWeb后端开发——Tomcat

目录 1.Web服务器概念 2.tomcat 1.Web服务器概念 服务器&#xff1a;安装了服务器软件的计算机服务器软件&#xff1a;接收用户的请求&#xff0c;处理请求&#xff0c;做出响应web服务器软件&#xff1a;在web服务器软件中&#xff0c;可以部署web项目&#xff0c;让用户通…

【Linux系列】Linux 和 Unix 系统中的`set`命令与错误处理

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…

Nuxt.js 应用中的 nitro:config 事件钩子详解

title: Nuxt.js 应用中的 nitro:config 事件钩子详解 date: 2024/11/2 updated: 2024/11/2 author: cmdragon excerpt: nitro:config 是 Nuxt 3 中的一个生命周期钩子,允许开发者在初始化 Nitro 之前自定义 Nitro 的配置。Nitro 是 Nuxt 3 的服务器引擎,负责处理请求、渲…

[论文阅读]LOGAN: Membership Inference Attacks Against Generative Models

LOGAN: Membership Inference Attacks Against Generative Models https://arxiv.org/abs/1705.07663v4 Proceedings on Privacy Enhancing Technologies &#xff08;PoPETs&#xff09;&#xff0c;第 2019 卷&#xff0c;第 1 期。 这篇文章是17年的一篇文章&#xff0c;…

使用Vite构建现代化前端应用

&#x1f493; 博客主页&#xff1a;瑕疵的CSDN主页 &#x1f4dd; Gitee主页&#xff1a;瑕疵的gitee主页 ⏩ 文章专栏&#xff1a;《热点资讯》 使用Vite构建现代化前端应用 引言 Vite 简介 安装 Vite 创建项目 启动开发服务器 项目结构 配置 Vite 开发模式 生产构建 使用插…

Node.js:模块 包

Node.js&#xff1a;模块 & 包 模块module对象 包npm安装包配置文件镜像源 分类 模块 模块化是指解决一个复杂问题时&#xff0c;自顶向下逐层把系统划分成若干模块的过程。对于整个系统来说&#xff0c;模块是可组合、分解和更换的单元。 简单来说&#xff0c;就是把一个…