Python实用工具: pyahocorasick库全方位使用教程

pip install pyahocorasick

import pyahocorasick
print("pyahocorasick版本：", pyahocorasick.__version__)

# 安装预编译包（需先安装wheel工具）
pip install wheel
pip install pyahocorasick --only-binary :all:

import pyahocorasick

# 1. 创建AC自动机对象（AhoCorasick类的实例）
# 参数'return_length'设为True，表示匹配结果会返回关键词的长度
ac = pyahocorasick.Automaton(return_length=True)

# 2. 向AC自动机中添加关键词
# 格式：ac.add_word(关键词, 关键词标识)，标识可自定义（如关键词本身）
keywords = ["人工智能", "大数据", "Python", "机器学习"]
for keyword in keywords:
    ac.add_word(keyword, keyword)  # 这里用关键词本身作为标识

# 3. 构建AC自动机（必须执行这一步，否则无法进行匹配）
ac.make_automaton()

# 4. 定义待匹配的文本
text = "随着人工智能技术的发展，大数据分析在各行业的应用越来越广泛。Python作为一门流行的编程语言，在机器学习领域发挥着重要作用。"

# 5. 执行匹配：遍历AC自动机的search方法，获取匹配结果
print("文本中匹配到的关键词：")
for end_index, (keyword, keyword_length) in ac.iter(text):
    # 计算关键词的起始位置（结束位置 - 关键词长度 + 1，注意中文每个字符占1个索引）
    start_index = end_index - keyword_length + 1
    # 输出结果：关键词、起始位置、结束位置、关键词在文本中的片段
    print(f"关键词：{keyword} | 起始位置：{start_index} | 结束位置：{end_index} | 文本片段：{text[start_index:end_index+1]}")

文本中匹配到的关键词：
关键词：人工智能 | 起始位置：2 | 结束位置：5 | 文本片段：人工智能
关键词：大数据 | 起始位置：10 | 结束位置：12 | 文本片段：大数据
关键词：Python | 起始位置：26 | 结束位置：31 | 文本片段：Python
关键词：机器学习 | 起始位置：38 | 结束位置：41 | 文本片段：机器学习

import pyahocorasick

# 1. 创建AC自动机对象
ac = pyahocorasick.Automaton(return_length=True)

# 2. 添加关键词（全部转为小写，标识保留原始关键词）
keywords = ["Python", "PYTHON", "python"]
for keyword in keywords:
    lower_keyword = keyword.lower()  # 转为小写
    ac.add_word(lower_keyword, keyword)  # 标识用原始关键词

# 3. 构建AC自动机
ac.make_automaton()

# 4. 待匹配文本（包含不同大小写的关键词）
text = "Python是一门简单的语言，PYTHON适合初学者，python在数据分析中很常用。"

# 5. 匹配时将文本转为小写，确保大小写一致
lower_text = text.lower()
print("忽略大小写匹配结果：")
for end_index, (original_keyword, keyword_length) in ac.iter(lower_text):
    start_index = end_index - keyword_length + 1
    # 注意：这里的start_index和end_index是基于lower_text的，与原文本索引一致（因仅大小写变化，长度不变）
    print(f"原始关键词：{original_keyword} | 匹配位置：{start_index}-{end_index} | 文本片段：{text[start_index:end_index+1]}")

忽略大小写匹配结果：
原始关键词：Python | 匹配位置：0-5 | 文本片段：Python
原始关键词：PYTHON | 匹配位置：13-18 | 文本片段：PYTHON
原始关键词：python | 匹配位置：26-31 | 文本片段：python

import pyahocorasick

def init_automaton(keywords):
    """初始化AC自动机并返回"""
    ac = pyahocorasick.Automaton(return_length=True)
    for keyword in keywords:
        ac.add_word(keyword, keyword)
    ac.make_automaton()
    return ac

