简介这是一份面向Python爬虫入门学习者与数据分析初学者的实战源码包以抓取东京奥运会奖牌国家及地区旗帜图片为案例演示从网页解析、图片下载到数据存储的完整流程。压缩包内含两个文件——一个HTML文件和一个Jupyter Notebook文件前者用于交互式讲解与结果展示后者提供可运行的Python代码逻辑压缩包整体约54KB便于快速下载与本地调试。目前已有217人浏览学习适合希望结合具体场景掌握requests、BeautifulSoup、CSS选择器及异常处理等爬虫核心技术的读者。通过学习该源码可以直观理解爬虫的请求发送、HTML结构解析、图片批量采集与文件保存等关键环节同时熟悉Jupyter Notebook的工程化写法与网络数据抓取的规范意识为后续数据采集与信息提取任务打下基础。1. 先看这包源码解决的核心问题旗帜图片远不止下载一份名为「部分国家及地区旗帜图片数据爬取源代码.zip」的项目摆在眼前乍一看你会觉得批量下载旗帜图片是最没技术含量的爬虫工作——一个循环、一个 URL 一批文件。真上手才会翻车同一面旗帜在不同来源里有时是 SVG、有时是 PNG列表页里缩略图和原图混在一起抓回来才发现只有 24 像素宽重跑两次还可能因为页面结构调整把已入库的图又原样下载一遍。这份源代码的价值不在「爬」而在「怎么保证爬回来的图能用」覆盖范围由清单文件控制格式用文件头校验重复图用感知哈希去重文件名用区域代码做全局主键——一条小流水线把下载变成了可校验、可重跑的数据任务。适合正在建标准化图库、做可视化配图或整理训练集的开发者。下文按这条链路拆开讲先定数据源再跑通请求与落盘接着用校验和去重收口最后给出增量更新的做法参数与代码照着改就能用。2. 数据来源与解析策略先定清单再谈抓取这类项目的第一步往往不是写爬虫而是先确定数据源。旗帜数据散落各处单一来源一旦改版或限流整条采集链路就断了。常见做法是同时配置两到三个来源抓完之后统一走一遍去重校验让任何一个来源失效都不至于影响全量图库。先看三类典型来源的差异再定解析规则。2.1 三类旗帜图片来源怎么选来源类型覆盖度常见格式稳定性典型麻烦开放百科的旗帜列表页高SVG 与 PNG 混合页面结构常改、偶尔限流缩略图与原始图链接混排图片走懒加载专业旗帜图库站中高多为 PNG、SVG访问频率限制严格图片可能带水印或额外边框机构官方发布页低但权威PNG、SVG 为主结构差异大、无统一规律命名和路径规则分散需要逐个适配选型原则并不复杂以开放百科列表页作为主数据源图库站做校验和补漏机构官方发布页只在个别旗帜缺失或颜色存疑时人工核对。这样组合的好处是覆盖度有保障又留了一条人工兜底路径——图库站挂了不影响全量百科改版了还有图库站的存量打底。判断一个列表页值不值得解析先看两样东西是否在同一页给出了全部旗帜入口以及原图链接是否可直接访问。如果原图来自另一个域名就要在代码里把跨域规则提前写好如果缩略图和原图只有尺寸参数不同解析时优先保留参数更大的那个。这个判断直接影响后续下载量和清洗成本值得在写代码前多花十分钟。2.2 用区域代码做全局主键清单决定爬取范围在写请求之前先把唯一标识定下来。这里的惯例是用两位字母的区域代码做文件名主键。它比中文名或英文名稳定不随语言切换而变化不会出现同名歧义也不会在跨系统传输时因为空格和特殊字符出问题。清单文件通常是一份制表符分隔的文本第一列是代码第二列是展示名。# load_codes.py from pathlib import Path def load_region_codes(path: Path, sep: str \t) - dict[str, str]: 读取区域代码映射表返回 {代码: 展示名}。 result: dict[str, str] {} for line in path.read_text(encodingutf-8).splitlines(): line line.strip() if not line or line.startswith(#): continue code, name, *_ line.split(sep) if code in result: raise ValueError(f重复的代码: {code}) result[code.strip()] name.strip() return result映射表加载时必须对重复代码直接报错不能在后续静默覆盖。这个检查能拦截最隐蔽的命名事故两个不同区域被映射成同一个代码后写入的图片把先写入的覆盖掉等你发现时已经丢了存量数据。sep默认是 Tab因为有些展示名里本身带空格用空格做分隔符很容易切错字段。这份映射表同时决定了爬取范围。缺了某个区域就去清单里补一行不想要的就把行注释掉——这就是标题里「部分国家及地区」的控制方式。展示名只用于日志和侧车文件不参与底层命名后期无论把展示名改成简体还是繁体存量文件名的唯一性都不受影响。2.3 列表页解析从 HTML 里稳定取出原图链接旗帜列表页的 HTML 结构并不统一但有几个高概率特征图片节点带 flag 相关类名原始图存在src或懒加载字段># parsers.py from urllib.parse import urljoin from bs4 import BeautifulSoup FLAG_CANDIDATE img[src*flag], img.flag, img[data-src*flag] def extract_original_urls(html: str, page_url: str) - list[str]: soup BeautifulSoup(html, html.parser) urls [] for img in soup.select(FLAG_CANDIDATE): # 懒加载优先取>def is_thumbnail(img) - bool: 通过宽高属性判断是否为缩略图拿不到属性时返回 False 由下载层兜底。 w img.get(width) or img.get(data-width) or 0 h img.get(height) or img.get(data-height) or 0 try: return int(w) 64 or int(h) 64 except ValueError: return False有些列表页不给宽高属性此时只能靠下载后的像素校验兜底不要在解析层为它写第二套逻辑。多源抓取时同一个区域代码可能拿到三条链接分别指向缩略图、中等尺寸图、原图不要全部下载。先按域名和格式排优先级原图优先于缩略图SVG/PNG 优先于 JPEG访问速度更快的源优先。排序规则写进配置后续换源不用改主流程。3. 抓取主流程落地限速、重试与命名落盘的完整实现数据源和主键定下来之后抓取层本身反而不复杂关键是几个参数要预先想清楚。旗帜图片单张体积普遍不大但数量一多对端服务的限流策略就会成为主要矛盾。这一章给出按常见模块划分的最小工程结构再逐个拆解请求、下载和命名三个环节。3.1 最小工程结构模块边界决定了排查效率我一般会把这类项目拆成下面几块每个模块只负责一件事。目录本身不复杂但边界清晰能省掉大量排查时间flag_downloader/ ├── config.py # 全局参数与阈值 ├── load_codes.py # 读取区域代码映射表 ├── parsers.py # HTML 列表页解析 ├── downloader.py # 请求、重试、临时落盘 ├── validator.py # 格式与内容校验 ├── dedupe.py # 感知哈希去重 ├── runner.py # 主流程编排 ├── tmp/ # 临时下载目录 └── output/ # 最终图片与报告核心参数集中在config.py别散落在各个函数里。下面的配置项是从此类项目里抽出来的最小集合# config.py REQUEST_TIMEOUT 15 # 单次请求超时单位秒 CONCURRENCY 3 # 并发线程数宁可小不要大 RATE_LIMIT_SECONDS 0.3 # 两次请求之间的最小间隔 MAX_RETRIES 4 # 失败重试次数 RETRY_BACKOFF 2.0 # 退避基数单位秒 MIN_IMAGE_BYTES 256 # 小于该体积的文件视为无效这几个参数里最影响成功率的不是超时而是并发与限速的配合。对多数公开列表页并发 3 已经够快一旦看到 503 或连续超时先把RATE_LIMIT_SECONDS提到 1.0再把CONCURRENCY降到 1 或 2。网络状况差的时候把REQUEST_TIMEOUT调到 30 并不丢人——超时太短只会让重试把对端负载进一步放大。3.2 请求层Session、UA 轮换与指数退避重试请求层的核心是「失败后如何退避」。旗帜项目里最常见的失败不是网络断开而是对端限流返回 429 或 503这种失败连续重试只会火上浇油。# downloader.py import random import time import requests UA_POOL [ Mozilla/5.0 (compatible; FlagBot/1.0; https://example.invalid/bot), Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36, ] def fetch_with_retry(session: requests.Session, url: str, cfg) - bytes: 带指数退避的 GET 请求仅对超时、429、5xx 做重试。 for attempt in range(1, cfg.MAX_RETRIES 1): try: resp session.get( url, headers{User-Agent: random.choice(UA_POOL)}, timeoutcfg.REQUEST_TIMEOUT, ) resp.raise_for_status() return resp.content except (requests.Timeout, requests.HTTPError) as e: status getattr(getattr(e, response, None), status_code, None) # 4xx 客户端错误重试没有意义429 属于限流需要单独放行 if status and 400 status 500 and status ! 429: raise wait cfg.RETRY_BACKOFF ** attempt random.uniform(0.2, 0.8) time.sleep(wait) raise RuntimeError(f重试 {cfg.MAX_RETRIES} 次仍失败: {url})重试条件只放行超时、429 和 5xx其余 4xx 直接抛错原因是 4xx 通常代表 URL 拼接错误或资源不存在重试多少次结果都一样。退避时间用2 ** attempt再加随机抖动是为了防止多个线程同时醒来再次请求形成一个虚假的并发峰值。请求要复用同一个requests.Session几百张图的任务能省掉大量 TCP 握手时间。3.3 下载层流式写文件与临时文件机制下载层不能直接把响应内容写进最终文件名。常见做法是先写带随机名的.part临时文件通过校验后再改名。# downloader.py import uuid from pathlib import Path def download_to_tmp(session: requests.Session, url: str, tmp_dir: Path, cfg) - Path: 流式下载到临时文件返回临时文件路径。 resp session.get( url, headers{User-Agent: random.choice(UA_POOL)}, streamTrue, timeoutcfg.REQUEST_TIMEOUT, ) resp.raise_for_status() tmp tmp_dir / f.{uuid.uuid4().hex}.part with tmp.open(wb) as f: for chunk in resp.iter_content(chunk_size65536): if chunk: f.write(chunk) return tmpstreamTrue避免大图一次性读进内存64KB 分块在图片场景下不会太碎也不会吃掉太多内存。临时文件以点开头并统一放在tmp/目录好处是清理脚本可以按.part后缀精准删除永远不会误伤正式图片。uuid.uuid4()生成随机名并发下载时不会互相覆盖。下载完成后不要立刻改名把校验放在改名之前。提示临时目录和正式目录必须分开这是避免误删的关键前提。3.4 命名落盘文件名只保留代码与格式后缀通过校验的图片最终命名为区域代码.格式后缀例如AA.png或AA.svg。落盘函数保持简单不给文件名附加任何展示信息。# runner.py import shutil from pathlib import Path def finalize(tmp_path: Path, code: str, fmt: str, output_dir: Path) - Path: 把临时文件移动到最终位置文件名 区域代码.后缀。 final_path output_dir / f{code}.{fmt} shutil.move(str(tmp_path), str(final_path)) return final_path展示名不放进文件名连接线空格逗号都不行。如果项目要求归档时能看到名称把展示名写进一个 sidecar JSON文件名保持纯 ASCII 的区域代码后续做跨平台同步才不会遇到编码歧义。如果两个来源返回了同一面旗帜的不同格式我这边的做法是统一按配置转成 PNG 输出但转换必须放在校验之后而不是之前——错误页可能以.svg结尾提前转换只会产出一张没意义的纯色图这个顺序就是第 4 章要展开的。4. 校验与去重把「看起来对」变成「确实能用」下载成功不代表入库成功。旗帜图库的验收标准是格式统一、无占位图、无重复图。很多初版爬虫栽在「浏览器里打开正常一进图库就是一堆坏图」原因是只做了文件名后缀判断没做文件内容判断。这一章把三道校验讲透。4.1 文件头魔数识别真实格式别信扩展名源站返回的Content-Type和文件扩展名都不可信有的站把 SVG 内容存成.png扩展名有的一言不合就返回 HTML 格式的错误页文件名后缀完全看不出来。可靠做法是读取文件头部字节按魔数判断真实格式。# validator.py from pathlib import Path MAGIC_PATTERNS [ (b\x89PNG\r\n\x1a\n, png), (b\xff\xd8\xff, jpg), (b?xml, svg), (bsvg, svg), ] def sniff_format(path: Path) - str: 读取文件头返回真实格式无法识别时返回 unknown。 head path.open(rb).read(64).lstrip() for magic, fmt in MAGIC_PATTERNS: if head.startswith(magic): return fmt return unknown.lstrip()是为了去掉文件开头的 BOM否则带 BOM 的 SVG 会被漏判。SVG 只有两种常见开头?xml声明或直接svg两种都要覆盖。JPEG 不需要检查 JFIF 标记\xff\xd8\xff三个字节已经足够。如果一批图片全部返回unknown基本可以断定对端在拦截脚本返回的是压缩或加密内容这时候排查重点要放到请求头而不是校验函数。4.2 体积、尺寸与内容特征拦下占位图和错误页光有格式还不够还要拦内容质量。占位图和错误页是两类最常见的垃圾文件它们都有鲜明特征文件过小、像素过小、SVG 文本里存在明显关键字。# validator.py from PIL import Image def validate_image(path: Path, fmt: str, cfg) - tuple[bool, str]: 综合校验文件体积、像素尺寸与 SVG 内容返回 (是否通过, 原因)。 stat path.stat() if stat.st_size cfg.MIN_IMAGE_BYTES: return False, f文件过小: {stat.st_size} bytes if fmt in (png, jpg): try: with Image.open(path) as im: w, h im.size if w 32 or h 32: return False, f图片过小: {w}x{h} except Exception: return False, 图片损坏无法解码 elif fmt svg: text path.read_text(utf-8, errorsignore).lower() if placeholder in text or not found in text: return False, 占位图或错误页 SVG return True, okMIN_IMAGE_BYTES设在 256 字节是经验值真实旗帜图片几乎都大于这个数错误页和 1x1 像素占位图基本都小于它。PNG/JPG 用 Pillow 打开后校验宽高不仅能拦缩略图还能顺便发现“文件头正确但内容损坏”的半截文件。SVG 是文本格式直接搜关键字最省事但关键字列表要按实际源站调整不同图库的占位图措辞不一样。4.3 感知哈希去重同一面旗帜多种来源只留一份多源抓取必然产生重复但同样一面旗帜经过不同压缩率、不同尺寸变换后SHA-256 完全不相等逐字节比较没有意义。这时要用感知哈希相似图片得到相近的指纹通过汉明距离判断是否同一面旗帜。# dedupe.py import imagehash from PIL import Image def phash_of(path: Path, hash_size: int 16) - str: 计算图片的平均哈希指纹返回十六进制字符串。 with Image.open(path) as im: im im.convert(RGB).resize((hash_size * 2, hash_size * 2), Image.LANCZOS) return str(imagehash.average_hash(im, hash_sizehash_size))average_hashaHash对旗帜这种大面积色块图足够用而且计算速度比 pHash 快hash_size16会得到 256 位指纹汉明距离小于 10 基本可以判定为同一面旗帜。如果在实际使用中误杀较多换成imagehash.phash即可原理一样只是对灰度变化的容忍度更强。去重落地时候选指纹要和已知指纹集合逐一比较# dedupe.py def is_duplicate(candidate_hash: str, known_hashes: set[str], threshold: int 10) - bool: 汉明距离小于阈值时认为与已知图片重复。 for h in known_hashes: diff sum(a ! b for a, b in zip(candidate_hash, h)) if diff threshold: return True return False几千张图规模下逐一遍历完全够用几万张以上再考虑 BK-tree 或倒排索引。一个前提条件SVG 在进入这个函数前必须先栅格化成 PNG因为文本格式没有像素可以比较。可以用cairosvg转成 512px 宽的标准图再做指纹这样跨格式也能识别重复。注意SVG 必须先转栅格化再交给感知哈希否则比较没有意义。5. 旗帜爬取常见坑与排查从限流到格式混排这个环节是血泪经验最集中的地方。看起来简单的旗帜爬取跑起来后会有各种「玄学」问题——浏览器里明明能看到图代码解析结果却是空的昨天还正常的任务今天突然全军覆没。以下五类问题我在不同项目里都遇到过每条按现象、原因、解决讲清楚。5.1 全部超时或一连串 503限流与并发设置不当现象任务运行几分钟后日志里开始出现大量超时和 503间隔重试后成功率仍然很低。原因并发开得太高把对端的限流策略触发了。旗帜列表页通常是静态页面但扛不住每秒十几个请求的扫描式抓取。UA 太单一也会加重嫌疑。解决把CONCURRENCY降到 1 或 2RATE_LIMIT_SECONDS提到 0.8 到 1.2。如果仍然出现 503检查 UA 池是否只有一条加入常规桌面浏览器指纹再重试。另外要确认 503 页面是否被下载成了 HTML 文件——这类错误页会通过文件头校验暴露出来然后在去重阶段被清除。5.2 浏览器有图、代码解析为空懒加载与动态渲染现象同一个列表页在浏览器里打开所有旗帜都能看到但代码里soup.select结果为零。原因图片走了懒加载。HTML 里的src本身就是占位图真实地址放在># index.py import json from pathlib import Path def load_index(path: Path) - dict[str, dict]: 读取指纹库返回 {区域代码: 记录}。 if not path.exists(): return {} rows [json.loads(line) for line in path.read_text(encodingutf-8).splitlines() if line.strip()] return {row[code]: row for row in rows} def append_entry(path: Path, entry: dict) - None: 追加一条指纹记录用追加而非覆盖保留历史痕迹。 with path.open(a, encodingutf-8) as f: f.write(json.dumps(entry, ensure_asciiFalse) \n)增量重跑时逻辑就很清晰目标文件已存在且 SHA-256 一致直接跳过不下载指纹不一致说明源站换了新图重新下载并覆盖指纹库里没有这个区域代码走完整的新增流程。# runner.py old load_index(INDEX_PATH) for code, url in pending_list: existing output_dir / f{code}.png if code in old and existing.exists(): if old[code].get(sha256) sha256_of(existing): continue # 走完整下载、校验、去重、落盘流程这份指纹库本身就是图库的「账本」审计时能说清楚每一面旗帜是什么时候下载的、内容指纹是什么。另一个不可省的习惯是失败报告。每失败一次就往failed.csv写一行字段至少包含区域代码、URL、失败原因# runner.py import csv with open(failed.csv, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnames[code, url, reason]) writer.writeheader() writer.writerows(failed_rows)失败报告是打破采集黑匣子的入口。同一来源全部失败说明列表页结构变了单个区域反复失败说明那条链接已经失效报告里的规律比日志里的单条错误更有价值。我现在的习惯是每次跑完只看两样东西指纹库新增了多少记录以及failed.csv里有没有成片的同类失败——很多问题都是在这份报告里发现的而不是在密密麻麻的日志里。把「下载」做成「可校验、可去重、可重跑」的小流水线这套思路能让你养出一份长期稳定的旗帜图库希望帮到你。本文还有配套的精品资源点击获取
