使用Python實(shí)現(xiàn)一個(gè)簡(jiǎn)單的文件搜索引擎

更新時(shí)間：2024年05月02日 07:29:44 作者：默默且聽(tīng)風(fēng)

這篇文章主要為大家詳細(xì)介紹了Python中文件操作的基礎(chǔ)和進(jìn)階知識(shí)并基于以上知識(shí)實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的文件搜索引擎,感興趣的小伙伴可以參考一下

文本是關(guān)于Python文件操作的基礎(chǔ)和進(jìn)階知識(shí)，包括讀寫(xiě)文件、文件和目錄管理、錯(cuò)誤處理、文件路徑操作、文件編碼、處理大文件、臨時(shí)文件、文件權(quán)限以及一個(gè)簡(jiǎn)單的文件搜索引擎示例。進(jìn)階部分涉及了文件模式、緩沖、文件鎖、高級(jí)文件搜索技巧、文件系統(tǒng)監(jiān)控、跨平臺(tái)文件路徑處理、性能考慮、安全性，以及一個(gè)進(jìn)一步優(yōu)化的文件搜索引擎示例。

基礎(chǔ)

讀寫(xiě)文件

示例代碼：

# 讀取文件
with open('example.txt', 'r') as file:
    content = file.read()
    print(content)

# 寫(xiě)入文件
with open('example.txt', 'w') as file:
    file.write('Hello, World!')

無(wú)需額外安裝包，Python內(nèi)置的open函數(shù)就可以進(jìn)行文件的讀寫(xiě)操作。

文件和目錄管理

示例代碼：

import os
import shutil

# 創(chuàng)建目錄
os.mkdir('new_directory')

# 重命名目錄
os.rename('new_directory', 'renamed_directory')

# 刪除文件
os.remove('old_file.txt')

# 復(fù)制文件
shutil.copy('source.txt', 'destination.txt')

# 列出目錄內(nèi)容
print(os.listdir('.'))

包簡(jiǎn)介：

os模塊：提供了豐富的方法用來(lái)處理文件和目錄。
shutil模塊：提供了一系列對(duì)文件和文件集合的高級(jí)操作。

錯(cuò)誤處理

在進(jìn)行文件操作時(shí)，處理潛在的錯(cuò)誤非常重要。例如，嘗試打開(kāi)一個(gè)不存在的文件會(huì)引發(fā)FileNotFoundError。使用try和except語(yǔ)句可以幫助您優(yōu)雅地處理這些情況：

try:
    with open('non_existent_file.txt', 'r') as file:
        content = file.read()
except FileNotFoundError:
    print("文件不存在。")

上下文管理器

Python的with語(yǔ)句提供了一種管理資源的簡(jiǎn)潔方式，特別是對(duì)于文件操作。使用with可以確保文件在使用后正確關(guān)閉，即便在文件操作過(guò)程中發(fā)生了異常。

with open('example.txt', 'r') as file:
    content = file.read()
    print(content)

文件路徑操作

雖然os模塊提供了基本的路徑操作功能，但pathlib模塊提供了一種更面向?qū)ο蟮姆绞絹?lái)處理文件路徑。使用pathlib可以使路徑操作更加直觀和易于維護(hù)：

from pathlib import Path

# 當(dāng)前目錄路徑
current_dir = Path('.')
# 列出當(dāng)前目錄中的所有文件
for file in current_dir.iterdir():
    print(file)

# 讀取文件
file_path = current_dir / 'example.txt'
with file_path.open('r') as file:
    content = file.read()

文件編碼

當(dāng)處理文本文件時(shí)，考慮文件的編碼非常重要。默認(rèn)情況下，Python使用系統(tǒng)默認(rèn)的編碼打開(kāi)文件，這可能會(huì)導(dǎo)致在不同系統(tǒng)之間移植代碼時(shí)出現(xiàn)問(wèn)題。指定編碼可以確保文件正確讀寫(xiě)：

# 使用UTF-8編碼打開(kāi)文件
with open('example.txt', 'r', encoding='utf-8') as file:
    content = file.read()

處理大文件

對(duì)于非常大的文件，一次性讀取它們的內(nèi)容可能會(huì)消耗大量?jī)?nèi)存。使用迭代器逐行讀取可以減少內(nèi)存使用：

with open('large_file.txt', 'r') as file:
    for line in file:
        process(line)  # 處理每一行

臨時(shí)文件

有時(shí)，您可能需要?jiǎng)?chuàng)建臨時(shí)文件來(lái)存儲(chǔ)數(shù)據(jù)，這些數(shù)據(jù)在程序結(jié)束后不再需要。tempfile模塊提供了創(chuàng)建臨時(shí)文件和目錄的方法：

import tempfile

# 創(chuàng)建臨時(shí)文件
with tempfile.TemporaryFile('w+t') as temp_file:
    temp_file.write('Hello, World!')
    temp_file.seek(0)  # 回到文件開(kāi)頭
    print(temp_file.read())

文件權(quán)限

