Python+JS自动化批量下载金山文档：从网页分析到并发下载实战 1. 项目缘起一个看似简单却暗藏玄机的需求最近在整理团队过往的项目资料发现大量历史文档都沉淀在金山文档的协作空间里。这些文档格式不一有表格、有文字零零散散加起来有几百个。领导一句话“把这些都下载下来本地备份一份方便归档和离线查阅。” 听起来是个简单的“批量下载”任务对吧我一开始也是这么想的心想着用浏览器的“另存为”或者金山文档自带的导出功能一个个点过去顶多费点时间。但实际操作起来才发现自己太天真了。金山文档的在线编辑体验很流畅但其批量下载功能尤其是针对非表格类文档如金山文字在网页端的支持并不友好。你无法像操作本地文件夹一样按住Shift或Ctrl键多选然后一键下载。更棘手的是当文档数量庞大时手动操作不仅效率低下还极易出错比如漏掉某个版本或者下载了错误的格式。于是这个“体力活”自然而然地转化成了一个技术需求如何自动化、批量化地将金山文档空间里的文件下载到本地这个需求的核心痛点在于“批量”和“自动化”目标是将人力从重复、机械的点击操作中解放出来并确保操作的准确性和一致性。这不仅仅是“下载”更是一个小型的数据归档与迁移工程。2. 技术路径选择为什么是Python JS的组合拳面对这个需求技术选型是第一步。浏览器的自动化专门的下载器还是自己写脚本我评估了几种常见方案方案一纯浏览器自动化工具如Selenium、Playwright这类工具可以模拟真人操作浏览器点击、跳转、等待、下载。理论上能解决所有问题。但缺点也很明显速度慢、资源占用高、稳定性受网页结构变化影响大。对于需要登录、且页面交互复杂的金山文档来说脚本会变得非常臃肿维护成本高。更重要的是下载动作通常需要处理浏览器的原生下载对话框这在不同浏览器和环境下的处理方式各异增加了不确定性。方案二调用官方API最理想的方式。如果金山文档提供了完善的开放API我们可以直接通过API获取文件列表、文件ID并调用下载接口。然而经过一番调研查阅金山办公开放平台文档我发现其API主要面向深度集成和开发对于普通用户批量下载自己空间内的文档权限申请、OAuth认证流程较为复杂且可能有频率限制。对于一次性或偶发性的归档需求显得有些“杀鸡用牛刀”。方案三基于网络请求分析的“轻量爬虫”这是最终选择的方案也是本文重点。其核心思想是直接分析金山文档网页在加载和下载时浏览器向服务器发送了哪些HTTP请求。然后我们用脚本Python去模拟这些请求从而绕过浏览器界面直接获取文件数据。但这里有一个关键前提你需要先获得所有目标文档的链接列表。如何自动获取这个列表呢这就是JavaScriptJS出场的时候了。为什么是Python JS这是一个非常巧妙的组合分工明确JS运行于浏览器控制台负责在已登录的金山文档页面内部执行利用浏览器已有的登录态Cookie、Session获取我们肉眼可见的文档列表数据。因为同源策略浏览器中的JS可以轻松访问当前页面的DOM和网络数据这是外部Python脚本难以直接做到的。Python运行于本地负责“脏活累活”。接收JS收集到的文档链接和元信息然后模拟网络请求进行并发下载、错误重试、文件重命名、本地存储管理等一系列自动化操作。Python在数据处理、网络请求和文件操作方面的库非常强大且易用。简单说JS是“内应”负责在堡垒内部收集情报链接列表Python是“主力部队”根据情报执行批量下载任务。两者结合既利用了浏览器的登录状态又发挥了Python自动化的强大能力。3. 实战第一步用JS在浏览器内部收集文档链接我们的首要任务是拿到所有待下载文档的“门牌号”——也就是它们的唯一访问链接甚至是直接的文件下载链接。我们假设你已经登录了金山文档并进入了包含所有目标文档的某个页面比如“我的文档”首页或者一个共享文件夹。注意以下操作请在金山文档的网页版进行。不同时期金山文档的页面结构可能微调需要灵活应对。3.1 理解金山文档页面的数据加载方式打开浏览器开发者工具F12切换到“网络(Network)”标签页然后刷新或滚动你的金山文档页面。你会看到大量网络请求。其中最关键的是那些返回文档列表数据的请求通常是XHR或Fetch请求响应体是JSON格式。你需要找到那个负责加载文档列表的请求。可以通过过滤XHR/Fetch请求并观察Preview或Response内容来判断。这个请求的URL可能包含list、files、items等关键词。找到它后记录下它的Request URL、Request Method通常是GET以及重要的Request Headers如Authorization,Cookie等。然而直接让Python去模拟这个请求可能比较复杂因为它依赖于当前浏览器的完整登录态。更简单的方法是直接让浏览器里的JS帮我们提取页面上已经渲染出来的链接。3.2 编写并执行文档链接抓取脚本以下是一个增强版的JS脚本你可以在浏览器开发者工具的“控制台(Console)”标签页中直接粘贴运行。它做了几件事获取当前页面所有文档卡片、提取链接和标题、处理滚动加载懒加载、并生成一个便于Python处理的输出。(function() { // 配置要收集的文档类型对应的选择器根据实际页面结构调整 const docItemSelector .docs-list-item, .file-item, [rolelistitem]; // 多个可能的选择器 // 目标域名用于过滤非金山文档的链接如果有 const targetDomain kdocs.cn; let allItems []; let retryCount 0; const maxRetry 5; /** * 滚动页面以触发懒加载 */ function scrollToLoad() { return new Promise((resolve) { const scrollHeight document.documentElement.scrollHeight; const clientHeight document.documentElement.clientHeight; const scrollStep clientHeight * 0.8; let scrolledHeight 0; function scroll() { window.scrollBy(0, scrollStep); scrolledHeight scrollStep; // 等待一小段时间让新内容加载 setTimeout(() { const newScrollHeight document.documentElement.scrollHeight; // 如果还能继续滚动或者页面高度增加了说明有新内容加载 if (scrolledHeight scrollHeight || newScrollHeight scrollHeight) { scroll(); } else { // 如果滚动到底且高度没变化尝试等待再检查一次防止网络延迟 setTimeout(() { const finalScrollHeight document.documentElement.scrollHeight; if (finalScrollHeight scrollHeight) { // 高度又变了继续滚 scrollHeight finalScrollHeight; scroll(); } else { resolve(); } }, 1000); } }, 500); // 滚动后等待时间可根据网络调整 } scroll(); }); } /** * 主收集函数 */ async function collectLinks() { console.log(开始收集文档链接...); // 先滚动加载所有可能的内容 await scrollToLoad(); console.log(页面滚动加载完成。); // 获取所有文档元素 const items document.querySelectorAll(docItemSelector); console.log(当前找到 ${items.length} 个文档元素。); if (items.length 0 retryCount maxRetry) { console.warn(未找到文档元素尝试调整选择器或页面。第${retryCount 1}次重试...); // 可以尝试其他常见选择器 const alternativeSelectors [ div[data-testidfile-list-item], a[href*/l/], .list-item ]; for (let selector of alternativeSelectors) { const altItems document.querySelectorAll(selector); if (altItems.length 0) { console.log(使用备选选择器 ${selector} 找到 ${altItems.length} 个元素。); items altItems; break; } } retryCount; } for (let item of items) { try { // 寻找链接优先找a标签其次找包含onclick或者data-link属性的元素 let linkElement item.querySelector(a); let href ; let title ; if (linkElement linkElement.href) { href linkElement.href; // 提取标题从链接的title属性、内部文本、或者相邻的标题元素中获取 title linkElement.title || linkElement.textContent.trim() || item.querySelector(.title, .name, [data-testidfile-name])?.textContent.trim(); } else { // 如果没有a标签可能链接是通过JS触发的尝试从data属性或onclick中解析 const dataLink item.getAttribute(data-link) || item.getAttribute(data-url); if (dataLink) { href dataLink.startsWith(http) ? dataLink : https://${targetDomain}${dataLink.startsWith(/) ? dataLink : / dataLink}; } else { // 尝试解析onclick事件中的链接常见于SPA应用 const onclickAttr item.getAttribute(onclick); if (onclickAttr onclickAttr.includes(/l/)) { const match onclickAttr.match(/(https:\/\/[^\s]*\/l\/[^\s]*)/); if (match) href match[0]; } } title item.querySelector(.title, .name, .file-name)?.textContent.trim() || item.textContent.trim().split(\n)[0]; } // 过滤和清洗 if (!href || !href.includes(targetDomain) || !href.includes(/l/)) { continue; // 不是目标文档链接 } // 清洗标题移除多余空白和换行 title title.replace(/\s/g, ).trim(); // 生成一个安全的文件名 const safeFileName title.replace(/[:/\\|?*]/g, _).substring(0, 100); // 限制长度 allItems.push({ url: href, title: title, safeName: safeFileName }); } catch (e) { console.error(处理单个元素时出错:, e, item); } } // 去重根据URL const uniqueItems []; const seenUrls new Set(); for (const item of allItems) { // 标准化URL去除可能的查询参数和哈希 const normalizedUrl new URL(item.url).origin new URL(item.url).pathname; if (!seenUrls.has(normalizedUrl)) { seenUrls.add(normalizedUrl); uniqueItems.push(item); } } console.log(收集完成共获得 ${uniqueItems.length} 个唯一文档链接。); // 将结果以JSON格式输出到控制台并复制到剪贴板 const outputJson JSON.stringify(uniqueItems, null, 2); console.log(文档列表JSON:); console.log(outputJson); // 尝试复制到剪贴板需要用户交互这里仅提供提示 navigator.clipboard.writeText(outputJson).then(() { console.log(文档列表已复制到剪贴板。); }).catch(err { console.log(自动复制失败请手动复制上面的JSON数据。); }); return uniqueItems; } // 执行并返回结果 return collectLinks(); })();脚本使用要点与避坑指南选择器是关键docItemSelector变量中的选择器需要根据金山文档的实际页面HTML结构进行调整。如果运行后items.length为0你需要打开开发者工具的“元素(Elements)”面板仔细查看一个文档卡片对应的HTML结构找到其最外层的、具有唯一性的CSS选择器。懒加载处理现代网页大量使用滚动懒加载。脚本中的scrollToLoad函数会模拟滚动到底部触发更多内容加载。等待时间setTimeout中的500ms和1000ms可能需要根据你的网络速度调整。链接提取逻辑脚本尝试了多种方式提取链接a标签、>pip install requests beautifulsoup4 tqdm # 如果需要异步高速下载额外安装 pip install aiohttp aiodns4.2 核心脚本解析同步下载版本我们先实现一个逻辑清晰、易于调试的同步版本。这个版本会按顺序下载文件适合理解整个流程。import os import json import time import requests from urllib.parse import urlparse, unquote from bs4 import BeautifulSoup from tqdm import tqdm import re class KdocsBatchDownloader: def __init__(self, list_json_path, output_dir./downloaded_kdocs): 初始化下载器 :param list_json_path: 从浏览器JS脚本获取的JSON文件路径或JSON字符串 :param output_dir: 文件输出目录 self.output_dir output_dir os.makedirs(self.output_dir, exist_okTrue) # 加载文档列表 if os.path.exists(list_json_path): with open(list_json_path, r, encodingutf-8) as f: self.doc_list json.load(f) else: # 假设传入的是JSON字符串 self.doc_list json.loads(list_json_path) # 配置请求头模拟浏览器 self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } # 初始化一个Session可以保持部分状态如Cookies self.session requests.Session() self.session.headers.update(self.headers) def _extract_download_url(self, doc_url): 核心函数从文档分享页中解析出真实的下载链接。 金山文档的下载链接通常隐藏在页面JS或特定的网络请求中。 这里采用解析HTML和模拟点击的思路。 try: print(f正在解析: {doc_url}) resp self.session.get(doc_url, timeout15) resp.raise_for_status() soup BeautifulSoup(resp.text, html.parser) # 方法1: 查找包含下载信息的meta标签或JS变量常见于SPA download_url None for script in soup.find_all(script): if script.string and downloadUrl in script.string: # 尝试匹配类似 downloadUrl: https://... match re.search(rdownloadUrl[\]?\s*:\s*[\]([^\])[\], script.string) if match: download_url match.group(1) break # 方法2: 查找“下载”按钮的链接传统页面 if not download_url: download_btn soup.find(a, hrefTrue, textre.compile(r下载|导出|Download, re.I)) if download_btn: download_url download_btn[href] # 处理相对路径 if download_url.startswith(/): parsed_url urlparse(doc_url) download_url f{parsed_url.scheme}://{parsed_url.netloc}{download_url} # 方法3: 直接匹配常见的下载接口模式通过分析网络请求获得 # 例如金山文档的下载接口可能形如/api/v3/documents/{file_id}/export if not download_url: # 尝试从页面URL中提取文件ID file_id_match re.search(r/l/([a-zA-Z0-9]), doc_url) if file_id_match: file_id file_id_match.group(1) # 这是一个假设的接口实际需要抓包分析确认 potential_api_url fhttps://www.kdocs.cn/api/v3/documents/{file_id}/export?typepdf # 假设导出PDF # 可以尝试请求这个API但可能需要额外的认证头 # 这里仅作为思路展示不直接请求 pass if download_url: print(f 找到下载链接: {download_url[:100]}...) return download_url else: print(f 警告未在页面中找到明确的下载链接。) # 可以尝试返回文档的打印页或预览页有时可以直接保存 return None except requests.exceptions.RequestException as e: print(f 请求出错: {e}) return None except Exception as e: print(f 解析过程出错: {e}) return None def _download_file(self, url, file_path): 下载文件到指定路径 try: # 流式下载适合大文件 with self.session.get(url, streamTrue, timeout30) as r: r.raise_for_status() total_size int(r.headers.get(content-length, 0)) with open(file_path, wb) as f: if total_size 0: f.write(r.content) else: # 使用tqdm显示进度条 with tqdm(totaltotal_size, unitB, unit_scaleTrue, descos.path.basename(file_path), leaveFalse) as pbar: for chunk in r.iter_content(chunk_size8192): if chunk: f.write(chunk) pbar.update(len(chunk)) return True except Exception as e: print(f 下载失败: {e}) return False def run_sync(self): 同步执行下载任务 success_count 0 fail_list [] for idx, doc_info in enumerate(self.doc_list, 1): doc_url doc_info.get(url) doc_title doc_info.get(title, fdoc_{idx}) safe_name doc_info.get(safeName, doc_title) if not doc_url: print(f[{idx}/{len(self.doc_list)}] 跳过无有效URL) continue print(f\n[{idx}/{len(self.doc_list)}] 处理: {doc_title}) # 步骤1: 获取真实下载地址 download_url self._extract_download_url(doc_url) if not download_url: print(f 跳过无法获取下载地址。) fail_list.append({title: doc_title, url: doc_url, reason: 无法解析下载链接}) continue # 步骤2: 确定文件扩展名和保存路径 # 从下载链接或Content-Type推断文件类型 parsed_download_url urlparse(download_url) path unquote(parsed_download_url.path) # 尝试从路径中获取扩展名 ext_match re.search(r\.(pdf|docx?|xlsx?|pptx?|txt|md)$, path, re.I) if ext_match: file_ext ext_match.group(1).lower() else: # 默认扩展名可以根据需要修改 file_ext pdf # 假设默认下载为PDF # 构建文件名避免重复 base_filename f{safe_name}.{file_ext} file_path os.path.join(self.output_dir, base_filename) counter 1 while os.path.exists(file_path): base_filename f{safe_name}_{counter}.{file_ext} file_path os.path.join(self.output_dir, base_filename) counter 1 # 步骤3: 执行下载 print(f 开始下载 - {base_filename}) if self._download_file(download_url, file_path): print(f 下载成功: {base_filename}) success_count 1 else: print(f 下载失败: {base_filename}) fail_list.append({title: doc_title, url: download_url, reason: 下载请求失败}) # 礼貌性延迟避免请求过快被封 time.sleep(1) # 总结报告 print(f\n{*50}) print(f下载完成) print(f成功: {success_count} / 总数: {len(self.doc_list)}) if fail_list: print(f失败列表:) for fail in fail_list: print(f - {fail[title]}: {fail[reason]}) print(f文件保存在: {os.path.abspath(self.output_dir)}) # 使用示例 if __name__ __main__: # 方式1: 从文件读取JS脚本输出的JSON downloader KdocsBatchDownloader(./doc_list.json) # 方式2: 直接传入JSON字符串从剪贴板粘贴过来 # json_str [{url: https://..., title: ..., safeName: ...}, ...] # downloader KdocsBatchDownloader(json_str) downloader.run_sync()4.3 核心脚本解析异步高速下载版本当文档数量成百上千时同步下载的等待时间是不可接受的。我们可以使用asyncio和aiohttp进行异步并发下载效率提升十倍不止。import aiohttp import asyncio from aiohttp import ClientTimeout, TCPConnector import aiofiles class KdocsBatchDownloaderAsync(KdocsBatchDownloader): 继承同步下载器重写下载部分为异步 def __init__(self, list_json_path, output_dir./downloaded_kdocs, max_concurrent5): super().__init__(list_json_path, output_dir) self.max_concurrent max_concurrent # 最大并发数 self.semaphore asyncio.Semaphore(max_concurrent) async def _async_download_file(self, session, url, file_path, pbar): 异步下载单个文件 async with self.semaphore: # 控制并发量 try: timeout ClientTimeout(total60, connect30) # 设置超时 async with session.get(url, timeouttimeout) as response: response.raise_for_status() total_size int(response.headers.get(content-length, 0)) async with aiofiles.open(file_path, wb) as f: if total_size 0: content await response.read() await f.write(content) if pbar: pbar.update(len(content)) else: downloaded 0 async for chunk in response.content.iter_chunked(8192): if chunk: await f.write(chunk) downloaded len(chunk) if pbar: pbar.update(len(chunk)) return True, None except Exception as e: return False, str(e) async def _process_single_doc(self, session, doc_info, idx, total, pbar): 异步处理单个文档解析链接并下载 doc_url doc_info.get(url) doc_title doc_info.get(title, fdoc_{idx}) safe_name doc_info.get(safeName, doc_title) if not doc_url: return {success: False, title: doc_title, reason: 无URL} # 解析下载链接这部分目前是同步的可以后续也改为异步但解析通常很快 download_url self._extract_download_url(doc_url) # 注意这里调用了同步方法 if not download_url: return {success: False, title: doc_title, reason: 无法解析下载链接} # 确定文件名 parsed_url urlparse(download_url) path unquote(parsed_url.path) ext_match re.search(r\.(pdf|docx?|xlsx?|pptx?|txt|md)$, path, re.I) file_ext ext_match.group(1).lower() if ext_match else pdf base_filename f{safe_name}.{file_ext} file_path os.path.join(self.output_dir, base_filename) counter 1 while os.path.exists(file_path): base_filename f{safe_name}_{counter}.{file_ext} file_path os.path.join(self.output_dir, base_filename) counter 1 # 异步下载 success, error_msg await self._async_download_file(session, download_url, file_path, pbar) if success: return {success: True, title: doc_title, file: base_filename} else: return {success: False, title: doc_title, reason: f下载失败: {error_msg}} async def run_async(self): 异步执行主函数 connector TCPConnector(limitself.max_concurrent, sslFalse) # 限制总连接数 timeout ClientTimeout(total300) # 总超时时间 async with aiohttp.ClientSession(headersself.headers, connectorconnector, timeouttimeout) as session: tasks [] results [] fail_list [] success_count 0 print(f开始异步下载 {len(self.doc_list)} 个文档并发数: {self.max_concurrent}) # 创建总进度条 with tqdm(totallen(self.doc_list), desc总进度) as pbar_total: # 为每个文档创建处理任务 for idx, doc_info in enumerate(self.doc_list, 1): task asyncio.create_task(self._process_single_doc(session, doc_info, idx, len(self.doc_list), pbar_total)) tasks.append(task) # 等待所有任务完成并收集结果 for task in asyncio.as_completed(tasks): result await task results.append(result) if result[success]: success_count 1 else: fail_list.append(result) # 输出报告 print(f\n{*50}) print(f异步下载完成) print(f成功: {success_count} / 总数: {len(self.doc_list)}) if fail_list: print(f失败列表:) for fail in fail_list: print(f - {fail[title]}: {fail[reason]}) # 异步使用示例 async def main_async(): downloader KdocsBatchDownloaderAsync(./doc_list.json, max_concurrent10) await downloader.run_async() if __name__ __main__: # 运行异步版本 asyncio.run(main_async())4.4 关键环节的深度解析与避坑1. 下载链接解析的“黑盒”挑战_extract_download_url函数是整个脚本最脆弱的部分。金山文档的前端技术栈可能变化下载按钮的定位方式、真实下载地址的隐藏位置都可能不同。上述脚本提供了三种解析思路分析JS变量最有效。在页面HTML的script标签里搜索downloadUrl、fileUrl、exportUrl等关键词用正则表达式提取。这需要你仔细查看页面源码。模拟点击按钮较通用。找到“下载”或“导出”按钮的a标签或button获取其href属性或onclick事件里的URL。但按钮可能被动态生成。网络请求抓包最可靠。在开发者工具的“网络(Network)”面板手动点击一个文档的下载按钮观察哪个请求最终返回了文件流Content-Type是application/pdf、application/octet-stream等。然后让Python脚本直接模拟这个请求。这需要分析请求的URL、Headers尤其是Authorization、Referer等和可能的请求体。提示如果遇到无法解析的情况一个退而求其次的方案是直接请求文档的“打印页”或“预览页”然后保存为PDF。很多浏览器的“打印”功能可以生成PDF。但这需要更复杂的模拟如使用pyppeteer或playwright控制无头浏览器超出了本文“轻量”的范畴。2. 会话(Session)与请求头(Headers)的重要性使用requests.Session()或aiohttp.ClientSession()可以自动管理Cookies在连续请求中保持登录状态。此外务必设置合理的User-Agent、Accept等请求头让服务器认为请求来自真实的浏览器降低被反爬机制拦截的风险。3. 并发控制与礼貌延迟即使是异步版本也通过Semaphore限制了最大并发数max_concurrent。过高的并发请求会对服务器造成压力可能导致IP被暂时限制。在同步版本的循环中加入了time.sleep(1)也是出于“礼貌”的考虑。对于公开服务建议将并发数设置在5-10延迟设置在0.5-1秒。4. 文件名处理与重复规避从网页提取的标题可能包含Windows/Linux文件名禁止的字符如\/:*?|。脚本中的safeName生成逻辑和_download_file方法里的文件名清洗replace(/[:/\\|?*]/g, _)至关重要。同时检查本地是否已存在同名文件并自动添加后缀_1,_2可以避免文件被意外覆盖。5. 错误处理与日志记录脚本中对网络请求、解析、下载等各个环节都进行了try...except捕获。将失败的任务记录到fail_list并在最后统一输出方便后续手动重试或排查问题。在生产环境中可以考虑将日志写入文件而不是仅仅打印到控制台。5. 进阶策略与疑难排错即使有了上面的脚本在实际操作中你仍可能遇到各种问题。这里分享一些进阶策略和常见问题的排查思路。问题1JS脚本无法获取到文档列表items.length始终为0。原因页面结构已更新选择器失效。解决打开开发者工具使用元素选择器CtrlShiftC点击一个文档查看其HTML结构。找到能唯一标识文档列表项的最外层元素观察其class或>