使用Python实现一个简单的文件搜索引擎
作者：默默且听风
文本是关于Python文件操作的基础和进阶知识，包括读写文件、文件和目录管理、错误处理、文件路径操作、文件编码、处理大文件、临时文件、文件权限以及一个简单的文件搜索引擎示例。进阶部分涉及了文件模式、缓冲、文件锁、高级文件搜索技巧、文件系统监控、跨平台文件路径处理、性能考虑、安全性，以及一个进一步优化的文件搜索引擎示例。
基础
读写文件
示例代码：
# 读取文件
with open('example.txt', 'r') as file:
content = file.read()
print(content)
# 写入文件
with open('example.txt', 'w') as file:
file.write('Hello, World!')
无需额外安装包，Python内置的open函数就可以进行文件的读写操作。
文件和目录管理
示例代码：
import os
import shutil
# 创建目录
os.mkdir('new_directory')
# 重命名目录
os.rename('new_directory', 'renamed_directory')
# 删除文件
os.remove('old_file.txt')
# 复制文件
shutil.copy('source.txt', 'destination.txt')
# 列出目录内容
print(os.listdir('.'))
包简介：
os模块：提供了丰富的方法用来处理文件和目录。shutil模块：提供了一系列对文件和文件集合的高级操作。
错误处理
在进行文件操作时，处理潜在的错误非常重要。例如，尝试打开一个不存在的文件会引发FileNotFoundError。使用try和except语句可以帮助您优雅地处理这些情况：
try:
with open('non_existent_file.txt', 'r') as file:
content = file.read()
except FileNotFoundError:
print("文件不存在。")
上下文管理器
Python的with语句提供了一种管理资源的简洁方式，特别是对于文件操作。使用with可以确保文件在使用后正确关闭，即便在文件操作过程中发生了异常。
with open('example.txt', 'r') as file:
content = file.read()
print(content)
文件路径操作
虽然os模块提供了基本的路径操作功能，但pathlib模块提供了一种更面向对象的方式来处理文件路径。使用pathlib可以使路径操作更加直观和易于维护：
from pathlib import Path
# 当前目录路径
current_dir = Path('.')
# 列出当前目录中的所有文件
for file in current_dir.iterdir():
print(file)
# 读取文件
file_path = current_dir / 'example.txt'
with file_path.open('r') as file:
content = file.read()
文件编码
当处理文本文件时，考虑文件的编码非常重要。默认情况下，Python使用系统默认的编码打开文件，这可能会导致在不同系统之间移植代码时出现问题。指定编码可以确保文件正确读写：
# 使用UTF-8编码打开文件
with open('example.txt', 'r', encoding='utf-8') as file:
content = file.read()
处理大文件
对于非常大的文件，一次性读取它们的内容可能会消耗大量内存。使用迭代器逐行读取可以减少内存使用：
with open('large_file.txt', 'r') as file:
for line in file:
process(line) # 处理每一行
临时文件
有时，您可能需要创建临时文件来存储数据，这些数据在程序结束后不再需要。tempfile模块提供了创建临时文件和目录的方法：
import tempfile
# 创建临时文件
with tempfile.TemporaryFile('w+t') as temp_file:
temp_file.write('Hello, World!')
temp_file.seek(0) # 回到文件开头
print(temp_file.read())
文件权限
在Linux和UNIX系统上，文件权限对于文件安全至关重要。使用os模块，您可以检查和修改文件的权限：
import os
# 修改文件权限（只读）
os.chmod('example.txt', 0o444)
综合示例——一个简单的文件搜索引擎
一个文件搜索引擎，允许用户指定一个根目录和一个文件名（或部分文件名），然后在该目录及其所有子目录中搜索匹配该名称的文件。
import os
import time
def find_files(directory, filename):
matches = []
# 遍历根目录
for root, dirnames, filenames in os.walk(directory):
for name in filenames:
# 检查文件名是否包含搜索关键字
if filename.lower() in name.lower():
matches.append(os.path.join(root, name))
return matches
# 用户输入
root_directory = input("请输入要搜索的根目录: ")
file_to_find = input("请输入要搜索的文件名（支持部分匹配）: ")
# 记录开始时间
start_time = time.time()
# 搜索文件
found_files = find_files(root_directory, file_to_find)
# 记录结束时间
end_time = time.time()
# 输出结果
print(f"找到 {len(found_files)} 个文件:")
for file in found_files:
print(file)
# 输出耗时
print(f"搜索耗时: {end_time - start_time:.2f} 秒")
这个脚本使用了os.walk()函数，该函数可以遍历指定目录下的所有子目录。脚本将所有找到的匹配文件的完整路径添加到一个列表中，并在搜索完成后将这些路径打印出来。
用户首先被提示输入要搜索的根目录和文件名。然后，脚本会调用find_files函数来执行搜索。搜索结果将显示找到的文件数量以及它们的路径。
请注意，这个脚本在文件名匹配时不区分大小写，因为它使用了.lower()方法来将文件名转换为小写。这意味着搜索是大小写不敏感的。
$ python3 r1.py
请输入要搜索的根目录: /DB6/project
请输入要搜索的文件名（支持部分匹配）: index.vue
找到 531 个文件:
/DB6/project/blog/BlogSSR/node_modules/@kangc/v-md-editor/src/components/scrollbar/index.vue
......
搜索耗时: 46.71 秒
进阶
文件模式详解
使用open函数时，可以通过不同的模式来打开文件，这些模式决定了文件的读写权限及行为。
# 写入模式，如果文件存在，覆盖原有内容
with open('example.txt', 'w') as file:
file.write('Hello, Python!')
# 追加模式，写入的内容会添加到文件末尾
with open('example.txt', 'a') as file:
file.write('\nAppend text.')
# 二进制写入模式
with open('example.bin', 'wb') as file:
file.write(b'\x00\xFF')
缓冲
缓冲是文件操作中的一个重要概念，它影响数据写入文件的时机。Python允许你控制文件的缓冲行为。
# 使用无缓冲模式打开文件
with open('example.txt', 'r', buffering=0) as file:
print(file.read())
文件锁
在多线程或多进程环境中，为了避免数据冲突，可以使用文件锁。
import portalocker
with open('example.txt', 'a') as file:
portalocker.lock(file, portalocker.LOCK_EX)
file.write('Locked file.\n')
portalocker.unlock(file)
高级文件搜索技巧
结合os.walk和正则表达式，可以实现复杂的文件搜索逻辑。
import os
import re
def search_files(directory, pattern):
regex = re.compile(pattern)
for root, _, files in os.walk(directory):
for name in files:
if regex.search(name):
print(os.path.join(root, name))
search_files('.', 'example.*')
文件系统监控
使用watchdog库可以监控文件系统的变化，这对于需要根据文件更新实时做出响应的应用非常有用。
from watchdog.observers import Observer from watchdog.events import LoggingEventHandler event_handler = LoggingEventHandler() observer = Observer() observer.schedule(event_handler, path='.', recursive=True) observer.start()
跨平台文件路径处理
pathlib模块提供了一种面向对象的方式来处理文件路径。
from pathlib import Path
p = Path('example.txt')
with p.open('r') as file:
print(file.read())
性能考虑
使用mmap模块可以通过内存映射的方式提高大文件的处理效率。
import mmap
import os
with open('example.txt', 'r+b') as f:
mm = mmap.mmap(f.fileno(), 0)
print(mm.readline())
mm.close()
安全性
在处理文件路径时，尤其是那些来自用户的路径时，需要特别小心，以避免安全漏洞。
from pathlib import Path
def safe_open(file_path, root_directory):
root = Path(root_directory).resolve()
absolute_path = (root / file_path).resolve()
if root not in absolute_path.parents:
raise ValueError("不允许访问根目录之外的文件")
return open(absolute_path, 'r')
user_path = '../outside.txt'
try:
file = safe_open(user_path, '.')
print(file.read())
except ValueError as e:
print(e)
综合示例——进一步修改文件搜索引擎
import os
import re
import time
from concurrent.futures import ThreadPoolExecutor
def search_files(directory, pattern):
"""
在指定目录中搜索匹配正则表达式的文件。
"""
matches = []
regex = re.compile(pattern)
for root, dirnames, filenames in os.walk(directory):
for name in filenames:
if regex.search(name):
matches.append(os.path.join(root, name))
return matches
def search_directory(directory, pattern):
"""
搜索单个目录。
"""
try:
return search_files(directory, pattern)
except PermissionError:
return [] # 忽略权限错误
def main(root_directory, pattern):
"""
主函数：并行搜索目录并汇总结果。
"""
start_time = time.time()
matches = []
# 使用ThreadPoolExecutor来并行搜索
with ThreadPoolExecutor() as executor:
futures = []
for root, dirs, files in os.walk(root_directory):
for dirname in dirs:
future = executor.submit(search_directory, os.path.join(root, dirname), pattern)
futures.append(future)
# 等待所有线程完成并汇总结果
for future in futures:
matches.extend(future.result())
end_time = time.time()
# 打印搜索结果
print(f"找到 {len(matches)} 个文件:")
# for match in matches:
# print(match)
print(f"搜索耗时: {end_time - start_time:.2f} 秒")
if __name__ == "__main__":
import sys
if len(sys.argv) != 3:
print("用法: python search_engine.py [根目录] [搜索模式]")
else:
main(sys.argv[1], sys.argv[2])
os: 用于与操作系统交互，包括遍历目录树。re: 用于正则表达式匹配，以便按模式搜索文件名。time: 用于测量搜索操作的开始和结束时间，以计算总耗时。concurrent.futures.ThreadPoolExecutor: 用于并行化搜索任务，提高搜索效率。
search_files 函数
这个函数接受两个参数：directory（要搜索的目录路径）和pattern（正则表达式模式），并返回匹配该模式的所有文件的完整路径列表。
- 首先，创建一个空列表
matches来存储找到的匹配文件路径。 - 使用
re.compile(pattern)编译正则表达式模式，以便在搜索中使用。 - 使用
os.walk(directory)遍历指定目录及其所有子目录。对于每个目录，os.walk返回一个三元组(root, dirnames, filenames)，其中root是当前目录的路径，dirnames是该目录下所有子目录的名称列表，filenames是该目录下所有文件的名称列表。 - 在每个目录中，遍历所有文件名，使用正则表达式的
.search(name)方法检查文件名是否与给定模式匹配。如果匹配，将文件的完整路径（使用os.path.join(root, name)构建）添加到matches列表中。 - 函数返回
matches列表，包含所有找到的匹配文件的路径。
search_directory 函数
这个函数封装了search_files函数，以便在单个目录中进行搜索，并处理可能发生的PermissionError。
- 接受和
search_files相同的参数。 - 尝试调用
search_files函数进行搜索，如果遇到PermissionError（例如，因为没有足够的权限访问某个目录），则捕获该异常并返回一个空列表，表示没有找到匹配的文件。
main 函数
这是脚本的主函数，负责初始化并行搜索，汇总结果，并打印搜索耗时和找到的匹配文件。
- 首先记录搜索开始时间。
- 创建一个空列表
matches来存储所有找到的匹配文件路径。 - 使用
ThreadPoolExecutor创建一个线程池，以并行执行搜索任务。这通过遍历根目录及其所有子目录，并为每个子目录提交一个search_directory任务到线程池来实现。 - 使用
executor.submit提交任务，并将返回的Future对象添加到futures列表中。 - 使用
future.result()等待所有任务完成并收集结果，将每个任务找到的匹配文件路径扩展到matches列表中。 - 记录搜索结束时间，并计算总耗时。
- 打印找到的匹配文件总数和搜索耗时。注释掉的部分可以取消注释以打印每个匹配文件的路径。
脚本入口
- 检查命令行参数的数量。如果不等于3（脚本名称、根目录和搜索模式），则打印使用说明。
- 如果参数数量正确，调用
main函数并传入根目录和搜索模式。
运行一下看看效果
$ python3 r2.py /DB6/project index.*
找到 1409008 个文件:
搜索耗时: 147.67 秒
以上就是使用Python实现一个简单的文件搜索引擎的详细内容，更多关于Python文件搜索引擎的资料请关注脚本之家其它相关文章！