如何在Python中解析分层数据并将其格式化为TSV文件？

import re

data = """A09100 Metabolism
B
B  09101 Carbohydrate metabolism
C    00010 Glycolysis / Gluconeogenesis [PATH:ko00010]
D      K00844  HK; hexokinase [EC:2.7.1.1]
D      K12407  GCK; glucokinase [EC:2.7.1.2]
D      K00001  E1.1.1.1, adh; alcohol dehydrogenase [EC:1.1.1.1]
B  09103 Lipid metabolism
C    00071 Fatty acid degradation [PATH:ko00071]
D      K00001  E1.1.1.1, adh; alcohol dehydrogenase [EC:1.1.1.1]
A09120 Genetic Information Processing
B
B  09121 Transcription
C    03020 RNA polymerase [PATH:ko03020]
D      K03043  rpoB; DNA-directed RNA polymerase subunit beta [EC:2.7.7.6]
D      K13797  rpoBC; DNA-directed RNA polymerase subunit beta-beta' [EC:2.7.7.6]"""

lines = data.split('\n')

result = []

ko = None
unique_ko_hierarchy = {}

current_hierarchy = []
current_group = None

for line in lines:
    parts = line.strip().split()
    if parts:
        if parts[0].startswith('A'):
            current_group = " ".join(parts[1:])
        elif parts[0].startswith('B'):
            current_hierarchy = [current_group, " ".join(parts[1:])]
        elif parts[0] == 'K':
            ko = parts[1]
            description = " ".join(parts[2:])
            if ko in unique_ko_hierarchy:
                unique_ko_hierarchy[ko]["metadata_KEGG_Pathways"].append(",".join(current_hierarchy))
            else:
                unique_ko_hierarchy[ko] = {
                    "metadata_KEGG_Description": description,
                    "metadata_KEGG_Pathways": [",".join(current_hierarchy)],
                }
        elif parts[0] == 'D' and len(parts) >= 3:
            ko = parts[1]
            description = " ".join(parts[2:])
            if ko in unique_ko_hierarchy:
                unique_ko_hierarchy[ko]["metadata_KEGG_Pathways"].append(",".join(current_hierarchy))
            else:
                unique_ko_hierarchy[ko] = {
                    "metadata_KEGG_Description": description,
                    "metadata_KEGG_Pathways": [",".join(current_hierarchy)],
                }
        else:
            current_hierarchy.append(" ".join(parts[1:]))

# Add the header row
result.append(["KO", "metadata_KEGG_Description", "metadata_KEGG_Pathways"])

for ko, data in unique_ko_hierarchy.items():
    # Remove numeric prefixes and "[PATH:...]" patterns from the pathway string
    pathways = [re.sub(r'\d+\s*', '', pathway.split('[PATH:')[0].strip()) for pathway in data["metadata_KEGG_Pathways"]]
    result.append([ko, data["metadata_KEGG_Description"], "|".join(pathways)])

# Specify the filename for the TSV file
tsv_filename = "output_data.tsv"

with open(tsv_filename, 'w') as tsv_file:
    for row in result:
        tsv_file.write("\t".join(row) + "\n")

print(f"Data saved to {tsv_filename}")

我建议您检查处理 KEGG 文件的 Orange Bioinformatics 的 DBGETEntryParser。否则，如果你想在一些正则表达式的帮助下使用 pandas，你可以试试这个：

import re

with open("file.keg") as f:
    pat = r"^([A-D]) *(\S+)\s*(.+?)\s*(\[.+\])?(?=$)"
    data = re.findall(pat, f.read(), flags=re.MULTILINE)  regex101-demo

import pandas as pd

tmp = pd.DataFrame(data, columns=["section", "name", "attribute", "path"])
mA = tmp["section"].eq("A"); mD = tmp["section"].eq("D")
df = tmp.assign(entry= tmp["attribute"].where(mA).ffill()).loc[~mA]

parents = (df["entry"].str.cat(df["attribute"].groupby(
        mD.ne(mD.shift()).cumsum(), sort=False)
            .transform(", ".join).where(~mD).ffill(), sep=", ")
            .rename("parents"))

edges = df[["name"]].join(parents).loc[mD, ["parents", "name"]]

out = (df.join(parents).loc[mD].assign(metadata_KEGG_Pathways=
        lambda x: x["attribute"].str.cat(x["path"], sep=" "))
           .groupby("name", sort=False, as_index=False).agg(
               metadata_KEGG_Description=("metadata_KEGG_Pathways", "first"),
               metadata_KEGG_Pathways=("parents", "|".join)))

# out.to_csv("file.tsv", sep="\t", index=False) # uncomment to make a `.tsv`

输出（表格格式）：

使用 networkx 的图形可视化：

import networkx as nx
from itertools import chain, pairwise

G = nx.from_edgelist(
    chain.from_iterable(
        [pairwise(vals) for vals in edges.agg(
            ",".join, axis=1).str.split(",").to_numpy()]),
    create_using=nx.DiGraph
)