脚本如何实现文件内容模糊粗糙调度

本文目录导读：

1. 核心概念
2. Python实现方案
3. 高级模糊调度算法
4. 性能优化建议

模糊粗糙调度的实现方式。

核心概念

模糊粗糙调度是指基于文件内容相似度或模式匹配,对文件进行分组、排序或移动的自动化处理。

Python实现方案

1 基于文本相似度的调度

import os
import shutil
from pathlib import Path
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import re
from typing import List, Dict, Tuple
import hashlib
class FuzzyFileScheduler:
    def __init__(self, base_path: str):
        self.base_path = Path(base_path)
        self.file_registry = {}
    def extract_content(self, file_path: Path) -> str:
        """提取文件内容"""
        try:
            if file_path.suffix in ['.txt', '.md', '.py', '.js', '.html', '.css']:
                with open(file_path, 'r', encoding='utf-8') as f:
                    return f.read()
            return ""
        except:
            return ""
    def get_file_similarity(self, file1: Path, file2: Path) -> float:
        """计算文件相似度"""
        content1 = self.extract_content(file1)
        content2 = self.extract_content(file2)
        if not content1 or not content2:
            return 0.0
        # TF-IDF向量化
        vectorizer = TfidfVectorizer(max_features=100)
        tfidf_matrix = vectorizer.fit_transform([content1, content2])
        # 计算余弦相似度
        similarity = cosine_similarity(tfidf_matrix[0:1], tfidf_matrix[1:2])
        return similarity[0][0]
    def cluster_files(self, threshold: float = 0.3) -> Dict[str, List[Path]]:
        """基于内容聚类文件"""
        all_files = list(self.base_path.rglob('*'))
        clusters = {}
        processed = set()
        for i, file1 in enumerate(all_files):
            if file1 in processed or not file1.is_file():
                continue
            cluster_id = f"cluster_{i}"
            clusters[cluster_id] = [file1]
            processed.add(file1)
            for file2 in all_files[i+1:]:
                if file2 in processed or not file2.is_file():
                    continue
                similarity = self.get_file_similarity(file1, file2)
                if similarity >= threshold:
                    clusters[cluster_id].append(file2)
                    processed.add(file2)
        return clusters
    def fuzzy_schedule(self, rules: List[Dict]) -> None:
        """模糊调度规则执行"""
        for rule in rules:
            pattern = rule.get('pattern', '')
            action = rule.get('action', 'move')
            destination = rule.get('destination', '')
            for file_path in self.base_path.rglob('*'):
                if file_path.is_file():
                    content = self.extract_content(file_path)
                    # 模糊匹配
                    if self.fuzzy_match(content, pattern):
                        self.execute_action(file_path, action, destination)
    def fuzzy_match(self, content: str, pattern: str) -> bool:
        """模糊匹配内容"""
        if not content:
            return False
        # 关键词匹配（带权重的模糊匹配）
        keywords = pattern.split()
        matched_keywords = 0
        for keyword in keywords:
            if keyword.lower() in content.lower():
                matched_keywords += 1
        # 匹配度阈值
        match_ratio = matched_keywords / len(keywords) if keywords else 0
        return match_ratio > 0.5  # 50%以上关键词匹配即认为是模糊匹配
    def execute_action(self, file_path: Path, action: str, destination: str):
        """执行调度动作"""
        dest_path = Path(destination)
        if not dest_path.exists():
            dest_path.mkdir(parents=True, exist_ok=True)
        try:
            if action == 'move':
                shutil.move(str(file_path), str(dest_path / file_path.name))
                print(f"Moved: {file_path.name} -> {destination}")
            elif action == 'copy':
                shutil.copy2(str(file_path), str(dest_path / file_path.name))
                print(f"Copied: {file_path.name} -> {destination}")
        except Exception as e:
            print(f"Error processing {file_path.name}: {e}")
# 使用示例
if __name__ == "__main__":
    scheduler = FuzzyFileScheduler("/path/to/files")
    # 定义调度规则
    rules = [
        {
            'pattern': 'error log warning',  # 模糊匹配关键词
            'action': 'move',
            'destination': '/path/to/logs'
        },
        {
            'pattern': 'data report statistics',
            'action': 'copy',
            'destination': '/path/to/reports'
        }
    ]
    # 执行调度
    scheduler.fuzzy_schedule(rules)
    # 聚类分析
    clusters = scheduler.cluster_files(threshold=0.3)
    for cluster_id, files in clusters.items():
        print(f"{cluster_id}: {len(files)} files")

