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_KEY=your_api_key_here GET_NOTES_USER_ID=your_user_id EXPORT_DIR=./exports

2. 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, json=payload) if response.status_code == 200: self.token = response.json().get("access_token") self.token_expiry = datetime.now() + timedelta(days=6) # 提前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": f"Bearer {token}", "Content-Type": "application/json" } payload = { "attachments": [{ "type": "link", "url": link }], "entry_type": "ai", "note_type": "link" } try: response = requests.post(url, headers=headers, json=payload, timeout=10) if response.status_code == 200: return response.json().get("id") # 返回笔记ID elif response.status_code == 429: raise Exception("API调用过于频繁，请降低请求速率") else: raise Exception(f"API错误: {response.status_code}") except requests.exceptions.Timeout: raise Exception("请求超时，请检查网络连接")

3. 批量处理架构设计

处理2万+文章需要精心设计任务队列和错误处理机制。以下是经过实战验证的架构方案：

3.1 任务队列实现

使用CSV文件作为任务队列，包含以下字段：

• article_url
• status (pending/processing/completed/failed)
• retry_count
• last_processed

import pandas as pd from pathlib import Path class ArticleQueue: def __init__(self, queue_file="article_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, index=False) 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, index=False) def get_next_batch(self, batch_size=50): 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(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) 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)}") raise

4. 导出与后处理流程

Get笔记的导出功能有一些特殊限制，需要特别注意：

4.1 批量导出策略

def export_notes(note_ids, export_format="html"): export_url = "https://get-notes.luojilab.com/voicenotes/web/export/batch" headers = {"Authorization": f"Bearer {token}"} payload = { "note_ids": note_ids, "format": export_format, "export_type": "batch" } response = requests.post(export_url, headers=headers, json=payload) 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, interval=30, timeout=3600): status_url = f"https://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, headers=headers) 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_ok=True) 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, check=True)

5. 性能优化与实战技巧

处理海量数据时，以下几个技巧可以显著提升效率：

5.1 并发处理

使用线程池控制并发数：

from concurrent.futures import ThreadPoolExecutor, as_completed def process_batch_concurrently(urls, max_workers=5): auth = GetNotesAuth() results = [] with ThreadPoolExecutor(max_workers=max_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 results

5.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_calls=25, period=60) def submit_article_limited(url, token): return submit_article(url, token)

在实际项目中，处理2万篇文章的完整流程大约需要8-12小时，主要时间花费在摘要生成和导出转换阶段。建议在夜间运行完整流程，白天进行小批量测试和调试。