1. 为什么需要从Excel中提取隐藏的图片和文件链接在日常工作中我们经常会遇到这样的场景同事发来一个包含数百条产品信息的Excel表格里面混杂着文字描述、JSON数据和图片链接。更让人头疼的是这些图片链接往往不是标准的超链接格式而是嵌入在单元格文本或JSON字符串中。手动一个个复制粘贴不仅效率低下还容易出错。我曾经接手过一个电商项目需要处理包含3000多款商品数据的Excel文件。每款商品的图片链接都藏在类似{main_img:http://example.com/1.jpg,thumb_img:http://example.com/1_small.jpg}的JSON字符串里。如果手动处理至少需要3天时间而用Python自动化脚本只用了不到10分钟就完成了全部下载。2. 准备工作与环境配置2.1 安装必要的Python库在开始之前我们需要确保安装了以下Python库pip install pandas openpyxl requestspandas强大的数据处理库可以轻松读取Excel文件openpyxl专门处理xlsx格式的Excel文件requests用于下载网络资源如果你需要处理xls格式的老版Excel文件还需要安装xlrd库pip install xlrd1.2.0 # 注意要安装1.2.0版本新版不支持xls格式2.2 准备测试数据为了更好地演示我准备了一个示例Excel文件包含以下几种常见的数据格式直接超链接最简单的情况文本中包含的URL如产品图片http://example.com/1.jpgJSON格式字符串如{images:[http://example.com/1.jpg,http://example.com/2.jpg]}HTML片段如img srchttp://example.com/1.jpg3. 基础方法处理直接超链接我们先从最简单的情况开始 - Excel中标准的超链接列。这种情况处理起来最直接import pandas as pd import requests # 读取Excel文件 df pd.read_excel(products.xlsx, sheet_nameSheet1) # 简单的下载函数 def download_file(url, save_path): response requests.get(url) if response.status_code 200: with open(save_path, wb) as f: f.write(response.content) return True return False # 假设超链接在图片链接列 for index, row in df.iterrows(): if pd.notna(row[图片链接]): file_name fimages/product_{index1}.jpg download_file(row[图片链接], file_name)这种方法虽然简单但有很大的局限性只能处理标准的超链接列没有错误处理机制文件名没有区分度4. 进阶技巧提取文本中的隐藏链接4.1 使用正则表达式匹配URL当链接隐藏在普通文本中时我们需要使用正则表达式来提取import re def extract_urls(text): # 匹配常见的URL格式 pattern rhttp[s]?://(?:[a-zA-Z]|[0-9]|[$-_.]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F])) return re.findall(pattern, text) # 示例用法 sample_text 产品主图请查看http://example.com/1.jpg 备用图http://example.com/2.jpg print(extract_urls(sample_text)) # 输出: [http://example.com/1.jpg, http://example.com/2.jpg]4.2 处理HTML片段中的图片链接有时候单元格中可能包含HTML代码我们需要从中提取img标签的src属性from bs4 import BeautifulSoup def extract_html_images(html): soup BeautifulSoup(html, html.parser) return [img[src] for img in soup.find_all(img) if src in img.attrs] # 示例用法 html_content divimg srchttp://example.com/1.jpgp产品展示/p/div print(extract_html_images(html_content)) # 输出: [http://example.com/1.jpg]5. 高级应用解析JSON数据中的嵌套链接5.1 识别和解析JSON字符串很多现代系统会将结构化数据以JSON格式存储在Excel单元格中。我们需要先判断单元格内容是否是合法的JSONimport json def is_valid_json(text): try: json.loads(text) return True except ValueError: return False def extract_json_urls(json_str): urls [] data json.loads(json_str) # 递归遍历JSON结构 def traverse(obj): if isinstance(obj, dict): for value in obj.values(): traverse(value) elif isinstance(obj, list): for item in obj: traverse(item) elif isinstance(obj, str) and obj.startswith((http://, https://)): urls.append(obj) traverse(data) return urls # 示例用法 json_data {main_img:http://example.com/1.jpg,gallery:[http://example.com/2.jpg,http://example.com/3.jpg]} print(extract_json_urls(json_data)) # 输出: [http://example.com/1.jpg, http://example.com/2.jpg, http://example.com/3.jpg]5.2 处理复杂的嵌套结构在实际项目中JSON结构可能非常复杂。比如电商系统中常见的商品数据结构{ product: { id: 123, images: { primary: http://example.com/main.jpg, alternatives: [ {url: http://example.com/alt1.jpg, type: angle}, {url: http://example.com/alt2.jpg, type: detail} ] }, documents: [ {type: manual, url: http://example.com/manual.pdf}, {type: spec, url: http://example.com/spec.pdf} ] } }针对这种复杂结构我们需要改进我们的遍历函数def extract_deep_json_urls(json_str): urls [] data json.loads(json_str) def traverse(obj): if isinstance(obj, dict): # 特别处理包含url或link的键 for key, value in obj.items(): if key.lower() in [url, link, src] and isinstance(value, str) and value.startswith((http://, https://)): urls.append(value) else: traverse(value) elif isinstance(obj, list): for item in obj: traverse(item) elif isinstance(obj, str) and obj.startswith((http://, https://)): urls.append(obj) traverse(data) return urls6. 构建健壮的批量下载脚本现在我们把所有技术组合起来创建一个完整的解决方案import os import re import json import pandas as pd import requests from bs4 import BeautifulSoup from urllib.parse import urlparse class ExcelLinkDownloader: def __init__(self, excel_path, output_dirdownloads): self.df pd.read_excel(excel_path) self.output_dir output_dir os.makedirs(output_dir, exist_okTrue) staticmethod def is_valid_json(text): try: json.loads(text) return True except ValueError: return False staticmethod def extract_urls(text): # 处理普通文本中的URL url_pattern rhttp[s]?://(?:[a-zA-Z]|[0-9]|[$-_.]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F])) urls re.findall(url_pattern, text) # 处理HTML中的图片 if img in text: soup BeautifulSoup(text, html.parser) urls.extend([img[src] for img in soup.find_all(img) if src in img.attrs]) # 处理JSON数据 if ExcelLinkDownloader.is_valid_json(text): data json.loads(text) def traverse(obj): found [] if isinstance(obj, dict): for key, value in obj.items(): if key.lower() in [url, link, src] and isinstance(value, str) and value.startswith((http://, https://)): found.append(value) else: found.extend(traverse(value)) elif isinstance(obj, list): for item in obj: found.extend(traverse(item)) elif isinstance(obj, str) and obj.startswith((http://, https://)): found.append(obj) return found urls.extend(traverse(data)) return list(set(urls)) # 去重 def download_file(self, url, subfolderNone): try: response requests.get(url, timeout10) if response.status_code 200: # 从URL中提取文件名 path urlparse(url).path filename os.path.basename(path) or unknown_file # 确定保存路径 save_dir os.path.join(self.output_dir, subfolder) if subfolder else self.output_dir os.makedirs(save_dir, exist_okTrue) save_path os.path.join(save_dir, filename) # 处理文件名冲突 counter 1 while os.path.exists(save_path): name, ext os.path.splitext(filename) save_path os.path.join(save_dir, f{name}_{counter}{ext}) counter 1 with open(save_path, wb) as f: f.write(response.content) return True except Exception as e: print(f下载失败 {url}: {str(e)}) return False def process_column(self, column_name, subfolderNone): if column_name not in self.df.columns: print(f警告: 列 {column_name} 不存在) return for index, row in self.df.iterrows(): cell_content row[column_name] if pd.isna(cell_content): continue urls self.extract_urls(str(cell_content)) for url in urls: print(f处理行 {index1}: 下载 {url}) self.download_file(url, subfolder) # 使用示例 if __name__ __main__: downloader ExcelLinkDownloader(products.xlsx, downloaded_files) downloader.process_column(产品描述, products) downloader.process_column(技术参数, specs) downloader.process_column(图片链接, images)这个脚本具有以下特点自动识别多种URL格式纯文本、HTML、JSON支持子文件夹分类存储自动处理文件名冲突完善的错误处理机制支持从指定列提取链接7. 错误处理与性能优化7.1 常见的错误及解决方案在实际运行中你可能会遇到以下问题SSL证书错误有些网站的证书可能有问题# 在requests.get()中添加verifyFalse参数 response requests.get(url, verifyFalse)连接超时网络不稳定可能导致连接超时# 设置超时时间 response requests.get(url, timeout10)大文件下载需要分块下载以防内存不足with requests.get(url, streamTrue) as r: with open(save_path, wb) as f: for chunk in r.iter_content(chunk_size8192): f.write(chunk)重定向问题有些URL会重定向# 禁用重定向 response requests.get(url, allow_redirectsFalse)7.2 性能优化技巧当处理大量文件时可以考虑以下优化多线程下载from concurrent.futures import ThreadPoolExecutor def download_all(urls): with ThreadPoolExecutor(max_workers5) as executor: executor.map(download_file, urls)进度显示from tqdm import tqdm for url in tqdm(urls, desc下载进度): download_file(url)断点续传if os.path.exists(save_path): file_size os.path.getsize(save_path) headers {Range: fbytes{file_size}-} response requests.get(url, headersheaders, streamTrue) mode ab # 追加模式 else: response requests.get(url, streamTrue) mode wb # 写入模式缓存已处理的行processed_file processed_rows.txt # 读取已处理的行 try: with open(processed_file, r) as f: processed set(int(line) for line in f.read().splitlines()) except FileNotFoundError: processed set() # 处理新行 for index, row in df.iterrows(): if index not in processed: # 处理逻辑... with open(processed_file, a) as f: f.write(f{index}\n)8. 实际案例电商产品图片批量下载让我们看一个真实的电商应用场景。假设我们有一个包含产品信息的Excel表格结构如下产品ID产品名称价格图片信息1001无线耳机199{main:http://example.com/1001.jpg,details:[http://example.com/1001_1.jpg,http://example.com/1001_2.jpg]}1002智能手表599我们可以这样处理downloader ExcelLinkDownloader(ecommerce_products.xlsx, product_images) # 为每个产品创建单独文件夹 for index, row in downloader.df.iterrows(): product_id row[产品ID] product_name row[产品名称] # 清理产品名称中的特殊字符 safe_name re.sub(r[\\/*?:|], _, product_name) folder_name f{product_id}_{safe_name} # 处理图片信息列 if pd.notna(row[图片信息]): urls downloader.extract_urls(str(row[图片信息])) for url in urls: downloader.download_file(url, folder_name)这个脚本会为每个产品创建一个单独的文件夹如1001_无线耳机并将所有相关图片下载到该文件夹中保持清晰的组织结构。9. 扩展应用处理其他文件类型虽然我们主要讨论了图片下载但同样的技术也适用于其他文件类型PDF文档# 只需修改保存的文件扩展名 save_path os.path.join(save_dir, fdocument_{index}.pdf)ZIP压缩包save_path os.path.join(save_dir, farchive_{index}.zip)CSV数据文件save_path os.path.join(save_dir, fdata_{index}.csv)视频文件save_path os.path.join(save_dir, fvideo_{index}.mp4)对于这些文件类型下载代码基本相同唯一的区别是保存时使用的文件扩展名。10. 将脚本打包为可执行文件如果你想将脚本分享给不会Python的同事可以使用PyInstaller将其打包为exe文件首先安装PyInstallerpip install pyinstaller创建打包脚本build.pyimport PyInstaller.__main__ PyInstaller.__main__.run([ excel_link_downloader.py, --onefile, --windowed, --nameExcelLinkDownloader, --iconapp.ico # 可选图标文件 ])运行打包命令python build.py打包完成后你会在dist文件夹中找到ExcelLinkDownloader.exe文件可以将其分发给其他Windows用户使用。11. 最佳实践与注意事项在实际使用中我总结了以下几点经验文件命名策略使用有意义的文件名如产品ID_类型_序号.jpg避免使用特殊字符和空格保持文件名一致性便于后续处理文件夹结构downloads/ ├── 2023-10-01_批量1/ │ ├── 产品A/ │ │ ├── main.jpg │ │ └── detail_1.jpg │ └── 产品B/ │ ├── main.jpg │ └── spec.pdf └── 2023-10-02_批量2/ ├── 产品C/ └── 产品D/日志记录import logging logging.basicConfig( filenamedownload.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) try: # 下载代码... logging.info(f成功下载 {url} 到 {save_path}) except Exception as e: logging.error(f下载 {url} 失败: {str(e)})性能考量对于大量文件1000考虑使用数据库记录下载状态设置合理的延迟如time.sleep(1)避免被封IP使用User-Agent轮换模拟不同浏览器法律合规只下载你有权访问的内容尊重robots.txt限制不要过度请求同一个网站12. 替代方案比较除了Python还有其他工具可以完成类似任务工具/方法优点缺点Python脚本灵活可定制处理复杂结构需要编程知识Excel宏(VBA)无需额外安装集成在Excel中功能有限复杂结构处理困难专业下载软件图形界面易用通常收费定制能力有限在线转换工具无需安装简单易用隐私风险文件大小限制对于需要处理复杂数据结构或大量文件的情况Python仍然是最佳选择。我曾经比较过几种方法在处理包含嵌套JSON的Excel文件时Python脚本的效率和成功率明显高于其他方案。13. 常见问题解答Q: 如何处理需要登录才能访问的链接A: 你可以使用requests的Session对象保持登录状态session requests.Session() session.post(login_url, data{username: xxx, password: xxx}) response session.get(protected_url)Q: 下载的文件名乱码怎么办A: 可以从Content-Disposition头获取文件名或使用URL解码from urllib.parse import unquote filename unquote(os.path.basename(urlparse(url).path))Q: 如何跳过已经下载过的文件A: 可以在下载前检查文件是否存在if os.path.exists(save_path): file_size os.path.getsize(save_path) if file_size 0: # 确保不是空文件 continueQ: 如何处理不同的字符编码问题A: 在读取Excel时指定编码df pd.read_excel(file.xlsx, engineopenpyxl) # 或者对于csv df pd.read_csv(file.csv, encodingutf-8-sig)Q: 脚本在公司的代理环境下无法工作怎么办A: 可以配置requests使用代理proxies { http: http://proxy.example.com:8080, https: http://proxy.example.com:8080 } response requests.get(url, proxiesproxies)14. 完整代码示例以下是整合了所有功能的完整脚本import os import re import json import time import logging import pandas as pd import requests from bs4 import BeautifulSoup from urllib.parse import urlparse, unquote from concurrent.futures import ThreadPoolExecutor, as_completed from tqdm import tqdm class AdvancedExcelLinkDownloader: def __init__(self, excel_path, output_dirdownloads, max_workers5): self.excel_path excel_path self.output_dir output_dir self.max_workers max_workers self.processed_rows set() self.load_processed_rows() os.makedirs(output_dir, exist_okTrue) self.setup_logging() try: self.df pd.read_excel(excel_path, engineopenpyxl) except Exception as e: logging.error(f读取Excel文件失败: {str(e)}) raise def setup_logging(self): logging.basicConfig( filenameos.path.join(self.output_dir, download.log), levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) console logging.StreamHandler() console.setLevel(logging.INFO) logging.getLogger().addHandler(console) def load_processed_rows(self): self.processed_file os.path.join(self.output_dir, processed_rows.txt) try: with open(self.processed_file, r) as f: self.processed_rows set(int(line.strip()) for line in f if line.strip()) except FileNotFoundError: self.processed_rows set() def save_processed_row(self, row_index): self.processed_rows.add(row_index) with open(self.processed_file, a) as f: f.write(f{row_index}\n) staticmethod def is_valid_json(text): try: json.loads(text) return True except ValueError: return False staticmethod def extract_urls(text): if not isinstance(text, str) or not text.strip(): return [] urls set() # 提取普通URL url_pattern rhttp[s]?://(?:[a-zA-Z]|[0-9]|[$-_.]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F])) urls.update(re.findall(url_pattern, text)) # 提取HTML中的资源 if any(tag in text for tag in [img, a href, script src]): soup BeautifulSoup(text, html.parser) for tag in soup.find_all([img, a, script, link]): attr src if tag.name in [img, script] else href if attr in tag.attrs and tag[attr].startswith((http://, https://)): urls.add(tag[attr]) # 提取JSON中的URL if AdvancedExcelLinkDownloader.is_valid_json(text): data json.loads(text) def traverse(obj): found set() if isinstance(obj, dict): for key, value in obj.items(): if isinstance(value, str) and value.startswith((http://, https://)): found.add(value) else: found.update(traverse(value)) elif isinstance(obj, list): for item in obj: found.update(traverse(item)) return found urls.update(traverse(data)) return list(urls) def download_file(self, url, subfolderNone, retry3): for attempt in range(retry): try: # 获取文件名 path urlparse(url).path filename unquote(os.path.basename(path)) or unknown_file # 清理文件名 filename re.sub(r[\\/*?:|], _, filename) if not filename.lower().endswith((.jpg, .jpeg, .png, .gif, .pdf, .zip)): filename .dat # 给未知类型添加扩展名 # 确定保存路径 save_dir os.path.join(self.output_dir, subfolder) if subfolder else self.output_dir os.makedirs(save_dir, exist_okTrue) save_path os.path.join(save_dir, filename) # 检查文件是否已存在 if os.path.exists(save_path): file_size os.path.getsize(save_path) if file_size 1024: # 大于1KB认为文件完整 logging.info(f文件已存在跳过: {save_path}) return True # 下载文件 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 } response requests.get(url, headersheaders, streamTrue, timeout30) if response.status_code 200: # 处理内容类型 content_type response.headers.get(content-type, ) if image in content_type and not filename.lower().endswith((.jpg, .jpeg, .png, .gif)): ext .jpg if jpeg in content_type else f.{content_type.split(/)[-1]} save_path os.path.splitext(save_path)[0] ext # 写入文件 with open(save_path, wb) as f: for chunk in response.iter_content(chunk_size8192): if chunk: f.write(chunk) logging.info(f成功下载: {save_path}) return True else: logging.warning(f下载失败HTTP状态码 {response.status_code}: {url}) except Exception as e: logging.error(f下载出错 (尝试 {attempt 1}/{retry}): {url} - {str(e)}) time.sleep(2) # 重试前等待 return False def process_column(self, column_name, subfolderNone): if column_name not in self.df.columns: logging.warning(f列不存在: {column_name}) return # 准备下载任务 tasks [] for index, row in self.df.iterrows(): if index in self.processed_rows: continue cell_content row[column_name] if pd.isna(cell_content): continue urls self.extract_urls(str(cell_content)) if not urls: continue # 创建子文件夹 folder subfolder if subfolder is None: # 尝试使用其他列作为文件夹名 for col in [产品ID, 名称, 型号]: if col in self.df.columns and pd.notna(row[col]): folder str(row[col]) folder re.sub(r[\\/*?:|], _, folder) break for url in urls: tasks.append((index, url, folder)) # 多线程下载 with ThreadPoolExecutor(max_workersself.max_workers) as executor: futures { executor.submit(self.download_file, url, folder): (index, url) for index, url, folder in tasks } with tqdm(totallen(futures), desc下载进度) as pbar: for future in as_completed(futures): index, url futures[future] try: if future.result(): self.save_processed_row(index) except Exception as e: logging.error(f任务出错: {url} - {str(e)}) finally: pbar.update(1) # 使用示例 if __name__ __main__: try: downloader AdvancedExcelLinkDownloader( excel_pathproducts.xlsx, output_dirdownloaded_files, max_workers5 ) # 处理不同列 downloader.process_column(图片链接, images) downloader.process_column(产品描述) # 自动创建文件夹 downloader.process_column(技术参数, specs) logging.info(所有任务完成) except Exception as e: logging.error(f程序异常终止: {str(e)})这个高级版本包含了多线程下载支持完善的日志记录断点续传功能自动重试机制进度条显示更智能的文件名和文件夹处理处理各种内容类型用户代理轮换15. 总结与进阶建议经过多次项目实践我发现处理Excel中的隐藏链接有几个关键点预处理很重要在开始下载前先检查Excel文件的结构和质量处理异常数据和格式问题。错误处理是关键网络请求可能因为各种原因失败完善的错误处理可以避免整个脚本中断。日志必不可少详细的日志可以帮助你追踪问题特别是在处理大量文件时。性能考量对于超过1000个文件的下载任务考虑使用数据库记录状态而不仅仅是文本文件。合规使用确保你有权下载和使用这些文件特别是当处理商业数据时。如果你想进一步扩展这个脚本可以考虑添加GUI界面使非技术人员也能使用集成到工作流自动化平台如Airflow添加对云存储如S3、Google Drive的支持实现更智能的内容识别和分类添加对需要认证的API的支持在实际项目中这个脚本已经帮助我和我的团队节省了数百小时的手动工作。特别是在处理电商产品数据、媒体资源管理和文档归档等场景时自动化下载可以显著提高工作效率和数据准确性。