使用Python和Pandas提取特定对象、转置并合并多个复杂嵌套JSON文件为CSV的实现方案

问题背景 #

我知道有不少帖子讨论这个主题，先提前说声抱歉，我已经读了很多内容也试了好多次了。

我把三个示例JSON文件存在fildir目录里：

https://data.sec.gov/api/xbrl/companyfacts/CIK0000320193.json
https://data.sec.gov/api/xbrl/companyfacts/CIK0001722010.json
https://data.sec.gov/api/xbrl/companyfacts/CIK0001722606.json

这些JSON的主键是cik、entityName、facts，在facts下面有"dei"和"us-gaap"两个对象，而这两个对象里的Concept（带引号的那些）没有固定的键名，这有点麻烦。

下面是其中一个JSON文件的片段：

{
  "cik": 320193,
  "entityName": "Apple Inc.",
  "facts": {
    "dei": {
      "EntityCommonStockSharesOutstanding": {
        "label": "Entity Common Stock, Shares Outstanding",
        "description": "Indicate number of shares or other units outstanding of each of registrant'.......etc....",
        "units": {
          "shares": [
            {
              "end": "2009-06-27",
              "val": 895816758,
              "accn": "0001193125-09-153165",
              "fy": 2009,
              "fp": "Q3",
              "form": "10-Q",
              "filed": "2009-07-22",
              "frame": "CY2009Q2I"
            }
          ]
        }
      }
    },
    "us-gaap": {
      "AccountsPayable": {
        "label": "Accounts Payable (Deprecated 2009-01-31)",
        "description": "Carrying value as of the balance sheet date of liabilities incurred (and for which invoices have typically been received) and payable to vendors for goods and services received that are used in an entity's business. For classified balance sheets, used to reflect the current portion of the liabilities (due within one year or within the normal operating cycle if longer); for unclassified balance sheets, used to reflect the total liabilities (regardless of due date).",
        "units": {
          "USD": [
            {
              "end": "2008-09-27",
              "val": 5520000000,
              "accn": "0001193125-09-153165",
              "fy": 2009,
              "fp": "Q3",
              "form": "10-Q",
              "filed": "2009-07-22",
              "frame": "CY2008Q3I"
            }
          ]
        }
      }
    }
  }
}

我之前的尝试 #

这是我试的其中一种写法：

import pandas as pd
import os
import glob
import json

OUTPUT_PATH = "../output/concepts/csv/"
fildir = '../resources/companyfacts/'
fils = os.path.join(fildir, '*.json')
filist = glob.glob(fils)

for fils in filist:
    i = open(fils, "r")
    sec_data = json.loads(i.read())
    try:
        for item in sec_data["facts"]["dei"]:
            i.close()
    except:
        continue
    try:
        for item in sec_data["facts"]["us-gaap"]:
            i.close()
    except:
        continue
    dfs = [pd.read_json(fils) for sec_data in filist]
    data = {"concepts": {}}
    for item in sec_data["facts"]["dei"]:
        if f"{item}" not in data:
            data[f"{item}"] = {}
        data[f"{item}"] = item
    for item in sec_data["facts"]["us-gaap"]:
       if f"{item}" not in data:
            data[f"{item}"] = {}
       data[f"{item}"] = item
    df = pd.concat(dfs, ignore_index=True)
    df = pd.DataFrame(data).transpose()
    df.to_csv(OUTPUT_PATH + "CONCEPTS.csv")

这段代码的输出结果是一个单列的概念列表：

concepts
EntityCommonStockSharesOutstanding
EntityPublicFloat
AccruedLiabilitiesCurrent
AdditionalPaidInCapital
CommonStockParOrStatedValuePerShare
CommonStockSharesAuthorized
......
......
......etc

后来我又试了另一种写法，能得到所有概念的大列表，但还是单列，没法按每个CIK分列：

import pandas as pd
import os
import json
import glob

OUTPUT_PATH = "../output/concepts/csv/"
fildir = '../resources/companyfacts/'
fils = os.path.join(fildir, '*.json')
filist = glob.glob(fils)

