import os from datetime import datetime import mysql.connector import requests from lxml import html import pandas as pd from common import db_config, upload_to_minio, save_to_db_import_record, get_md5 # 定义页面 URL url = 'https://radhome.gsfc.nasa.gov/radhome/papers/TIDPart.html' # 创建文件夹以保存文件 os.makedirs('downloaded_files', exist_ok=True) # CSV 表头 csv_header = [ "序号", "试验对象类型", "试验开始日期", "试验结束日期", "试验对象名称", "试验对象型号", "试验对象数量", "试验性质", "试验目的", "装置名称", "数据提供单位", "试验委托单位", "失效判据", "失效数量", "试验结果描述", "成果", "来源项目名称", "来源项目类型", "分类", "元器件名称", "元器件型号", "元器件批号", "生产单位", "是否国产", "元器件成熟度", "晶圆材料", "晶圆批号", "封装材料", "封装技术", "是否倒装", "制造工艺", "工艺特征尺寸", "工艺平台", "工艺代号", "工艺版本", "质量等级", "加固措施", "工作原理", "供货能力", "应用经历", "规范手册", "器件图片", "电子系统分类", "电子系统名称", "电子系统型号", "生产单位", "电子系统功能", "电子系统加固措施", "电子系统图片", "材料名称", "材料型号", "材料组分", "材料用途", "材料生产单位", "材料物理结构", "材料使用经历", "辐照试验大纲", "大纲审核专家类别", "辐照试验所依据的标准规范", "试验步骤（过程）描述", "辐照过程是否加电", "直流偏置条件描述", "交流偏置条件描述", "时钟频率", "测试图形", "其他偏置条件", "辐照偏置原理图", "测试方式", "测试原理图", "试验用仪器名称", "试验用仪器型号", "试验用仪器生产厂家", "试验用仪器检定证书", "试验用软件名称", "试验用软件开发单位", "试验用软件版本号", "试验现场照片", "测试人员姓名", "测试人员单位", "测试人员电话", "装置运行人员", "第三方人员", "第三方人员单位", "第三方人员电话", "其他需要说明的事项", "是否采用铅铝屏蔽", "剂量率", "总剂量", "剂量等效材料", "试验对象编号", "测试参数名称", "测试参数单位", "测试参数结果", "是否为加速试验后数据", "是否为退火数据", "退火温度", "退火时间", "原始数据", "数据处理方法", "其他需要说明的事项" ] def scrape(): nasa2_connection = mysql.connector.connect(**db_config) try: # 发送请求 response = requests.get(url) response.raise_for_status() # 使用 lxml 解析 HTML tree = html.fromstring(response.text) # 找到目标表格 table = tree.xpath('/html/body/table[3]')[0] # 获取第一个表格 # 遍历表格行，跳过表头 for row in table.xpath('.//tr')[1:]: # 跳过表头 columns = row.xpath('.//td') if len(columns) < 8: continue # 跳过不完整的行 table_id= get_md5('NASA-' + columns[0].text_content().strip()) part_number =table_id # columns[2].text_content().strip() # 第三列 Part Number file_link_tag = columns[7].xpath('//a') # 第八列的标签 # 获取文件名和下载链接 file_url = 'https://radhome.gsfc.nasa.gov/' + file_link_tag[0].get('href') # 创建目录 folder_path = os.path.join('downloaded_files', part_number) os.makedirs(folder_path, exist_ok=True) # 创建 DataFrame 并保存为 Excel 文件 xlsx_file_path = os.path.join(folder_path, 'data.xlsx') # 填写数据 data_row = [ "", # 序号 "", # 试验对象类型 columns[8].text_content().strip(), # 试验开始日期 "", # 试验结束日期 "", # 试验对象名称 columns[2].text_content().strip(), # 试验对象型号 "", # 试验对象数量 columns[4].text_content().strip(), # 试验性质 "", # 试验目的 "", # 装置名称 "", # 数据提供单位 "", # 试验委托单位 "", # 失效判据 "", # 失效数量 "", # 试验结果描述 "", # 成果 "", # 来源项目名称 "", # 来源项目类型 "", # 分类 "", # 元器件名称 "", # 元器件型号 "", # 元器件批号 columns[5].text_content().strip(), # 生产单位 "", # 是否国产 "", # 元器件成熟度 "", # 晶圆材料 "", # 晶圆批号 "", # 封装材料 "", # 封装技术 "", # 是否倒装 "", # 制造工艺 "", # 工艺特征尺寸 "", # 工艺平台 "", # 工艺代号 "", # 工艺版本 "", # 质量等级 "", # 加固措施 "", # 工作原理 "", # 供货能力 "", # 应用经历 "", # 规范手册 "", # 器件图片 "", # 电子系统分类 "", # 电子系统名称 "", # 电子系统型号 "", # 生产单位 "", # 电子系统功能 "", # 电子系统加固措施 "", # 电子系统图片 "", # 材料名称 "", # 材料型号 "", # 材料组分 "", # 材料用途 "", # 材料生产单位 "", # 材料物理结构 "", # 材料使用经历 "", # 辐照试验大纲 "", # 大纲审核专家类别 "", # 辐照试验所依据的标准规范 "", # 试验步骤（过程）描述 "", # 辐照过程是否加电 "", # 直流偏置条件描述 "", # 交流偏置条件描述 "", # 时钟频率 "", # 测试图形 "", # 其他偏置条件 "", # 辐照偏置原理图 "", # 测试方式 "", # 测试原理图 "", # 试验用仪器名称 "", # 试验用仪器型号 "", # 试验用仪器生产厂家 "", # 试验用仪器检定证书 "", # 试验用软件名称 "", # 试验用软件开发单位 "", # 试验用软件版本号 "", # 试验现场照片 "", # 测试人员姓名 "", # 测试人员单位 "", # 测试人员电话 "", # 装置运行人员 "", # 第三方人员 "", # 第三方人员单位 "", # 第三方人员电话 "", # 其他需要说明的事项 "", # 是否采用铅铝屏蔽 "", # 剂量率 "", # 总剂量 "", # 剂量等效材料 "", # 试验对象编号 "", # 测试参数名称 "", # 测试参数单位 "", # 测试参数结果 "", # 是否为加速试验后数据 "", # 是否为退火数据 "", # 退火温度 "", # 退火时间 "", # 原始数据 "", # 数据处理方法 "", # 其他需要说明的事项 ] # 创建 DataFrame df = pd.DataFrame([data_row], columns=csv_header) df.to_excel(xlsx_file_path, index=False) try: # 下载文件 file_response = requests.get(file_url) file_response.raise_for_status() # 保存文件 file_path = os.path.join(folder_path, os.path.basename(file_url)) with open(file_path, 'wb') as file: file.write(file_response.content) print(f"NASA2 Download file: {file_path}") except requests.RequestException as e: print(f"NASA2 Download file error: {file_url}，error: {e}") upload_ids = upload_to_minio(nasa2_connection, folder_path, 'NASA') device_type = columns[4].text_content().strip() device_name = columns[2].text_content().strip() device_mode = columns[2].text_content().strip() device_function = columns[3].text_content().strip() device_batch = None manufacturer = None experiment_date = columns[8].text_content().strip() origin_data = ','.join([c.text_content().strip() for c in columns]) data_db = [table_id, 'Crawler', datetime.now(), None, None, None, device_type, device_name, device_mode, device_function, device_batch, manufacturer, experiment_date, 'NASA', None, None, str(upload_ids),origin_data ] save_to_db_import_record(nasa2_connection, data_db) except Exception as e: print(f"error: {e}") finally: # 关闭游标和连接 nasa2_connection.close() # 输出数据 # for row in table.xpath('.//tr')[1:]: # columns = row.xpath('.//td') # print([col.text_content().strip() for col in columns]) # 打印每一行的数据