def match_text(ac, text, text_name):
    """
    匹配单篇文本中的关键词
    参数：ac（AC自动机对象）、text（待匹配文本）、text_name（文本名称，用于标识）
    返回：匹配结果列表
    """
    results = []
    for end_idx, (kw, kw_len) in ac.iter(text):
        start_idx = end_idx - kw_len + 1
        results.append({
            "文本名称": text_name,
            "关键词": kw,
            "起始位置": start_idx,
            "结束位置": end_idx,
            "文本片段": text[start_idx:end_idx+1]
        })
    return results

def save_results(results, save_path):
    """将匹配结果保存到文件"""
    with open(save_path, "w", encoding="utf-8") as f:
        # 写入表头
        f.write("文本名称,关键词,起始位置,结束位置,文本片段\n")
        # 写入每一条结果
        for res in results:
            line = f"{res['文本名称']},{res['关键词']},{res['起始位置']},{res['结束位置']},{res['文本片段']}\n"
            f.write(line)
    print(f"结果已保存到：{save_path}")

# ---------------------- 主程序 ----------------------
if __name__ == "__main__":
    # 1. 定义关键词和批量文本
    target_keywords = ["互联网", "电商", "直播"]
    texts = [
        {
            "name": "文章1",
            "content": "互联网行业发展迅速，电商平台如雨后春笋般涌现，直播带货成为新的消费模式。"
        },
        {
            "name": "文章2",
            "content": "随着5G技术普及，互联网速度大幅提升，电商的用户体验也随之改善。"
        },
        {
            "name": "文章3",
            "content": "直播行业竞争激烈，主播需要不断创新内容才能吸引观众。"
        }
    ]

    # 2. 初始化AC自动机（只需初始化一次，避免重复构建）
    ac_automaton = init_automaton(target_keywords)

    # 3. 批量匹配所有文本
    all_results = []
    for text_info in texts:
        text_name = text_info["name"]
        text_content = text_info["content"]
        print(f"正在匹配：{text_name}")
        result = match_text(ac_automaton, text_content, text_name)
        all_results.extend(result)  # 将单篇结果添加到总结果中

    # 4. 保存结果到文件
    save_results(all_results, "match_result.txt")

正在匹配：文章1
正在匹配：文章2
正在匹配：文章3
结果已保存到：match_result.txt

文本名称,关键词,起始位置,结束位置,文本片段
文章1,互联网,0,3,互联网
文章1,电商,8,10,电商
文章1,直播,18,20,直播
文章2,互联网,8,11,互联网
文章2,电商,16,18,电商
文章3,直播,0,2,直播

import pyahocorasick

# 1. 创建AC自动机对象
ac = pyahocorasick.Automaton(return_length=True)

# 2. 添加关键词及附加信息（用元组存储：(关键词, 类别, 优先级)）
keyword_info = [
    ("人工智能", "技术", 1),
    ("大数据", "技术", 2),
    ("电商", "行业", 1),
    ("直播", "行业", 2)
]
for kw, category, priority in keyword_info:
    # 标识设为元组，包含关键词、类别、优先级
    ac.add_word(kw, (kw, category, priority))

# 3. 构建AC自动机
ac.make_automaton()

# 4. 待匹配文本
text = "人工智能和大数据推动电商行业发展，直播成为电商新渠道。"

# 5. 执行匹配并获取附加信息
print("带附加信息的匹配结果：")
print("关键词 | 类别 | 优先级 | 起始位置 | 结束位置")
print("-" * 60)
for end_idx, (kw, category, priority), kw_len in ac.iter(text):
    start_idx = end_idx - kw_len + 1
    print(f"{kw} | {category} | {priority} | {start_idx} | {end_idx}")

带附加信息的匹配结果：
关键词 | 类别 | 优先级 | 起始位置 | 结束位置
------------------------------------------------------------
人工智能 | 技术 | 1 | 0 | 3
大数据 | 技术 | 2 | 6 | 8
电商 | 行业 | 1 | 12 | 14
直播 | 行业 | 2 | 19 | 21
电商 | 行业 | 1 | 25 | 27

