影刀RPA 网页图片批量下载：URL提取与保存
影刀RPA 网页图片批量下载：URL提取与保存
作者：林焱
什么情况用什么
需要把网页上的图片批量下载到本地——商品图片、文章配图、证件照片、设计素材。在影刀RPA里需要先提取图片URL，再批量下载保存，还要处理重命名、去重、懒加载等问题。
适用场景：电商商品图片批量保存、新闻配图下载、网页素材采集、图片素材库构建。
怎么做
拼多多店群自动化报活动上架！
提取图片URL
frombs4importBeautifulSoupimportrequestsdefextract_image_urls(url):"""提取网页所有图片URL"""headers={'User-Agent':'Mozilla/5.0...'}response=requests.get(url,headers=headers,timeout=10)soup=BeautifulSoup(response.text,'html.parser')image_urls=[]forimginsoup.find_all('img'):# 优先取data-src（懒加载真实地址）src=(img.get('data-src')orimg.get('data-original')orimg.get('src',''))ifnotsrcorsrc.startswith('data:'):continue# 跳过base64图片# 处理相对路径ifsrc.startswith('//'):src='https:'+srcelifsrc.startswith('/'):src=requests.compat.urljoin(url,src)elifnotsrc.startswith('http'):src=requests.compat.urljoin(url,src)# 去重ifsrcnotinimage_urls:image_urls.append(src)returnimage_urls# 使用urls=extract_image_urls("https://example.com/products")print(f"共找到{len(urls)}张图片")批量下载图片
importosimporttimefromdatetimeimportdatetimedefbatch_download_images(image_urls,save_dir,naming_rule='index'):""" 批量下载图片 naming_rule: 'index'=序号, 'md5'=URL的MD5, 'original'=原文件名 """os.makedirs(save_dir,exist_ok=True)results=[]fori,urlinenumerate(image_urls,1):try:# 生成文件名ifnaming_rule=='index':ext=get_extension_from_url(url)filename=f"image_{i:04d}{ext}"elifnaming_rule=='md5':importhashlib ext=get_extension_from_url(url)filename=hashlib.md5(url.encode()).hexdigest()+extelifnaming_rule=='original':filename=url.split('/')[-1].split('?')[0]ifnotfilename:filename=f"image_{i}{get_extension_from_url(url)}"filepath=os.path.join(save_dir,filename)# 下载headers={'User-Agent':'Mozilla/5.0...'}response=requests.get(url,headers=headers,timeout=15,stream=True)ifresponse.status_code==200:withopen(filepath,'wb')asf:forchunkinresponse.iter_content(8192):f.write(chunk)file_size=os.path.getsize(filepath)results.append({'url':url,'filename':filename,'path':filepath,'size':file_size,'status':'成功'})print(f"[{i}/{len(image_urls)}]{filename}({file_size//1024}KB)")else:results.append({'url':url,'status':f'HTTP{response.status_code}'})exceptExceptionase:results.append({'url':url,'status':f'错误:{e}'})time.sleep(0.5)# 延迟避免封IPreturnresultsdefget_extension_from_url(url):"""从URL获取文件扩展名"""importos path=url.split('?')[0]# 去掉参数ext=os.path.splitext(path)[1].lower()ifextin('.jpg','.jpeg','.png','.gif','.webp','.bmp','.svg'):returnextreturn'.jpg'# 默认jpg影刀RPA完整流程
【设置变量】 page_url = "https://example.com/products" save_dir = r"C:\Images\产品图片" 【打开网页】page_url 【等待元素出现】→ .product-image 【执行Python代码】 # 获取页面HTML html = yd_var['page_source'] # 提取图片URL from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') urls = [] for img in soup.select('.product-image img'): src = img.get('data-src') or img.get('src', '') if src and not src.startswith('data:'): if src.startswith('//'): src = 'https:' + src urls.append(src) yd_var['image_urls'] = urls 【执行Python代码】 # 批量下载 results = batch_download_images( yd_var['image_urls'], yd_var['save_dir'] ) yd_var['download_results'] = results 【输出结果】 成功：{成功数量}张 失败：{失败数量}张图片下载进阶
defdownload_images_advanced(urls,save_dir,referer=None,min_size=1024):""" 高级图片下载 min_size: 最小文件大小（字节），过滤占位图 """results=[]fori,urlinenumerate(urls,1):try:headers={'User-Agent':'Mozilla/5.0...'}ifreferer:headers['Referer']=referer# 防盗链response=requests.get(url,headers=headers,timeout=15,stream=True)ifresponse.status_code!=200:continue# 检查Content-Typecontent_type=response.headers.get('Content-Type','')if'image'notincontent_type:continue# 下载到临时文件ext=get_extension_from_url(url)filename=f"img_{i:04d}{ext}"filepath=os.path.join(save_dir,filename)withopen(filepath,'wb')asf:forchunkinresponse.iter_content(8192):f.write(chunk)# 检查文件大小file_size=os.path.getsize(filepath)iffile_size<min_size:os.remove(filepath)# 太小，可能是占位图results.append({'url':url,'status':'文件太小，已删除'})continueresults.append({'url':url,'filename':filename,'size':file_size,'status':'成功'})exceptExceptionase:results.append({'url':url,'status':str(e)})returnresults图片去重
defdeduplicate_images(image_dir):"""按文件MD5去重"""importhashlib files=[fforfinos.listdir(image_dir)iff.endswith(('.jpg','.png','.gif','.webp'))]md5_map={}duplicates=[]forfilenameinfiles:filepath=os.path.join(image_dir,filename)withopen(filepath,'rb')asf:file_md5=hashlib.md5(f.read()).hexdigest()iffile_md5inmd5_map:duplicates.append(filepath)os.remove(filepath)# 删除重复文件print(f"删除重复:{filename}(与{md5_map[file_md5]}相同)")else:md5_map[file_md5]=filenameprint(f"去重完成：共{len(files)}张，删除{len(duplicates)}张重复，剩余{len(md5_map)}张")returnlen(md5_map)有什么坑
TEMU店群矩阵自动化运营核价报活动
坑1：图片防盗链
# 问题：直接请求图片URL返回403# 原因：服务器检查Referer头# 解决：设置Referer为来源页面headers={'User-Agent':'Mozilla/5.0...','Referer':'https://example.com/'# 设置来源页面}response=requests.get(image_url,headers=headers)坑2：懒加载图片src是占位图
# 问题：img的src是loading.gif，真实图片在data-src里# 但有些网站用更复杂的懒加载# 解决1：优先取data-srcsrc=img.get('data-src')orimg.get('data-original')orimg.get('data-lazy')orimg.get('src')# 解决2：在影刀中先滚动触发加载# 【执行JavaScript代码】# window.scrollTo(0, document.body.scrollHeight);# 【等待】2秒# 然后再提取# 解决3：用Intersection Observer的网站# 需要逐个滚动到图片位置触发加载js_scroll_to_img=""" var imgs = document.querySelectorAll('img[data-src]'); imgs.forEach(function(img) { img.src = img.dataset.src; // 直接替换src触发加载 }); """坑3：图片URL带token过期
# 问题：图片URL包含临时token，几分钟后过期# 如：https://cdn.example.com/img.jpg?token=abc123&expire=1234567890# 解决：提取URL后立即下载，不要存起来等后面下forurlinimage_urls:# 立即下载response=requests.get(url,timeout=10)# 不要先存URL列表，循环到一半token就过期了坑4：WebP格式图片打不开
# 问题：下载的.webp图片在某些软件里打不开# 解决1：下载后转换为jpgfromPILimportImagedefconvert_webp_to_jpg(filepath):"""WebP转JPG"""img=Image.open(filepath)ifimg.modein('RGBA','P'):img=img.convert('RGB')jpg_path=filepath.rsplit('.',1)[0]+'.jpg'img.save(jpg_path,'JPEG',quality=95)os.remove(filepath)# 删除原webpreturnjpg_path# 解决2：在影刀中用Pillow批量转换forfinos.listdir(save_dir):iff.endswith('.webp'):convert_webp_to_jpg(os.path.join(save_dir,f))坑5：下载到错误内容（HTML而非图片）
# 问题：URL返回的是HTML错误页面而不是图片# 下载的"图片"打开后是乱码# 解决：检查Content-Type和文件头defis_valid_image(filepath):"""检查文件是否是有效图片"""# 检查文件头withopen(filepath,'rb')asf:header=f.read(10)# JPEG: FF D8# PNG: 89 50 4E 47# GIF: 47 49 46# WebP: 52 49 46 46ifheader[:2]==b'\xff\xd8':returnTrue# JPEGifheader[:4]==b'\x89PNG':returnTrue# PNGifheader[:3]==b'GIF':returnTrue# GIFifheader[:4]==b'RIFF':returnTrue# WebPreturnFalse# 下载后验证ifnotis_valid_image(filepath):os.remove(filepath)print(f"无效图片已删除:{filename}")