108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
# backend/script_groups/example_group/x1.py
|
|
from backend.script_groups.base_script import BaseScript
|
|
import os
|
|
from pathlib import Path
|
|
import json
|
|
import csv
|
|
from datetime import datetime
|
|
|
|
class FileCounter(BaseScript):
|
|
"""
|
|
File Analysis
|
|
Analyzes files in directory with configurable filters and reporting
|
|
"""
|
|
|
|
def run(self, work_dir: str, profile: dict) -> dict:
|
|
try:
|
|
# Get configuration
|
|
config = self.get_config(work_dir, "example_group")
|
|
|
|
# Process configuration values
|
|
exclude_dirs = [d.strip() for d in config.get("exclude_dirs", "").split(",") if d.strip()]
|
|
count_hidden = config.get("count_hidden", False)
|
|
min_size = config.get("min_size", 0)
|
|
save_report = config.get("save_report", True)
|
|
report_format = config.get("report_format", "json")
|
|
|
|
# Initialize counters
|
|
extension_counts = {}
|
|
total_files = 0
|
|
total_size = 0
|
|
skipped_files = 0
|
|
|
|
# Walk through directory
|
|
for root, dirs, files in os.walk(work_dir):
|
|
# Skip excluded directories
|
|
dirs[:] = [d for d in dirs if d not in exclude_dirs]
|
|
|
|
for file in files:
|
|
file_path = Path(root) / file
|
|
|
|
# Skip hidden files if not counting them
|
|
if not count_hidden and file.startswith('.'):
|
|
skipped_files += 1
|
|
continue
|
|
|
|
# Check file size
|
|
try:
|
|
file_size = file_path.stat().st_size
|
|
if file_size < min_size:
|
|
skipped_files += 1
|
|
continue
|
|
except:
|
|
continue
|
|
|
|
# Count file
|
|
total_files += 1
|
|
total_size += file_size
|
|
ext = file_path.suffix.lower() or 'no extension'
|
|
extension_counts[ext] = extension_counts.get(ext, 0) + 1
|
|
|
|
# Prepare results
|
|
results = {
|
|
"scan_time": datetime.now().isoformat(),
|
|
"total_files": total_files,
|
|
"total_size": total_size,
|
|
"skipped_files": skipped_files,
|
|
"extension_counts": extension_counts
|
|
}
|
|
|
|
# Save report if configured
|
|
if save_report:
|
|
report_path = Path(work_dir) / f"file_analysis.{report_format}"
|
|
if report_format == "json":
|
|
with open(report_path, 'w') as f:
|
|
json.dump(results, f, indent=2)
|
|
elif report_format == "csv":
|
|
with open(report_path, 'w', newline='') as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow(["Extension", "Count"])
|
|
for ext, count in sorted(extension_counts.items()):
|
|
writer.writerow([ext, count])
|
|
else: # txt
|
|
with open(report_path, 'w') as f:
|
|
f.write(f"File Analysis Report\n")
|
|
f.write(f"Generated: {results['scan_time']}\n\n")
|
|
f.write(f"Total Files: {total_files}\n")
|
|
f.write(f"Total Size: {total_size:,} bytes\n")
|
|
f.write(f"Skipped Files: {skipped_files}\n\n")
|
|
f.write("Extension Counts:\n")
|
|
for ext, count in sorted(extension_counts.items()):
|
|
f.write(f"{ext}: {count}\n")
|
|
|
|
return {
|
|
"status": "success",
|
|
"data": results,
|
|
"output": f"Found {total_files:,} files ({total_size:,} bytes)\n" +
|
|
f"Skipped {skipped_files} files\n\n" +
|
|
"Extensions:\n" + "\n".join(
|
|
f"{ext}: {count:,} files"
|
|
for ext, count in sorted(extension_counts.items())
|
|
)
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"error": str(e)
|
|
} |