import pyahocorasick
import re

def init_email_automaton(common_suffixes):
    """初始化邮箱后缀的AC自动机"""
    ac = pyahocorasick.Automaton(return_length=True)
    for suffix in common_suffixes:
        ac.add_word(suffix, suffix)
    ac.make_automaton()
    return ac

def find_emails(text, ac):
    """找到文本中的所有邮箱（基于AC自动机匹配后缀）"""
    emails = []
    # 先找到所有包含@的片段，再用AC自动机匹配后缀
    at_positions = [i for i, char in enumerate(text) if char == "@"]
    for at_pos in at_positions:
        # 从@位置开始往后匹配邮箱后缀
        for end_idx, suffix, suffix_len in ac.iter(text[at_pos:]):
            # 邮箱前缀：从@前第一个非字母/数字/下划线的位置到@前
            start_idx_prefix = at_pos - 1
            while start_idx_prefix >= 0 and (text[start_idx_prefix].isalnum() or text[start_idx_prefix] == "_"):
                start_idx_prefix -= 1
            start_idx = start_idx_prefix + 1
            end_idx_full = at_pos + end_idx  # 转换为原始文本的结束索引
            email = text[start_idx:end_idx_full + 1]
            emails.append((start_idx, end_idx_full, email))
    # 去重（避免同一邮箱被多次匹配）
    unique_emails = list(set(emails))
    return unique_emails

def filter_sensitive_info(log_content, email_ac):
    """过滤日志中的手机号和邮箱"""
    # 1. 匹配并替换手机号（用正则表达式）
    # 手机号正则：1开头，后跟10位数字
    phone_pattern = re.compile(r"1\d{10}")
    filtered_content = phone_pattern.sub("[敏感信息]", log_content)

    # 2. 匹配并替换邮箱（用AC自动机）
    # 注意：需要重新处理替换后的文本，避免索引偏移
    emails = find_emails(filtered_content, email_ac)
    # 按结束索引从大到小排序，避免替换时前面的替换影响后面的索引
    emails.sort(key=lambda x: x[1], reverse=True)
    for start, end, email in emails:
        filtered_content = filtered_content[:start] + "[敏感信息]" + filtered_content[end+1:]

    return filtered_content

def process_log_file(input_path, output_path, email_suffixes):
    """处理日志文件：读取→过滤→保存"""
    # 初始化邮箱AC自动机
    email_ac = init_email_automaton(email_suffixes)

    # 读取日志文件
    with open(input_path, "r", encoding="utf-8") as f:
        log_content = f.read()

    # 过滤敏感信息
    filtered_content = filter_sensitive_info(log_content, email_ac)

    # 保存过滤后的日志
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(filtered_content)

    print(f"日志过滤完成，已保存到：{output_path}")

# ---------------------- 主程序 ----------------------
if __name__ == "__main__":
    # 常见邮箱后缀（可根据实际需求扩展）
    common_email_suffixes = [
        "@qq.com", "@163.com", "@126.com", "@gmail.com", 
        "@outlook.com", "@sina.com", "@aliyun.com"
    ]

    # 处理日志文件
    process_log_file(
        input_path="app.log",
        output_path="filtered_app.log",
        email_suffixes=common_email_suffixes
    )

2023-10-01 08:30:00 用户登录 - 手机号：13800138000，邮箱：user1@qq.com
2023-10-01 08:35:00 数据提交 - 联系人：张三，电话：13912345678，邮箱：zhangsan@163.com
2023-10-01 08:40:00 系统警告 - 检测到异常访问，来源邮箱：hacker@gmail.com

2023-10-01 08:30:00 用户登录 - 手机号：[敏感信息]，邮箱：[敏感信息]
2023-10-01 08:35:00 数据提交 - 联系人：张三，电话：[敏感信息]，邮箱：[敏感信息]
2023-10-01 08:40:00 系统警告 - 检测到异常访问，来源邮箱：[敏感信息]