2 Bash脚本实现

#!/bin/bash
# 模糊文件调度脚本
SCHEDULE_DIR="/path/to/files"
RULE_FILE="schedule_rules.txt"
# 读取规则文件
while IFS='|' read -r keyword_folder action dest_pattern; do
    # 遍历所有文件
    find "$SCHEDULE_DIR" -type f | while read -r file; do
        # 获取文件内容的前1000字符
        content_head=$(head -1000 "$file")
        # 模糊匹配关键词
        match_count=0
        for keyword in $(echo "$keyword_folder" | tr ',' ' '); do
            if echo "$content_head" | grep -qi "$keyword"; then
                ((match_count++))
            fi
        done
        # 如果匹配数大于阈值
        keyword_count=$(echo "$keyword_folder" | tr ',' ' ' | wc -w)
        threshold=$((keyword_count / 2))
        if [ $match_count -gt $threshold ]; then
            # 生成目标路径
            dest_path=$(echo "$dest_pattern" | sed "s/{keyword}/$keyword_folder/g")
            mkdir -p "$dest_path"
            case $action in
                move)
                    mv "$file" "$dest_path/"
                    echo "Moved: $file -> $dest_path"
                    ;;
                copy)
                    cp "$file" "$dest_path/"
                    echo "Copied: $file -> $dest_path"
                    ;;
            esac
        fi
    done
done < "$RULE_FILE"

3 规则文件格式 (schedule_rules.txt)

# 格式: 关键词|动作|目标路径
error,log,warning|move|/archive/logs/{keyword}
data,report,statistics|copy|/processed/reports/{keyword}
backup,archive,old|move|/storage/archive/{keyword}

高级模糊调度算法

1 基于编辑距离的调度

def levenshtein_distance(s1: str, s2: str) -> int:
    """计算编辑距离"""
    m, n = len(s1), len(s2)
    dp = [[0] * (n + 1) for _ in range(m + 1)]
    for i in range(m + 1):
        dp[i][0] = i
    for j in range(n + 1):
        dp[0][j] = j
    for i in range(1, m + 1):
        for j in range(1, n + 1):
            if s1[i-1] == s2[j-1]:
                dp[i][j] = dp[i-1][j-1]
            else:
                dp[i][j] = min(dp[i-1][j-1], dp[i-1][j], dp[i][j-1]) + 1
    return dp[m][n]
def fuzzy_schedule_by_edit_distance(content: str, patterns: List[str]) -> str:
    """基于编辑距离的模糊调度"""
    best_match = None
    min_distance = float('inf')
    for pattern in patterns:
        distance = levenshtein_distance(content[:100], pattern[:100])
        if distance < min_distance:
            min_distance = distance
            best_match = pattern
    # 阈值判断
    threshold = len(content) * 0.3  # 30%的编辑距离阈值
    return best_match if min_distance < threshold else None

2 基于N-gram的调度

def extract_ngrams(content: str, n: int = 3) -> set:
    """提取N-gram特征"""
    return set(content[i:i+n] for i in range(len(content)-n+1))
def ngram_similarity(content1: str, content2: str) -> float:
    """N-gram相似度计算"""
    ngrams1 = extract_ngrams(content1)
    ngrams2 = extract_ngrams(content2)
    if not ngrams1 or not ngrams2:
        return 0.0
    intersection = ngrams1 & ngrams2
    union = ngrams1 | ngrams2
    return len(intersection) / len(union) if union else 0.0
def schedule_by_ngram(directory: str, patterns: Dict[str, str], threshold: float = 0.3):
    """基于N-gram的模糊调度"""
    for file_path in Path(directory).rglob('*'):
        if not file_path.is_file():
            continue
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
        for pattern, action in patterns.items():
            similarity = ngram_similarity(content, pattern)
            if similarity >= threshold:
                # 执行调度动作
                print(f"Scheduled: {file_path.name} -> {action} (相似度: {similarity:.2f})")

性能优化建议

1. 缓存机制：缓存已分析文件的内容哈希
2. 增量处理：只处理新增或修改的文件
3. 并行处理：多线程/多进程处理大量文件
4. 采样分析：对大文件只分析头部/尾部内容

这个实现提供了基础的模糊调度框架,可以根据具体需求进行调整和扩展。