Python中读取结构化TXT文件并反序列化为dict或存入SQL数据库的最优方案

看起来你需要处理一个非标准的结构化文本文件——这种格式没法直接用csv或json库解析，得靠自定义逻辑来处理。下面我给你一套分步实现的方案，先把文本转成Python字典方便内存操作，再演示如何存入SQL数据库，代码都是可直接运行的，你可以根据实际需求调整细节。

一、解析文本为Python字典 #

核心思路是用状态机跟踪当前解析的上下文（比如正在处理城市基础信息、北区街区还是居民数据），逐行读取并处理内容：

实现代码 #

import re

def parse_city_txt(file_path):
    city_data = {
        "name": "",
        "founded_year": "",
        "parts": []  # 存储南北区数据
    }
    current_part = None
    current_quarter = None
    current_street = None
    current_section = "city"  # 初始状态：解析城市基础信息

    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:  # 跳过空白行
                continue

            # 切换上下文状态
            if line == "NORTH PART" or line == "SOUTH PART":
                current_section = "part"
                part_name = line.split()[0]
                current_part = {
                    "name": part_name,
                    "size_percent": "",
                    "quarters": [],
                    "streets": [],
                    "citizens": []
                }
                city_data["parts"].append(current_part)
                continue
            elif line == "QUATERS":
                current_section = "quarter"
                continue
            elif line == "STREETS":
                current_section = "street"
                continue
            elif line == "CITIZENS":
                current_section = "citizen"
                continue
            elif line == "~same story~":
                # 假设南区结构和北区完全一致，这里可以直接复用北区的解析逻辑
                continue

            # 根据当前状态处理行内容
            if current_section == "city":
                if line.startswith("City name is"):
                    city_data["name"] = line.split("is")[-1].strip()
                elif line.startswith("It was build in"):
                    # 忽略原文拼写错误（build→built），直接提取年份
                    city_data["founded_year"] = line.split("in")[-1].strip()
            elif current_section == "part":
                if line.startswith("Size"):
                    size_str = line.split()[1]
                    current_part["size_percent"] = size_str.replace("%", "")
            elif current_section == "quarter":
                if line.startswith("Quarter name"):
                    current_quarter = {
                        "name": line.split("name")[-1].strip(),
                        "size": ""
                    }
                    current_part["quarters"].append(current_quarter)
                elif line.startswith("Size"):
                    current_quarter["size"] = line.split()[-1].strip()
            elif current_section == "street":
                if line.startswith("Street name"):
                    current_street = {
                        "name": line.split("name")[-1].strip(),
                        "house_count": "",
                        "located_in": "",
                        "zip_code": ""
                    }
                    current_part["streets"].append(current_street)
                elif line.startswith("Number of houses"):
                    current_street["house_count"] = line.split()[-1].strip()
                elif line.startswith("Is located in"):
                    current_street["located_in"] = line.split("in")[-1].strip()
                elif line.startswith("ZipCode"):
                    current_street["zip_code"] = line.split()[-1].strip()
            elif current_section == "citizen":
                # 用正则表达式提取居民信息，比字符串分割更可靠
                name_match = re.search(r"My name is (\w+),", line)
                age_match = re.search(r"i'm (\d+) years old", line)
                address_match = re.search(r"live in (.+) in (\d+) house, flat (\d+), room (\d+)", line)
                occupation_match = re.search(r"I am (.+)$", line)

                citizen = {}
                if name_match:
                    citizen["name"] = name_match.group(1)
                if age_match:
                    citizen["age"] = int(age_match.group(1))
                if address_match:
                    citizen["street_name"] = address_match.group(1)
                    citizen["house_number"] = int(address_match.group(2))
                    citizen["flat_number"] = int(address_match.group(3))
                    citizen["room_number"] = int(address_match.group(4))
                if occupation_match:
                    citizen["occupation"] = occupation_match.group(1)
                current_part["citizens"].append(citizen)

    return city_data

# 测试解析
if __name__ == "__main__":
    city_dict = parse_city_txt("city_info.txt")
    import pprint
    pprint.pprint(city_dict)

关键说明 #

• 用current_section变量跟踪当前解析的区块，避免不同层级数据混乱；
• 居民信息用正则表达式提取，比手动分割字符串更鲁棒，能适应小格式变化；
• 南区的~same story~部分，你可以直接复制北区的解析逻辑，或者封装成复用函数减少冗余。

