Training on usage of repository
A new line is added to test commit
以下是实现所需功能的HTML和JavaScript代码：
<!DOCTYPE html>
<html>
<head>
<title>数字跳转工具</title>
</head>
<body>
<input type="text" id="numberInput" placeholder="请输入6-8位数字">
<button onclick="handleRedirect()">立即跳转</button>
<script>
function handleRedirect() {
// 获取用户输入内容
const input = document.getElementById('numberInput').value.trim();
// 验证输入格式
if (!/^\d{6,8}$/.test(input)) {
alert("请输入6-8位的纯数字");
return;
}
// 转换为数字并验证数值范围
const number = parseInt(input, 10);
if (number < 100000 || number > 99999999) {
alert("数字范围应为100000到99999999");
return;
}
// 计算ID值
const calculatedID = number - 50;
// 验证计算结果
if (calculatedID < 0) {
alert("计算结果不能为负数");
return;
}
// 构建并打开URL
window.open(`https://abc.xyz/def.aspx?ID=${calculatedID}`, '_blank');
}
</script>
<style>
body {
font-family: Arial, sans-serif;
padding: 20px;
}
input {
padding: 8px;
margin-right: 10px;
width: 200px;
}
button {
padding: 8px 16px;
background-color: #4CAF50;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
}
button:hover {
background-color: #45a049;
}
</style>
</body>
</html>代码特点说明：
-
输入验证：
- 使用正则表达式
/^\d{6,8}$/确保输入为6-8位纯数字 - 数值范围验证（100000-99999999）
- 计算结果非负验证
- 使用正则表达式
-
用户体验：
- 清晰的错误提示
- 输入框和按钮的基础样式
- 按钮悬停效果
-
安全性：
- 自动去除输入内容前后空格
- 显式指定数字解析的十进制格式
- 使用模板字符串构建URL防止注入
使用方法：
- 在输入框输入6-8位数字
- 点击按钮后：
- 如果输入合法 → 新标签页打开目标URL
- 如果输入不合法 → 显示对应错误提示
注意：实际使用时请将URL中的https://abc.xyz/def.aspx替换为真实的目标地址。
import os
import argparse
from openpyxl import load_workbook
import xlrd
def column_to_letter(col):
"""将列索引转换为Excel列字母（例如：0 -> A，26 -> AA）"""
letter = ''
while col >= 0:
letter = chr(col % 26 + 65) + letter
col = col // 26 - 1
return letter
def process_xlsx(file_path, keywords):
"""处理.xlsx文件"""
results = []
try:
wb = load_workbook(filename=file_path, read_only=True, data_only=True)
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
for row in ws.iter_rows():
for cell in row:
if cell.value is not None:
cell_value = str(cell.value)
matched_keywords = [k for k in keywords if k in cell_value]
for keyword in matched_keywords:
results.append({
'file': file_path,
'sheet': sheet_name,
'cell_address': cell.coordinate,
'keyword': keyword
})
wb.close()
except Exception as e:
print(f"Error processing {file_path}: {str(e)}")
return results
def process_xls(file_path, keywords):
"""处理.xls文件"""
results = []
try:
wb = xlrd.open_workbook(file_path)
for sheet in wb.sheets():
sheet_name = sheet.name
for row_idx in range(sheet.nrows):
for col_idx in range(sheet.ncols):
cell = sheet.cell(row_idx, col_idx)
cell_value = str(cell.value)
matched_keywords = [k for k in keywords if k in cell_value]
for keyword in matched_keywords:
row_number = row_idx + 1
col_letter = column_to_letter(col_idx)
cell_address = f"{col_letter}{row_number}"
results.append({
'file': file_path,
'sheet': sheet_name,
'cell_address': cell_address,
'keyword': keyword
})
except Exception as e:
print(f"Error processing {file_path}: {str(e)}")
return results
def search_excel_files(folder_path, keywords):
"""递归搜索文件夹中的Excel文件"""
results = []
for root, _, files in os.walk(folder_path):
for file in files:
file_path = os.path.join(root, file)
if file.endswith('.xlsx'):
results.extend(process_xlsx(file_path, keywords))
elif file.endswith('.xls'):
results.extend(process_xls(file_path, keywords))
return results
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='在Excel文件中搜索关键词')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
args = parser.parse_args()
search_results = search_excel_files(args.folder, args.keywords)
for result in search_results:
print(f"文件：{result['file']}")
print(f"工作表：{result['sheet']}")
print(f"单元格：{result['cell_address']}")
print(f"关键词：{result['keyword']}\n{'='*50}")再来
import os
import argparse
from openpyxl import load_workbook
import xlrd
from tqdm import tqdm
def column_to_letter(col):
"""将列索引转换为Excel列字母（例如：0 -> A，26 -> AA）"""
letter = ''
while col >= 0:
letter = chr(col % 26 + 65) + letter
col = col // 26 - 1
return letter
def process_xlsx(file_path, keywords):
"""处理.xlsx文件"""
results = []
try:
wb = load_workbook(filename=file_path, read_only=True, data_only=True)
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
for row in ws.iter_rows():
for cell in row:
if cell.value is not None:
cell_value = str(cell.value)
matched_keywords = [k for k in keywords if k in cell_value]
for keyword in matched_keywords:
results.append({
'file': file_path,
'sheet': sheet_name,
'cell_address': cell.coordinate,
'keyword': keyword
})
wb.close()
except Exception as e:
print(f"\nError processing {file_path}: {str(e)}")
return results
def process_xls(file_path, keywords):
"""处理.xls文件"""
results = []
try:
wb = xlrd.open_workbook(file_path)
for sheet in wb.sheets():
sheet_name = sheet.name
for row_idx in range(sheet.nrows):
for col_idx in range(sheet.ncols):
cell = sheet.cell(row_idx, col_idx)
cell_value = str(cell.value)
matched_keywords = [k for k in keywords if k in cell_value]
for keyword in matched_keywords:
row_number = row_idx + 1
col_letter = column_to_letter(col_idx)
cell_address = f"{col_letter}{row_number}"
results.append({
'file': file_path,
'sheet': sheet_name,
'cell_address': cell_address,
'keyword': keyword
})
except Exception as e:
print(f"\nError processing {file_path}: {str(e)}")
return results
def get_excel_files(folder_path):
"""获取所有Excel文件路径"""
excel_files = []
for root, _, files in os.walk(folder_path):
for file in files:
if file.lower().endswith(('.xlsx', '.xls')):
excel_files.append(os.path.join(root, file))
return excel_files
def search_excel_files_with_progress(folder_path, keywords):
"""带进度条的搜索函数"""
print("正在扫描Excel文件...")
file_list = get_excel_files(folder_path)
total_files = len(file_list)
results = []
with tqdm(total=total_files, desc="搜索进度", unit="file", ncols=100) as pbar:
for file_path in file_list:
if file_path.endswith('.xlsx'):
results.extend(process_xlsx(file_path, keywords))
elif file_path.endswith('.xls'):
results.extend(process_xls(file_path, keywords))
pbar.update(1)
pbar.set_postfix({"当前文件": os.path.basename(file_path)[:20]})
return results
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='在Excel文件中搜索关键词')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
args = parser.parse_args()
search_results = search_excel_files_with_progress(args.folder, args.keywords)
print("\n搜索结果：")
for result in search_results:
print(f"文件：{result['file']}")
print(f"工作表：{result['sheet']}")
print(f"单元格：{result['cell_address']}")
print(f"关键词：{result['keyword']}\n{'='*50}")
再来3
import os
import argparse
from tqdm import tqdm
from python_calamine import CalamineWorkbook
def search_excel_files(folder_path, keywords):
"""带进度条的搜索函数"""
print("正在扫描Excel文件...")
file_list = []
for root, _, files in os.walk(folder_path):
for file in files:
if file.lower().endswith('.xlsx'):
file_list.append(os.path.join(root, file))
results = []
total_files = len(file_list)
with tqdm(total=total_files, desc="搜索进度", unit="file", ncols=100) as pbar:
for file_path in file_list:
try:
# 使用calamine读取Excel文件
wb = CalamineWorkbook.from_path(file_path)
for sheet in wb.sheets:
df = sheet.to_python(skip_empty_rows=False)
for row_idx, row in enumerate(df, 1):
for col_idx, cell in enumerate(row, 1):
cell_value = str(cell) if cell is not None else ""
for keyword in keywords:
if keyword in cell_value:
cell_address = f"{column_to_letter(col_idx-1)}{row_idx}"
results.append({
'file': file_path,
'sheet': sheet.name,
'cell_address': cell_address,
'keyword': keyword
})
except Exception as e:
print(f"\n错误处理文件 {file_path}: {str(e)}")
finally:
pbar.update(1)
pbar.set_postfix({"当前文件": os.path.basename(file_path)[:20]})
return results
def column_to_letter(col):
"""将列索引转换为Excel列字母（例如：0 -> A，26 -> AA）"""
letter = ''
while col >= 0:
letter = chr(col % 26 + 65) + letter
col = col // 26 - 1
return letter or 'A'
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='在Excel文件中搜索关键词')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
args = parser.parse_args()
search_results = search_excel_files(args.folder, args.keywords)
print("\n搜索结果：")
for result in search_results:
print(f"文件：{result['file']}")
print(f"工作表：{result['sheet']}")
print(f"单元格：{result['cell_address']}")
print(f"关键词：{result['keyword']}\n{'='*50}")再4
import os
import argparse
import pandas as pd
from tqdm import tqdm
def convert_size_to_bytes(size_str):
"""将人类可读的文件大小转换为字节数"""
units = {'B': 1, 'K': 1024, 'M': 1024**2, 'G': 1024**3}
unit = size_str[-1].upper()
if unit.isdigit():
return int(size_str)
value = float(size_str[:-1])
return int(value * units[unit])
def column_to_letter(col):
"""将列索引转换为Excel列字母"""
letter = ''
while col >= 0:
letter = chr(col % 26 + 65) + letter
col = col // 26 - 1
return letter or 'A'
def search_excel_files(folder_path, keywords, min_size=None, max_size=None):
"""带进度条和文件大小过滤的搜索函数"""
print("正在扫描并筛选Excel文件...")
file_list = []
# 转换文件大小单位为字节
min_bytes = convert_size_to_bytes(min_size) if min_size else 0
max_bytes = convert_size_to_bytes(max_size) if max_size else float('inf')
# 遍历文件并筛选
for root, _, files in os.walk(folder_path):
for file in files:
if file.lower().endswith('.xlsx'):
file_path = os.path.join(root, file)
file_size = os.path.getsize(file_path)
if min_bytes <= file_size <= max_bytes:
file_list.append(file_path)
results = []
total_files = len(file_list)
with tqdm(total=total_files, desc="搜索进度", unit="file", ncols=100) as pbar:
for file_path in file_list:
try:
# 使用Pandas读取Excel
xls = pd.ExcelFile(file_path, engine='openpyxl')
for sheet_name in xls.sheet_names:
df = xls.parse(sheet_name, header=None, dtype=str)
# 遍历所有单元格
for row_idx, row in df.iterrows():
for col_idx, cell in enumerate(row):
cell_value = str(cell) if pd.notnull(cell) else ""
for keyword in keywords:
if keyword in cell_value:
cell_address = f"{column_to_letter(col_idx)}{row_idx+1}"
results.append({
'file': file_path,
'sheet': sheet_name,
'cell_address': cell_address,
'keyword': keyword
})
xls.close()
except Exception as e:
print(f"\n错误处理文件 {file_path}: {str(e)}")
finally:
pbar.update(1)
pbar.set_postfix({"当前文件": os.path.basename(file_path)[:20]})
return results
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='在Excel文件中搜索关键词')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
parser.add_argument('--min-size', help='最小文件大小（例如：1M, 500K）')
parser.add_argument('--max-size', help='最大文件大小（例如：10M, 2G）')
args = parser.parse_args()
search_results = search_excel_files(
args.folder,
args.keywords,
min_size=args.min_size,
max_size=args.max_size
)
print("\n搜索结果：")
for result in search_results:
print(f"文件：{result['file']}")
print(f"工作表：{result['sheet']}")
print(f"单元格：{result['cell_address']}")
print(f"关键词：{result['keyword']}\n{'='*50}")
再5
import os
import argparse
import pandas as pd
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor
from python_calamine.pandad import pandas_monkeypatch
# 应用calamine补丁替换pandas的Excel引擎
pandas_monkeypatch()
def convert_size_to_bytes(size_str):
"""将人类可读的文件大小转换为字节数"""
units = {'B': 1, 'K': 1024, 'M': 1024**2, 'G': 1024**3}
if not size_str: return 0
unit = size_str[-1].upper()
if unit.isdigit():
return int(size_str)
value = float(size_str[:-1])
return int(value * units.get(unit, 1))
def column_to_letter(col):
"""将列索引转换为Excel列字母"""
letter = ''
while col >= 0:
letter = chr(col % 26 + 65) + letter
col = col // 26 - 1
return letter or 'A'
def process_single_file(file_path, keywords):
"""处理单个文件并返回结果"""
results = []
try:
with pd.ExcelFile(file_path, engine="calamine") as xls:
for sheet_name in xls.sheet_names:
df = xls.parse(
sheet_name,
header=None,
dtype=str,
na_filter=False,
engine="calamine"
)
# 遍历所有单元格
for row_idx, row in df.iterrows():
for col_idx, cell in enumerate(row):
if pd.isna(cell):
continue
cell_value = str(cell)
for keyword in keywords:
if keyword in cell_value:
cell_address = f"{column_to_letter(col_idx)}{row_idx+1}"
results.append({
'file': file_path,
'sheet': sheet_name,
'cell_address': cell_address,
'keyword': keyword
})
except Exception as e:
print(f"\n错误处理文件 {file_path}: {str(e)}")
return results
def search_excel_files(folder_path, keywords, min_size=None, max_size=None, workers=4):
"""多线程搜索函数"""
print("正在扫描并筛选Excel文件...")
file_list = []
# 转换文件大小单位
min_bytes = convert_size_to_bytes(min_size) if min_size else 0
max_bytes = convert_size_to_bytes(max_size) if max_size else float('inf')
# 遍历文件并筛选
for root, _, files in os.walk(folder_path):
for file in files:
if file.lower().endswith('.xlsx'):
file_path = os.path.join(root, file)
file_size = os.path.getsize(file_path)
if min_bytes <= file_size <= max_bytes:
file_list.append(file_path)
total_files = len(file_list)
results = []
with ThreadPoolExecutor(max_workers=workers) as executor:
# 创建进度条
with tqdm(total=total_files, desc="搜索进度", unit="file", ncols=100) as pbar:
# 提交任务
futures = []
for file_path in file_list:
future = executor.submit(process_single_file, file_path, keywords)
future.add_done_callback(lambda _: pbar.update(1))
futures.append(future)
# 收集结果
for future in futures:
results.extend(future.result())
pbar.set_postfix({"当前进度": f"{pbar.n}/{pbar.total}"})
return results
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='多线程Excel文件搜索工具')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
parser.add_argument('--min-size', help='最小文件大小（例如：1M, 500K）')
parser.add_argument('--max-size', help='最大文件大小（例如：10M, 2G）')
parser.add_argument('-w', '--workers', type=int, default=4,
help='线程池工作线程数（默认：4）')
args = parser.parse_args()
search_results = search_excel_files(
args.folder,
args.keywords,
min_size=args.min_size,
max_size=args.max_size,
workers=args.workers
)
print("\n搜索结果：")
for result in search_results:
print(f"文件：{result['file']}")
print(f"工作表：{result['sheet']}")
print(f"单元格：{result['cell_address']}")
print(f"关键词：{result['keyword']}\n{'='*50}")再6
import os
import argparse
import pandas as pd
import threading
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor
from python_calamine.pandad import pandas_monkeypatch
# 应用calamine补丁替换pandas的Excel引擎
pandas_monkeypatch()
# 全局锁用于保证文件写入安全
result_lock = threading.Lock()
status_lock = threading.Lock()
class SearchManager:
def __init__(self, folder, result_file="search_results.log", status_file="search_status.txt"):
self.folder = folder
self.result_file = os.path.join(folder, result_file)
self.status_file = os.path.join(folder, status_file)
# 初始化已完成文件记录
self.completed_files = set()
if os.path.exists(self.status_file):
with open(self.status_file, 'r') as f:
self.completed_files = set(f.read().splitlines())
# 初始化结果文件（追加模式）
if not os.path.exists(self.result_file):
with open(self.result_file, 'w') as f:
f.write("文件路径,工作表名,单元格地址,关键词\n")
def record_result(self, result):
"""线程安全地记录结果"""
with result_lock:
# 实时显示结果
print(f"[实时发现] {result['file']} -> {result['sheet']}!{result['cell_address']} ({result['keyword']})")
# 保存到结果文件
with open(self.result_file, 'a') as f:
f.write(f"{result['file']},{result['sheet']},{result['cell_address']},{result['keyword']}\n")
def mark_completed(self, file_path):
"""标记文件已完成处理"""
with status_lock:
with open(self.status_file, 'a') as f:
f.write(f"{file_path}\n")
self.completed_files.add(file_path)
def convert_size_to_bytes(size_str):
"""将人类可读的文件大小转换为字节数"""
units = {'B': 1, 'K': 1024, 'M': 1024**2, 'G': 1024**3}
if not size_str: return 0
unit = size_str[-1].upper()
if unit.isdigit():
return int(size_str)
value = float(size_str[:-1])
return int(value * units.get(unit, 1))
def column_to_letter(col):
"""将列索引转换为Excel列字母"""
letter = ''
while col >= 0:
letter = chr(col % 26 + 65) + letter
col = col // 26 - 1
return letter or 'A'
def process_single_file(file_path, keywords, manager):
"""处理单个文件并返回结果"""
results = []
try:
with pd.ExcelFile(file_path, engine="calamine") as xls:
for sheet_name in xls.sheet_names:
df = xls.parse(
sheet_name,
header=None,
dtype=str,
na_filter=False,
engine="calamine"
)
# 遍历所有单元格
for row_idx, row in df.iterrows():
for col_idx, cell in enumerate(row):
if pd.isna(cell):
continue
cell_value = str(cell)
for keyword in keywords:
if keyword in cell_value:
cell_address = f"{column_to_letter(col_idx)}{row_idx+1}"
result = {
'file': file_path,
'sheet': sheet_name,
'cell_address': cell_address,
'keyword': keyword
}
results.append(result)
manager.record_result(result)
# 标记文件完成处理
manager.mark_completed(file_path)
except Exception as e:
print(f"\n错误处理文件 {file_path}: {str(e)}")
return results
def search_excel_files(folder_path, keywords, min_size=None, max_size=None, workers=4):
"""支持断点续查的多线程搜索函数"""
manager = SearchManager(folder_path)
print("正在扫描并筛选Excel文件...")
file_list = []
# 转换文件大小单位
min_bytes = convert_size_to_bytes(min_size) if min_size else 0
max_bytes = convert_size_to_bytes(max_size) if max_size else float('inf')
# 遍历文件并筛选
for root, _, files in os.walk(folder_path):
for file in files:
if file.lower().endswith('.xlsx'):
file_path = os.path.join(root, file)
# 跳过已处理文件
if file_path in manager.completed_files:
continue
file_size = os.path.getsize(file_path)
if min_bytes <= file_size <= max_bytes:
file_list.append(file_path)
total_files = len(file_list)
results = []
with ThreadPoolExecutor(max_workers=workers) as executor:
# 创建进度条
with tqdm(total=total_files, desc="搜索进度", unit="file", ncols=100) as pbar:
# 提交任务
futures = []
for file_path in file_list:
future = executor.submit(process_single_file, file_path, keywords, manager)
future.add_done_callback(lambda _: pbar.update(1))
futures.append(future)
# 收集结果
for future in futures:
try:
results.extend(future.result())
except Exception as e:
print(f"\n任务执行错误: {str(e)}")
pbar.set_postfix({"剩余文件": f"{total_files - pbar.n}"})
# 完成后清理状态文件（如需保留注释此行）
# os.remove(manager.status_file)
return results
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='支持断点续查的多线程Excel搜索工具')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
parser.add_argument('--min-size', help='最小文件大小（例如：1M, 500K）')
parser.add_argument('--max-size', help='最大文件大小（例如：10M, 2G）')
parser.add_argument('-w', '--workers', type=int, default=4,
help='线程池工作线程数（默认：4）')
args = parser.parse_args()
search_results = search_excel_files(
args.folder,
args.keywords,
min_size=args.min_size,
max_size=args.max_size,
workers=args.workers
)
print("\n最终搜索结果已保存至：")
print(f"- 实时日志：{os.path.join(args.folder, 'search_results.log')}")
print(f"- 断点状态：{os.path.join(args.folder, 'search_status.txt')}")
再7
import os
import argparse
import pandas as pd
import threading
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor
from python_calamine.pandad import pandas_monkeypatch
# 应用calamine补丁替换pandas的Excel引擎
pandas_monkeypatch()
# 全局锁用于保证文件写入安全
result_lock = threading.Lock()
status_lock = threading.Lock()
class SearchManager:
def __init__(self, folder, result_file="search_results.log", status_file="search_status.txt"):
self.folder = folder
self.result_file = os.path.join(folder, result_file)
self.status_file = os.path.join(folder, status_file)
# 初始化已完成文件记录
self.completed_files = set()
if os.path.exists(self.status_file):
with open(self.status_file, 'r') as f:
self.completed_files = set(f.read().splitlines())
# 初始化结果文件（追加模式）
if not os.path.exists(self.result_file):
with open(self.result_file, 'w', encoding='utf-8') as f:
f.write("文件路径,工作表名,单元格地址,单元格内容,关键词\n")
def record_result(self, result):
"""线程安全地记录结果"""
with result_lock:
# 实时显示结果（显示前100个字符）
display_content = result['cell_content'][:100].replace('\n', ' ')
print(f"[实时发现] {result['file']} -> {result['sheet']}!{result['cell_address']}")
print(f" 内容片段：{display_content}...")
print(f" 匹配关键词：{result['keyword']}\n{'━'*50}")
# 保存到结果文件（完整内容）
with open(self.result_file, 'a', encoding='utf-8') as f:
content = result['cell_content'].replace('\n', '\\n').replace(',', '，')
f.write(f"{result['file']},{result['sheet']},{result['cell_address']},\"{content}\",{result['keyword']}\n")
def mark_completed(self, file_path):
"""标记文件已完成处理"""
with status_lock:
with open(self.status_file, 'a') as f:
f.write(f"{file_path}\n")
self.completed_files.add(file_path)
def convert_size_to_bytes(size_str):
"""将人类可读的文件大小转换为字节数"""
units = {'B': 1, 'K': 1024, 'M': 1024**2, 'G': 1024**3}
if not size_str: return 0
unit = size_str[-1].upper()
if unit.isdigit():
return int(size_str)
value = float(size_str[:-1])
return int(value * units.get(unit, 1))
def column_to_letter(col):
"""将列索引转换为Excel列字母"""
letter = ''
while col >= 0:
letter = chr(col % 26 + 65) + letter
col = col // 26 - 1
return letter or 'A'
def process_single_file(file_path, keywords, manager):
"""处理单个文件并返回结果"""
results = []
try:
with pd.ExcelFile(file_path, engine="calamine") as xls:
for sheet_name in xls.sheet_names:
df = xls.parse(
sheet_name,
header=None,
dtype=str,
na_filter=False,
engine="calamine"
)
# 遍历所有单元格
for row_idx, row in df.iterrows():
for col_idx, cell in enumerate(row):
if pd.isna(cell):
continue
cell_value = str(cell)
for keyword in keywords:
if keyword in cell_value:
cell_address = f"{column_to_letter(col_idx)}{row_idx+1}"
result = {
'file': file_path,
'sheet': sheet_name,
'cell_address': cell_address,
'cell_content': cell_value,
'keyword': keyword
}
results.append(result)
manager.record_result(result)
# 标记文件完成处理
manager.mark_completed(file_path)
except Exception as e:
print(f"\n错误处理文件 {file_path}: {str(e)}")
return results
def search_excel_files(folder_path, keywords, min_size=None, max_size=None, workers=4):
"""支持断点续查的多线程搜索函数"""
manager = SearchManager(folder_path)
print("正在扫描并筛选Excel文件...")
file_list = []
# 转换文件大小单位
min_bytes = convert_size_to_bytes(min_size) if min_size else 0
max_bytes = convert_size_to_bytes(max_size) if max_size else float('inf')
# 遍历文件并筛选
for root, _, files in os.walk(folder_path):
for file in files:
if file.lower().endswith('.xlsx'):
file_path = os.path.join(root, file)
# 跳过已处理文件
if file_path in manager.completed_files:
continue
file_size = os.path.getsize(file_path)
if min_bytes <= file_size <= max_bytes:
file_list.append(file_path)
total_files = len(file_list)
results = []
with ThreadPoolExecutor(max_workers=workers) as executor:
# 创建进度条
with tqdm(total=total_files, desc="搜索进度", unit="file", ncols=100) as pbar:
# 提交任务
futures = []
for file_path in file_list:
future = executor.submit(process_single_file, file_path, keywords, manager)
future.add_done_callback(lambda _: pbar.update(1))
futures.append(future)
# 收集结果
for future in futures:
try:
results.extend(future.result())
except Exception as e:
print(f"\n任务执行错误: {str(e)}")
pbar.set_postfix({"剩余文件": f"{total_files - pbar.n}"})
print("\n最终搜索结果已保存至：")
print(f"- 实时日志：{os.path.join(folder_path, 'search_results.log')}")
print(f"- 断点状态：{os.path.join(folder_path, 'search_status.txt')}")
return results
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='支持单元格内容输出的Excel搜索工具')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
parser.add_argument('--min-size', help='最小文件大小（例如：1M, 500K）')
parser.add_argument('--max-size', help='最大文件大小（例如：10M, 2G）')
parser.add_argument('-w', '--workers', type=int, default=4,
help='线程池工作线程数（默认：4）')
args = parser.parse_args()
search_results = search_excel_files(
args.folder,
args.keywords,
min_size=args.min_size,
max_size=args.max_size,
workers=args.workers
)再8
import os
import argparse
import pandas as pd
import threading
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor
from python_calamine.pandas import pandas_monkeypatch
# 应用calamine补丁替换pandas的Excel引擎
pandas_monkeypatch()
# 全局锁用于保证文件写入安全
result_lock = threading.Lock()
status_lock = threading.Lock()
class SearchManager:
def __init__(self, folder, result_file="search_results.log", status_file="search_status.txt"):
# ...（保持原有初始化逻辑不变）...
def record_result(self, result):
# ...（保持原有记录逻辑不变）...
def mark_completed(self, file_path):
# ...（保持原有标记逻辑不变）...
def convert_size_to_bytes(size_str):
# ...（保持原有转换逻辑不变）...
def column_to_letter(col):
# ...（保持原有转换逻辑不变）...
def process_single_file(file_path, keywords, exclude_keywords, manager):
"""处理单个文件并返回结果（增加排除关键词逻辑）"""
results = []
try:
with pd.ExcelFile(file_path, engine="calamine") as xls:
for sheet_name in xls.sheet_names:
df = xls.parse(
sheet_name,
header=None,
dtype=str,
na_filter=False,
engine="calamine"
)
# 遍历所有单元格
for row_idx, row in df.iterrows():
for col_idx, cell in enumerate(row):
if pd.isna(cell):
continue
cell_value = str(cell)
# 检查排除关键词
if any(ex_kw in cell_value for ex_kw in exclude_keywords):
continue
# 检查包含关键词
matched_keywords = [kw for kw in keywords if kw in cell_value]
for keyword in matched_keywords:
cell_address = f"{column_to_letter(col_idx)}{row_idx+1}"
result = {
'file': file_path,
'sheet': sheet_name,
'cell_address': cell_address,
'cell_content': cell_value,
'keyword': keyword
}
results.append(result)
manager.record_result(result)
manager.mark_completed(file_path)
except Exception as e:
print(f"\n错误处理文件 {file_path}: {str(e)}")
return results
def search_excel_files(folder_path, keywords, exclude_keywords, min_size=None, max_size=None, workers=4):
"""支持排除关键词的搜索函数"""
manager = SearchManager(folder_path)
print("正在扫描并筛选Excel文件...")
file_list = []
# 转换文件大小单位
min_bytes = convert_size_to_bytes(min_size) if min_size else 0
max_bytes = convert_size_to_bytes(max_size) if max_size else float('inf')
# 遍历文件并筛选
for root, _, files in os.walk(folder_path):
for file in files:
if file.lower().endswith('.xlsx'):
file_path = os.path.join(root, file)
if file_path in manager.completed_files:
continue
file_size = os.path.getsize(file_path)
if min_bytes <= file_size <= max_bytes:
file_list.append(file_path)
total_files = len(file_list)
results = []
with ThreadPoolExecutor(max_workers=workers) as executor:
with tqdm(total=total_files, desc="搜索进度", unit="file", ncols=100) as pbar:
futures = []
for file_path in file_list:
future = executor.submit(
process_single_file,
file_path,
keywords,
exclude_keywords, # 传递排除关键词
manager
)
future.add_done_callback(lambda _: pbar.update(1))
futures.append(future)
for future in futures:
try:
results.extend(future.result())
except Exception as e:
print(f"\n任务执行错误: {str(e)}")
pbar.set_postfix({"剩余文件": f"{total_files - pbar.n}"})
print("\n最终搜索结果已保存至：")
print(f"- 实时日志：{os.path.join(folder_path, 'search_results.log')}")
print(f"- 断点状态：{os.path.join(folder_path, 'search_status.txt')}")
return results
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='支持关键词排除的Excel搜索工具')
parser.add_argument('folder', help='要搜索的文件夹路径')
parser.add_argument('-k', '--keywords', nargs='+', required=True,
help='要搜索的关键词列表（用空格分隔）')
parser.add_argument('-e', '--exclude', nargs='+', default=[],
help='要排除的关键词列表（用空格分隔）')
parser.add_argument('--min-size', help='最小文件大小（例如：1M, 500K）')
parser.add_argument('--max-size', help='最大文件大小（例如：10M, 2G）')
parser.add_argument('-w', '--workers', type=int, default=4,
help='线程池工作线程数（默认：4）')
args = parser.parse_args()
search_results = search_excel_files(
args.folder,
args.keywords,
exclude_keywords=args.exclude, # 新增排除关键词参数
min_size=args.min_size,
max_size=args.max_size,
workers=args.workers
)import os
import shutil
from tqdm import tqdm
def move_directory_with_progress(src, dst):
"""移动目录并显示进度条
Args:
src (str): 源目录路径
dst (str): 目标目录路径
"""
# 获取源目录的基本名称
src_base = os.path.basename(src)
# 确定最终目标路径
if os.path.exists(dst) and os.path.isdir(dst):
final_dst = os.path.join(dst, src_base)
else:
final_dst = dst
# 确保目标父目录存在
os.makedirs(os.path.dirname(final_dst), exist_ok=True)
# 计算总文件数用于进度条
total_files = 0
for root, _, files in os.walk(src):
total_files += len(files)
# 使用 tqdm 创建进度条
with tqdm(total=total_files, unit='file', desc='Moving files') as pbar:
# 自定义复制函数用于更新进度条
def copy_with_progress(src_path, dst_path):
shutil.copy2(src_path, dst_path)
pbar.update(1)
# 执行目录移动（复制+删除）
try:
# 复制目录结构
shutil.copytree(
src,
final_dst,
copy_function=copy_with_progress,
dirs_exist_ok=True
)
# 删除原目录
shutil.rmtree(src)
except Exception as e:
# 错误处理：清理部分复制的文件
if os.path.exists(final_dst):
shutil.rmtree(final_dst)
raise RuntimeError(f"移动失败: {str(e)}")
if __name__ == "__main__":
# 配置你的路径
pathA = "/path/to/source/directory" # 替换为你的源目录
pathB = "/path/to/destination" # 替换为你的目标目录
# 执行移动操作
move_directory_with_progress(pathA, pathB)
print(f"\n操作完成！'{pathA}' 已移动到 '{pathB}'")import os
import shutil
import sys
from tqdm import tqdm
def move_directory_with_progress(src, dst):
"""移动目录并显示进度条（兼容所有 Python 3.x 版本）"""
# 获取源目录名称
src_name = os.path.basename(src.rstrip(os.sep))
# 确定最终目标路径
if os.path.isdir(dst):
final_dst = os.path.join(dst, src_name)
else:
final_dst = dst
# 确保目标父目录存在
os.makedirs(os.path.dirname(final_dst), exist_ok=True)
# 计算总文件数
total_files = 0
for root, _, files in os.walk(src):
total_files += len(files)
# 使用 tqdm 创建进度条
with tqdm(total=total_files, unit='file', desc='Moving files') as pbar:
# 手动复制目录结构
for root, dirs, files in os.walk(src):
# 计算相对路径
rel_path = os.path.relpath(root, src)
dest_dir = os.path.join(final_dst, rel_path)
# 创建目标目录（如果不存在）
if not os.path.exists(dest_dir):
os.makedirs(dest_dir)
# 复制文件
for file in files:
src_file = os.path.join(root, file)
dest_file = os.path.join(dest_dir, file)
# 复制文件并保留元数据
shutil.copy2(src_file, dest_file)
pbar.update(1)
# 复制完成后删除源目录
shutil.rmtree(src)
if __name__ == "__main__":
# 配置你的路径
pathA = "/path/to/source/directory" # 替换为你的源目录
pathB = "/path/to/destination" # 替换为你的目标目录
try:
# 检查路径是否存在
if not os.path.exists(pathA):
print(f"错误：源路径 '{pathA}' 不存在")
sys.exit(1)
print(f"开始移动: {pathA} -> {pathB}")
move_directory_with_progress(pathA, pathB)
print(f"\n操作完成！'{pathA}' 已成功移动到 '{pathB}'")
except Exception as e:
print(f"\n移动过程中出错: {str(e)}")
sys.exit(1)