n8n Python 代码节点常用模式指南

# 常用模式 - Python 代码节点 经过生产环境测试的 n8n Python 代码节点模式。 --- ## ⚠️ 重要提示：首选 JavaScript **95% 的用例应优先使用 JavaScript。** n8n 中的 Python **没有外部库**（没有 requests, pandas, numpy）。 仅在以下情况使用 Python： - 你有复杂的 Python 特定逻辑 - 你需要 Python 标准库的功能 - 你对 Python 的熟悉程度远高于 JavaScript 对于大多数工作流，**JavaScript 是更好的选择**。 --- ## 模式概览 这 10 种模式涵盖了使用 Python 的常见 n8n 代码节点场景： 1. **多源数据聚合** - 合并来自多个节点的数据 2. **基于正则的过滤** - 使用模式匹配过滤项目 3. **Markdown 转结构化数据** - 将 markdown 解析为结构化格式 4. **JSON 对象比较** - 比较两个 JSON 对象的差异 5. **CRM 数据转换** - 将 CRM 数据转换为标准格式 6. **发布日志处理** - 解析并分类发布日志 7. **数组转换** - 重塑数组并提取字段 8. **字典查找** - 创建并使用查找字典 9. **Top N 过滤** - 按分数/数值获取前几名项目 10. **字符串聚合** - 聚合字符串并进行格式化 --- ## 模式 1：多源数据聚合 **使用场景**：合并来自多个源（API、Webhook、数据库）的数据。 **具体情境**：聚合来自多个源的新闻文章。 ### 代码实现 ```python from datetime import datetime all_items = _input.all() processed_articles = [] for item in all_items: source_name = item["json"].get("name", "Unknown") source_data = item["json"] # 处理 Hacker News 源 if source_name == "Hacker News" and source_data.get("hits"): for hit in source_data["hits"]: processed_articles.append({ "title": hit.get("title", "No title"), "url": hit.get("url", ""), "summary": hit.get("story_text") or "No summary", "source": "Hacker News", "score": hit.get("points", 0), "fetched_at": datetime.now().isoformat() }) # 处理 Reddit 源 elif source_name == "Reddit" and source_data.get("data"): for post in source_data["data"].get("children", []): post_data = post.get("data", {}) processed_articles.append({ "title": post_data.get("title", "No title"), "url": post_data.get("url", ""), "summary": post_data.get("selftext", "")[:200], "source": "Reddit", "score": post_data.get("score", 0), "fetched_at": datetime.now().isoformat() }) # 按分数降序排序 processed_articles.sort(key=lambda x: x["score"], reverse=True) # 以 n8n 项目格式返回 return [{"json": article} for article in processed_articles] ``` ### 核心技术 - 在一个循环中处理多个数据源 - 标准化不同的数据结构 - 使用 datetime 处理时间戳 - 按标准排序 - 返回格式正确的项目 --- ## 模式 2：基于正则表达式的过滤 **使用场景**：根据文本字段中的模式匹配过滤项目。 **具体情境**：按优先级关键词过滤支持工单。 ### 代码实现 ```python import re all_items = _input.all() priority_tickets = [] # 高优先级关键词模式 high_priority_pattern = re.compile( r'b(urgent|critical|emergency|asap|down|outage|broken)b', re.IGNORECASE ) for item in all_items: ticket = item["json"] # 检查主题和描述 subject = ticket.get("subject", "") description = ticket.get("description", "") combined_text = f"{subject} {description}" # 查找匹配项 matches = high_priority_pattern.findall(combined_text) if matches: priority_tickets.append({ "json": { **ticket, "priority": "high", "matched_keywords": list(set(matches)), "keyword_count": len(matches) } }) else: priority_tickets.append({ "json": { **ticket, "priority": "normal", "matched_keywords": [], "keyword_count": 0 } }) # 按关键词数量排序（最紧急的排在前面） priority_tickets.sort(key=lambda x: x["json"]["keyword_count"], reverse=True) return priority_tickets ``` ### 核心技术 - 使用 `re.compile()` 实现模式复用 - `re.IGNORECASE` 实现不区分大小写的匹配 - 合并多个文本字段进行搜索 - 提取并去重匹配项 - 按优先级指标排序 --- ## 模式 3：Markdown 转结构化数据 **使用场景**：将 Markdown 文本解析为结构化数据。 **具体情境**：从 Markdown 检查列表中提取任务。 ### 代码实现 ```python import re markdown_text = _input.first()["json"]["body"].get("markdown", "") # 解析 Markdown 检查列表 tasks = [] lines = markdown_text.split("n") for line in lines: # 匹配: - [ ] 任务 或 - [x] 任务 match = re.match(r'^s*-s*[([ x])]s*(.+)$', line, re.IGNORECASE) if match: checked = match.group(1).lower() == 'x' task_text = match.group(2).strip() # 如果存在优先级则提取 (例如 [P1], [HIGH]) priority_match = re.search(r'[(Pd|HIGH|MEDIUM|LOW)]', task_text, re.IGNORECASE) priority = priority_match.group(1).upper() if priority_match else "NORMAL" # 从文本中移除优先级标签 clean_text = re.sub(r'[(Pd|HIGH|MEDIUM|LOW)]', '', task_text, flags=re.IGNORECASE).strip() tasks.append({ "text": clean_text, "completed": checked, "priority": priority, "original_line": line.strip() }) return [{ "json": { "tasks": tasks, "total": len(tasks), "completed": sum(1 for t in tasks if t["completed"]), "pending": sum(1 for t in tasks if not t["completed"]) } }] ``` ### 核心技术 - 逐行解析 - 使用多个正则模式进行提取 - 从文本中提取元数据 - 计算汇总统计信息 - 返回结构化数据 --- ## 模式 4：JSON 对象比较 **使用场景**：比较两个 JSON 对象以查找差异。 **具体情境**：比较旧的和新的用户个人资料数据。 ### 代码实现 ```python import json all_items = _input.all() # 假设第一个项目是旧数据，第二个是新数据 old_data = all_items[0]["json"] if len(all_items) > 0 else {} new_data = all_items[1]["json"] if len(all_items) > 1 else {} changes = { "added": {}, "removed": {}, "modified": {}, "unchanged": {} } # 查找所有唯一的键 all_keys = set(old_data.keys()) | set(new_data.keys()) for key in all_keys: old_value = old_data.get(key) new_value = new_data.get(key) if key not in old_data: # 新增字段 changes["added"][key] = new_value elif key not in new_data: # 移除字段 changes["removed"][key] = old_value elif old_value != new_value: # 修改字段 changes["modified"][key] = { "old": old_value, "new": new_value } else: # 未改变字段 changes["unchanged"][key] = old_value return [{ "json": { "changes": changes, "summary": { "added_count": len(changes["added"]), "removed_count": len(changes["removed"]), "modified_count": len(changes["modified"]), "unchanged_count": len(changes["unchanged"]), "has_changes": len(changes["added"]) > 0 or len(changes["removed"]) > 0 or len(changes["modified"]) > 0 } } }] ``` ### 核心技术 - 使用集合操作进行键比较 - 使用字典 `.get()` 进行安全访问 - 按类型对更改进行分类 - 创建汇总统计信息 - 返回详细的对比结果 --- ## 模式 5：CRM 数据转换 **使用场景**：将 CRM 数据转换为标准格式。 **具体情境**：标准化来自不同 CRM 系统的数据。 ### 代码实现 ```python from datetime import datetime import re all_items = _input.all() normalized_contacts = [] for item in all_items: raw_contact = item["json"] source = raw_contact.get("source", "unknown") # 标准化邮箱 email = raw_contact.get("email", "").lower().strip() # 标准化电话（移除所有非数字字符） phone_raw = raw_contact.get("phone", "") phone = re.sub(r'D', '', phone_raw) # 解析姓名 if "full_name" in raw_contact: name_parts = raw_contact["full_name"].split(" ", 1) first_name = name_parts[0] if len(name_parts) > 0 else "" last_name = name_parts[1] if len(name_parts) > 1 else "" else: first_name = raw_contact.get("first_name", "") last_name = raw_contact.get("last_name", "") # 标准化状态 status_raw = raw_contact.get("status", "").lower() status = "active" if status_raw in ["active", "enabled", "true", "1"] else "inactive" # 创建标准化的联系人 normalized_contacts.append({ "json": { "id": raw_contact.get("id", ""), "first_name": first_name.strip(), "last_name": last_name.strip(), "full_name": f"{first_name} {last_name}".strip(), "email": email, "phone": phone, "status": status, "source": source, "normalized_at": datetime.now().isoformat(), "original_data": raw_contact } }) return normalized_contacts ``` ### 核心技术 - 处理多种字段名变体 - 字符串清洗与标准化 - 使用正则清洗电话号码 - 姓名解析逻辑 - 状态标准化 - 保留原始数据 --- ## 模式 6：发布日志处理 **使用场景**：解析发布日志并对更改进行分类。 **具体情境**：从发布日志中提取功能、修复和破坏性更改。 ### 代码实现 ```python import re release_notes = _input.first()["json"]["body"].get("notes", "") categories = { "features": [], "fixes": [], "breaking": [], "other": [] } # 分割为行 lines = release_notes.split("n") for line in lines: line = line.strip() # 跳过空行和标题 if not line or line.startswith("#"): continue # 移除列表符号 clean_line = re.sub(r'^[*-+]s*', '', line) # 分类 if re.search(r'b(feature|add|new)b', clean_line, re.IGNORECASE): categories["features"].append(clean_line) elif re.search(r'b(fix|bug|patch|resolve)b', clean_line, re.IGNORECASE): categories["fixes"].append(clean_line) elif re.search(r'b(breaking|deprecated|remove)b', clean_line, re.IGNORECASE): categories["breaking"].append(clean_line) else: categories["other"].append(clean_line) return [{ "json": { "categories": categories, "summary": { "features": len(categories["features"]), "fixes": len(categories["fixes"]), "breaking": len(categories["breaking"]), "other": len(categories["other"]), "total": sum(len(v) for v in categories.values()) } } }] ``` ### 核心技术 - 逐行解析 - 基于模式的分类 - 移除列表符号 - 跳过标题和空行 - 汇总统计信息 --- ## 模式 7：数组转换 **使用场景**：重塑数组并提取特定字段。 **具体情境**：转换用户数据数组以提取特定字段。 ### 代码实现 ```python all_items = _input.all() # 提取并转换 transformed = [] for item in all_items: user = item["json"] # 提取嵌套字段 profile = user.get("profile", {}) settings = user.get("settings", {}) transformed.append({ "json": { "user_id": user.get("id"), "email": user.get("email"), "name": profile.get("name", "Unknown"), "avatar": profile.get("avatar_url"), "bio": profile.get("bio", "")[:100], # 截断至 100 字符 "notifications_enabled": settings.get("notifications", True), "theme": settings.get("theme", "light"), "created_at": user.get("created_at"), "last_login": user.get("last_login_at") } }) return transformed ``` ### 核心技术 - 从嵌套对象中提取字段 - 使用 `.get()` 设置默认值 - 字符串截断 - 扁平化嵌套结构 --- ## 模式 8：字典查找 **使用场景**：创建查找字典以实现快速数据访问。 **具体情境**：通过 ID 查找用户详情。 ### 代码实现 ```python all_items = _input.all() # 构建查找字典 users_by_id = {} for item in all_items: user = item["json"] user_id = user.get("id") if user_id: users_by_id[user_id] = { "name": user.get("name"), "email": user.get("email"), "status": user.get("status") } # 示例：查找特定用户 lookup_ids = [1, 3, 5] looked_up = [] for user_id in lookup_ids: if user_id in users_by_id: looked_up.append({ "json": { "id": user_id, **users_by_id[user_id], "found": True } }) else: looked_up.append({ "json": { "id": user_id, "found": False } }) return looked_up ``` ### 核心技术 - 字典推导式的替代方案 - O(1) 查找时间 - 优雅处理缺失的键 - 保留查找顺序 --- ## 模式 9：Top N 过滤 **使用场景**：按分数或数值获取前几名项目。 **具体情境**：获取销售额前 10 名的产品。 ### 代码实现 ```python all_items = _input.all() # 提取带有销售额的产品 products = [] for item in all_items: product = item["json"] products.append({ "id": product.get("id"), "name": product.get("name"), "sales": product.get("sales", 0), "revenue": product.get("revenue", 0.0), "category": product.get("category") }) # 按销售额降序排序 products.sort(key=lambda p: p["sales"], reverse=True) # 获取前 10 名 top_10 = products[:10] return [ { "json": { **product, "rank": index + 1 } } for index, product in enumerate(top_10) ] ``` ### 核心技术 - 使用自定义键进行列表排序 - 使用切片获取 Top N - 添加排名信息 - 使用 `enumerate` 获取索引 --- ## 模式 10：字符串聚合 **使用场景**：聚合字符串并进行格式化。 **具体情境**：从多个项目中创建汇总文本。 ### 代码实现 ```python all_items = _input.all() # 收集消息 messages = [] for item in all_items: data = item["json"] user = data.get("user", "Unknown") message = data.get("message", "") timestamp = data.get("timestamp", "") # 格式化每条消息 formatted = f"[{timestamp}] {user}: {message}" messages.append(formatted) # 使用换行符连接 summary = "n".join(messages) # 创建统计信息 total_length = sum(len(msg) for msg in messages) average_length = total_length / len(messages) if messages else 0 return [{ "json": { "summary": summary, "message_count": len(messages), "total_characters": total_length, "average_length": round(average_length, 2) } }] ``` ### 核心技术 - 使用 f-strings 进行字符串格式化 - 使用分隔符连接列表 - 计算字符串统计信息 - 处理空列表 --- ## 模式对比：Python vs JavaScript ### 数据访问 ```python # Python all_items = _input.all() first_item = _input.first() current = _input.item webhook_data = _json["body"] # JavaScript const allItems = $input.all(); const firstItem = $input.first(); const current = $input.item; const webhookData = $json.body; ``` ### 字典/对象访问 ```python # Python - 字典键访问 name = user["name"] # 可能引发 KeyError name = user.get("name", "?") # 安全访问，带默认值 # JavaScript - 对象属性访问 const name = user.name; // 可能是 undefined const name = user.name || "?"; // 安全访问，带默认值 ``` ### 数组操作 ```python # Python - 列表推导式 filtered = [item for item in items if item["active"]] # JavaScript - 数组方法 const filtered = items.filter(item => item.active); ``` ### 排序 ```python # Python items.sort(key=lambda x: x["score"], reverse=True) # JavaScript items.sort((a, b) => b.score - a.score); ``` --- ## 最佳实践 ### 1. 使用 .get() 进行安全访问 ```python # ✅ 安全：使用 .get() 和默认值 name = user.get("name", "Unknown") email = user.get("email", "[email protected]") # ❌ 风险：直接通过键访问 name = user["name"] # 如果缺失会触发 KeyError！ ``` ### 2. 处理空列表 ```python # ✅ 安全：处理前先检查 items = _input.all() if items: first = items[0] else: return [{"json": {"error": "No items"}}] # ❌ 风险：假设项目存在 first = items[0] # 如果为空会触发 IndexError！ ``` ### 3. 使用列表推导式 ```python # ✅ Pythonic：列表推导式 active = [item for item in items if item["json"].get("active")] # ❌ 冗长：传统循环 active = [] for item in items: if item["json"].get("active"): active.append(item) ``` ### 4. 返回正确的格式 ```python # ✅ 正确：带有 "json" 键的对象数组 return [{"json": {"field": "value"}}] # ❌ 错误：仅返回数据 return {"field": "value"} # ❌ 错误：没有 "json" 包装的数组 return [{"field": "value"}] ``` ### 5. 使用标准库 ```python # ✅ 推荐：使用标准库 import statistics average = statistics.mean(numbers) # ✅ 同样推荐：内置函数 average = sum(numbers) / len(numbers) if numbers else 0 # ❌ 无法实现：外部库 import numpy as np # ModuleNotFoundError! ``` --- ## 各模式适用场景 | 模式 | 何时使用 | |---------|-------------| | 多源聚合 | 合并来自不同节点/源的数据 | | 正则过滤 | 文本模式匹配、验证、提取 | | Markdown 解析 | 将格式化文本处理为结构化数据 | | JSON 比较 | 检测对象之间的更改 | | CRM 转换 | 标准化来自不同系统的数据 | | 发布日志 | 按关键词对文本进行分类 | | 数组转换 | 重塑数据、提取字段 | | 字典查找 | 快速基于 ID 的查找 | | Top N 过滤 | 按标准获取最好/最差的项目 | | 字符串聚合 | 创建格式化的文本摘要 | --- ## 总结 **核心要点**： - 使用 `.get()` 进行安全的字典访问 - 列表推导式既 Pythonic 又高效 - 处理空列表/None 值 - 使用标准库（json, datetime, re） - 返回正确的 n8n 格式：`[{"json": {...}}]` **请记住**： - 95% 的用例建议使用 JavaScript - Python 没有外部库 - 复杂操作优先使用 n8n 节点 - 代码节点用于数据转换，而非 API 调用