状态机编程实战｜如何更优雅地处理字符串

1943年，数学家 McCulloch 和 Pitts 提出了第一个自动机模型。他们可能没有想到，这个最初用来模拟神经网络的数学工具，如今会成为现代软件开发中字符串处理的核心技术。从你每天使用的搜索引擎，到手机里的输入法，再到各种编程语言的编译器——自动机算法无处不在，默默地让我们的数字世界运转得更加流畅。

自动机算法：从理论到实践的处理利器

什么是自动机？

自动机（Automaton）是一个数学模型，用来描述在不同输入下系统状态的变化过程。简单来说，它就像一个"状态转换器"：根据当前所处的状态和接收到的输入，决定下一步应该转换到哪个状态。

想象一下你在使用自动售货机购买饮料的过程：

初始状态：等待投币
投币后：选择商品状态
选择商品后：出货状态
最终状态：交易完成

这个过程中，每一个动作（投币、选择、出货）都会触发状态的转换，而系统的行为完全由当前状态和输入决定。这就是自动机的基本思想。

为什么选择自动机？

在字符串处理领域，我们通常有几种选择：

正则表达式：简洁但难以处理复杂嵌套结构
手工编写解析器：灵活但容易出错，维护困难
自动机方法：结构清晰，易于调试和扩展

自动机算法的优势在于：

逻辑清晰：每个状态代表解析过程中的一个明确阶段
易于调试：可以精确跟踪状态转换过程
高度可扩展：添加新的解析规则只需要增加相应的状态和转换
性能优异：时间复杂度通常为O(n)，空间占用也很合理

实际应用场景

自动机算法在现实世界中有着广泛的应用：

编译器前端：词法分析器使用自动机识别关键字、标识符、数字等词法单元

网络协议：HTTP解析器、JSON解析器都大量使用状态机来处理复杂的协议格式

用户输入验证：邮箱、电话号码、身份证号等格式验证

文本处理工具：Markdown解析器、模板引擎等都依赖状态机来处理文本转换

本文将带你探索什么？

在接下来的内容中，我们将从最基础的概念开始，逐步深入到实际应用：

🔧 基础概念：状态、转换、触发器的核心原理
💻 实战演练：使用 Python的 transitions 库构建状态机
🎯 真实案例：邮箱验证、URL解析、JSON处理等实用示例
⚡ 高级技巧：条件转换、回调函数、错误处理
🚀 复杂应用：构建Lean定理解析器的完整方案

无论你是初学者还是有经验的开发者，这篇文章都将为你打开自动机算法的大门，让你掌握这个强大而优雅的字符串处理工具。

让我们开始这段从理论到实践的探索之旅吧！

1. 安装和基本概念

1	pip install transitions

基本概念：

状态(States)：系统可能处在的不同情况
转换(Transitions)：从一个状态到另一个状态的变化
触发器(Triggers)：引起状态转换的事件/方法
条件(Conditions)：转换是否发生的判断条件
回调(Callbacks)：状态转换时执行的动作

2. 最简单的例子：开关

from transitions import Machine

class Switch:
    # 定义所有可能的状态
    states = ['off', 'on']
    
    def __init__(self):
        # 创建状态机，初始状态为'off'
        self.machine = Machine(
            model=self,           # 状态机绑定到当前对象
            states=Switch.states, # 状态列表
            initial='off'         # 初始状态
        )
        
        # 添加状态转换
        # 添加从'off'到'on'的转换，触发器名为'turn_on'
        self.machine.add_transition('turn_on', 'off', 'on')
        # 添加从'on'到'off'的转换，触发器名为'turn_off'
        self.machine.add_transition('turn_off', 'on', 'off')

# 使用示例
switch = Switch()
print(f"初始状态: {switch.state}")  # off

switch.turn_on()  # 触发转换
print(f"开启后: {switch.state}")    # on

switch.turn_off() # 触发转换
print(f"关闭后: {switch.state}")    # off

# 尝试非法转换（会抛出异常）
try:
    switch.turn_off()  # 已经是off状态，再次关闭会出错
except Exception as e:
    print(f"错误: {e}")

3. 带条件的状态转换

class Door:
    states = ['closed', 'open', 'locked']
    
    def __init__(self):
        self.has_key = False
        
        self.machine = Machine(
            model=self,
            states=Door.states,
            initial='closed'
        )
        
        # 简单转换
        self.machine.add_transition('open_door', 'closed', 'open')
        self.machine.add_transition('close_door', 'open', 'closed')
        
        # 带条件的转换
        self.machine.add_transition(
            'lock_door', 
            'closed', 
            'locked',
            conditions='has_key_condition'  # 需要满足条件
        )
        
        self.machine.add_transition(
            'unlock_door',
            'locked',
            'closed', 
            conditions='has_key_condition'
        )
    
    def has_key_condition(self):
        """条件函数：检查是否有钥匙"""
        return self.has_key
    
    def get_key(self):
        """获得钥匙"""
        self.has_key = True
        print("获得了钥匙!")
    
    def lose_key(self):
        """丢失钥匙"""
        self.has_key = False
        print("丢失了钥匙!")

