
简介WebDetector网站下载资源探测器源代码是一份面向网站管理员、SEO优化人员及爬虫开发者的实用工具源码核心功能是输入网站URL后自动扫描并提取页面中的全部下载链接整理为XML格式保存至本地便于快速掌握站点资源分布。资源包共108个文件约510KB以C#源码cs为主体辅以xml配置与数据文件、dll类库、resx资源文件、csproj工程文件及少量exe可执行程序构成一套完整的Visual Studio解决方案可直接编译运行或二次开发。程序采用多线程技术并发处理多个网站的探测任务在网络请求等I/O密集场景下能显著缩短整体执行时间XML输出则兼顾可读性与易解析性。目前已有318人学习下载。由于缺少说明文档初次上手需结合源码理解配置与调用逻辑但这也为开发者提供了研究网页抓取、链接提取、线程管理与数据存储的实战样本适合具备一定编程基础、希望提升相关技能的学习者参考。1. WebDetector 网站下载资源探测器从页面里把可下载的东西捞干净做数据采集、竞品监控或者资源归档的同行大概都遇到过这种局面目标站点没有开放 API也没有规整的下载列表下载入口散落在正文链接、按钮的onclick、动态渲染的卡片里甚至藏在一段 Base64 字符串里。手动点一遍能拿到量一上来就彻底崩。WebDetector 这类「网站下载资源探测器」要解决的就是这件事——给它一个页面地址它把页面里所有可能指向可下载文件的线索抽出来去重、分类、验证最后给你一份能直接用的清单。它适合三类人做批量资源归档的运维、做竞品素材收集的运营开发、以及写爬虫时被「下载链接到底在哪」反复折磨的工程师。核心不是「爬」而是「识别」——识别哪些 URL 是资源、哪些是导航、哪些是伪装成资源的陷阱。下面按「先立住原理、再动手复现、最后讲坑」的顺序拆开讲代码用 Python 写依赖尽量少方便你直接抄。2. 探测器到底在探测什么资源识别的四层信号2.1 从「链接」到「资源」判定标准先定死很多人一上来就写正则抓.zip|.rar|.pdf跑两个站就发现漏一大半。原因是现代站点的下载链接经常不带扩展名比如/download?id1024、/file/abc123甚至是一个 POST 接口。所以判定不能只看后缀要分层。我一般把信号分成四层优先级从高到低层级信号来源典型特征可信度L1HTTP 响应头Content-Disposition: attachment、Content-Type: application/octet-stream最高L2URL 路径与查询含download、file、attach、export等词或带id/fid参数高L3页面语义a download、按钮文案「下载」「导出」、>import re, json, requests from urllib.parse import urljoin HEADERS { User-Agent: Mozilla/5.0 (compatible; ResourceProbe/1.0), Accept: text/html,application/xhtmlxml, } def fetch_html(url, timeout10): resp requests.get(url, headersHEADERS, timeouttimeout) resp.raise_for_status() # 部分站点不声明编码手动兜底避免中文乱码影响后续正则 if not resp.encoding or resp.encoding.lower() iso-8859-1: resp.encoding resp.apparent_encoding return resp.text def extract_embedded_json(html): 抽取页面内嵌 JSONSPA 首屏数据常在这里 results [] # 匹配 script typeapplication/json.../script for m in re.finditer(rscript[^]*type[\]application/json[\][^]*(.*?)/script, html, re.S | re.I): try: results.append(json.loads(m.group(1).strip())) except json.JSONDecodeError: continue # 匹配 window.__INITIAL_STATE__ {...}; for m in re.finditer(rwindow\.__[A-Z_]__\s*\s*(\{.*?\});, html, re.S): try: results.append(json.loads(m.group(1))) except json.JSONDecodeError: continue return results这段代码做了两件事fetch_html负责稳定拿到文本并修正编码extract_embedded_json负责把内嵌数据挖出来。参数上timeout别设太大10 秒足够设 30 秒只会让批量任务卡死apparent_encoding依赖chardet装requests时一般会带上没有就pip install chardet。注意正则里的.*?配re.S是贪婪控制的常见写法遇到嵌套大括号会截断所以解析失败要continue而不是抛异常——内嵌 JSON 格式千奇百怪容错比精确更重要。2.3 候选链接的抽取与归一化拿到 HTML 或 JSON 后要把所有可能是资源的 URL 抽出来。这里的关键是「归一化」相对路径转绝对、去重复、去掉锚点和无关查询参数。很多站点的同一个文件会有?utm_sourcexxx这种跟踪参数不去掉就会重复计数。from urllib.parse import urljoin, urlparse, urlunparse, parse_qsl, urlencode RESOURCE_HINTS re.compile( r(download|file|attach|export|resource|media|asset|getfile), re.I) STATIC_EXT re.compile( r\.(zip|rar|7z|tar|gz|pdf|docx?|xlsx?|pptx?|csv|mp4|mp3|png|jpe?g|apk|exe)(\?|$), re.I) def normalize(url, base): 相对转绝对去掉 fragment 和跟踪参数 absolute urljoin(base, url.strip()) parsed urlparse(absolute) # 只保留有意义的查询参数剔除 utm_/from/spm 等跟踪项 kept [(k, v) for k, v in parse_qsl(parsed.query) if not k.lower().startswith((utm_, spm, from, ref))] clean parsed._replace(queryurlencode(kept), fragment) return urlunparse(clean) def collect_candidates(html, base): candidates set() # 1. 常规 href / src for m in re.finditer(r(?:href|src|data-href|data-url)[\]([^\])[\], html, re.I): candidates.add(m.group(1)) # 2. 内联 JS 里的字符串 URL for m in re.finditer(r[\](https?://[^\]|\/[^\])[\], html): candidates.add(m.group(1)) # 3. 过滤 归一化 result [] for c in candidates: if c.startswith((javascript:, #, mailto:)): continue full normalize(c, base) if RESOURCE_HINTS.search(full) or STATIC_EXT.search(full): result.append(full) return sorted(set(result))RESOURCE_HINTS和STATIC_EXT是两个过滤器前者抓语义后者抓后缀命中任一即保留。normalize里剔除utm_、spm、from、ref这几类参数是血泪经验——不剔的话一个文件在清单里出现七八次去重逻辑直接失效。collect_candidates里第二步用宽松正则扫 JS 字符串会引入一些噪声但配合后面的验证能过滤掉宁可多捞不可漏捞。注意>def verify(url, timeout8): HEAD 验证返回 (是否资源, 元信息) try: resp requests.head(url, headersHEADERS, timeouttimeout, allow_redirectsTrue) except requests.RequestException as e: return False, {error: str(e)} ctype resp.headers.get(Content-Type, ).lower() disp resp.headers.get(Content-Disposition, ).lower() size resp.headers.get(Content-Length) is_resource ( resp.status_code 200 and (attachment in disp or not ctype.startswith(text/html)) ) return is_resource, { status: resp.status_code, content_type: ctype, size: int(size) if size and size.isdigit() else None, final_url: resp.url, }allow_redirectsTrue很重要很多下载链接是 302 跳转到 CDN 的不跟随就永远拿不到真实类型。Content-Disposition里带attachment是最强信号直接判定为资源。Content-Length可能为空分块传输所以size允许为None别在这里做强制校验。超时设 8 秒比抓取短因为验证是批量并发跑的单个卡住会拖垮整体。3.2 并发验证与限速别把人家站点打挂候选集动辄几百上千条串行验证慢到无法接受必须并发。但并发不等于无脑开线程要限速、要重试、要控制并发数。我一般用concurrent.futures配一个信号量把并发压在 8 到 16 之间同时对同一域名做间隔控制。import time, threading from concurrent.futures import ThreadPoolExecutor, as_completed class RateLimiter: 按域名限速最小间隔 min_interval 秒 def __init__(self, min_interval0.3): self.min_interval min_interval self._last {} self._lock threading.Lock() def wait(self, domain): with self._lock: now time.time() last self._last.get(domain, 0) delta now - last if delta self.min_interval: time.sleep(self.min_interval - delta) self._last[domain] time.time() def batch_verify(urls, workers12, min_interval0.3): limiter RateLimiter(min_interval) results {} def task(u): limiter.wait(urlparse(u).netloc) ok, meta verify(u) return u, ok, meta with ThreadPoolExecutor(max_workersworkers) as pool: futures [pool.submit(task, u) for u in urls] for fut in as_completed(futures): u, ok, meta fut.result() results[u] {is_resource: ok, **meta} return resultsRateLimiter用字典记录每个域名上次请求时间wait里加锁保证多线程下计时准确。min_interval0.3是保守值对大多数站点够礼貌如果目标站有明确的Crawl-delay按它来。workers12是经验值再高容易触发风控再低吞吐不够。注意as_completed是无序返回的如果你需要保持顺序得在结果里带原始索引别依赖字典顺序。3.3 分类与输出让清单能直接用验证完的清单要分类否则用户拿到一堆 URL 还是不知道哪个是哪个。分类维度我一般用三个资源类型文档/压缩包/媒体/安装包、来源层级L1 响应头判定 / L2 路径判定、大小区间。输出成 JSON 和 CSV 两份JSON 给程序用CSV 给人看。import csv, json TYPE_MAP { zip: archive, rar: archive, 7z: archive, tar: archive, gz: archive, pdf: document, doc: document, docx: document, xls: document, xlsx: document, ppt: document, pptx: document, csv: document, mp4: media, mp3: media, png: image, jpg: image, jpeg: image, apk: package, exe: package, } def classify(url, meta): path urlparse(url).path.lower() ext path.rsplit(., 1)[-1] if . in path else rtype TYPE_MAP.get(ext, unknown) size meta.get(size) if size is None: bucket unknown elif size 1024 * 1024: bucket small elif size 100 * 1024 * 1024: bucket medium else: bucket large return {type: rtype, size_bucket: bucket} def export(results, json_pathresources.json, csv_pathresources.csv): rows [] for url, meta in results.items(): if not meta.get(is_resource): continue row {url: url, **classify(url, meta), content_type: meta.get(content_type), size: meta.get(size)} rows.append(row) with open(json_path, w, encodingutf-8) as f: json.dump(rows, f, ensure_asciiFalse, indent2) with open(csv_path, w, encodingutf-8, newline) as f: writer csv.DictWriter(f, fieldnames[url, type, size_bucket, content_type, size]) writer.writeheader() writer.writerows(rows) return len(rows)TYPE_MAP覆盖常见格式没命中的归unknown别硬猜。size_bucket分三档是为了让用户快速筛掉超大文件或锁定小文件。export里只导出is_resourceTrue的验证失败的留在内存里供排查不污染最终清单。CSV 用newline是 Windows 下避免空行的标准写法跨平台跑不会出问题。4. 避坑与排查探测器跑不通时先看这几条4.1 现象候选集为空但浏览器里明明有下载按钮原因通常是页面动态渲染requests拿到的 HTML 里没有链接。解决顺序先查内嵌 JSONextract_embedded_json再抓 XHR 接口最后才上无头浏览器。别一上来就开浏览器慢且容易被识别。如果内嵌 JSON 里也没有打开开发者工具的 Network 面板找返回数据里带下载地址的那个接口直接请求它。4.2 现象验证阶段大量超时或 403原因多半是请求头太「裸」或者并发太高触发风控。解决补全User-Agent、Referer、Accept-Language把workers降到 4 到 6min_interval提到 1 秒以上。如果目标站校验Referer把来源页地址带上。注意别用假 UA 池乱轮换很多站点对 UA 与行为不一致的请求更敏感稳定用一个真实浏览器 UA 反而更安全。4.3 现象同一个文件在清单里出现多次原因是 URL 带了跟踪参数或大小写不一致。解决normalize里剔除utm_、spm、from、ref等参数同时对路径部分做小写归一查询参数保留原样因为有些站点参数区分大小写。去重时用归一化后的 URL 做 key别用原始 URL。4.4 现象HEAD 返回 200 但内容是 HTML 登录页原因是站点对未登录请求返回登录页而非 403。解决在verify里加一层判断如果Content-Type是text/html且响应体里含「登录」「login」等关键词标记为need_auth而非资源。需要登录的场景得先维护会话requests.Session带 Cookie这一步没有通用解按站点来。4.5 现象大文件验证很慢拖垮整体原因是 HEAD 虽然不传 body但部分服务器对 HEAD 支持差会退化成 GET。解决给verify加streamTrue的 GET 兜底只读响应头就关闭连接不下载 body。同时给单个验证设硬超时超时直接标记失败别让它无限等。5. 进阶把探测器做成可复用的采集管线单次探测只是起点真正省事的是把它接成管线输入一批页面 URL输出结构化资源清单中间带缓存和增量。缓存用 URL 的哈希做 key存验证结果避免重复请求增量靠对比上次清单的差异只报新增和失效。import hashlib, os, json CACHE_DIR .probe_cache os.makedirs(CACHE_DIR, exist_okTrue) def cache_key(url): return hashlib.md5(url.encode()).hexdigest() def cached_verify(url): key cache_key(url) path os.path.join(CACHE_DIR, key .json) if os.path.exists(path): with open(path, encodingutf-8) as f: return json.load(f) ok, meta verify(url) data {is_resource: ok, **meta} with open(path, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse) return data def diff(prev_rows, curr_rows): prev {r[url]: r for r in prev_rows} curr {r[url]: r for r in curr_rows} added [u for u in curr if u not in prev] removed [u for u in prev if u not in curr] return {added: added, removed: removed}cached_verify用 MD5 做文件名简单够用缓存没有过期机制适合短期批量任务长期跑要加 TTL。diff对比两次清单输出新增和失效配合定时任务就是一套资源监控。这里有个我踩过的坑缓存目录别放在项目根目录下跟源码混一起.probe_cache这种隐藏目录加进.gitignore不然提交时一堆哈希文件。验证方法上我习惯拿一个已知结构的页面做基准跑完对比人工点出来的清单看漏了哪些、多了哪些。漏的通常是动态加载或内嵌 JSON 里的多的通常是导航页和登录跳转。把这两类误差压到 5% 以内这套探测器就算能用了。最后说个习惯每次改抽取规则先在小样本上跑别直接上全量。我早期图快改完正则直接跑几千个页面结果一个贪婪匹配把整页 HTML 吞进去验证阶段全超时白等半小时。现在固定先跑 20 个页面看候选集和验证结果确认没问题再放量。希望帮到你。本文还有配套的精品资源点击获取