简介面向工业相机开发与机器视觉入门者的Python图像采集脚本定位于解决工业场景中实时抓取图像数据的基础需求。脚本基于OpenCV的VideoCapture模块完成相机初始化、逐帧读取、曝光与增益等参数配置、图像显示与保存并规范释放相机资源适用于质量检测、自动化产线监控等场景的快速原型验证。压缩包内仅1个py文件整体大小约4KB代码量精简、结构清晰便于直接阅读和移植改造。已有2768人学习下载说明该资源受到相关开发者的关注。通过研读此脚本可掌握工业相机与Python交互的核心流程理解ret与frame返回值处理、相机属性设置方法以及摄像头资源管理要点为后续图像处理与分析打下扎实基础尤其适合希望快速上手OpenCV相机编程的初学者。1. 图片抓取脚本为什么值得单独写一个 GrabImage.py做数据采集的人都有一个感受抓网页文本容易抓图片反而总在翻车。文本抓坏了顶多丢几个字段图片抓坏了是整张图花掉、半截文件、或者 Content-Type 对不上但保存成了 .jpg等清点数据时才发现几千张图全是 0 字节。GrabImage.py 就是一个把「从网页或 API 按规则批量下载图片到本地」这件事做稳的 Python 脚本它解决的不只是「能下载」而是「下载下来的每一张图都能用」。这个脚本适合三类人做数据集清洗的算法工程师、做商品图或素材备份的爬虫开发、以及需要定时同步远程图片到自己服务器的运维。它不是一个几百行的框架而是把下载、校验、命名、去重、断点续传和并发控制收敛在一个文件里拿着就能改、改完就能跑。下面我会从选型逻辑讲起落到可直接运行的最小版本再讲并发参数和那些不跑一遍绝对不知道的坑。2. 图片抓取和网页抓取不是一回事三个必须先想清楚的底层差异2.1 字节流才是本体为什么 requests.get 之后不能直接 r.content 存盘很多第一次写图片下载脚本的人会把图片当文本处理resp requests.get(url); open(a.jpg, wb).write(resp.content)。小图没问题图一大就暴露问题。图片是二进制流requests默认会等整个响应体收完才返回.content如果对方服务器断流、超时、或者中间代理掐了连接resp.content本身就是不完整的你拿到一段残缺字节流还照样写进了文件。一个可靠的做法是用流式读取。把streamTrue打开让响应体以块为单位进入你的内存边读边写。这样即使连接中途断开你至少能拿到已经落盘的部分配合断点续传还能接着下。另一个理由是内存几十 MB 的大图直接.content会一次性吃进内存流式写盘可以把峰值占用压到几 KB。import requests def download_one(url, save_path, timeout(5, 10)): try: with requests.get(url, streamTrue, timeouttimeout) as resp: resp.raise_for_status() with open(save_path, wb) as f: for chunk in resp.iter_content(chunk_size8192): if chunk: f.write(chunk) except requests.exceptions.RequestException as exc: print(f[failed] {url} - {exc})参数说明timeout(5, 10)第一个数是连接超时第二个数是读超时。图片服务器经常出现「连上了但迟迟不给数据」的情况读超时必须有。iter_content(chunk_size8192)是每次从 socket 读取的块大小8 KB 是通用值内网高带宽可以调到 65536服务器带宽一般就保持 8192避免单次读取过大把内存顶上去。resp.raise_for_status()是必须的否则 404 页面会把 HTML 文本当图片存下来。2.2 Content-Type 校验只靠文件后缀名判断类型是给自己埋雷图片抓取脚本最常见的脏数据来源就是 URL 以.jpg结尾但服务器返回的是 WebP 或 GIF。反过来也常见URL 没有任何图片后缀但响应头写着image/jpeg。如果只按 URL 后缀决定存成什么格式最后这批文件打不开的概率极高。我的习惯是下载前先发一个GET请求拿响应头校验Content-Type是否以image/开头再决定要不要真的把 body 拉下来。这一步多一次请求但能拦截掉大量广告页、404 陷阱页和防盗链占位图。注意有些图床返回的是application/octet-stream这种属于「不告诉你类型但确实给图」的情况我会额外把 URL 后缀解析出来兜底。from urllib.parse import urlparse import os ALLOWED_PREFIX image/ def get_real_type(url, session, default_ext.jpg): try: resp session.head(url, timeout5, allow_redirectsTrue) ctype resp.headers.get(Content-Type, ) if ctype.startswith(ALLOWED_PREFIX): return . ctype.split(/)[-1].split(;)[0].strip().lower() path_ext os.path.splitext(urlparse(url).path)[1].lower() return path_ext if path_ext in {.jpg, .jpeg, .png, .gif, .webp, .bmp} else default_ext except requests.exceptions.RequestException: return default_ext参数说明allow_redirectsTrue很关键很多图床会把图片链接 302 到 CDN 地址不跟随的话你 HEAD 拿到的是跳转页的Content-Type。Content-Type里可能带; charset之类的参数所以要split(;)取第一段。session用同一个连接池对象能复用 TCP 连接批量下载时速度差好几倍。这段逻辑的坑在下一章踩坑部分会详细说。2.3 小文件高并发图片下载的瓶颈不在带宽在握手和线程切换文本抓取常常是单线程跑因为单个页面就几十 KB响应快。图片不一样默认一张图一个 TCP 连接连接握手和 TLS 协商占的时间比传输还多。如果你一张一张顺序下1000 张图可能要 20 分钟开 8 个线程同时下同样的量能压到 3 分钟。所以 GrabImage.py 的主干不是下载函数而是并发调度。我一般用concurrent.futures.ThreadPoolExecutor不用asyncio——图片下载是 IO 密集型但requests是同步库硬套协程反而要换httpx或aiohttp对多数人来说改造成本高于收益。线程池的每个 worker 里跑独立的requests.Session避免 Session 对象在线程间共享带来的连接串线问题。3. 把 GrabImage.py 跑起来最小可运行版本与四个必调参数3.1 输入、下载、保存三步拆分先让脚本在本地端到端跑通先给一个不追求并发、但结构完整的最小版本。它读一个文本文件里的 URL 列表逐行下载按「序号_原文件名」的规则保存到指定目录。这个版本是为了先验证你的网络环境、目标站点是否允许抓取以及图片链接是否真的能直接访问。#!/usr/bin/env python3 import os import sys import time import requests from urllib.parse import urlparse def load_urls(path): with open(path, r, encodingutf-8) as f: return [line.strip() for line in f if line.strip() and not line.startswith(#)] def gen_save_path(index, url, out_dir): name os.path.basename(urlparse(url).path) or fimg_{index} ext os.path.splitext(name)[1].lower() if ext not in {.jpg, .jpeg, .png, .gif, .webp, .bmp}: ext .jpg return os.path.join(out_dir, f{index:05d}_{name if ext in name else download ext}) def grab_one(session, url, save_path, timeout(5, 10)): try: with session.get(url, streamTrue, timeouttimeout) as resp: resp.raise_for_status() with open(save_path, wb) as f: for chunk in resp.iter_content(chunk_size8192): if chunk: f.write(chunk) return True except requests.exceptions.RequestException as exc: print(f[x] {url} - {exc}) return False def main(): if len(sys.argv) ! 3: print(usage: python grabimage.py url_list.txt out_dir) sys.exit(1) url_file, out_dir sys.argv[1], sys.argv[2] os.makedirs(out_dir, exist_okTrue) urls load_urls(url_file) print(f[*] total {len(urls)} urls) session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 }) ok, fail 0, 0 for i, url in enumerate(urls): save_path gen_save_path(i, url, out_dir) if grab_one(session, url, save_path): ok 1 else: fail 1 time.sleep(0.2) print(f[*] done. ok{ok} fail{fail}) if __name__ __main__: main()逻辑说明load_urls里startswith(#)的支持是为了让你在 URL 列表里写注释行这比另开一个配置文件省事。gen_save_path用urlparse(url).path而不是直接url.split(/)[-1]是为了避开 URL 里 query 参数对文件名的污染——比如a.jpg?token123直接切分会把token123也塞进文件名。打印进度用print而不是logging是为了让新手先看清楚每张图的状态后面再换结构化日志。参数说明time.sleep(0.2)是单线程版本的礼貌限速200ms 间隔对大多数站点够用不会触发太激情的反爬。session.headers.update里的 User-Agent 最好换成一个真实浏览器版本全默认的python-requests/x.y在很多站点会直接被拒。gen_save_path里做了两次扩展名校验一次在name上一次在ext集合判断上少一次都会出现「URL 路径没有扩展名导致文件存成无后缀」的问题。3.2 日志和失败重试不重试的下载脚本不是脚本是废品没有任何一个图片服务器能保证 100% 不超时所以脚本必须内置重试。重试不是简单地把下载函数再调一遍而是要考虑三个参数重试次数、退避策略、以及什么错误值得重试。常见的做法是用requests的urllib3.Retry配合HTTPAdapter挂到 Session 上。这样做的好处是重试逻辑在 HTTP 层完成不污染业务代码。但要注意Retry默认只会重试连接错误和 5xx 状态码4xx 一律不重试因为 404 和 403 重试多少次结果都一样。import requests from urllib3.util.retry import Retry from requests.adapters import HTTPAdapter def build_session(retries3, backoff0.5): s requests.Session() retry Retry( totalretries, connectretries, readretries, statusretries, backoff_factorbackoff, status_forcelist[500, 502, 503, 504], allowed_methods[GET, HEAD] ) adapter HTTPAdapter(pool_connections10, pool_maxsize10, max_retriesretry) s.mount(http://, adapter) s.mount(https://, adapter) return s参数说明total是所有类型的重试次数上限connect和read分别控制连接阶段和读取阶段的重试。backoff_factor0.5表示第一次重试前等0.5秒第二次等0.5 * 2**1秒第三次等0.5 * 2**2秒指数退避是防止你重试太猛把自己打成攻击流量。status_forcelist里 500、502、503、504 值得重试但 501、505 不要加。allowed_methods只放 GET 和 HEAD——POST重试可能导致重复提交图片抓取用不到。配了HTTPAdapter之后pool_connections和pool_maxsize是需要按并发数调的。如果你开了 8 个线程但池子最大只有 2那 6 个线程会排队等连接白白浪费线程。经验值是pool_maxsize 线程数 2留点余量给 HEAD 请求和重试产生的临时连接。遇到 SSL 握手报错的站点可以把重试挂到https://和http://两套 mount 上有些图床 CDN 会从 HTTPS 302 到 HTTP少 mount 一个就少一层保护。4. 并发与限速把 GrabImage.py 从「能跑」改成「跑得快且不被封」4.1 生产者-消费者队列控制图片 URL 的读取速率比控制下载线程数更重要上一版脚本是「读一行下一张」但真实场景往往是URL 列表有几万行图片分布在不同域名下有的域名给得快有的域名限速严格。如果所有线程一股脑地拉 URL限速严格的域名会被打爆你的 IP 就上了对方黑名单。改进方式是把「读 URL」和「下载图片」解耦。主线程只负责读文件、把 URL 丢进队列消费者线程从队列里取任务执行下载。这样你能在队列层面做限速——比如每秒最多从队列取出 N 个 URL而不是让每个线程各自闷头跑。常见的做法是给队列加一个令牌桶消费者线程在取任务前先获取令牌没有令牌就等。import threading import queue import time class RateLimiter: def __init__(self, qps): self.qps qps self.tokens qps self.lock threading.Lock() self.last_refill time.time() def acquire(self): with self.lock: now time.time() elapsed now - self.last_refill self.tokens min(self.qps, self.tokens elapsed * self.qps) if self.tokens 1: self.tokens - 1 return wait (1 - self.tokens) / self.qps time.sleep(wait) self.tokens 0 self.last_refill time.time()参数说明qps是每秒允许通过的请求数注意这里的请求是「从队列里取任务」的次数不是下载请求本身。如果一个任务需要 HEAD 一次、GET 一次实际打到服务器的请求是两倍 QPS。所以如果你想把服务器压力控制在 5 QPS队列令牌应该设 2 到 3。time.sleep(wait)只锁住了当前线程不会阻塞其他线程的令牌获取这在 GIL 下是安全的因为time.sleep会释放 GIL。另一种常见方案是直接用threading.Semaphore限制同时下载的线程数。这个更简单但它控制的是并发度而不是速率。并发度高但每个请求间隔很短整体速率仍然可以很吓人。我的习惯是两个一起用Semaphore管并发上限RateLimiter管整体速率。这样即使某个域名响应特别快整体也不会超过你设定的阈值。4.2 线程数、超时、重试三个参数的联动照着这张表调不会翻车很多人在并发脚本里单独调线程数忽略超时和重试的联动关系。比如你设了 30 秒读超时、5 次重试、20 个线程最坏情况下一个任务要占线程 150 秒20 个线程全部被慢任务占满后面的任务全部排队。这不是并发是自锁。我一般按目标站点类型把参数配成一组而不是单独调一个值。站点类型线程数读超时(秒)重试次数队列QPS适用场景自家服务器 / 内网1652不限制数据集本地迁移公开图床 CDN81035通用素材采集反爬严格站点41522商品图、社交头像境外慢速图站63033大图原图下载给参数组合而不是单个参数定值是因为这几个参数会互相影响超时太短会让慢速站点的正常响应被误判为超时从而疯狂重试超时太长又会占满线程池。上面这张表里「反爬严格站点」的读超时反而比「公开图床」长是因为严格站点经常在响应前先做浏览器检测耗时久但并非不可用误杀代价更高。线程数也不是越大越好。图片下载的瓶颈大多数在对方服务器的带宽和连接数限制超过一定阈值后新线程只会增加重试率而不是吞吐。一个简单的验证方法把线程数从 4 开始翻倍跑观察每秒完成数和错误率。如果错误率开始上升而完成数不涨说明已经到极限了。用这个「翻倍试探法」比拍脑袋设 32 线程可靠得多。from concurrent.futures import ThreadPoolExecutor, as_completed def run_batch(urls, out_dir, workers8, qps5): os.makedirs(out_dir, exist_okTrue) session build_session(retries3, backoff0.5) limiter RateLimiter(qps) semaphore threading.Semaphore(workers * 2) def task(url, idx): with semaphore: limiter.acquire() save_path gen_save_path(idx, url, out_dir) return url, grab_one(session, url, save_path) with ThreadPoolExecutor(max_workersworkers) as pool: futures [pool.submit(task, url, i) for i, url in enumerate(urls)] for future in as_completed(futures): url, ok future.result() if not ok: print(f[retry-later] {url})逻辑说明Semaphore(workers * 2)是给任务队列一个缓冲避免线程空闲时队列也空着。qps5意味着全局每秒最多从队列消费 5 个 URL但每个任务的下载可能产生 2 个请求所以实际下行约为 10 QPS。as_completed(futures)按完成顺序处理结果而不是按提交顺序这能让你尽早看到失败的任务而不必等全部跑完。5. GrabImage.py 避坑指南现象、原因与解决5.1 保存的文件全部是 0 字节误判了 Content-Length 为 0 的响应这个坑我踩过不止一次。现象是脚本跑完没有任何报错日志全绿但输出目录里的文件大小全是 0。原因出在服务器对 HEAD 请求返回Content-Length: 0但对 GET 请求正常返回图片。如果你在下载前用 HEAD 做了一次类型校验然后判断「如果 Content-Length 为 0 就不下载」就会把所有这种站点全部跳过。解决不要在 HEAD 请求阶段依赖Content-Length只依赖Content-Type。Content-Length的可靠性在图片 CDN 上非常差——有些服务器根本不实现 HEAD有些实现了但故意返回 0 来省流量。正确做法是 HEAD 只取类型长度留给 GET 阶段去判断如果 GET 阶段读到的 body 也是空的再判定失败。5.2 文件名反复出现「_1」后缀URL 里查询参数污染了保存路径很多图片链接长这样https://cdn.example.com/images/photo.jpg?width800height600。如果你用url.split(/)[-1]取文件名得到的是photo.jpg?width800height600它在 Windows 和 Linux 上都不是合法文件名。更隐蔽的情况是 CDN 会在重定向后的 URL 里再追加签名参数导致同一张图每次生成的保存路径都不一样重复下载。解决用urlparse(url).path来取路径部分再用os.path.splitext拆扩展名。如果path本身没有扩展名不要硬猜统一存成.jpg并在文件名里加序号。加序号这件事不要省——不同 URL 可能指向同名文件比如avatar.jpg没有序号就会互相覆盖。我上面的gen_save_path已经用{index:05d}_做了前缀索引从 0 到 99999 足够容纳绝大部分批量任务。5.3 高并发下部分图片损坏重跑一遍又好了没有做内容完整性校验现象是并发开到 16 后下载成功的图里有一小部分打不开报「文件过早结束」。重跑同一批 URL坏的文件又变了。这类问题最隐蔽因为它时好时坏很难复现。原因有两层第一层是连接被服务器或中间代理重置iter_content还没读完就抛异常你捕获异常后认为「失败了」但with open(save_path, wb)已经创建了文件里面残留着半截数据第二层是raise_for_status()只能保证 HTTP 状态码正常不能保证 body 完整。解决在grab_one里加一个事后校验——文件大小不为 0且文件头两个字节匹配图片魔数JPEG 是FF D8PNG 是89 50。这个校验很便宜但对脏数据的拦截率极高。如果校验不通过删除文件并记录到失败列表方便后续重试。可以只对「预期是图片但校验失败」的情况重试不要对任何 URL 无脑重试整个下载。MAGIC_BYTES { .jpg: b\xff\xd8, .jpeg: b\xff\xd8, .png: b\x89\x50, .gif: b\x47\x49\x46, .webp: b\x52\x49\x46, .bmp: b\x42\x4d, } def verify_image(path, ext): if os.path.getsize(path) 0: return False expected MAGIC_BYTES.get(ext) if not expected: return True with open(path, rb) as f: header f.read(2) return header expected代码逻辑verify_image返回False时调用方删除文件并把它记为失败。MAGIC_BYTES里.webp的魔数其实是RIFF四个字节这里只校验前两个字节52 49因为 WebP 头部前四字节固定是RIFF前两字节足够区分大部分误存。.bmp的42 4d是 ASCII 的BM注意大小写敏感读取前两字节即可判断。5.4 下载到一半 IP 被限制限速只放在了「取任务」层没有放在「发包」层用前面的RateLimiter限速后仍然被站点封是因为限速发生在从队列取任务时而一个任务内部有 HEAD 和 GET 两次请求且Retry重试也会产生额外请求。如果服务器看的是每秒钟实际到达的请求数你设了qps5但真实请求可能是 15 到 20。解决把限速放到真正的请求层。用requests的Session包装一层自定义适配器在send()里挂一个全局的请求锁。或者更简单——把RateLimiter.acquire()同时在 HEAD 和 GET 前各调用一次把限速粒度从任务级降到请求级。但这会在高并发下牺牲一定吞吐因为两个请求之间释放了 GIL 但令牌获取是原子的。另一个经验是如果发现站点的响应头里有Retry-After务必让重试逻辑尊重这个值而不是按你自己的backoff_factor重试。5.5 目标站点返回了验证码页面但脚本全存成了 HTML这可能是最极端的情况网站上了人机验证所有图片请求返回的是一个 HTML 挑战页面但状态码是 200。你的脚本完全不会报错因为raise_for_status()只看状态码。等图片全部下完打开文件夹发现无一例外全是 HTML 文件那一刻的心情只能用「翻车」形容。解决在verify_image那一步之外再加一道防线——下载完成后检查文件的前几个字节是否为 ASCII 文本HTML 通常以!DOCTYPE或html开头。这个检查对 JPEG 和 PNG 魔数的校验可以并行做都是读取文件头几个字节几乎不增加耗时。一旦发现批量文件全是 HTML优先怀疑被验证码拦截而不是怀疑代码逻辑。此时应该停止任务、检查浏览器能否正常访问目标 URL而不是盲目调大重试次数。6. 增量抓取与断点续传让 GrabImage.py 变成能挂着跑一周的常驻工具批量图片采集往往不是一次跑完而是今天抓一批、明天补一批。每次全量重跑会重复下载大量已存在的文件浪费带宽也增加被限速的概率。我给 GrabImage.py 加增量能力的做法很简单在保存文件前先检查目标路径是否已存在如果存在且通过verify_image直接跳过。但这只解决「文件名稳定」的情况如果 URL 带时间戳签名参数每次 URL 都不同单靠路径存在判断不够需要用文件内容哈希来去重。import hashlib def sha1_head(path, chunk8192): h hashlib.sha1() with open(path, rb) as f: while True: data f.read(chunk) if not data: break h.update(data) return h.hexdigest()哈希去重的代价是必须把已下载的文件全部读一遍文件多了会拖慢速度。我的方案是只对「文件大小相同」的候选做哈希对比先按(size, ext)建立索引命中后再比哈希。这样大部分文件在大小层面就排除了重复哈希只需要计算真正可疑的那一小批。断点续传的逻辑则更直接——下载时先写到save_path .part完整校验通过后再os.rename成正式文件名。重跑任务时发现.part文件存在可以尝试从已有字节数往后请求Range头。最后分享一个个人习惯每跑完一批任务我会把失败 URL 单独存成一个.txt文件名带时间戳。下次重跑时只需要把grab_one的返回值从True/False改成「失败时返回 URL」然后用追加模式写文件。这个习惯救了我很多次因为图片服务器挂了之后恢复时间不确定等第二天重跑失败列表比全量重跑省事得多。如果你也打算拿 GrabImage.py 做每周定时同步建议把build_session、verify_image、RateLimiter三个函数独立出来它们之间的耦合度很低留好接口后续加数据库去重也方便。希望帮到你。本文还有配套的精品资源点击获取