physical3/physical-boot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
ls
2024-10-23 16:56:01 +08:00
parent 0f65911496
commit 5e0bbbd757
3 changed files with 277 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

489
scripts/esa.py
View File

@@ -1,9 +1,14 @@
import os
import requests
import csv
import mysql.connector
from bs4 import BeautifulSoup
import hashlib
import os
import re
from datetime import datetime
import mysql.connector
import requests
from bs4 import BeautifulSoup
from minio import Minio
from minio.error import S3Error
# 设置下载目录
download_dir = 'downloaded_files'
@@ -11,235 +16,289 @@ os.makedirs(download_dir, exist_ok=True)
# MySQL 连接配置
db_config = {
'host': 'localhost',
'user': 'your_username',
'password': 'your_password',
'database': 'your_database'
'host': '192.168.50.100',
'port': 23306,
'user': 'root',
'password': '123456',
'database': 'physical-boot'
}
connection = mysql.connector.connect(**db_config)
cursor = connection.cursor()
# minio 配置
minio_client = Minio(
"192.168.50.100:29000", # MinIO服务器地址或IP
access_key="root", # 替换为你的Access Key
secret_key="12345678", # 替换为你的Secret Key
secure=False # 如果使用的是http则为False
)
bucket_name = 'physical'
def get_md5(input_string):
# 创建MD5对象
md5_obj = hashlib.md5()
# 更新对象，注意字符串需要编码为字节
md5_obj.update(input_string.encode('utf-8'))
# 返回MD5值的十六进制字符串
return md5_obj.hexdigest()
def save_to_mysql(data):
"""保存数据到 MySQL 数据库"""
connection = mysql.connector.connect(**db_config)
cursor = connection.cursor()
try:
"""保存数据到 MySQL 数据库"""
insert_query = """
INSERT INTO your_table (
sequence_number, test_object_type, test_start_date, test_end_date, test_object_name,
test_object_model, test_object_quantity, test_nature, test_purpose, device_name,
data_provider, test_commissioner, failure_criteria, failure_quantity, test_result_description,
outcome, source_project_name, source_project_type, classification, component_name,
component_model, component_batch_number, manufacturer, is_domestic, component_maturity,
wafer_material, wafer_batch_number, package_material, package_technology, is_flip_chip,
manufacturing_process, process_feature_size, process_platform, process_code, process_version,
quality_grade, reinforcement_measures, working_principle, supply_capacity, application_experience,
specification_manual, device_image, electronic_system_classification, electronic_system_name,
electronic_system_model, manufacturer, electronic_system_function, electronic_system_reinforcement_measures,
electronic_system_image, material_name, material_model, material_components, material_purpose,
material_manufacturer, material_physical_structure, material_usage_experience, irradiation_test_outline,
outline_expert_category, standard_specifications, test_steps_description, irradiation_process_power,
dc_bias_condition_description, ac_bias_condition_description, clock_frequency, test_pattern,
other_bias_conditions, irradiation_bias_schematic, test_method, test_schematic,
test_instrument_name, test_instrument_model, test_instrument_manufacturer, test_instrument_certificate,
test_software_name, test_software_developer, test_software_version, test_site_photo,
tester_name, tester_organization, tester_phone, device_operator, third_party_person, third_party_organization,
third_party_phone, other_notes, lead_aluminum_shielding, dose_rate, total_dose, dose_equivalent_material,
test_object_number, test_parameter_name, test_parameter_unit, test_parameter_result,
is_accelerated_test_data, is_annealing_data, annealing_temperature, annealing_time, raw_data,
data_processing_method, other_notes_additional
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
insert_query = """INSERT INTO `import_record` (`id`, `create_by`, `create_time`, `update_by`, `update_time`, `sys_org_code`, `device_type`, `device_name`, `device_mode`, `device_function`, `device_batch`, `manufacturer`, `experiment_date`, `data_source`, `experiment_user`, `total_count`, `file_list`)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);"""
cursor.execute(insert_query, data)
connection.commit()
cursor.close()
connection.close()
cursor.execute(insert_query, data)
connection.commit()
except Exception as e:
# 处理 ZeroDivisionError 的代码
print(e)
# 获取网页内容
url = 'https://esarad.esa.int/'
def save_to_db_file(data):
try:
"""保存数据到 MySQL 数据库"""
insert_query = """INSERT INTO `oss_file` (`id`,`file_name`,`url`,`create_by`,`create_time` )
VALUES (%s, %s, %s, %s, %s);"""
cursor.execute(insert_query, data)
connection.commit()
except Exception as e:
# 处理 ZeroDivisionError 的代码
print(e)
def upload_to_minio(folder_path):
if not minio_client.bucket_exists(bucket_name):
minio_client.make_bucket(bucket_name)
folder_name = os.path.basename(folder_path)
# 遍历文件夹中的所有文件，并上传
file_ids = []
for file_name in os.listdir(folder_path):
file_path = os.path.join(folder_path, file_name)
# 检查是否是文件，忽略非文件类型
if os.path.isfile(file_path):
object_name = f"ESA/{folder_name}/{file_name}"
try:
# 上传文件到 MinIO
minio_client.fput_object(bucket_name, object_name, file_path)
print(f"已上传: {file_path} -> {bucket_name}/{object_name}")
file_id = get_md5(object_name)
file_ids.append(file_id)
db_file = [file_id, file_name,
'http://58.215.212.230:8005/oss/' + bucket_name + '/' + object_name, 'admin', datetime.now()]
save_to_db_file(db_file)
except S3Error as err:
print(f"上传 {file_name} 时出错: {err}")
return file_ids
def scrape():
response = requests.get(url)
try:
response = requests.get(url)
# 确保请求成功
if response.status_code == 200:
# 解析网页内容
soup = BeautifulSoup(response.content, 'html.parser')
# 确保请求成功
if response.status_code == 200:
# 解析网页内容
soup = BeautifulSoup(response.content, 'html.parser')
# 找到所有表格
tables = soup.find_all('table')
# 找到所有表格
tables = soup.find_all('table')
# 提取第一个表格的内容
for row in tables[0].find_all('tr')[2:]: # 跳过标题行
cells = row.find_all(['td', 'th'])
if len(cells) >= 6: # 确保有至少六列
folder_name = cells[1].get_text(strip=True) # 第2列
file_id = cells[0].get_text(strip=True) # 第1列
download_url = f'https://esarad.esa.int/?id={file_id}&handler=DownloadDb'
# 提取第一个表格的内容
for row in tables[0].find_all('tr')[2:]: # 跳过标题行
cells = row.find_all(['td', 'th'])
if len(cells) >= 6: # 确保有至少六列
folder_name = cells[1].get_text(strip=True) # 第2列
file_id = cells[0].get_text(strip=True) # 第1列
download_url = f'https://esarad.esa.int/?id={file_id}&handler=DownloadDb'
# 创建文件夹
folder_path = os.path.join(download_dir, folder_name)
os.makedirs(folder_path, exist_ok=True)
# 创建文件夹
folder_path = os.path.join(download_dir, folder_name)
os.makedirs(folder_path, exist_ok=True)
# 下载文件并获取文件名
file_response = requests.get(download_url)
if file_response.status_code == 200:
# 从响应头获取文件名
content_disposition = file_response.headers.get('Content-Disposition')
filename = ''
if content_disposition:
match = re.search(r'filename="([^"]+)"', content_disposition)
if match:
filename = match.group(1)
# 下载文件并获取文件名
file_response = requests.get(download_url)
if file_response.status_code == 200:
# 从响应头获取文件名
content_disposition = file_response.headers.get('Content-Disposition')
filename = ''
if content_disposition:
match = re.search(r'filename="([^"]+)"', content_disposition)
if match:
filename = match.group(1)
if not filename: # 如果没有找到，则使用默认文件名
filename = f'{file_id}.pdf'
if not filename: # 如果没有找到，则使用默认文件名
filename = f'{file_id}.pdf'
file_path = os.path.join(folder_path, filename)
with open(file_path, 'wb') as f:
f.write(file_response.content)
print(f'ESA Downloaded: {file_path}')
file_path = os.path.join(folder_path, filename)
with open(file_path, 'wb') as f:
f.write(file_response.content)
print(f'ESA Downloaded: {file_path}')
# 创建 CSV 文件
csv_file_path = os.path.join(folder_path, 'data.csv')
with open(csv_file_path, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile)
# 写入标题行
writer.writerow([
'序号', '试验对象类型', '试验开始日期', '试验结束日期', '试验对象名称',
'试验对象型号', '试验对象数量', '试验性质', '试验目的', '装置名称',
'数据提供单位', '试验委托单位', '失效判据', '失效数量', '试验结果描述',
'成果', '来源项目名称', '来源项目类型', '分类', '元器件名称',
'元器件型号', '元器件批号', '生产单位', '是否国产', '元器件成熟度',
'晶圆材料', '晶圆批号', '封装材料', '封装技术', '是否倒装',
'制造工艺', '工艺特征尺寸', '工艺平台', '工艺代号', '工艺版本',
'质量等级', '加固措施', '工作原理', '供货能力', '应用经历',
'规范手册', '器件图片', '电子系统分类', '电子系统名称', '电子系统型号',
'生产单位', '电子系统功能', '电子系统加固措施', '电子系统图片', '材料名称',
'材料型号', '材料组分', '材料用途', '材料生产单位', '材料物理结构',
'材料使用经历', '辐照试验大纲', '大纲审核专家类别', '辐照试验所依据的标准规范',
'试验步骤（过程）描述', '辐照过程是否加电', '直流偏置条件描述', '交流偏置条件描述',
'时钟频率', '测试图形', '其他偏置条件', '辐照偏置原理图', '测试方式',
'测试原理图', '试验用仪器名称', '试验用仪器型号', '试验用仪器生产厂家',
'试验用仪器检定证书', '试验用软件名称', '试验用软件开发单位',
'试验用软件版本号', '试验现场照片', '测试人员姓名', '测试人员单位',
'测试人员电话', '装置运行人员', '第三方人员', '第三方人员单位',
'第三方人员电话', '其他需要说明的事项', '是否采用铅铝屏蔽', '剂量率',
'总剂量', '剂量等效材料', '试验对象编号', '测试参数名称',
'测试参数单位', '测试参数结果', '是否为加速试验后数据', '是否为退火数据',
'退火温度', '退火时间', '原始数据', '数据处理方法', '其他需要说明的事项'
])
# 创建 CSV 文件
csv_file_path = os.path.join(folder_path, 'data.csv')
with open(csv_file_path, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile)
# 写入标题行
writer.writerow([
'序号', '试验对象类型', '试验开始日期', '试验结束日期', '试验对象名称',
'试验对象型号', '试验对象数量', '试验性质', '试验目的', '装置名称',
'数据提供单位', '试验委托单位', '失效判据', '失效数量', '试验结果描述',
'成果', '来源项目名称', '来源项目类型', '分类', '元器件名称',
'元器件型号', '元器件批号', '生产单位', '是否国产', '元器件成熟度',
'晶圆材料', '晶圆批号', '封装材料', '封装技术', '是否倒装',
'制造工艺', '工艺特征尺寸', '工艺平台', '工艺代号', '工艺版本',
'质量等级', '加固措施', '工作原理', '供货能力', '应用经历',
'规范手册', '器件图片', '电子系统分类', '电子系统名称', '电子系统型号',
'生产单位', '电子系统功能', '电子系统加固措施', '电子系统图片', '材料名称',
'材料型号', '材料组分', '材料用途', '材料生产单位', '材料物理结构',
'材料使用经历', '辐照试验大纲', '大纲审核专家类别', '辐照试验所依据的标准规范',
'试验步骤（过程）描述', '辐照过程是否加电', '直流偏置条件描述', '交流偏置条件描述',
'时钟频率', '测试图形', '其他偏置条件', '辐照偏置原理图', '测试方式',
'测试原理图', '试验用仪器名称', '试验用仪器型号', '试验用仪器生产厂家',
'试验用仪器检定证书', '试验用软件名称', '试验用软件开发单位',
'试验用软件版本号', '试验现场照片', '测试人员姓名', '测试人员单位',
'测试人员电话', '装置运行人员', '第三方人员', '第三方人员单位',
'第三方人员电话', '其他需要说明的事项', '是否采用铅铝屏蔽', '剂量率',
'总剂量', '剂量等效材料', '试验对象编号', '测试参数名称',
'测试参数单位', '测试参数结果', '是否为加速试验后数据', '是否为退火数据',
'退火温度', '退火时间', '原始数据', '数据处理方法', '其他需要说明的事项'
])
# 写入数据行并保存到数据库
data_row = [
file_id, # 序号
'', # 试验对象类型（可以根据需要填充）
cells[11].get_text(strip=True), # 试验开始日期（第12列）
'', # 试验结束日期（可以根据需要填充）
'', # 试验对象名称（可以根据需要填充）
cells[1].get_text(strip=True), # 试验对象型号（第2列）
'', # 试验对象数量（可以根据需要填充）
cells[5].get_text(strip=True), # 试验性质（第6列）
'', # 试验目的（可以根据需要填充）
'', # 装置名称（可以根据需要填充）
'', # 数据提供单位（可以根据需要填充）
'', # 试验委托单位（可以根据需要填充）
'', # 失效判据（可以根据需要填充）
'', # 失效数量（可以根据需要填充）
'', # 试验结果描述（可以根据需要填充）
'', # 成果（可以根据需要填充）
'', # 来源项目名称（可以根据需要填充）
'', # 来源项目类型（可以根据需要填充）
'', # 分类（可以根据需要填充）
'', # 元器件名称（可以根据需要填充）
'', # 元器件型号（可以根据需要填充）
'', # 元器件批号（可以根据需要填充）
cells[3].get_text(strip=True), # 生产单位（第4列）
'', # 是否国产（可以根据需要填充）
'', # 元器件成熟度（可以根据需要填充）
'', # 晶圆材料（可以根据需要填充）
'', # 晶圆批号（可以根据需要填充）
'', # 封装材料（可以根据需要填充）
'', # 封装技术（可以根据需要填充）
'', # 是否倒装（可以根据需要填充）
'', # 制造工艺（可以根据需要填充）
'', # 工艺特征尺寸（可以根据需要填充）
'', # 工艺平台（可以根据需要填充）
'', # 工艺代号（可以根据需要填充）
'', # 工艺版本（可以根据需要填充）
'', # 质量等级（可以根据需要填充）
'', # 加固措施（可以根据需要填充）
'', # 工作原理（可以根据需要填充）
'', # 供货能力（可以根据需要填充）
'', # 应用经历（可以根据需要填充）
'', # 规范手册（可以根据需要填充）
'', # 器件图片（可以根据需要填充）
'', # 电子系统分类（可以根据需要填充）
'', # 电子系统名称（可以根据需要填充）
'', # 电子系统型号（可以根据需要填充）
'', # 生产单位（可以根据需要填充）
'', # 电子系统功能（可以根据需要填充）
'', # 电子系统加固措施（可以根据需要填充）
'', # 电子系统图片（可以根据需要填充）
'', # 材料名称（可以根据需要填充）
'', # 材料型号（可以根据需要填充）
'', # 材料组分（可以根据需要填充）
'', # 材料用途（可以根据需要填充）
'', # 材料生产单位（可以根据需要填充）
'', # 材料物理结构（可以根据需要填充）
'', # 材料使用经历（可以根据需要填充）
'', # 辐照试验大纲（可以根据需要填充）
'', # 大纲审核专家类别（可以根据需要填充）
'', # 辐照试验所依据的标准规范（可以根据需要填充）
'', # 试验步骤（过程）描述（可以根据需要填充）
'', # 辐照过程是否加电（可以根据需要填充）
'', # 直流偏置条件描述（可以根据需要填充）
'', # 交流偏置条件描述（可以根据需要填充）
'', # 时钟频率（可以根据需要填充）
'', # 测试图形（可以根据需要填充）
'', # 其他偏置条件（可以根据需要填充）
'', # 辐照偏置原理图（可以根据需要填充）
'', # 测试方式（可以根据需要填充）
'', # 测试原理图（可以根据需要填充）
'', # 试验用仪器名称（可以根据需要填充）
'', # 试验用仪器型号（可以根据需要填充）
'', # 试验用仪器生产厂家（可以根据需要填充）
'', # 试验用仪器检定证书（可以根据需要填充）
'', # 试验用软件名称（可以根据需要填充）
'', # 试验用软件开发单位（可以根据需要填充）
'', # 试验用软件版本号（可以根据需要填充）
'', # 试验现场照片（可以根据需要填充）
'', # 测试人员姓名（可以根据需要填充）
'', # 测试人员单位（可以根据需要填充）
'', # 测试人员电话（可以根据需要填充）
'', # 装置运行人员（可以根据需要填充）
'', # 第三方人员（可以根据需要填充）
'', # 第三方人员单位（可以根据需要填充）
'', # 第三方人员电话（可以根据需要填充）
'', # 其他需要说明的事项（可以根据需要填充）
'', # 是否采用铅铝屏蔽（可以根据需要填充）
'', # 剂量率（可以根据需要填充）
'', # 总剂量（可以根据需要填充）
'', # 剂量等效材料（可以根据需要填充）
'', # 试验对象编号（可以根据需要填充）
'', # 测试参数名称（可以根据需要填充）
'', # 测试参数单位（可以根据需要填充）
'', # 测试参数结果（可以根据需要填充）
'', # 是否为加速试验后数据（可以根据需要填充）
'', # 是否为退火数据（可以根据需要填充）
'', # 退火温度（可以根据需要填充）
'', # 退火时间（可以根据需要填充）
'', # 原始数据（可以根据需要填充）
'', # 数据处理方法（可以根据需要填充）
'' # 其他需要说明的事项（可以根据需要填充）
]
writer.writerow(data_row)
# (`id`, `create_by`, `create_time`, `update_by`, `update_time`, `sys_org_code`, `device_type`,
# `device_name`, `device_mode`, `device_function`, `device_batch`, `manufacturer`, `experiment_date`, `data_source`, `experiment_user`, `total_count`, `file_list`)
# 写入数据行并保存到数据库
data_row = [
file_id, # 序号
'', # 试验对象类型（可以根据需要填充）
cells[11].get_text(strip=True), # 试验开始日期（第12列）
'', # 试验结束日期（可以根据需要填充）
'', # 试验对象名称（可以根据需要填充）
cells[1].get_text(strip=True), # 试验对象型号（第2列）
'', # 试验对象数量（可以根据需要填充）
cells[5].get_text(strip=True), # 试验性质（第6列）
'', # 试验目的（可以根据需要填充）
'', # 装置名称（可以根据需要填充）
'', # 数据提供单位（可以根据需要填充）
'', # 试验委托单位（可以根据需要填充）
'', # 失效判据（可以根据需要填充）
'', # 失效数量（可以根据需要填充）
'', # 试验结果描述（可以根据需要填充）
'', # 成果（可以根据需要填充）
'', # 来源项目名称（可以根据需要填充）
'', # 来源项目类型（可以根据需要填充）
'', # 分类（可以根据需要填充）
'', # 元器件名称（可以根据需要填充）
'', # 元器件型号（可以根据需要填充）
'', # 元器件批号（可以根据需要填充）
cells[3].get_text(strip=True), # 生产单位（第4列）
'', # 是否国产（可以根据需要填充）
'', # 元器件成熟度（可以根据需要填充）
'', # 晶圆材料（可以根据需要填充）
'', # 晶圆批号（可以根据需要填充）
'', # 封装材料（可以根据需要填充）
'', # 封装技术（可以根据需要填充）
'', # 是否倒装（可以根据需要填充）
'', # 制造工艺（可以根据需要填充）
'', # 工艺特征尺寸（可以根据需要填充）
'', # 工艺平台（可以根据需要填充）
'', # 工艺代号（可以根据需要填充）
'', # 工艺版本（可以根据需要填充）
'', # 质量等级（可以根据需要填充）
'', # 加固措施（可以根据需要填充）
'', # 工作原理（可以根据需要填充）
'', # 供货能力（可以根据需要填充）
'', # 应用经历（可以根据需要填充）
'', # 规范手册（可以根据需要填充）
'', # 器件图片（可以根据需要填充）
'', # 电子系统分类（可以根据需要填充）
'', # 电子系统名称（可以根据需要填充）
'', # 电子系统型号（可以根据需要填充）
'', # 生产单位（可以根据需要填充）
'', # 电子系统功能（可以根据需要填充）
'', # 电子系统加固措施（可以根据需要填充）
'', # 电子系统图片（可以根据需要填充）
'', # 材料名称（可以根据需要填充）
'', # 材料型号（可以根据需要填充）
'', # 材料组分（可以根据需要填充）
'', # 材料用途（可以根据需要填充）
'', # 材料生产单位（可以根据需要填充）
'', # 材料物理结构（可以根据需要填充）
'', # 材料使用经历（可以根据需要填充）
'', # 辐照试验大纲（可以根据需要填充）
'', # 大纲审核专家类别（可以根据需要填充）
'', # 辐照试验所依据的标准规范（可以根据需要填充）
'', # 试验步骤（过程）描述（可以根据需要填充）
'', # 辐照过程是否加电（可以根据需要填充）
'', # 直流偏置条件描述（可以根据需要填充）
'', # 交流偏置条件描述（可以根据需要填充）
'', # 时钟频率（可以根据需要填充）
'', # 测试图形（可以根据需要填充）
'', # 其他偏置条件（可以根据需要填充）
'', # 辐照偏置原理图（可以根据需要填充）
'', # 测试方式（可以根据需要填充）
'', # 测试原理图（可以根据需要填充）
'', # 试验用仪器名称（可以根据需要填充）
'', # 试验用仪器型号（可以根据需要填充）
'', # 试验用仪器生产厂家（可以根据需要填充）
'', # 试验用仪器检定证书（可以根据需要填充）
'', # 试验用软件名称（可以根据需要填充）
'', # 试验用软件开发单位（可以根据需要填充）
'', # 试验用软件版本号（可以根据需要填充）
'', # 试验现场照片（可以根据需要填充）
'', # 测试人员姓名（可以根据需要填充）
'', # 测试人员单位（可以根据需要填充）
'', # 测试人员电话（可以根据需要填充）
'', # 装置运行人员（可以根据需要填充）
'', # 第三方人员（可以根据需要填充）
'', # 第三方人员单位（可以根据需要填充）
'', # 第三方人员电话（可以根据需要填充）
'', # 其他需要说明的事项（可以根据需要填充）
'', # 是否采用铅铝屏蔽（可以根据需要填充）
'', # 剂量率（可以根据需要填充）
'', # 总剂量（可以根据需要填充）
'', # 剂量等效材料（可以根据需要填充）
'', # 试验对象编号（可以根据需要填充）
'', # 测试参数名称（可以根据需要填充）
'', # 测试参数单位（可以根据需要填充）
'', # 测试参数结果（可以根据需要填充）
'', # 是否为加速试验后数据（可以根据需要填充）
'', # 是否为退火数据（可以根据需要填充）
'', # 退火温度（可以根据需要填充）
'', # 退火时间（可以根据需要填充）
'', # 原始数据（可以根据需要填充）
'', # 数据处理方法（可以根据需要填充）
'' # 其他需要说明的事项（可以根据需要填充）
]
writer.writerow(data_row)
save_to_mysql(data_row)
print(f'CSV created and data saved to MySQL: {csv_file_path}')
else:
print(f'Failed to download: {download_url}')
else:
print(f'Error: {response.status_code}')
else:
print(f'Failed to download: {download_url}')
upload_ids = upload_to_minio(folder_path)
data_db = ['ESA-' + cells[0].get_text(strip=True), 'Crawler', datetime.now(), None, None, None,
cells[5].get_text(strip=True),
cells[1].get_text(strip=True), cells[1].get_text(strip=True),
cells[8].get_text(strip=True),
cells[7].get_text(strip=True), cells[2].get_text(strip=True),
cells[11].get_text(strip=True),
'ESA', None, None, ','.join(upload_ids)
]
save_to_mysql(data_db)
else:
print(f'Error: {response.status_code}')
finally:
# 关闭游标和连接
cursor.close()
connection.close()

2
scripts/requirements.txt
View File

@@ -4,4 +4,4 @@ urllib3
flask
lxml
mysql-connector-python
gunicorn
minio