在Linux和UNIX系統(tǒng)上，文件權(quán)限對(duì)于文件安全至關(guān)重要。使用os模塊，您可以檢查和修改文件的權(quán)限：

import os

# 修改文件權(quán)限（只讀）
os.chmod('example.txt', 0o444)

綜合示例——一個(gè)簡(jiǎn)單的文件搜索引擎

一個(gè)文件搜索引擎，允許用戶指定一個(gè)根目錄和一個(gè)文件名（或部分文件名），然后在該目錄及其所有子目錄中搜索匹配該名稱的文件。

import os
import time

def find_files(directory, filename):
    matches = []
    # 遍歷根目錄
    for root, dirnames, filenames in os.walk(directory):
        for name in filenames:
            # 檢查文件名是否包含搜索關(guān)鍵字
            if filename.lower() in name.lower():
                matches.append(os.path.join(root, name))
    return matches

# 用戶輸入
root_directory = input("請(qǐng)輸入要搜索的根目錄: ")
file_to_find = input("請(qǐng)輸入要搜索的文件名（支持部分匹配）: ")

# 記錄開(kāi)始時(shí)間
start_time = time.time()

# 搜索文件
found_files = find_files(root_directory, file_to_find)

# 記錄結(jié)束時(shí)間
end_time = time.time()

# 輸出結(jié)果
print(f"找到 {len(found_files)} 個(gè)文件:")
for file in found_files:
    print(file)

# 輸出耗時(shí)
print(f"搜索耗時(shí): {end_time - start_time:.2f} 秒")

這個(gè)腳本使用了os.walk()函數(shù)，該函數(shù)可以遍歷指定目錄下的所有子目錄。腳本將所有找到的匹配文件的完整路徑添加到一個(gè)列表中，并在搜索完成后將這些路徑打印出來(lái)。

用戶首先被提示輸入要搜索的根目錄和文件名。然后，腳本會(huì)調(diào)用find_files函數(shù)來(lái)執(zhí)行搜索。搜索結(jié)果將顯示找到的文件數(shù)量以及它們的路徑。

請(qǐng)注意，這個(gè)腳本在文件名匹配時(shí)不區(qū)分大小寫(xiě)，因?yàn)樗褂昧?code>.lower()方法來(lái)將文件名轉(zhuǎn)換為小寫(xiě)。這意味著搜索是大小寫(xiě)不敏感的。

$ python3 r1.py
請(qǐng)輸入要搜索的根目錄: /DB6/project
請(qǐng)輸入要搜索的文件名（支持部分匹配）: index.vue
找到 531 個(gè)文件:
/DB6/project/blog/BlogSSR/node_modules/@kangc/v-md-editor/src/components/scrollbar/index.vue
......
搜索耗時(shí): 46.71 秒

進(jìn)階

文件模式詳解

使用open函數(shù)時(shí)，可以通過(guò)不同的模式來(lái)打開(kāi)文件，這些模式?jīng)Q定了文件的讀寫(xiě)權(quán)限及行為。

# 寫(xiě)入模式，如果文件存在，覆蓋原有內(nèi)容
with open('example.txt', 'w') as file:
    file.write('Hello, Python!')

# 追加模式，寫(xiě)入的內(nèi)容會(huì)添加到文件末尾
with open('example.txt', 'a') as file:
    file.write('\nAppend text.')

# 二進(jìn)制寫(xiě)入模式
with open('example.bin', 'wb') as file:
    file.write(b'\x00\xFF')

緩沖

緩沖是文件操作中的一個(gè)重要概念，它影響數(shù)據(jù)寫(xiě)入文件的時(shí)機(jī)。Python允許你控制文件的緩沖行為。

# 使用無(wú)緩沖模式打開(kāi)文件
with open('example.txt', 'r', buffering=0) as file:
    print(file.read())

文件鎖

在多線程或多進(jìn)程環(huán)境中，為了避免數(shù)據(jù)沖突，可以使用文件鎖。

import portalocker

with open('example.txt', 'a') as file:
    portalocker.lock(file, portalocker.LOCK_EX)
    file.write('Locked file.\n')
    portalocker.unlock(file)

高級(jí)文件搜索技巧

結(jié)合os.walk和正則表達(dá)式，可以實(shí)現(xiàn)復(fù)雜的文件搜索邏輯。

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.*')

文件系統(tǒng)監(jiān)控

使用watchdog庫(kù)可以監(jiān)控文件系統(tǒng)的變化，這對(duì)于需要根據(jù)文件更新實(shí)時(shí)做出響應(yīng)的應(yīng)用非常有用。

from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

event_handler = LoggingEventHandler()
observer = Observer()
observer.schedule(event_handler, path='.', recursive=True)
observer.start()

跨平臺(tái)文件路徑處理

pathlib模塊提供了一種面向?qū)ο蟮姆绞絹?lái)處理文件路徑。

from pathlib import Path

p = Path('example.txt')
with p.open('r') as file:
    print(file.read())

性能考慮

使用mmap模塊可以通過(guò)內(nèi)存映射的方式提高大文件的處理效率。

import mmap
import os