# 使用示例
door = Door()
print(f"初始状态: {door.state}")  # closed

door.open_door()
print(f"开门后: {door.state}")    # open

door.close_door()
print(f"关门后: {door.state}")    # closed

# 尝试没有钥匙时锁门
if not door.lock_door():
    print(f"无法锁门: {door.state}")

# 获得钥匙后锁门
door.get_key()
door.lock_door()
print(f"锁门后: {door.state}")    # locked

# 开锁
door.unlock_door()
print(f"开锁后: {door.state}")    # closed

4. 带回调函数的状态机

class TrafficLight:
    states = ['red', 'yellow', 'green']
    
    def __init__(self):
        self.timer = 0
        
        self.machine = Machine(
            model=self,
            states=TrafficLight.states,
            initial='red'
        )
        
        # 添加带回调的转换
        self.machine.add_transition(
            'next_light',           # 触发器名称
            'red',                  # 源状态
            'green',                # 目标状态
            before='on_before_change',    # 转换前回调
            after='on_after_change'       # 转换后回调
        )
        
        self.machine.add_transition(
            'next_light',
            'green', 
            'yellow',
            before='on_before_change',
            after='on_after_change'
        )
        
        self.machine.add_transition(
            'next_light',
            'yellow',
            'red', 
            before='on_before_change',
            after='on_after_change'
        )
    
    def on_before_change(self):
        """转换前的回调"""
        print(f"calling on before, 当前状态 {self.state}")
        self.timer = 0
    
    def on_after_change(self):
        """转换后的回调"""
        print(f"calling on after, 已切换到 {self.state}")
        if self.state == 'red':
            print("停止! 红灯30秒")
        elif self.state == 'yellow':
            print("注意! 黄灯3秒")
        elif self.state == 'green':
            print("通行! 绿灯25秒")

# 使用示例
light = TrafficLight()
print(f"初始状态: {light.state}")

for i in range(5): # red -> green -> yello -> red...
    print(f"\n第{i+1}次切换:")
    light.next_light()

# red -- on_before_change -> green -> on_after_change

5. 复杂状态机：文件下载器

class FileDownloader:
    states = [
        'idle',        # 空闲
        'connecting',  # 连接中
        'downloading', # 下载中
        'paused',      # 暂停
        'completed',   # 完成
        'failed'       # 失败
    ]
    
    def __init__(self):
        self.progress = 0
        self.error_count = 0
        self.file_size = 100  # 假设文件大小
        
        self.machine = Machine(
            model=self,
            states=FileDownloader.states,
            initial='idle'
        )
        
        # 定义所有可能的状态转换
        self._setup_transitions()
    
    def _setup_transitions(self):
        # 开始下载
        self.machine.add_transition(
            'start_download', 
            'idle', 
            'connecting',
            after='on_start_connecting'
        )
        
        # 连接成功
        self.machine.add_transition(
            'connected', 
            'connecting', 
            'downloading',
            after='on_start_downloading'
        )
        
        # 连接失败
        self.machine.add_transition(
            'connection_failed',
            'connecting', 
            'failed',
            after='on_failed'
        )
        
        # 暂停下载
        self.machine.add_transition(
            'pause', 
            'downloading', 
            'paused',
            after='on_paused'
        )
        
        # 恢复下载
        self.machine.add_transition(
            'resume', 
            'paused', 
            'downloading',
            after='on_resumed'
        )
        
        # 下载完成
        self.machine.add_transition(
            'download_complete',
            'downloading', 
            'completed',
            conditions='is_download_finished',
            after='on_completed'
        )
        
        # 下载失败
        self.machine.add_transition(
            'download_failed',
            'downloading',
            'failed', 
            after='on_failed'
        )
        
        # 重试 (从失败状态回到空闲)
        self.machine.add_transition(
            'retry',
            'failed',
            'idle',
            conditions='can_retry',
            after='on_retry'
        )
        
        # 重置 (从任何状态回到空闲)
        self.machine.add_transition(
            'reset',
            '*',  # 通配符，表示从任何状态
            'idle',
            after='on_reset'
        )
    
    # 条件函数
    def is_download_finished(self):
        return self.progress >= self.file_size
    
    def can_retry(self):
        return self.error_count < 3
    
    # 回调函数
    def on_start_connecting(self):
        print("正在连接服务器...")
    
    def on_start_downloading(self):
        print("开始下载文件...")
    
    def on_paused(self):
        print(f"下载已暂停，进度: {self.progress}/{self.file_size}")
    
    def on_resumed(self):
        print("恢复下载...")
    
    def on_completed(self):
        print("下载完成!")
    
    def on_failed(self):
        self.error_count += 1
        print(f"下载失败! 错误次数: {self.error_count}")
    
    def on_retry(self):
        print("重试下载...")
    
    def on_reset(self):
        self.progress = 0
        self.error_count = 0
        print("下载器已重置")
    
    # 辅助方法
    def simulate_download_progress(self, amount=10):
        """模拟下载进度"""
        if self.state == 'downloading':
            self.progress += amount
            print(f"下载进度: {self.progress}/{self.file_size}")
            
            if self.progress >= self.file_size:
                self.download_complete()

