尧图网页设计 
Python后端爬虫专题12数据库里有结果还不够——保存原始HTML与离线重放上一篇练习完整答案上一课的状态图可以写成API 创建queued成功投递后仍保持queuedWorker 领取任务改为running运行结束按结果改为completed或partial未能读取任何列表页则是failedAPI 投递队列失败时写dispatch_failed。API 负责创建和投递两条边Worker 负责执行期的三条边任何一方都不能越过tenant_id更新别人的记录。“先落库再入队”有两个主要崩溃窗口数据库提交后、消息发送前进程退出会留下永远排队的记录消息发送成功后、回写queue_task_id前退出任务会执行但接口暂时看不到队列编号。最小 outbox 表需要id、event_type、aggregate_id、payload、created_at、published_at、attempts并与crawl_tasks在同一事务写入。独立发布器发送消息后再写published_at重复发布则依靠业务task_id幂等。迁移检查不是查看迁移文件而是真的在空库执行$dbJoin-Path$env:TEMPjobradar-migration.db$env:JOBRADAR_DATABASE_URL sqlitepysqlite:///$($db-replace\\,/).\.venv\Scripts\alembic.exe upgrade head.\.venv\Scripts\python.exe-cfrom sqlalchemy import create_engine,inspect; import os; print(inspect(create_engine(os.environ[JOBRADAR_DATABASE_URL])).get_table_names())输出应包含alembic_version、crawl_tasks和jobs。这段答案同时验证了脚本可执行、连接地址正确、迁移真的建表而不是只证明 Python 能导入 Alembic。一次“字段为什么变空”的事故假设周一抓到的职位描述有 800 字周二数据库中的描述只剩一句“登录后查看”。问题可能来自页面改版、登录失效、选择器选错、服务端灰度也可能是源站真的改了内容。只保存最终JobItem时这几种原因看起来完全一样保存下载时的原始响应后我们可以在不再次访问源站的情况下重放解析器确认当时服务器究竟返回了什么。因此快照不是备份数据库的替代品。数据库保存“当前可查询事实”快照保存“产生事实的输入证据”。前者适合分页、统计和 API后者适合调试解析器、审计内容变化、构造回归 Fixture。两份数据目的不同保留周期也不应一刀切。JobRadar 的快照合同FileSnapshotStore.save(url, body, headers)接收最终响应 URL、原始字节和响应头返回SnapshotReference。快照标识由 URL 与正文共同计算元数据另存 JSON记录 URL、白名单响应头、抓取时间和正文 SHA-256。这里刻意不保存请求 Cookie、Authorization也不把全部响应头照单全收防止调试证据变成凭据仓库。读取时使用load(snapshot_id)返回主体与类型化元数据。调用者不需要知道磁盘目录结构第 27 篇切换到 MinIO 时流水线仍然只依赖相同的save合同。MinioSnapshotStore用同一个快照标识作为对象名前缀先确保 bucket 存在再上传.html与.json两个对象。从线上失败回到离线解析先运行本篇检查点cd project.\.venv\Scripts\python.exe-m pytest tests\test_snapshots.py::test_file_snapshot_round_trip_preserves_body_and_safe_metadata-q然后可以用保存结果离线重放。注意parse_detail要接收原 URL因为相对链接、来源身份和报错定位都依赖它frompathlibimportPathfromjobradar.parsingimportparse_detailfromjobradar.snapshotsimportFileSnapshotStore storeFileSnapshotStore(Path(data/snapshots))storedstore.load(把实际 snapshot_id 填在这里)htmlstored.body.decode(utf-8)jobparse_detail(html,stored.metadata.url)print(job.model_dump_json(indent2))如果 HTML 编码不是 UTF-8不能无条件 decode应依据已验证的 Content-Type 或站点合同处理。课程 TargetLab 固定 UTF-8便于把注意力放在架构上。生产系统还要对快照做访问控制和生命周期管理因为公开页面也可能包含联系方式等个人信息。为什么先保存再解析流水线对有正文的响应先save再parse_detail。这样即使解析失败坏页面也留下证据若反过来最需要调查的响应恰好不会保存。304 没有正文不能创建一个空快照覆盖旧引用它只说明已保存版本仍有效。列表页同样保存因为分页入口改版会让整个任务发现数骤降。哈希命名还带来一个实用性质保存动作是幂等的。进程在写完 body、尚未写 metadata 时崩溃下一次相同正文仍落到同一标识实现用临时文件后原子替换避免读到半截 JSON。文件系统不是无限可靠真正重要的数据仍要依靠对象存储版本、备份和校验策略。保留策略不是“永远保存”可以把快照分成三档解析失败与字段突变的证据保留 90 天正常成功样本保留 14 天被法律或业务标记的审计样本按批准期限保留。清理时必须先判断 jobs 表是否仍引用某快照或者允许引用变为墓碑。课程没有自动删除功能避免把数据生命周期决定偷偷藏进教学代码。本篇完整快照模块阅读时先看两个返回模型再看文件实现最后看 MinIO 实现。它们共享输入输出却不共享内部 IO这就是让本地练习和生产部署可以替换的边界。原始响应快照解析失败后可以脱离网络重放。fromdataclassesimportasdict,dataclassfromdatetimeimportdatetime,timezonefromhashlibimportsha256fromioimportBytesIOimportjsonfrompathlibimportPathfromtypingimportProtocol _SAFE_HEADERS{content-type,etag,last-modified,content-language}dataclass(frozenTrue)classSnapshotMetadata:snapshot_id:strurl:strcaptured_at:strheaders:dict[str,str]body_sha256:strdataclass(frozenTrue)classSnapshotReference:snapshot_id:strbody_path:Path metadata_path:Pathdataclass(frozenTrue)classStoredSnapshot:metadata:SnapshotMetadata body:bytesdataclass(frozenTrue)classObjectSnapshotReference:snapshot_id:strbody_object:strmetadata_object:strclassObjectStorageClient(Protocol):defbucket_exists(self,bucket:str)-bool:...defmake_bucket(self,bucket:str)-object:...defput_object(self,bucket:str,name:str,stream:object,length:int,**kwargs:object)-object:...defget_object(self,bucket:str,name:str)-object:...classFileSnapshotStore:开发环境快照实现文件名只使用服务端生成的哈希。def__init__(self,root:Path)-None:self._rootroot.resolve()self._root.mkdir(parentsTrue,exist_okTrue)defsave(self,url:str,body:bytes,headers:dict[str,str])-SnapshotReference:body_hashsha256(body).hexdigest()snapshot_idsha256(url.encode(utf-8)b\0body).hexdigest()safe_headers{key.casefold():valueforkey,valueinheaders.items()ifkey.casefold()in_SAFE_HEADERS}metadataSnapshotMetadata(snapshot_idsnapshot_id,urlurl,captured_atdatetime.now(timezone.utc).isoformat(),headerssafe_headers,body_sha256body_hash,)body_pathself._root/f{snapshot_id}.htmlmetadata_pathself._root/f{snapshot_id}.jsonifnotbody_path.exists():body_path.write_bytes(body)metadata_path.write_text(json.dumps(asdict(metadata),ensure_asciiFalse,indent2),encodingutf-8,)returnSnapshotReference(snapshot_id,body_path,metadata_path)defload(self,snapshot_id:str)-StoredSnapshot:iflen(snapshot_id)!64orany(chnotin0123456789abcdefforchinsnapshot_id):raiseValueError(snapshot_id must be a lowercase SHA-256 value)body_pathself._root/f{snapshot_id}.htmlmetadata_pathself._root/f{snapshot_id}.jsonbodybody_path.read_bytes()rawjson.loads(metadata_path.read_text(encodingutf-8))metadataSnapshotMetadata(**raw)ifsha256(body).hexdigest()!metadata.body_sha256:raiseValueError(snapshot body hash does not match metadata)returnStoredSnapshot(metadatametadata,bodybody)classMinioSnapshotStore:生产快照实现与文件存储使用相同元数据和哈希合同。def__init__(self,client:ObjectStorageClient,bucket:str)-None:ifnotbucket:raiseValueError(snapshot bucket must not be blank)self._clientclient self._bucketbucketifnotself._client.bucket_exists(bucket):self._client.make_bucket(bucket)defsave(self,url:str,body:bytes,headers:dict[str,str])-ObjectSnapshotReference:body_hashsha256(body).hexdigest()snapshot_idsha256(url.encode(utf-8)b\0body).hexdigest()safe_headers{key.casefold():valueforkey,valueinheaders.items()ifkey.casefold()in_SAFE_HEADERS}metadataSnapshotMetadata(snapshot_idsnapshot_id,urlurl,captured_atdatetime.now(timezone.utc).isoformat(),headerssafe_headers,body_sha256body_hash,)body_objectfsnapshots/{snapshot_id}.htmlmetadata_objectfsnapshots/{snapshot_id}.jsonmetadata_bytesjson.dumps(asdict(metadata),ensure_asciiFalse,indent2).encode(utf-8)self._client.put_object(self._bucket,body_object,BytesIO(body),len(body),content_typetext/html,)self._client.put_object(self._bucket,metadata_object,BytesIO(metadata_bytes),len(metadata_bytes),content_typeapplication/json,)returnObjectSnapshotReference(snapshot_id,body_object,metadata_object)defload(self,snapshot_id:str)-StoredSnapshot:iflen(snapshot_id)!64orany(chnotin0123456789abcdefforchinsnapshot_id):raiseValueError(snapshot_id must be a lowercase SHA-256 value)body_responseself._client.get_object(self._bucket,fsnapshots/{snapshot_id}.html)metadata_responseself._client.get_object(self._bucket,fsnapshots/{snapshot_id}.json)try:bodybody_response.read()# type: ignore[attr-defined]rawjson.loads(metadata_response.read().decode(utf-8))# type: ignore[attr-defined]finally:forresponsein(body_response,metadata_response):response.close()# type: ignore[attr-defined]response.release_conn()# type: ignore[attr-defined]metadataSnapshotMetadata(**raw)ifsha256(body).hexdigest()!metadata.body_sha256:raiseValueError(snapshot body hash does not match metadata)returnStoredSnapshot(metadatametadata,bodybody)本篇课后练习用tests/fixtures/job-detail.html保存快照再读取并交给parse_detail给出完整可运行脚本并断言职位标题。把同一正文保存两次解释为什么 snapshot_id 相同改动一个字符后再次保存确认标识变化。列出哪些请求/响应头可以进入元数据哪些必须排除并说明 Cookie 泄露后可能造成的后果。下一篇会利用已保存的 ETag 做增量采集。 Python后端爬虫专题12：数据库里有结果还不够——保存原始HTML与离线重放
Python后端爬虫专题12数据库里有结果还不够——保存原始HTML与离线重放上一篇练习完整答案上一课的状态图可以写成API 创建queued成功投递后仍保持queuedWorker 领取任务改为running运行结束按结果改为completed或partial未能读取任何列表页则是failedAPI 投递队列失败时写dispatch_failed。API 负责创建和投递两条边Worker 负责执行期的三条边任何一方都不能越过tenant_id更新别人的记录。“先落库再入队”有两个主要崩溃窗口数据库提交后、消息发送前进程退出会留下永远排队的记录消息发送成功后、回写queue_task_id前退出任务会执行但接口暂时看不到队列编号。最小 outbox 表需要id、event_type、aggregate_id、payload、created_at、published_at、attempts并与crawl_tasks在同一事务写入。独立发布器发送消息后再写published_at重复发布则依靠业务task_id幂等。迁移检查不是查看迁移文件而是真的在空库执行$dbJoin-Path$env:TEMPjobradar-migration.db$env:JOBRADAR_DATABASE_URL sqlitepysqlite:///$($db-replace\\,/).\.venv\Scripts\alembic.exe upgrade head.\.venv\Scripts\python.exe-cfrom sqlalchemy import create_engine,inspect; import os; print(inspect(create_engine(os.environ[JOBRADAR_DATABASE_URL])).get_table_names())输出应包含alembic_version、crawl_tasks和jobs。这段答案同时验证了脚本可执行、连接地址正确、迁移真的建表而不是只证明 Python 能导入 Alembic。一次“字段为什么变空”的事故假设周一抓到的职位描述有 800 字周二数据库中的描述只剩一句“登录后查看”。问题可能来自页面改版、登录失效、选择器选错、服务端灰度也可能是源站真的改了内容。只保存最终JobItem时这几种原因看起来完全一样保存下载时的原始响应后我们可以在不再次访问源站的情况下重放解析器确认当时服务器究竟返回了什么。因此快照不是备份数据库的替代品。数据库保存“当前可查询事实”快照保存“产生事实的输入证据”。前者适合分页、统计和 API后者适合调试解析器、审计内容变化、构造回归 Fixture。两份数据目的不同保留周期也不应一刀切。JobRadar 的快照合同FileSnapshotStore.save(url, body, headers)接收最终响应 URL、原始字节和响应头返回SnapshotReference。快照标识由 URL 与正文共同计算元数据另存 JSON记录 URL、白名单响应头、抓取时间和正文 SHA-256。这里刻意不保存请求 Cookie、Authorization也不把全部响应头照单全收防止调试证据变成凭据仓库。读取时使用load(snapshot_id)返回主体与类型化元数据。调用者不需要知道磁盘目录结构第 27 篇切换到 MinIO 时流水线仍然只依赖相同的save合同。MinioSnapshotStore用同一个快照标识作为对象名前缀先确保 bucket 存在再上传.html与.json两个对象。从线上失败回到离线解析先运行本篇检查点cd project.\.venv\Scripts\python.exe-m pytest tests\test_snapshots.py::test_file_snapshot_round_trip_preserves_body_and_safe_metadata-q然后可以用保存结果离线重放。注意parse_detail要接收原 URL因为相对链接、来源身份和报错定位都依赖它frompathlibimportPathfromjobradar.parsingimportparse_detailfromjobradar.snapshotsimportFileSnapshotStore storeFileSnapshotStore(Path(data/snapshots))storedstore.load(把实际 snapshot_id 填在这里)htmlstored.body.decode(utf-8)jobparse_detail(html,stored.metadata.url)print(job.model_dump_json(indent2))如果 HTML 编码不是 UTF-8不能无条件 decode应依据已验证的 Content-Type 或站点合同处理。课程 TargetLab 固定 UTF-8便于把注意力放在架构上。生产系统还要对快照做访问控制和生命周期管理因为公开页面也可能包含联系方式等个人信息。为什么先保存再解析流水线对有正文的响应先save再parse_detail。这样即使解析失败坏页面也留下证据若反过来最需要调查的响应恰好不会保存。304 没有正文不能创建一个空快照覆盖旧引用它只说明已保存版本仍有效。列表页同样保存因为分页入口改版会让整个任务发现数骤降。哈希命名还带来一个实用性质保存动作是幂等的。进程在写完 body、尚未写 metadata 时崩溃下一次相同正文仍落到同一标识实现用临时文件后原子替换避免读到半截 JSON。文件系统不是无限可靠真正重要的数据仍要依靠对象存储版本、备份和校验策略。保留策略不是“永远保存”可以把快照分成三档解析失败与字段突变的证据保留 90 天正常成功样本保留 14 天被法律或业务标记的审计样本按批准期限保留。清理时必须先判断 jobs 表是否仍引用某快照或者允许引用变为墓碑。课程没有自动删除功能避免把数据生命周期决定偷偷藏进教学代码。本篇完整快照模块阅读时先看两个返回模型再看文件实现最后看 MinIO 实现。它们共享输入输出却不共享内部 IO这就是让本地练习和生产部署可以替换的边界。原始响应快照解析失败后可以脱离网络重放。fromdataclassesimportasdict,dataclassfromdatetimeimportdatetime,timezonefromhashlibimportsha256fromioimportBytesIOimportjsonfrompathlibimportPathfromtypingimportProtocol _SAFE_HEADERS{content-type,etag,last-modified,content-language}dataclass(frozenTrue)classSnapshotMetadata:snapshot_id:strurl:strcaptured_at:strheaders:dict[str,str]body_sha256:strdataclass(frozenTrue)classSnapshotReference:snapshot_id:strbody_path:Path metadata_path:Pathdataclass(frozenTrue)classStoredSnapshot:metadata:SnapshotMetadata body:bytesdataclass(frozenTrue)classObjectSnapshotReference:snapshot_id:strbody_object:strmetadata_object:strclassObjectStorageClient(Protocol):defbucket_exists(self,bucket:str)-bool:...defmake_bucket(self,bucket:str)-object:...defput_object(self,bucket:str,name:str,stream:object,length:int,**kwargs:object)-object:...defget_object(self,bucket:str,name:str)-object:...classFileSnapshotStore:开发环境快照实现文件名只使用服务端生成的哈希。def__init__(self,root:Path)-None:self._rootroot.resolve()self._root.mkdir(parentsTrue,exist_okTrue)defsave(self,url:str,body:bytes,headers:dict[str,str])-SnapshotReference:body_hashsha256(body).hexdigest()snapshot_idsha256(url.encode(utf-8)b\0body).hexdigest()safe_headers{key.casefold():valueforkey,valueinheaders.items()ifkey.casefold()in_SAFE_HEADERS}metadataSnapshotMetadata(snapshot_idsnapshot_id,urlurl,captured_atdatetime.now(timezone.utc).isoformat(),headerssafe_headers,body_sha256body_hash,)body_pathself._root/f{snapshot_id}.htmlmetadata_pathself._root/f{snapshot_id}.jsonifnotbody_path.exists():body_path.write_bytes(body)metadata_path.write_text(json.dumps(asdict(metadata),ensure_asciiFalse,indent2),encodingutf-8,)returnSnapshotReference(snapshot_id,body_path,metadata_path)defload(self,snapshot_id:str)-StoredSnapshot:iflen(snapshot_id)!64orany(chnotin0123456789abcdefforchinsnapshot_id):raiseValueError(snapshot_id must be a lowercase SHA-256 value)body_pathself._root/f{snapshot_id}.htmlmetadata_pathself._root/f{snapshot_id}.jsonbodybody_path.read_bytes()rawjson.loads(metadata_path.read_text(encodingutf-8))metadataSnapshotMetadata(**raw)ifsha256(body).hexdigest()!metadata.body_sha256:raiseValueError(snapshot body hash does not match metadata)returnStoredSnapshot(metadatametadata,bodybody)classMinioSnapshotStore:生产快照实现与文件存储使用相同元数据和哈希合同。def__init__(self,client:ObjectStorageClient,bucket:str)-None:ifnotbucket:raiseValueError(snapshot bucket must not be blank)self._clientclient self._bucketbucketifnotself._client.bucket_exists(bucket):self._client.make_bucket(bucket)defsave(self,url:str,body:bytes,headers:dict[str,str])-ObjectSnapshotReference:body_hashsha256(body).hexdigest()snapshot_idsha256(url.encode(utf-8)b\0body).hexdigest()safe_headers{key.casefold():valueforkey,valueinheaders.items()ifkey.casefold()in_SAFE_HEADERS}metadataSnapshotMetadata(snapshot_idsnapshot_id,urlurl,captured_atdatetime.now(timezone.utc).isoformat(),headerssafe_headers,body_sha256body_hash,)body_objectfsnapshots/{snapshot_id}.htmlmetadata_objectfsnapshots/{snapshot_id}.jsonmetadata_bytesjson.dumps(asdict(metadata),ensure_asciiFalse,indent2).encode(utf-8)self._client.put_object(self._bucket,body_object,BytesIO(body),len(body),content_typetext/html,)self._client.put_object(self._bucket,metadata_object,BytesIO(metadata_bytes),len(metadata_bytes),content_typeapplication/json,)returnObjectSnapshotReference(snapshot_id,body_object,metadata_object)defload(self,snapshot_id:str)-StoredSnapshot:iflen(snapshot_id)!64orany(chnotin0123456789abcdefforchinsnapshot_id):raiseValueError(snapshot_id must be a lowercase SHA-256 value)body_responseself._client.get_object(self._bucket,fsnapshots/{snapshot_id}.html)metadata_responseself._client.get_object(self._bucket,fsnapshots/{snapshot_id}.json)try:bodybody_response.read()# type: ignore[attr-defined]rawjson.loads(metadata_response.read().decode(utf-8))# type: ignore[attr-defined]finally:forresponsein(body_response,metadata_response):response.close()# type: ignore[attr-defined]response.release_conn()# type: ignore[attr-defined]metadataSnapshotMetadata(**raw)ifsha256(body).hexdigest()!metadata.body_sha256:raiseValueError(snapshot body hash does not match metadata)returnStoredSnapshot(metadatametadata,bodybody)本篇课后练习用tests/fixtures/job-detail.html保存快照再读取并交给parse_detail给出完整可运行脚本并断言职位标题。把同一正文保存两次解释为什么 snapshot_id 相同改动一个字符后再次保存确认标识变化。列出哪些请求/响应头可以进入元数据哪些必须排除并说明 Cookie 泄露后可能造成的后果。下一篇会利用已保存的 ETag 做增量采集。