with open('example.txt', 'r+b') as f:
    mm = mmap.mmap(f.fileno(), 0)
    print(mm.readline())
    mm.close()

安全性

在處理文件路徑時(shí)，尤其是那些來(lái)自用戶的路徑時(shí)，需要特別小心，以避免安全漏洞。

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("不允許訪問(wèn)根目錄之外的文件")
    return open(absolute_path, 'r')

user_path = '../outside.txt'
try:
    file = safe_open(user_path, '.')
    print(file.read())
except ValueError as e:
    print(e)

綜合示例——進(jìn)一步修改文件搜索引擎

import os
import re
import time
from concurrent.futures import ThreadPoolExecutor

def search_files(directory, pattern):
    """
    在指定目錄中搜索匹配正則表達(dá)式的文件。
    """
    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):
    """
    搜索單個(gè)目錄。
    """
    try:
        return search_files(directory, pattern)
    except PermissionError:
        return []  # 忽略權(quán)限錯(cuò)誤

def main(root_directory, pattern):
    """
    主函數(shù)：并行搜索目錄并匯總結(jié)果。
    """
    start_time = time.time()
    matches = []

    # 使用ThreadPoolExecutor來(lái)并行搜索
    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)

        # 等待所有線程完成并匯總結(jié)果
        for future in futures:
            matches.extend(future.result())

    end_time = time.time()
    
    # 打印搜索結(jié)果
    print(f"找到 {len(matches)} 個(gè)文件:")
    # for match in matches:
    #     print(match)
    
    print(f"搜索耗時(shí): {end_time - start_time:.2f} 秒")

if __name__ == "__main__":
    import sys
    if len(sys.argv) != 3:
        print("用法: python search_engine.py [根目錄](méi) [搜索模式]")
    else:
        main(sys.argv[1], sys.argv[2])

os: 用于與操作系統(tǒng)交互，包括遍歷目錄樹(shù)。
re: 用于正則表達(dá)式匹配，以便按模式搜索文件名。
time: 用于測(cè)量搜索操作的開(kāi)始和結(jié)束時(shí)間，以計(jì)算總耗時(shí)。
concurrent.futures.ThreadPoolExecutor: 用于并行化搜索任務(wù)，提高搜索效率。

search_files 函數(shù)

這個(gè)函數(shù)接受兩個(gè)參數(shù)：directory（要搜索的目錄路徑）和pattern（正則表達(dá)式模式），并返回匹配該模式的所有文件的完整路徑列表。

首先，創(chuàng)建一個(gè)空列表matches來(lái)存儲(chǔ)找到的匹配文件路徑。
使用re.compile(pattern)編譯正則表達(dá)式模式，以便在搜索中使用。
使用os.walk(directory)遍歷指定目錄及其所有子目錄。對(duì)于每個(gè)目錄，os.walk返回一個(gè)三元組(root, dirnames, filenames)，其中root是當(dāng)前目錄的路徑，dirnames是該目錄下所有子目錄的名稱列表，filenames是該目錄下所有文件的名稱列表。
在每個(gè)目錄中，遍歷所有文件名，使用正則表達(dá)式的.search(name)方法檢查文件名是否與給定模式匹配。如果匹配，將文件的完整路徑（使用os.path.join(root, name)構(gòu)建）添加到matches列表中。
函數(shù)返回matches列表，包含所有找到的匹配文件的路徑。

search_directory 函數(shù)

這個(gè)函數(shù)封裝了search_files函數(shù)，以便在單個(gè)目錄中進(jìn)行搜索，并處理可能發(fā)生的PermissionError。

接受和search_files相同的參數(shù)。
嘗試調(diào)用search_files函數(shù)進(jìn)行搜索，如果遇到PermissionError（例如，因?yàn)闆](méi)有足夠的權(quán)限訪問(wèn)某個(gè)目錄），則捕獲該異常并返回一個(gè)空列表，表示沒(méi)有找到匹配的文件。

main 函數(shù)

這是腳本的主函數(shù)，負(fù)責(zé)初始化并行搜索，匯總結(jié)果，并打印搜索耗時(shí)和找到的匹配文件。

首先記錄搜索開(kāi)始時(shí)間。
創(chuàng)建一個(gè)空列表matches來(lái)存儲(chǔ)所有找到的匹配文件路徑。
使用ThreadPoolExecutor創(chuàng)建一個(gè)線程池，以并行執(zhí)行搜索任務(wù)。這通過(guò)遍歷根目錄及其所有子目錄，并為每個(gè)子目錄提交一個(gè)search_directory任務(wù)到線程池來(lái)實(shí)現(xiàn)。
使用executor.submit提交任務(wù)，并將返回的Future對(duì)象添加到futures列表中。
使用future.result()等待所有任務(wù)完成并收集結(jié)果，將每個(gè)任務(wù)找到的匹配文件路徑擴(kuò)展到matches列表中。
記錄搜索結(jié)束時(shí)間，并計(jì)算總耗時(shí)。
打印找到的匹配文件總數(shù)和搜索耗時(shí)。注釋掉的部分可以取消注釋以打印每個(gè)匹配文件的路徑。