当前位置：首页 > news >正文

python导入redis json数据，通过接口的方式

news 2026/3/26 20:31:39

python导入redis json数据，通过接口的方式

需求背景：
按这个请求格式： https://you ip/apigateway/cmsServer/console/article
POST请求
{"title":"test221","identifier":"test222","description":"test223","content":"<p>test224</p>","type":"scooter"}
数据来源redis_export.json，生成python脚本，可以直接导入
需要带上请求头：
authorization:

python脚本代码：

import json
import requests
import sys# 配置参数
CMS_API_ENDPOINT = "https://you ip/apigateway/cmsServer/console/article"
AUTH_TOKEN = "you token header params"
REDIS_JSON_FILE = "redis_export.json"  # 请确保文件在当前目录或修改为实际路径,从redis导出的数据json文件def import_articles_from_redis_json():"""从 Redis 导出的 JSON 文件中读取文章数据，并导入到 CMS 系统"""try:# 1. 读取并解析 Redis 导出的 JSON 文件with open(REDIS_JSON_FILE, 'r', encoding='utf-8') as f:redis_data = json.load(f)print(f"成功读取 {REDIS_JSON_FILE}，共找到 {len(redis_data)} 条数据")# 2. 设置请求头（包含认证Token）headers = {"Content-Type": "application/json","Authorization": AUTH_TOKEN}# 3. 遍历所有数据，筛选出文章数据并构建请求imported_count = 0failed_count = 0for key, value in redis_data.items():try:# 只处理文章数据（根据 key 模式判断）if key.startswith("test:cmsServer:article:"):# 解析嵌套的 JSON 字符串article_data = json.loads(value)# 构建 API 请求体payload = {"title": article_data.get("title", ""),"identifier": article_data.get("identifier", ""),"description": article_data.get("description", ""),"content": article_data.get("htmlContent", ""),"type": article_data.get("type", "scooter")  # 默认为 scooter 类型
                    }# 检查必需字段if not payload["title"] or not payload["identifier"]:print(f"⚠️  跳过: {key} - 缺少标题或标识符")continue# 发送 POST 请求print(f"\n正在导入文章: {payload['title']} (标识符: {payload['identifier']})")print(f"  Type: {payload['type']}")print(f"  描述: {payload['description'][:50]}..." if payload['description'] else "  描述: 空")response = requests.post(CMS_API_ENDPOINT,json=payload,headers=headers)# 检查响应print(f"  HTTP状态码: {response.status_code}")if response.status_code == 200:result = response.json()print(f"  响应内容: {json.dumps(result, ensure_ascii=False)}")if result.get("resCode") == "000000":print(f"  ✅ 导入成功")imported_count += 1else:error_msg = result.get("resMsg", "未知错误")print(f"  ❌ 导入失败: {error_msg}")failed_count += 1else:print(f"  ❌ HTTP 错误: {response.status_code}")print(f"  错误详情: {response.text}")failed_count += 1# 添加短暂延时以避免请求过于频繁import timetime.sleep(0.5)except json.JSONDecodeError as e:print(f"  ⚠️  数据解析失败 (key: {key}): {str(e)}")failed_count += 1except Exception as e:print(f"  ❌ 处理失败 (key: {key}): {str(e)}")failed_count += 1# 4. 输出统计结果print(f"\n{'='*60}")print(f"导入完成!")print(f"成功导入: {imported_count} 条")print(f"导入失败: {failed_count} 条")print(f"总计处理: {imported_count + failed_count} 条")return imported_count, failed_countexcept FileNotFoundError:print(f"❌ 错误: 找不到文件 {REDIS_JSON_FILE}")print("请确认文件路径是否正确，或修改脚本中的 REDIS_JSON_FILE 变量")sys.exit(1)except json.JSONDecodeError as e:print(f"❌ 错误: JSON 文件格式不正确 - {str(e)}")sys.exit(1)except Exception as e:print(f"❌ 错误: {str(e)}")sys.exit(1)if __name__ == "__main__":print("="*60)print("文章数据导入脚本 - CMS系统")print("="*60)print(f"目标API: {CMS_API_ENDPOINT}")print(f"数据源: {REDIS_JSON_FILE}")print(f"Token: {AUTH_TOKEN[:20]}...{AUTH_TOKEN[-20:]}")print("-"*60)# 显示将要处理的数据摘要try:with open(REDIS_JSON_FILE, 'r', encoding='utf-8') as f:redis_data = json.load(f)article_keys = [k for k in redis_data.keys() if k.startswith("node:cmsServer:article:")]print(f"检测到 {len(article_keys)} 篇待导入文章:")for i, key in enumerate(article_keys[:5], 1):  # 显示前5个try:data = json.loads(redis_data[key])print(f"  {i}. {data.get('title', '无标题')} (ID: {data.get('identifier', '无标识符')})")except:print(f"  {i}. {key}")if len(article_keys) > 5:print(f"  ... 还有 {len(article_keys)-5} 篇")except:print("无法预览数据内容，将继续执行导入...")print("-"*60)# 询问确认confirm = input("是否开始导入? (输入 'y' 或 'yes' 继续): ")if confirm.lower() in ['y', 'yes']:print("开始导入...")import_articles_from_redis_json()else:print("导入已取消")