# 使用示例
downloader = FileDownloader()

print("=== 正常下载流程 ===")
print(f"当前状态: {downloader.state}")

downloader.start_download()
print(f"状态: {downloader.state}")

downloader.connected()
print(f"状态: {downloader.state}")

# 模拟下载过程
for i in range(3):
    downloader.simulate_download_progress(30)
    
print(f"最终状态: {downloader.state}")

print("\n=== 暂停和恢复 ===")
downloader.reset()
downloader.start_download()
downloader.connected()
downloader.simulate_download_progress(20)
downloader.pause()
downloader.resume()
downloader.simulate_download_progress(80)

print("\n=== 失败和重试 ===")
downloader.reset()
downloader.start_download()
downloader.connection_failed()
print(f"是否可以重试: {downloader.can_retry()}")
downloader.retry()

6. 邮件验证器

from transitions import Machine

class EmailValidator:
    states = [
        'start',
        'username',
        'at_symbol', 
        'domain',
        'dot',
        'extension',
        'valid',
        'invalid'
    ]
    
    def __init__(self):
        self.machine = Machine(model=self, states=self.states, initial='start')
        self.current_char = ''
        self.buffer = ''
        self.setup_transitions()
    
    def setup_transitions(self):
        # 开始 -> 用户名
        self.machine.add_transition('process_char', 'start', 'username', 
                                  conditions=['is_username_start']) # 满足条件下，进入 username
        self.machine.add_transition('process_char', 'start', 'invalid') # 否则退出
        
        # 用户名 -> @ 或继续用户名
        self.machine.add_transition('process_char', 'username', 'at_symbol',
                                  conditions=['is_at_symbol']) 
        self.machine.add_transition('process_char', 'username', 'username',
                                  conditions=['is_username_char'])
        self.machine.add_transition('process_char', 'username', 'invalid')
        
        # @ -> 域名
        self.machine.add_transition('process_char', 'at_symbol', 'domain',
                                  conditions=['is_domain_start'])
        self.machine.add_transition('process_char', 'at_symbol', 'invalid')
        
        # 域名 -> . 或继续域名
        self.machine.add_transition('process_char', 'domain', 'dot',
                                  conditions=['is_dot'])
        self.machine.add_transition('process_char', 'domain', 'domain',
                                  conditions=['is_domain_char'])
        self.machine.add_transition('process_char', 'domain', 'invalid')
        
        # . -> 扩展名
        self.machine.add_transition('process_char', 'dot', 'extension',
                                  conditions=['is_extension_start'])
        self.machine.add_transition('process_char', 'dot', 'invalid')
        
        # 扩展名 -> 继续扩展名或完成
        self.machine.add_transition('process_char', 'extension', 'extension',
                                  conditions=['is_extension_char'])
        self.machine.add_transition('process_char', 'extension', 'invalid')
        
        # 结束时检查
        self.machine.add_transition('finish', 'extension', 'valid')
        self.machine.add_transition('finish', '*', 'invalid')
    
    # 条件函数
    def is_username_start(self):
        return self.current_char.isalnum()
    
    def is_username_char(self):
        return self.current_char.isalnum() or self.current_char in '._-'
    
    def is_at_symbol(self):
        return self.current_char == '@'
    
    def is_domain_start(self):
        return self.current_char.isalnum()
    
    def is_domain_char(self):
        return self.current_char.isalnum() or self.current_char == '-'
    
    def is_dot(self):
        return self.current_char == '.'
    
    def is_extension_start(self):
        return self.current_char.isalpha()
    
    def is_extension_char(self):
        return self.current_char.isalpha()
    
    def validate(self, email):
        """验证邮箱地址"""
        # 重置状态
        self.to_start()
        self.buffer = ''
        
        print(f"验证邮箱: {email}")
        
        for i, char in enumerate(email):
            self.current_char = char
            self.buffer += char
            
            print(f"  字符 '{char}' -> 状态: {self.state}", end='')
            
            try:
                self.process_char()
                print(f" -> {self.state}")
                
                if self.state == 'invalid':
                    return False
                    
            except Exception as e:
                print(f" -> 转换失败: {e}")
                return False
        
        # 检查最终状态
        try:
            self.finish()
            return self.state == 'valid'
        except:
            return False

# 测试邮箱验证器
validator = EmailValidator()

test_emails = [
    "user@domain.com",      # 有效
    "test.email@example.org", # 有效
    "invalid.email",        # 无效 - 没有@
    "@domain.com",          # 无效 - 没有用户名
    "user@.com",            # 无效 - 域名无效
    "user@domain.",         # 无效 - 没有扩展名
    "user@domain.c0m",      # 无效 - 扩展名包含数字
]

for email in test_emails:
    result = validator.validate(email)
    print(f"结果: {'✅ 有效' if result else '❌ 无效'}\n")