data = []
for fils in filist:
    i = open(fils, "r")
    sec_data = json.loads(i.read())
    cik = str(sec_data['cik'])
    padded_cik = cik.zfill(10)
    cickStr = f'CIK{padded_cik}-concepts'

    entList = []
    for k,v in sec_data['facts'].items():
        dei = list(sec_data['facts']['dei'])
        gaap = list(sec_data['facts']['us-gaap'])
        entList += list(dei)
        entList += list(gaap)
        data += list(entList)
        sec_data.update({cickStr:list(set(entList))})

#    dfs = pd.DataFrame(entList)
    dfs = pd.DataFrame(data)
    #df = pd.concat(dfs, ignore_index=True)
#    df.to_csv(OUTPUT_PATH + "CONCEPTS.csv")
#    for k,v in sec_data['facts'].items():
#        entList += list(v.keys())
#df = pd.DataFrame(dict([(k, pd.Series(v)) for k, v in sec_data.items()])) 
dfs.to_csv(OUTPUT_PATH + "CONCEPTS.csv")

我的需求 #

1. 跳过异常文件：如果某个JSON文件是空的，或者facts里没有"us-gaap"或"dei"对象，直接跳过它处理下一个文件，避免出现KeyError: 'us-gaap'这类报错。
2. 期望的CSV格式：我希望输出的CSV按每个CIK分列，每列是对应CIK的所有Concept列表，类似这样：

CIK0000320193-concepts:    CIK0001722010-concepts:   CIK0001722606-concepts:
EntityCommonStockSharesOutstanding       AccountsPayable      ......
EntityPublicFloat       ......      ......
......       ......      ......

解决方案代码 #

针对你的需求，我调整了代码逻辑，解决了异常处理和分列输出的问题：

import pandas as pd
import os
import json
import glob

OUTPUT_PATH = "../output/concepts/csv/"
fildir = '../resources/companyfacts/'
filist = glob.glob(os.path.join(fildir, '*.json'))

# 用来存储每个CIK对应的概念列表
concepts_dict = {}

for file_path in filist:
    try:
        with open(file_path, "r") as f:
            sec_data = json.load(f)
        
        # 提取CIK并格式化为要求的字符串（补零到10位）
        cik = str(sec_data.get('cik', ''))
        if not cik:
            print(f"跳过文件 {file_path}：未找到cik字段")
            continue
        padded_cik = cik.zfill(10)
        column_name = f'CIK{padded_cik}-concepts'
        
        # 提取facts中的dei和us-gaap概念
        facts = sec_data.get('facts', {})
        current_concepts = []
        
        # 处理dei下的概念
        dei_section = facts.get('dei', {})
        current_concepts.extend(list(dei_section.keys()))
        
        # 处理us-gaap下的概念
        us_gaap_section = facts.get('us-gaap', {})
        current_concepts.extend(list(us_gaap_section.keys()))
        
        # 去重并存储
        if current_concepts:
            concepts_dict[column_name] = list(set(current_concepts))
        else:
            print(f"文件 {file_path} 未提取到任何概念")
            
    except json.JSONDecodeError:
        print(f"跳过文件 {file_path}：JSON格式错误或文件为空")
    except Exception as e:
        print(f"处理文件 {file_path} 时出错：{str(e)}")
        continue

# 将字典转换为DataFrame，填充缺失值为空字符串
df = pd.DataFrame(dict([(k, pd.Series(v)) for k, v in concepts_dict.items()])).fillna('')

# 保存为CSV
os.makedirs(os.path.dirname(OUTPUT_PATH), exist_ok=True)
df.to_csv(os.path.join(OUTPUT_PATH, "CONCEPTS.csv"), index=False)

print(f"处理完成！结果已保存到 {os.path.join(OUTPUT_PATH, 'CONCEPTS.csv')}")

代码说明 #

• 异常处理：用try-except捕获了JSON解析错误、缺失字段等异常，遇到问题的文件会直接跳过并打印提示信息。
• CIK格式化：把提取到的cik补零到10位，生成符合要求的列名。
• 概念提取：分别从dei和us-gaap中提取所有概念键，去重后存储。
• 分列输出：把每个CIK的概念列表转成DataFrame的一列，缺失的行用空字符串填充，最后保存为CSV。

备注：内容来源于stack exchange，提问作者shrykullgod