Get笔记API + Python脚本：如何自动化处理2W+公众号文章，实现批量摘要与导出

Get笔记API Python脚本自动化处理2W公众号文章的技术实践每天被海量公众号文章淹没手动整理2万篇文章就像用勺子舀干大海。作为经历过这种痛苦的技术从业者我想分享一套完整的自动化解决方案——通过Get笔记API和Python脚本实现批量摘要与导出的全流程自动化。1. 环境准备与基础配置在开始编写自动化脚本前需要完成几个基础准备工作。首先确保你的开发环境已经安装Python 3.8版本这是大多数现代API库支持的最低版本要求。核心依赖库安装pip install requests beautifulsoup4 python-dotenv pandas tqdm这些库将分别用于requests处理HTTP请求beautifulsoup4解析HTML内容python-dotenv管理环境变量pandas数据处理tqdm进度条显示提示建议使用虚拟环境管理项目依赖避免与其他项目产生冲突创建一个.env文件来存储敏感信息GET_NOTES_API_KEYyour_api_key_here GET_NOTES_USER_IDyour_user_id EXPORT_DIR./exports2. Get笔记API深度解析与封装Get笔记的API文档并不完全公开但通过开发者工具可以捕获到核心接口。以下是经过实战验证的几个关键端点2.1 认证与令牌管理Get笔记使用Bearer Token进行认证有效期为7天。我们需要一个自动刷新令牌的机制import os from dotenv import load_dotenv import requests from datetime import datetime, timedelta class GetNotesAuth: def __init__(self): load_dotenv() self.token None self.token_expiry None def get_token(self): if self.token and datetime.now() self.token_expiry: return self.token auth_url https://get-notes.luojilab.com/auth/v2/login payload { username: os.getenv(GET_NOTES_USERNAME), password: os.getenv(GET_NOTES_PASSWORD) } response requests.post(auth_url, jsonpayload) if response.status_code 200: self.token response.json().get(access_token) self.token_expiry datetime.now() timedelta(days6) # 提前1天刷新 return self.token else: raise Exception(f认证失败: {response.status_code})2.2 文章提交与摘要生成核心的摘要生成API需要特别注意请求频率限制def submit_article(link, token): url https://get-notes.luojilab.com/voicenotes/web/notes/stream headers { Authorization: fBearer {token}, Content-Type: application/json } payload { attachments: [{ type: link, url: link }], entry_type: ai, note_type: link } try: response requests.post(url, headersheaders, jsonpayload, timeout10) if response.status_code 200: return response.json().get(id) # 返回笔记ID elif response.status_code 429: raise Exception(API调用过于频繁请降低请求速率) else: raise Exception(fAPI错误: {response.status_code}) except requests.exceptions.Timeout: raise Exception(请求超时请检查网络连接)3. 批量处理架构设计处理2万文章需要精心设计任务队列和错误处理机制。以下是经过实战验证的架构方案3.1 任务队列实现使用CSV文件作为任务队列包含以下字段article_urlstatus (pending/processing/completed/failed)retry_countlast_processedimport pandas as pd from pathlib import Path class ArticleQueue: def __init__(self, queue_filearticle_queue.csv): self.queue_file Path(queue_file) if not self.queue_file.exists(): pd.DataFrame(columns[ article_url, status, retry_count, last_processed ]).to_csv(self.queue_file, indexFalse) def add_articles(self, urls): df pd.read_csv(self.queue_file) new_urls set(urls) - set(df[article_url]) if new_urls: new_df pd.DataFrame({ article_url: list(new_urls), status: pending, retry_count: 0, last_processed: None }) pd.concat([df, new_df]).to_csv(self.queue_file, indexFalse) def get_next_batch(self, batch_size50): df pd.read_csv(self.queue_file) pending df[df[status].isin([pending, failed])] return pending.head(batch_size)[article_url].tolist()3.2 容错与重试机制from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def process_article_with_retry(url, auth): token auth.get_token() try: note_id submit_article(url, token) return note_id except Exception as e: print(f处理文章失败: {url} - {str(e)}) raise4. 导出与后处理流程Get笔记的导出功能有一些特殊限制需要特别注意4.1 批量导出策略导出类型文件格式限制条件适用场景单篇导出PDF/DOC无数量限制少量精选文章批量导出HTML压缩包每次最多500篇大规模归档API导出JSON需自定义开发结构化数据处理def export_notes(note_ids, export_formathtml): export_url https://get-notes.luojilab.com/voicenotes/web/export/batch headers {Authorization: fBearer {token}} payload { note_ids: note_ids, format: export_format, export_type: batch } response requests.post(export_url, headersheaders, jsonpayload) if response.status_code 202: task_id response.json().get(task_id) return monitor_export_task(task_id) else: raise Exception(f导出请求失败: {response.status_code}) def monitor_export_task(task_id, interval30, timeout3600): status_url fhttps://get-notes.luojilab.com/voicenotes/web/export/tasks/{task_id} start_time time.time() while time.time() - start_time timeout: response requests.get(status_url, headersheaders) status response.json().get(status) if status completed: return response.json().get(download_url) elif status failed: raise Exception(导出任务失败) time.sleep(interval) raise Exception(导出任务超时)4.2 HTML到PDF的转换使用wkhtmltopdf进行高质量转换# 先安装wkhtmltopdf sudo apt-get install wkhtmltopdf对应的Python封装import subprocess from pathlib import Path def convert_html_to_pdf(html_dir, output_dir): output_dir Path(output_dir) output_dir.mkdir(exist_okTrue) for html_file in Path(html_dir).glob(*.html): pdf_file output_dir / f{html_file.stem}.pdf cmd [ wkhtmltopdf, --encoding, utf-8, --quiet, str(html_file), str(pdf_file) ] subprocess.run(cmd, checkTrue)5. 性能优化与实战技巧处理海量数据时以下几个技巧可以显著提升效率5.1 并发处理使用线程池控制并发数from concurrent.futures import ThreadPoolExecutor, as_completed def process_batch_concurrently(urls, max_workers5): auth GetNotesAuth() results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures { executor.submit(process_article_with_retry, url, auth): url for url in urls } for future in as_completed(futures): url futures[future] try: note_id future.result() results.append((url, note_id)) except Exception as e: print(f最终处理失败: {url} - {str(e)}) return results5.2 速率限制策略Get笔记API的速率限制大约为认证API每分钟最多10次文章提交API每分钟最多30次导出API每分钟最多5次实现一个简单的速率限制器import time from collections import defaultdict class RateLimiter: def __init__(self, max_calls, period): self.max_calls max_calls self.period period self.calls defaultdict(list) def __call__(self, func): def wrapped(*args, **kwargs): now time.time() func_name func.__name__ # 清除过期记录 self.calls[func_name] [ t for t in self.calls[func_name] if now - t self.period ] if len(self.calls[func_name]) self.max_calls: sleep_time self.period - (now - self.calls[func_name][0]) time.sleep(sleep_time) result func(*args, **kwargs) self.calls[func_name].append(time.time()) return result return wrapped # 使用示例 RateLimiter(max_calls25, period60) def submit_article_limited(url, token): return submit_article(url, token)在实际项目中处理2万篇文章的完整流程大约需要8-12小时主要时间花费在摘要生成和导出转换阶段。建议在夜间运行完整流程白天进行小批量测试和调试。