本文目录导读：
模糊粗糙调度的实现方式。
核心概念
模糊粗糙调度是指基于文件内容相似度或模式匹配,对文件进行分组、排序或移动的自动化处理。
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})")
性能优化建议
- 缓存机制：缓存已分析文件的内容哈希
- 增量处理：只处理新增或修改的文件
- 并行处理：多线程/多进程处理大量文件
- 采样分析：对大文件只分析头部/尾部内容
这个实现提供了基础的模糊调度框架,可以根据具体需求进行调整和扩展。