二、将解析后的数据存入SQL数据库 #

我们用SQLite做示例（轻量无需额外安装），先设计合理的关联表结构，再把字典数据插入进去。

1. 数据库表结构设计 #

我们需要建立层级关联的表来存储数据：

• cities：存储城市基础信息
• city_parts：存储南北区信息（关联cities）
• quarters：存储街区信息（关联city_parts）
• streets：存储街道信息（关联quarters）
• citizens：存储居民信息（关联streets）

2. 实现代码 #

import sqlite3
from parse_city_txt import parse_city_txt  # 导入上面的解析函数

def init_db(db_path):
    # 创建数据库和表
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # 创建表
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS cities (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        founded_year TEXT
    )
    """)

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS city_parts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        city_id INTEGER NOT NULL,
        name TEXT NOT NULL,
        size_percent TEXT,
        FOREIGN KEY (city_id) REFERENCES cities(id)
    )
    """)

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS quarters (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        part_id INTEGER NOT NULL,
        name TEXT NOT NULL,
        size TEXT,
        FOREIGN KEY (part_id) REFERENCES city_parts(id)
    )
    """)

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS streets (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        quarter_id INTEGER NOT NULL,
        name TEXT NOT NULL,
        house_count INTEGER,
        zip_code TEXT,
        FOREIGN KEY (quarter_id) REFERENCES quarters(id)
    )
    """)

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS citizens (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        street_id INTEGER NOT NULL,
        name TEXT NOT NULL,
        age INTEGER,
        house_number INTEGER,
        flat_number INTEGER,
        room_number INTEGER,
        occupation TEXT,
        FOREIGN KEY (street_id) REFERENCES streets(id)
    )
    """)

    conn.commit()
    conn.close()

def insert_data_to_db(city_data, db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # 插入城市信息
    cursor.execute("INSERT INTO cities (name, founded_year) VALUES (?, ?)",
                   (city_data["name"], city_data["founded_year"]))
    city_id = cursor.lastrowid

    # 插入城市分区数据
    for part in city_data["parts"]:
        cursor.execute("INSERT INTO city_parts (city_id, name, size_percent) VALUES (?, ?, ?)",
                       (city_id, part["name"], part["size_percent"]))
        part_id = cursor.lastrowid

        # 插入街区数据
        for quarter in part["quarters"]:
            cursor.execute("INSERT INTO quarters (part_id, name, size) VALUES (?, ?, ?)",
                           (part_id, quarter["name"], quarter["size"]))
            quarter_id = cursor.lastrowid

            # 插入街道数据
            for street in part["streets"]:
                # 通过街区名称匹配ID，实际可以在解析字典时直接关联ID提升效率
                cursor.execute("SELECT id FROM quarters WHERE name = ?", (street["located_in"],))
                street_quarter_id = cursor.fetchone()[0]
                cursor.execute("INSERT INTO streets (quarter_id, name, house_count, zip_code) VALUES (?, ?, ?, ?)",
                               (street_quarter_id, street["name"], street["house_count"], street["zip_code"]))
                street_id = cursor.lastrowid

                # 插入居民数据
                for citizen in part["citizens"]:
                    if citizen["street_name"] == street["name"]:
                        cursor.execute("""
                        INSERT INTO citizens (street_id, name, age, house_number, flat_number, room_number, occupation)
                        VALUES (?, ?, ?, ?, ?, ?, ?)
                        """, (street_id, citizen["name"], citizen["age"], citizen["house_number"],
                              citizen["flat_number"], citizen["room_number"], citizen["occupation"]))

    conn.commit()
    conn.close()

# 运行示例
if __name__ == "__main__":
    db_path = "city_database.db"
    init_db(db_path)
    city_data = parse_city_txt("city_info.txt")
    insert_data_to_db(city_data, db_path)
    print("数据存入数据库完成！")

关键说明 #

• 用外键维护表之间的层级关系，保证数据一致性；
• 插入街道时通过街区名称匹配ID，实际使用中可以在解析字典时直接关联ID，减少数据库查询次数；
• 如果用MySQL或PostgreSQL，只需要修改数据库连接方式和少量SQL语法，核心逻辑完全通用。

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