From 4a579be2f5a5e312df64a687cd00788ed001f07e Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:07:54 -0300 Subject: [PATCH 01/29] ci: add lint and test workflow --- .github/workflows/ci.yml | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..c91f494 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,23 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + - name: Install dependencies + run: pip install -r requirements-dev.txt + - name: Lint + run: ruff check . && ruff format --check . + - name: Run tests + run: pytest -v From 570642886d1806b3d8d853d59f7e052b996c7687 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:07:58 -0300 Subject: [PATCH 02/29] build: add ruff configuration --- pyproject.toml | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 pyproject.toml diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e3e67d8 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,6 @@ +[tool.ruff] +line-length = 100 +exclude = [".venv"] + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "RUF"] From 8d03b18db7987ebbce31f0f007bec1cbbe7d362a Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:03 -0300 Subject: [PATCH 03/29] test: add pytest configuration --- pytest.ini | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 pytest.ini diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..f607c7b --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +python_files = test_*.py +testpaths = tests From f45ed6109f1ce1a29545704f0674fe7570c14e8c Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:08 -0300 Subject: [PATCH 04/29] build: add development requirements --- requirements-dev.txt | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 requirements-dev.txt diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..5de43ca --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,3 @@ +-r requirements.txt +pytest==9.1.1 +ruff==0.16.9 From 10af9a736877cda1f76592c7a3fcec453b9b8f3d Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:12 -0300 Subject: [PATCH 05/29] build: make scripts an importable package --- scripts/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 scripts/__init__.py diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..e69de29 From 9fedb48c78cffd66e7f605d8ee14f8fe31772344 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:22 -0300 Subject: [PATCH 06/29] refactor: translate file organizer and separate logic from CLI --- scripts/file_organizer.py | 98 +++++++++++++++++++-------------------- 1 file changed, 47 insertions(+), 51 deletions(-) diff --git a/scripts/file_organizer.py b/scripts/file_organizer.py index 73f42d1..591018f 100644 --- a/scripts/file_organizer.py +++ b/scripts/file_organizer.py @@ -1,83 +1,79 @@ -""" -File Organizer - organiza arquivos em pastas por categoria. - -Uso: python scripts/file_organizer.py -""" -import sys +import argparse import shutil +import sys from pathlib import Path - CATEGORIES = { - "Documentos": [".pdf", ".doc", ".docx", ".txt", ".odt", ".rtf"], - "Planilhas": [".xls", ".xlsx", ".csv", ".ods"], - "Apresentacoes": [".ppt", ".pptx", ".odp"], - "Imagens": [".jpg", ".jpeg", ".png", ".gif", ".bmp", ".svg", ".webp"], + "Documents": [".pdf", ".doc", ".docx", ".txt", ".odt", ".rtf"], + "Spreadsheets": [".xls", ".xlsx", ".csv", ".ods"], + "Presentations": [".ppt", ".pptx", ".odp"], + "Images": [".jpg", ".jpeg", ".png", ".gif", ".bmp", ".svg", ".webp"], "Videos": [".mp4", ".avi", ".mov", ".mkv", ".wmv", ".flv"], - "Audios": [".mp3", ".wav", ".flac", ".aac", ".ogg"], - "Compactados": [".zip", ".rar", ".7z", ".tar", ".gz"], - "Codigo": [".py", ".js", ".html", ".css", ".java", ".cpp", ".c", ".go"], - "Executaveis": [".exe", ".msi", ".dmg", ".deb", ".rpm"], + "Audio": [".mp3", ".wav", ".flac", ".aac", ".ogg"], + "Archives": [".zip", ".rar", ".7z", ".tar", ".gz"], + "Code": [".py", ".js", ".html", ".css", ".java", ".cpp", ".c", ".go"], + "Executables": [".exe", ".msi", ".dmg", ".deb", ".rpm"], } +DEFAULT_CATEGORY = "Other" def get_category(extension: str) -> str: - """Retorna a categoria de uma extensão.""" - ext_lower = extension.lower() + extension = extension.lower() for category, extensions in CATEGORIES.items(): - if ext_lower in extensions: + if extension in extensions: return category - return "Outros" + return DEFAULT_CATEGORY + + +def unique_destination(directory: Path, name: str) -> Path: + candidate = directory / name + stem, suffix = candidate.stem, candidate.suffix + counter = 1 + while candidate.exists(): + candidate = directory / f"{stem}_{counter}{suffix}" + counter += 1 + return candidate -def organize_folder(folder_path: str) -> dict: - """Organiza uma pasta movendo arquivos para subpastas por categoria.""" +def organize_folder(folder_path: str) -> dict[str, int]: folder = Path(folder_path).expanduser().resolve() if not folder.exists(): - print(f"❌ Pasta não encontrada: {folder}") - return {} - + raise FileNotFoundError(f"Folder not found: {folder}") if not folder.is_dir(): - print(f"❌ Não é uma pasta: {folder}") - return {} - - print(f"📁 Organizando: {folder}\n") + raise NotADirectoryError(f"Not a folder: {folder}") - stats = {} - for item in folder.iterdir(): + stats: dict[str, int] = {} + for item in sorted(folder.iterdir()): if not item.is_file(): continue category = get_category(item.suffix) - dest_dir = folder / category - dest_dir.mkdir(exist_ok=True) - - dest_path = dest_dir / item.name - # Evita sobrescrita - counter = 1 - while dest_path.exists(): - dest_path = dest_dir / f"{item.stem}_{counter}{item.suffix}" - counter += 1 + destination_dir = folder / category + destination_dir.mkdir(exist_ok=True) - shutil.move(str(item), str(dest_path)) + shutil.move(str(item), str(unique_destination(destination_dir, item.name))) stats[category] = stats.get(category, 0) + 1 - print(f" ✓ {item.name} → {category}/") - - print(f"\n📊 Resumo:") - for cat, count in sorted(stats.items()): - print(f" {cat}: {count} arquivo(s)") return stats -def main(): - if len(sys.argv) < 2: - print("Uso: python scripts/file_organizer.py ") - sys.exit(1) +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Organize files into folders by category") + parser.add_argument("folder", help="Folder to organize") + args = parser.parse_args(argv) + + try: + stats = organize_folder(args.folder) + except (FileNotFoundError, NotADirectoryError) as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 - organize_folder(sys.argv[1]) + for category, count in sorted(stats.items()): + print(f"{category}: {count} file(s)") + print(f"Organized {sum(stats.values())} file(s)") + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From b285eaadedb9e2faf9ebb32bc93524101ccde696 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:29 -0300 Subject: [PATCH 07/29] refactor: translate bulk rename, return results and guard against collisions --- scripts/bulk_rename.py | 81 ++++++++++++++++++------------------------ 1 file changed, 35 insertions(+), 46 deletions(-) diff --git a/scripts/bulk_rename.py b/scripts/bulk_rename.py index 859dc6d..ff41c5c 100644 --- a/scripts/bulk_rename.py +++ b/scripts/bulk_rename.py @@ -1,70 +1,59 @@ -""" -Bulk Rename - renomeia arquivos em lote usando regex. - -Uso: - python scripts/bulk_rename.py --pattern "regex" --replacement "novo_nome" - -Exemplos: - # Troca "IMG_" por "foto_" - python scripts/bulk_rename.py ./fotos --pattern "^IMG_" --replacement "foto_" - - # Captura grupos e reusa - python scripts/bulk_rename.py ./fotos --pattern "IMG_(\\d+)" --replacement "foto_{1}" -""" import argparse import re +import sys from pathlib import Path -def bulk_rename(folder: str, pattern: str, replacement: str, dry_run: bool = False): - """Renomeia arquivos que batem com o pattern regex.""" +def bulk_rename( + folder: str, pattern: str, replacement: str, dry_run: bool = False +) -> list[tuple[str, str]]: folder_path = Path(folder).expanduser().resolve() - if not folder_path.is_dir(): - print(f"❌ Pasta inválida: {folder_path}") - return + raise NotADirectoryError(f"Invalid folder: {folder_path}") regex = re.compile(pattern) - count = 0 - - # Converte {1}, {2} etc. para \1, \2 do regex - repl = re.sub(r"\{(\d+)\}", r"\\\1", replacement) + group_replacement = re.sub(r"\{(\d+)\}", r"\\\1", replacement) - print(f"🔍 Procurando em: {folder_path}") - print(f"📝 Pattern: {pattern}") - print(f"📝 Replacement: {replacement}\n") - - for item in folder_path.iterdir(): + renames = [] + for item in sorted(folder_path.iterdir()): if not item.is_file(): continue - new_name = regex.sub(repl, item.name) + new_name = regex.sub(group_replacement, item.name) if new_name == item.name: continue - new_path = item.parent / new_name + target = item.parent / new_name + if target.exists(): + raise FileExistsError(f"Target already exists: {target}") + + if not dry_run: + item.rename(target) + renames.append((item.name, new_name)) - if dry_run: - print(f" [DRY] {item.name} → {new_name}") - else: - item.rename(new_path) - print(f" ✓ {item.name} → {new_name}") - count += 1 + return renames - action = "Simulados" if dry_run else "Renomeados" - print(f"\n✅ {action}: {count} arquivo(s)") +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Bulk rename files using a regex") + parser.add_argument("folder", help="Folder containing the files") + parser.add_argument("--pattern", required=True, help="Regex to search for") + parser.add_argument("--replacement", required=True, help="Replacement text, use {1} for groups") + parser.add_argument("--dry-run", action="store_true", help="Preview without renaming") + args = parser.parse_args(argv) -def main(): - parser = argparse.ArgumentParser(description="Renomeação em lote com regex") - parser.add_argument("folder", help="Pasta com os arquivos") - parser.add_argument("--pattern", required=True, help="Regex para procurar") - parser.add_argument("--replacement", required=True, help="Texto substituto") - parser.add_argument("--dry-run", action="store_true", help="Simula sem renomear") + try: + renames = bulk_rename(args.folder, args.pattern, args.replacement, args.dry_run) + except (NotADirectoryError, FileExistsError, re.error) as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 - args = parser.parse_args() - bulk_rename(args.folder, args.pattern, args.replacement, args.dry_run) + for old_name, new_name in renames: + print(f"{old_name} -> {new_name}") + label = "Would rename" if args.dry_run else "Renamed" + print(f"{label}: {len(renames)} file(s)") + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From 1fd37ab2ce1fbf0ec77daf887a53d496abe63e98 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:36 -0300 Subject: [PATCH 08/29] style: shorten argument help --- scripts/bulk_rename.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/bulk_rename.py b/scripts/bulk_rename.py index ff41c5c..001c8a1 100644 --- a/scripts/bulk_rename.py +++ b/scripts/bulk_rename.py @@ -38,7 +38,7 @@ def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Bulk rename files using a regex") parser.add_argument("folder", help="Folder containing the files") parser.add_argument("--pattern", required=True, help="Regex to search for") - parser.add_argument("--replacement", required=True, help="Replacement text, use {1} for groups") + parser.add_argument("--replacement", required=True, help="Replacement ({1} for groups)") parser.add_argument("--dry-run", action="store_true", help="Preview without renaming") args = parser.parse_args(argv) From 59f19c1f74fb3e510f6bf4b5d3f6850833596ba6 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:46 -0300 Subject: [PATCH 09/29] refactor: translate duplicate finder and separate scanning from deletion --- scripts/duplicate_finder.py | 141 ++++++++++++++++-------------------- 1 file changed, 64 insertions(+), 77 deletions(-) diff --git a/scripts/duplicate_finder.py b/scripts/duplicate_finder.py index 45e5680..1de1dae 100644 --- a/scripts/duplicate_finder.py +++ b/scripts/duplicate_finder.py @@ -1,109 +1,96 @@ -""" -Duplicate Finder - encontra arquivos duplicados por hash MD5. - -Uso: python scripts/duplicate_finder.py [--delete] -""" -import sys -import hashlib import argparse -from pathlib import Path +import hashlib +import sys from collections import defaultdict +from pathlib import Path def file_hash(filepath: Path, chunk_size: int = 8192) -> str: - """Calcula MD5 de um arquivo em chunks (eficiente para arquivos grandes).""" - md5 = hashlib.md5() - with filepath.open("rb") as f: - while chunk := f.read(chunk_size): - md5.update(chunk) - return md5.hexdigest() - - -def find_duplicates(folder: str, delete: bool = False) -> dict: - """ - Encontra arquivos duplicados comparando hashes. - Estratégia otimizada: primeiro agrupa por tamanho, depois por hash. - """ + digest = hashlib.md5(usedforsecurity=False) + with filepath.open("rb") as handle: + while chunk := handle.read(chunk_size): + digest.update(chunk) + return digest.hexdigest() + + +def find_duplicates(folder: str) -> list[list[Path]]: folder_path = Path(folder).expanduser().resolve() if not folder_path.is_dir(): - print(f"❌ Pasta inválida: {folder_path}") - return {} - - print(f"🔍 Buscando duplicatas em: {folder_path}\n") + raise NotADirectoryError(f"Invalid folder: {folder_path}") - # Passo 1: agrupar por tamanho (rápido) by_size = defaultdict(list) - total_files = 0 for file in folder_path.rglob("*"): - if file.is_file(): - try: + try: + if file.is_file(): by_size[file.stat().st_size].append(file) - total_files += 1 - except (OSError, PermissionError): - continue + except OSError: + continue - # Passo 2: hash apenas onde há mesmo tamanho by_hash = defaultdict(list) for size, files in by_size.items(): - if len(files) < 2 or size == 0: + if size == 0 or len(files) < 2: continue for file in files: try: - h = file_hash(file) - by_hash[h].append(file) - except (OSError, PermissionError): + by_hash[file_hash(file)].append(file) + except OSError: continue - # Filtrar só duplicatas - duplicates = {h: files for h, files in by_hash.items() if len(files) > 1} + groups = [sorted(files) for files in by_hash.values() if len(files) > 1] + return sorted(groups) + + +def wasted_bytes(groups: list[list[Path]]) -> int: + return sum(group[0].stat().st_size * (len(group) - 1) for group in groups) - if not duplicates: - print(f"✅ Nenhum duplicado encontrado em {total_files:,} arquivos.") - return {} - total_duplicates = sum(len(files) - 1 for files in duplicates.values()) - total_waste = sum( - files[0].stat().st_size * (len(files) - 1) - for files in duplicates.values() - ) +def delete_duplicates(groups: list[list[Path]]) -> list[Path]: + removed = [] + for group in groups: + for file in group[1:]: + try: + file.unlink() + except OSError: + continue + removed.append(file) + return removed - print(f"📊 Resultado:") - print(f" Arquivos analisados: {total_files:,}") - print(f" Grupos de duplicatas: {len(duplicates)}") - print(f" Arquivos redundantes: {total_duplicates:,}") - print(f" Espaço desperdiçado: {total_waste / 1024 / 1024:.2f} MB\n") - for idx, (h, files) in enumerate(duplicates.items(), 1): - print(f"Grupo {idx} ({len(files)} cópias, hash {h[:8]}...):") - for i, file in enumerate(files): - marker = " [ORIGINAL]" if i == 0 else " [DUPLICATA]" - print(f" {marker} {file}") +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Find duplicate files by content hash") + parser.add_argument("folder", help="Folder to analyze") + parser.add_argument("--delete", action="store_true", help="Delete extra copies") + args = parser.parse_args(argv) - if delete and i > 0: - try: - file.unlink() - print(f" 🗑️ Removido: {file.name}") - except Exception as e: - print(f" ❌ Erro ao remover: {e}") - print() + if args.delete: + answer = input("Delete duplicate files? [y/N]: ") + if answer.strip().lower() != "y": + print("Cancelled.") + return 0 - return duplicates + try: + groups = find_duplicates(args.folder) + except NotADirectoryError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + if not groups: + print("No duplicates found.") + return 0 -def main(): - parser = argparse.ArgumentParser(description="Encontra arquivos duplicados") - parser.add_argument("folder", help="Pasta para analisar") - parser.add_argument("--delete", action="store_true", help="DELETA duplicatas (mantém a primeira)") - args = parser.parse_args() + for index, group in enumerate(groups, 1): + print(f"Group {index} ({len(group)} copies):") + for position, file in enumerate(group): + label = "original" if position == 0 else "duplicate" + print(f" [{label}] {file}") - if args.delete: - resp = input("⚠️ Tem certeza que quer DELETAR as duplicatas? [s/N]: ") - if resp.lower() != "s": - print("Cancelado.") - sys.exit(0) + print(f"\nWasted space: {wasted_bytes(groups) / 1024 / 1024:.2f} MB") - find_duplicates(args.folder, args.delete) + if args.delete: + removed = delete_duplicates(groups) + print(f"Removed {len(removed)} file(s)") + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From 84dbad4130c3918e8bb5c660dcc70bcc19ba77fd Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:08:58 -0300 Subject: [PATCH 10/29] refactor: translate email sender and report missing configuration --- scripts/email_sender.py | 111 ++++++++++++++++++++-------------------- 1 file changed, 56 insertions(+), 55 deletions(-) diff --git a/scripts/email_sender.py b/scripts/email_sender.py index 9511309..f149be5 100644 --- a/scripts/email_sender.py +++ b/scripts/email_sender.py @@ -1,86 +1,87 @@ -""" -Email Sender - envia e-mails em massa a partir de template. - -Configure credenciais no .env antes de usar: - SMTP_HOST=smtp.gmail.com - SMTP_PORT=587 - SMTP_USER=seu@email.com - SMTP_PASSWORD=sua_senha_de_app - -Uso: python scripts/email_sender.py destinatarios.csv template.txt - (destinatarios.csv deve ter coluna 'email' e 'name') -""" -import os -import sys +import argparse import csv +import os import smtplib -from email.mime.text import MIMEText +import sys from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText from pathlib import Path + from dotenv import load_dotenv load_dotenv() -def send_email(to: str, subject: str, body: str) -> bool: - """Envia um e-mail via SMTP.""" +def load_template(template_file: str) -> tuple[str, str]: + content = Path(template_file).read_text(encoding="utf-8") + first_line, _, rest = content.partition("\n") + subject = first_line.replace("Subject:", "").strip() + return subject, rest.strip() + + +def render_body(template: str, name: str, email: str) -> str: + return template.replace("{{name}}", name).replace("{{email}}", email) + + +def build_message(sender: str, recipient: str, subject: str, body: str) -> MIMEMultipart: + message = MIMEMultipart() + message["From"] = sender + message["To"] = recipient + message["Subject"] = subject + message.attach(MIMEText(body, "plain", "utf-8")) + return message + + +def send_email(recipient: str, subject: str, body: str) -> bool: try: - msg = MIMEMultipart() - msg["From"] = os.getenv("SMTP_USER") - msg["To"] = to - msg["Subject"] = subject - msg.attach(MIMEText(body, "plain", "utf-8")) + host = os.environ["SMTP_HOST"] + user = os.environ["SMTP_USER"] + password = os.environ["SMTP_PASSWORD"] + port = int(os.getenv("SMTP_PORT", "587")) - with smtplib.SMTP(os.getenv("SMTP_HOST"), int(os.getenv("SMTP_PORT", 587))) as server: + message = build_message(user, recipient, subject, body) + with smtplib.SMTP(host, port) as server: server.starttls() - server.login(os.getenv("SMTP_USER"), os.getenv("SMTP_PASSWORD")) - server.send_message(msg) - return True - except Exception as e: - print(f" ❌ Erro enviando para {to}: {e}") + server.login(user, password) + server.send_message(message) + except (KeyError, ValueError, OSError) as exc: + print(f"Failed to send to {recipient}: {exc!r}", file=sys.stderr) return False + return True -def send_bulk(csv_file: str, template_file: str): - """Envia e-mails em massa usando template.""" - # Carregar template - template = Path(template_file).read_text(encoding="utf-8") - subject_line = template.split("\n")[0].replace("Subject:", "").strip() - body_template = "\n".join(template.split("\n")[1:]).strip() +def send_bulk(csv_file: str, template_file: str) -> tuple[int, int]: + subject, body_template = load_template(template_file) - # Ler destinatários - success = 0 + sent = 0 failed = 0 - with open(csv_file, encoding="utf-8") as f: - reader = csv.DictReader(f) - for row in reader: + with open(csv_file, newline="", encoding="utf-8") as handle: + for row in csv.DictReader(handle): email = row.get("email", "").strip() name = row.get("name", "").strip() if not email: continue - # Substituir placeholders - body = body_template.replace("{{name}}", name).replace("{{email}}", email) - - print(f"📧 Enviando para {email}...") - if send_email(email, subject_line, body): - success += 1 + body = render_body(body_template, name, email) + if send_email(email, subject, body): + sent += 1 else: failed += 1 - print(f"\n✅ Enviados: {success}") - print(f"❌ Falhas: {failed}") + return sent, failed -def main(): - if len(sys.argv) < 3: - print("Uso: python scripts/email_sender.py destinatarios.csv template.txt") - print("\nTemplate deve começar com: Subject: Assunto aqui") - print("Placeholders suportados: {{name}} e {{email}}") - sys.exit(1) +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Send bulk emails from a CSV and a template") + parser.add_argument("recipients", help="CSV file with 'email' and 'name' columns") + parser.add_argument("template", help="Template file; first line is 'Subject: ...'") + args = parser.parse_args(argv) - send_bulk(sys.argv[1], sys.argv[2]) + sent, failed = send_bulk(args.recipients, args.template) + print(f"Sent: {sent}") + print(f"Failed: {failed}") + return 1 if failed else 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From 1fd15992f447ee4fca549efe51efcde32d4209e8 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:09:06 -0300 Subject: [PATCH 11/29] refactor: translate Excel merger and return a summary --- scripts/excel_merger.py | 83 +++++++++++++++++++++-------------------- 1 file changed, 42 insertions(+), 41 deletions(-) diff --git a/scripts/excel_merger.py b/scripts/excel_merger.py index e7f42a1..e2b2a3e 100644 --- a/scripts/excel_merger.py +++ b/scripts/excel_merger.py @@ -1,64 +1,65 @@ -""" -Excel Merger - combina múltiplas planilhas Excel em um único arquivo. - -Uso: python scripts/excel_merger.py -""" +import argparse import sys from pathlib import Path -from openpyxl import load_workbook, Workbook + +from openpyxl import Workbook, load_workbook + +MAX_SHEET_NAME_LENGTH = 31 -def merge_excels(input_dir: str, output_file: str): - """Combina todas as planilhas .xlsx de uma pasta.""" +def merge_excels(input_dir: str, output_file: str) -> dict: input_path = Path(input_dir).expanduser().resolve() output_path = Path(output_file).expanduser().resolve() if not input_path.is_dir(): - print(f"❌ Pasta inválida: {input_path}") - return + raise NotADirectoryError(f"Invalid folder: {input_path}") - excel_files = list(input_path.glob("*.xlsx")) + excel_files = sorted(input_path.glob("*.xlsx")) if not excel_files: - print(f"❌ Nenhum arquivo .xlsx encontrado em {input_path}") - return + raise FileNotFoundError(f"No .xlsx files found in {input_path}") - print(f"📊 Combinando {len(excel_files)} arquivo(s)...\n") - - merged_wb = Workbook() - merged_wb.remove(merged_wb.active) # Remove sheet padrão + merged = Workbook() + merged.remove(merged.active) total_rows = 0 - for excel_file in sorted(excel_files): - print(f" 📄 {excel_file.name}") - wb = load_workbook(excel_file, data_only=True) - - for sheet_name in wb.sheetnames: - source_sheet = wb[sheet_name] - # Nome único: arquivo + aba - new_name = f"{excel_file.stem}_{sheet_name}"[:31] # Excel limita 31 chars - new_sheet = merged_wb.create_sheet(title=new_name) - - for row in source_sheet.iter_rows(values_only=True): - new_sheet.append(row) + for excel_file in excel_files: + workbook = load_workbook(excel_file, data_only=True) + for sheet_name in workbook.sheetnames: + title = f"{excel_file.stem}_{sheet_name}"[:MAX_SHEET_NAME_LENGTH] + target = merged.create_sheet(title=title) + for row in workbook[sheet_name].iter_rows(values_only=True): + target.append(row) total_rows += 1 output_path.parent.mkdir(parents=True, exist_ok=True) - merged_wb.save(output_path) + merged.save(output_path) + + return { + "files": len(excel_files), + "sheets": len(merged.sheetnames), + "rows": total_rows, + "output": output_path, + } - print(f"\n✅ Combinado!") - print(f" Arquivos processados: {len(excel_files)}") - print(f" Abas criadas: {len(merged_wb.sheetnames)}") - print(f" Linhas totais: {total_rows:,}") - print(f" Saída: {output_path}") +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Merge Excel workbooks into a single file") + parser.add_argument("input_dir", help="Folder containing .xlsx files") + parser.add_argument("output_file", help="Path of the merged .xlsx file") + args = parser.parse_args(argv) -def main(): - if len(sys.argv) < 3: - print("Uso: python scripts/excel_merger.py ") - sys.exit(1) + try: + summary = merge_excels(args.input_dir, args.output_file) + except (NotADirectoryError, FileNotFoundError) as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 - merge_excels(sys.argv[1], sys.argv[2]) + print(f"Files processed: {summary['files']}") + print(f"Sheets created: {summary['sheets']}") + print(f"Total rows: {summary['rows']:,}") + print(f"Output: {summary['output']}") + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From 6023d6d03ca891ef90838a82f3091570a0e76bda Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:09:14 -0300 Subject: [PATCH 12/29] refactor: translate folder backup and return a result object --- scripts/folder_backup.py | 101 ++++++++++++++++++++++----------------- 1 file changed, 56 insertions(+), 45 deletions(-) diff --git a/scripts/folder_backup.py b/scripts/folder_backup.py index 655ceda..0bfd34a 100644 --- a/scripts/folder_backup.py +++ b/scripts/folder_backup.py @@ -1,62 +1,73 @@ -""" -Folder Backup - cria backup compactado de uma pasta com timestamp. - -Uso: python scripts/folder_backup.py -""" +import argparse import sys import zipfile -from pathlib import Path +from dataclasses import dataclass from datetime import datetime +from pathlib import Path + +@dataclass(frozen=True) +class BackupResult: + path: Path + files: int + original_bytes: int + zip_bytes: int -def backup_folder(source: str, dest_dir: str) -> str: - """Compacta pasta em ZIP com data no nome.""" + @property + def savings_percent(self) -> float: + if self.original_bytes == 0: + return 0.0 + return (1 - self.zip_bytes / self.original_bytes) * 100 + + +def backup_folder(source: str, dest_dir: str, now: datetime | None = None) -> BackupResult: source_path = Path(source).expanduser().resolve() dest_path = Path(dest_dir).expanduser().resolve() if not source_path.exists(): - print(f"❌ Origem não existe: {source_path}") - return "" + raise FileNotFoundError(f"Source does not exist: {source_path}") dest_path.mkdir(parents=True, exist_ok=True) - timestamp = datetime.now().strftime("%Y-%m-%d_%H%M") - zip_name = f"backup_{timestamp}_{source_path.name}.zip" - zip_path = dest_path / zip_name - - print(f"📦 Criando backup de {source_path}") - print(f"📍 Destino: {zip_path}\n") + timestamp = (now or datetime.now()).strftime("%Y-%m-%d_%H%M") + zip_path = dest_path / f"backup_{timestamp}_{source_path.name}.zip" - files_count = 0 - total_size = 0 - - with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as zf: - for file in source_path.rglob("*"): + files = 0 + original_bytes = 0 + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as archive: + for file in sorted(source_path.rglob("*")): if file.is_file(): - arcname = file.relative_to(source_path.parent) - zf.write(file, arcname) - files_count += 1 - total_size += file.stat().st_size - - zip_size = zip_path.stat().st_size - compression = (1 - zip_size / total_size) * 100 if total_size > 0 else 0 - - print(f"✅ Backup concluído!") - print(f" Arquivos: {files_count:,}") - print(f" Original: {total_size / 1024 / 1024:.2f} MB") - print(f" ZIP: {zip_size / 1024 / 1024:.2f} MB") - print(f" Economia: {compression:.1f}%") - - return str(zip_path) - - -def main(): - if len(sys.argv) < 3: - print("Uso: python scripts/folder_backup.py ") - sys.exit(1) - - backup_folder(sys.argv[1], sys.argv[2]) + archive.write(file, file.relative_to(source_path.parent)) + files += 1 + original_bytes += file.stat().st_size + + return BackupResult( + path=zip_path, + files=files, + original_bytes=original_bytes, + zip_bytes=zip_path.stat().st_size, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Create a timestamped ZIP backup of a folder") + parser.add_argument("source", help="Folder to back up") + parser.add_argument("destination", help="Folder where the ZIP is stored") + args = parser.parse_args(argv) + + try: + result = backup_folder(args.source, args.destination) + except FileNotFoundError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + + print(f"Backup created: {result.path}") + print(f"Files: {result.files:,}") + print(f"Original: {result.original_bytes / 1024 / 1024:.2f} MB") + print(f"ZIP: {result.zip_bytes / 1024 / 1024:.2f} MB") + print(f"Savings: {result.savings_percent:.1f}%") + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From 89eee7a2c04ce6b1153736384dd2f95297c2be69 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:09:26 -0300 Subject: [PATCH 13/29] refactor: translate log analyzer and split analysis from reporting --- scripts/log_analyzer.py | 173 +++++++++++++++++++--------------------- 1 file changed, 83 insertions(+), 90 deletions(-) diff --git a/scripts/log_analyzer.py b/scripts/log_analyzer.py index 470ac14..facff8d 100644 --- a/scripts/log_analyzer.py +++ b/scripts/log_analyzer.py @@ -1,120 +1,113 @@ -""" -Log Analyzer - analisa logs e extrai erros, warnings e estatísticas. - -Uso: python scripts/log_analyzer.py -""" +import argparse import re import sys -from pathlib import Path -from collections import Counter, defaultdict +from collections import Counter +from dataclasses import dataclass, field from datetime import datetime +from pathlib import Path - -# Padrões comuns de níveis de log -LEVEL_PATTERN = re.compile( - r"\b(ERROR|CRITICAL|FATAL|WARNING|WARN|INFO|DEBUG)\b", re.IGNORECASE -) -TIMESTAMP_PATTERN = re.compile( - r"(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2})" -) +LEVEL_PATTERN = re.compile(r"\b(ERROR|CRITICAL|FATAL|WARNING|WARN|INFO|DEBUG)\b", re.IGNORECASE) +TIMESTAMP_PATTERN = re.compile(r"(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2})") IP_PATTERN = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b") +LEVEL_ALIASES = {"WARN": "WARNING", "FATAL": "CRITICAL"} +LEVEL_ORDER = ["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"] +MAX_ERROR_SAMPLES = 20 +MAX_WARNING_SAMPLES = 10 +MAX_LINE_LENGTH = 200 -def analyze_log(filepath: str): - """Analisa arquivo de log e gera relatório.""" - log_path = Path(filepath).expanduser().resolve() +@dataclass +class LogReport: + total_lines: int = 0 + level_counts: Counter = field(default_factory=Counter) + errors: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + ips: Counter = field(default_factory=Counter) + hourly_activity: Counter = field(default_factory=Counter) + + +def normalize_level(level: str) -> str: + level = level.upper() + return LEVEL_ALIASES.get(level, level) + +def analyze_log(filepath: str) -> LogReport: + log_path = Path(filepath).expanduser().resolve() if not log_path.is_file(): - print(f"❌ Arquivo não encontrado: {log_path}") - return - - print(f"📊 Analisando: {log_path}") - print(f"📦 Tamanho: {log_path.stat().st_size / 1024:.2f} KB\n") - - level_counts = Counter() - error_messages = [] - warnings = [] - ips = Counter() - hourly_activity = defaultdict(int) - total_lines = 0 - - with log_path.open("r", encoding="utf-8", errors="ignore") as f: - for line in f: - total_lines += 1 - line = line.rstrip() + raise FileNotFoundError(f"File not found: {log_path}") + + report = LogReport() + with log_path.open("r", encoding="utf-8", errors="ignore") as handle: + for raw_line in handle: + report.total_lines += 1 + line = raw_line.rstrip() if not line: continue - # Contar níveis level_match = LEVEL_PATTERN.search(line) if level_match: - level = level_match.group(1).upper() - if level == "WARN": - level = "WARNING" - if level == "FATAL": - level = "CRITICAL" - level_counts[level] += 1 - - # Guardar exemplos - if level in ("ERROR", "CRITICAL") and len(error_messages) < 20: - error_messages.append(line[:200]) - elif level == "WARNING" and len(warnings) < 10: - warnings.append(line[:200]) - - # Timestamps → atividade por hora - ts_match = TIMESTAMP_PATTERN.search(line) - if ts_match: + level = normalize_level(level_match.group(1)) + report.level_counts[level] += 1 + if level in ("ERROR", "CRITICAL") and len(report.errors) < MAX_ERROR_SAMPLES: + report.errors.append(line[:MAX_LINE_LENGTH]) + elif level == "WARNING" and len(report.warnings) < MAX_WARNING_SAMPLES: + report.warnings.append(line[:MAX_LINE_LENGTH]) + + timestamp_match = TIMESTAMP_PATTERN.search(line) + if timestamp_match: try: - dt = datetime.fromisoformat(ts_match.group(1).replace(" ", "T")) - hourly_activity[dt.strftime("%Y-%m-%d %H:00")] += 1 + moment = datetime.fromisoformat(timestamp_match.group(1).replace(" ", "T")) except ValueError: pass + else: + report.hourly_activity[moment.strftime("%Y-%m-%d %H:00")] += 1 + + report.ips.update(IP_PATTERN.findall(line)) + + return report + - # IPs - for ip in IP_PATTERN.findall(line): - ips[ip] += 1 +def format_report(report: LogReport) -> str: + lines = [f"Total lines: {report.total_lines:,}", "", "Level distribution:"] + for level in LEVEL_ORDER: + count = report.level_counts.get(level, 0) + if count: + lines.append(f" {level:<10}: {count:,}") - # RELATÓRIO - print("=" * 60) - print(f"📈 RESUMO DO LOG") - print("=" * 60) - print(f"Total de linhas: {total_lines:,}\n") + if report.errors: + lines.extend(["", "Latest errors:"]) + lines.extend(f" {message}" for message in report.errors[-5:]) - print("📊 Distribuição por nível:") - for level in ["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"]: - count = level_counts.get(level, 0) - if count > 0: - emoji = {"CRITICAL": "🔴", "ERROR": "❌", "WARNING": "⚠️", "INFO": "ℹ️", "DEBUG": "🔧"}[level] - print(f" {emoji} {level:10s}: {count:,}") + if report.warnings: + lines.extend(["", "Latest warnings:"]) + lines.extend(f" {message}" for message in report.warnings[-3:]) - if error_messages: - print(f"\n❌ Últimos erros ({min(len(error_messages), 5)}):") - for err in error_messages[-5:]: - print(f" {err}") + if report.ips: + lines.extend(["", "Top 5 IPs:"]) + lines.extend(f" {ip}: {count:,}" for ip, count in report.ips.most_common(5)) - if warnings: - print(f"\n⚠️ Últimos warnings ({min(len(warnings), 3)}):") - for w in warnings[-3:]: - print(f" {w}") + if report.hourly_activity: + busiest = sorted(report.hourly_activity.items(), key=lambda item: -item[1])[:5] + lines.extend(["", "Busiest hours:"]) + lines.extend(f" {hour}: {count:,} events" for hour, count in busiest) - if ips: - print(f"\n🌐 Top 5 IPs:") - for ip, count in ips.most_common(5): - print(f" {ip}: {count:,} ocorrências") + return "\n".join(lines) - if hourly_activity: - print(f"\n⏰ Horas mais ativas (top 5):") - for hour, count in sorted(hourly_activity.items(), key=lambda x: -x[1])[:5]: - print(f" {hour}: {count:,} eventos") +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Summarize a log file") + parser.add_argument("logfile", help="Path to the log file") + args = parser.parse_args(argv) -def main(): - if len(sys.argv) < 2: - print("Uso: python scripts/log_analyzer.py ") - sys.exit(1) + try: + report = analyze_log(args.logfile) + except FileNotFoundError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 - analyze_log(sys.argv[1]) + print(format_report(report)) + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From 07429808dc83b026463b6a8e3bba22a66a07b11f Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:09:38 -0300 Subject: [PATCH 14/29] refactor: translate system monitor and make the loop bounded for testing --- scripts/system_monitor.py | 84 +++++++++++++++++++-------------------- 1 file changed, 42 insertions(+), 42 deletions(-) diff --git a/scripts/system_monitor.py b/scripts/system_monitor.py index bded670..670fc5d 100644 --- a/scripts/system_monitor.py +++ b/scripts/system_monitor.py @@ -1,88 +1,88 @@ -""" -System Monitor - monitora CPU, RAM e Disco e gera log estruturado. - -Uso: python scripts/system_monitor.py [--interval 60] [--log monitor.log] -""" import argparse +import os import time -import psutil from datetime import datetime from pathlib import Path +import psutil + +BYTES_PER_GB = 1024**3 + def get_system_stats() -> dict: - """Coleta estatísticas atuais do sistema.""" + memory = psutil.virtual_memory() + disk = psutil.disk_usage(os.path.abspath(os.sep)) return { "timestamp": datetime.now().isoformat(), "cpu_percent": psutil.cpu_percent(interval=1), "cpu_count": psutil.cpu_count(), - "ram_percent": psutil.virtual_memory().percent, - "ram_used_gb": round(psutil.virtual_memory().used / 1024**3, 2), - "ram_total_gb": round(psutil.virtual_memory().total / 1024**3, 2), - "disk_percent": psutil.disk_usage("/").percent, - "disk_free_gb": round(psutil.disk_usage("/").free / 1024**3, 2), + "ram_percent": memory.percent, + "ram_used_gb": round(memory.used / BYTES_PER_GB, 2), + "ram_total_gb": round(memory.total / BYTES_PER_GB, 2), + "disk_percent": disk.percent, + "disk_free_gb": round(disk.free / BYTES_PER_GB, 2), } def format_stats(stats: dict) -> str: - """Formata estatísticas para log.""" return ( f"[{stats['timestamp']}] " f"CPU: {stats['cpu_percent']:5.1f}% | " f"RAM: {stats['ram_percent']:5.1f}% ({stats['ram_used_gb']}/{stats['ram_total_gb']} GB) | " - f"Disco: {stats['disk_percent']:5.1f}% ({stats['disk_free_gb']} GB livre)" + f"Disk: {stats['disk_percent']:5.1f}% ({stats['disk_free_gb']} GB free)" ) -def check_alerts(stats: dict, cpu_threshold: float = 80, ram_threshold: float = 85) -> list: - """Retorna alertas se métricas ultrapassarem limites.""" +def check_alerts( + stats: dict, + cpu_threshold: float = 80, + ram_threshold: float = 85, + disk_threshold: float = 90, +) -> list[str]: alerts = [] if stats["cpu_percent"] >= cpu_threshold: - alerts.append(f"⚠️ CPU alta: {stats['cpu_percent']:.1f}%") + alerts.append(f"High CPU usage: {stats['cpu_percent']:.1f}%") if stats["ram_percent"] >= ram_threshold: - alerts.append(f"⚠️ RAM alta: {stats['ram_percent']:.1f}%") - if stats["disk_percent"] >= 90: - alerts.append(f"🔴 Disco crítico: {stats['disk_percent']:.1f}%") + alerts.append(f"High RAM usage: {stats['ram_percent']:.1f}%") + if stats["disk_percent"] >= disk_threshold: + alerts.append(f"Critical disk usage: {stats['disk_percent']:.1f}%") return alerts -def monitor(interval: int, log_file: str = None): - """Loop de monitoramento.""" +def monitor(interval: int, log_file: str | None = None, iterations: int | None = None) -> None: log_path = Path(log_file) if log_file else None if log_path: log_path.parent.mkdir(parents=True, exist_ok=True) - print(f"📝 Logando em: {log_path}") - - print(f"🔍 Monitor iniciado (intervalo: {interval}s). Ctrl+C para parar.\n") + completed = 0 try: while True: stats = get_system_stats() - line = format_stats(stats) - print(line) - - alerts = check_alerts(stats) - for alert in alerts: - print(f" {alert}") + lines = [format_stats(stats)] + lines.extend(f" {alert}" for alert in check_alerts(stats)) + print("\n".join(lines)) if log_path: - with log_path.open("a", encoding="utf-8") as f: - f.write(line + "\n") - for alert in alerts: - f.write(f" {alert}\n") + with log_path.open("a", encoding="utf-8") as handle: + handle.write("\n".join(lines) + "\n") + completed += 1 + if iterations is not None and completed >= iterations: + break time.sleep(interval) except KeyboardInterrupt: - print("\n\n✅ Monitor encerrado.") + print("\nMonitor stopped.") + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Monitor CPU, RAM and disk usage") + parser.add_argument("--interval", type=int, default=60, help="Interval in seconds") + parser.add_argument("--log", help="Log file path") + args = parser.parse_args(argv) -def main(): - parser = argparse.ArgumentParser(description="Monitor de sistema") - parser.add_argument("--interval", type=int, default=60, help="Intervalo em segundos") - parser.add_argument("--log", type=str, help="Arquivo de log") - args = parser.parse_args() monitor(args.interval, args.log) + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) From c7ebf976afcd526ea39d096723451316950de397 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:09:44 -0300 Subject: [PATCH 15/29] test: add tests package --- tests/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 tests/__init__.py diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 From d161e5abe5359208e48b01f3cbf584800468281c Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:09:53 -0300 Subject: [PATCH 16/29] test: cover file organizer --- tests/test_file_organizer.py | 87 ++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) create mode 100644 tests/test_file_organizer.py diff --git a/tests/test_file_organizer.py b/tests/test_file_organizer.py new file mode 100644 index 0000000..6e83ab3 --- /dev/null +++ b/tests/test_file_organizer.py @@ -0,0 +1,87 @@ +import pytest + +from scripts import file_organizer +from scripts.file_organizer import get_category, organize_folder, unique_destination + + +@pytest.mark.parametrize( + ("extension", "category"), + [ + (".pdf", "Documents"), + (".XLSX", "Spreadsheets"), + (".png", "Images"), + (".py", "Code"), + (".unknown", "Other"), + ("", "Other"), + ], +) +def test_get_category(extension, category): + assert get_category(extension) == category + + +def test_unique_destination_avoids_overwriting(tmp_path): + (tmp_path / "report.pdf").write_text("a") + (tmp_path / "report_1.pdf").write_text("b") + + assert unique_destination(tmp_path, "report.pdf") == tmp_path / "report_2.pdf" + + +def test_organize_folder_moves_files_into_category_folders(tmp_path): + (tmp_path / "notes.txt").write_text("notes") + (tmp_path / "photo.PNG").write_text("img") + (tmp_path / "data.bin").write_text("raw") + + stats = organize_folder(str(tmp_path)) + + assert stats == {"Documents": 1, "Images": 1, "Other": 1} + assert (tmp_path / "Documents" / "notes.txt").read_text() == "notes" + assert (tmp_path / "Images" / "photo.PNG").exists() + assert (tmp_path / "Other" / "data.bin").exists() + assert not (tmp_path / "notes.txt").exists() + + +def test_organize_folder_keeps_existing_files_on_name_clash(tmp_path): + (tmp_path / "Documents").mkdir() + (tmp_path / "Documents" / "a.txt").write_text("old") + (tmp_path / "a.txt").write_text("new") + + organize_folder(str(tmp_path)) + + assert (tmp_path / "Documents" / "a.txt").read_text() == "old" + assert (tmp_path / "Documents" / "a_1.txt").read_text() == "new" + + +def test_organize_folder_ignores_subfolders(tmp_path): + (tmp_path / "nested").mkdir() + + assert organize_folder(str(tmp_path)) == {} + assert (tmp_path / "nested").is_dir() + + +def test_organize_folder_rejects_missing_folder(tmp_path): + with pytest.raises(FileNotFoundError): + organize_folder(str(tmp_path / "missing")) + + +def test_organize_folder_rejects_file_path(tmp_path): + file = tmp_path / "file.txt" + file.write_text("x") + + with pytest.raises(NotADirectoryError): + organize_folder(str(file)) + + +def test_main_prints_summary(tmp_path, capsys): + (tmp_path / "a.txt").write_text("x") + + code = file_organizer.main([str(tmp_path)]) + + assert code == 0 + assert "Documents: 1 file(s)" in capsys.readouterr().out + + +def test_main_reports_errors(tmp_path, capsys): + code = file_organizer.main([str(tmp_path / "missing")]) + + assert code == 1 + assert "Error:" in capsys.readouterr().err From f36c35333d2bd063549a0328f5331a850ae2f8cb Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:10:02 -0300 Subject: [PATCH 17/29] test: cover bulk rename --- tests/test_bulk_rename.py | 76 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 tests/test_bulk_rename.py diff --git a/tests/test_bulk_rename.py b/tests/test_bulk_rename.py new file mode 100644 index 0000000..3fb7f7c --- /dev/null +++ b/tests/test_bulk_rename.py @@ -0,0 +1,76 @@ +import pytest + +from scripts import bulk_rename as module +from scripts.bulk_rename import bulk_rename + + +@pytest.fixture +def photos(tmp_path): + for name in ("IMG_001.jpg", "IMG_002.jpg", "notes.txt"): + (tmp_path / name).write_text(name) + return tmp_path + + +def test_renames_files_matching_pattern(photos): + renames = bulk_rename(str(photos), "^IMG_", "photo_") + + assert renames == [("IMG_001.jpg", "photo_001.jpg"), ("IMG_002.jpg", "photo_002.jpg")] + assert (photos / "photo_001.jpg").read_text() == "IMG_001.jpg" + assert not (photos / "IMG_001.jpg").exists() + assert (photos / "notes.txt").exists() + + +def test_replacement_supports_numbered_groups(photos): + bulk_rename(str(photos), r"IMG_(\d+)", "pic-{1}") + + assert (photos / "pic-001.jpg").exists() + + +def test_dry_run_does_not_touch_files(photos): + renames = bulk_rename(str(photos), "^IMG_", "photo_", dry_run=True) + + assert len(renames) == 2 + assert (photos / "IMG_001.jpg").exists() + assert not (photos / "photo_001.jpg").exists() + + +def test_ignores_directories(photos): + (photos / "IMG_dir").mkdir() + + bulk_rename(str(photos), "^IMG_", "photo_") + + assert (photos / "IMG_dir").is_dir() + + +def test_refuses_to_overwrite_existing_file(photos): + (photos / "photo_001.jpg").write_text("existing") + + with pytest.raises(FileExistsError): + bulk_rename(str(photos), "^IMG_", "photo_") + + assert (photos / "photo_001.jpg").read_text() == "existing" + + +def test_rejects_invalid_folder(tmp_path): + with pytest.raises(NotADirectoryError): + bulk_rename(str(tmp_path / "missing"), "a", "b") + + +def test_main_reports_renamed_count(photos, capsys): + code = module.main([str(photos), "--pattern", "^IMG_", "--replacement", "photo_"]) + + assert code == 0 + assert "Renamed: 2 file(s)" in capsys.readouterr().out + + +def test_main_dry_run_label(photos, capsys): + module.main([str(photos), "--pattern", "^IMG_", "--replacement", "x_", "--dry-run"]) + + assert "Would rename: 2 file(s)" in capsys.readouterr().out + + +def test_main_reports_invalid_regex(photos, capsys): + code = module.main([str(photos), "--pattern", "(", "--replacement", "x"]) + + assert code == 1 + assert "Error:" in capsys.readouterr().err From 8bf01975916bfa6c9c958fd28a820437d6b36bbb Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:10:15 -0300 Subject: [PATCH 18/29] test: cover duplicate finder --- tests/test_duplicate_finder.py | 108 +++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 tests/test_duplicate_finder.py diff --git a/tests/test_duplicate_finder.py b/tests/test_duplicate_finder.py new file mode 100644 index 0000000..4d76806 --- /dev/null +++ b/tests/test_duplicate_finder.py @@ -0,0 +1,108 @@ +import hashlib + +import pytest + +from scripts import duplicate_finder as module +from scripts.duplicate_finder import delete_duplicates, file_hash, find_duplicates, wasted_bytes + + +@pytest.fixture +def tree(tmp_path): + (tmp_path / "sub").mkdir() + (tmp_path / "a.txt").write_text("same content") + (tmp_path / "sub" / "b.txt").write_text("same content") + (tmp_path / "c.txt").write_text("different!!!!") + (tmp_path / "empty1.txt").write_text("") + (tmp_path / "empty2.txt").write_text("") + return tmp_path + + +def test_file_hash_matches_md5(tmp_path): + file = tmp_path / "data.bin" + file.write_bytes(b"x" * 20_000) + + assert file_hash(file, chunk_size=1024) == hashlib.md5(b"x" * 20_000).hexdigest() + + +def test_find_duplicates_groups_identical_files(tree): + groups = find_duplicates(str(tree)) + + assert groups == [[tree / "a.txt", tree / "sub" / "b.txt"]] + + +def test_find_duplicates_ignores_empty_files_and_same_size_files(tree): + flat = [file for group in find_duplicates(str(tree)) for file in group] + + assert tree / "c.txt" not in flat + assert tree / "empty1.txt" not in flat + + +def test_find_duplicates_returns_empty_list_when_unique(tmp_path): + (tmp_path / "a.txt").write_text("one") + (tmp_path / "b.txt").write_text("two") + + assert find_duplicates(str(tmp_path)) == [] + + +def test_find_duplicates_rejects_invalid_folder(tmp_path): + with pytest.raises(NotADirectoryError): + find_duplicates(str(tmp_path / "missing")) + + +def test_wasted_bytes_counts_redundant_copies(tree): + groups = find_duplicates(str(tree)) + + assert wasted_bytes(groups) == len("same content") + + +def test_delete_duplicates_keeps_first_copy(tree): + groups = find_duplicates(str(tree)) + + removed = delete_duplicates(groups) + + assert removed == [tree / "sub" / "b.txt"] + assert (tree / "a.txt").exists() + assert not (tree / "sub" / "b.txt").exists() + + +def test_main_lists_duplicates(tree, capsys): + code = module.main([str(tree)]) + + output = capsys.readouterr().out + assert code == 0 + assert "[original]" in output + assert "[duplicate]" in output + assert (tree / "sub" / "b.txt").exists() + + +def test_main_reports_when_nothing_found(tmp_path, capsys): + (tmp_path / "a.txt").write_text("only") + + assert module.main([str(tmp_path)]) == 0 + assert "No duplicates found." in capsys.readouterr().out + + +def test_main_delete_requires_confirmation(tree, monkeypatch, capsys): + monkeypatch.setattr("builtins.input", lambda prompt: "n") + + code = module.main([str(tree), "--delete"]) + + assert code == 0 + assert "Cancelled." in capsys.readouterr().out + assert (tree / "sub" / "b.txt").exists() + + +def test_main_delete_removes_copies_when_confirmed(tree, monkeypatch, capsys): + monkeypatch.setattr("builtins.input", lambda prompt: "y") + + module.main([str(tree), "--delete"]) + + assert "Removed 1 file(s)" in capsys.readouterr().out + assert not (tree / "sub" / "b.txt").exists() + + +def test_main_reports_invalid_folder(tmp_path, capsys): + code = module.main([str(tmp_path / "missing")]) + + assert code == 1 + assert "Error:" in capsys.readouterr().err From de2b79973f314fd88b5d2b914f7ef0cf69f975b5 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:10:23 -0300 Subject: [PATCH 19/29] test: make the non-duplicate fixture the same size --- tests/test_duplicate_finder.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_duplicate_finder.py b/tests/test_duplicate_finder.py index 4d76806..92571ac 100644 --- a/tests/test_duplicate_finder.py +++ b/tests/test_duplicate_finder.py @@ -11,7 +11,7 @@ def tree(tmp_path): (tmp_path / "sub").mkdir() (tmp_path / "a.txt").write_text("same content") (tmp_path / "sub" / "b.txt").write_text("same content") - (tmp_path / "c.txt").write_text("different!!!!") + (tmp_path / "c.txt").write_text("diff content") (tmp_path / "empty1.txt").write_text("") (tmp_path / "empty2.txt").write_text("") return tmp_path From efc8a3f649ebc3e6d09cb0022ffec769e59d47b5 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:10:38 -0300 Subject: [PATCH 20/29] test: cover email sender with a fake SMTP server --- tests/test_email_sender.py | 136 +++++++++++++++++++++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 tests/test_email_sender.py diff --git a/tests/test_email_sender.py b/tests/test_email_sender.py new file mode 100644 index 0000000..fca1d76 --- /dev/null +++ b/tests/test_email_sender.py @@ -0,0 +1,136 @@ +import smtplib + +import pytest + +from scripts import email_sender as module +from scripts.email_sender import ( + build_message, + load_template, + render_body, + send_bulk, + send_email, +) + + +class FakeSMTP: + def __init__(self, host, port, log, failing_recipients): + self.log = log + self.failing_recipients = failing_recipients + log.append(("connect", host, port)) + + def __enter__(self): + return self + + def __exit__(self, *exc_info): + return False + + def starttls(self): + self.log.append(("starttls",)) + + def login(self, user, password): + self.log.append(("login", user, password)) + + def send_message(self, message): + if message["To"] in self.failing_recipients: + raise smtplib.SMTPRecipientsRefused({message["To"]: (550, b"rejected")}) + self.log.append(("send", message["To"], message["Subject"])) + + +@pytest.fixture +def smtp(monkeypatch): + log = [] + failing_recipients = set() + monkeypatch.setenv("SMTP_HOST", "smtp.example.com") + monkeypatch.setenv("SMTP_PORT", "2525") + monkeypatch.setenv("SMTP_USER", "me@example.com") + monkeypatch.setenv("SMTP_PASSWORD", "secret") + monkeypatch.setattr( + module.smtplib, + "SMTP", + lambda host, port: FakeSMTP(host, port, log, failing_recipients), + ) + return log, failing_recipients + + +@pytest.fixture +def campaign(tmp_path): + template = tmp_path / "template.txt" + template.write_text("Subject: Hello\nHi {{name}} <{{email}}>\n", encoding="utf-8") + recipients = tmp_path / "recipients.csv" + recipients.write_text( + "email,name\nana@example.com,Ana\nbad@example.com,Bad\n,NoEmail\n", + encoding="utf-8", + ) + return str(recipients), str(template) + + +def test_load_template_splits_subject_and_body(tmp_path): + template = tmp_path / "t.txt" + template.write_text("Subject: Greetings\nLine one\nLine two\n", encoding="utf-8") + + assert load_template(str(template)) == ("Greetings", "Line one\nLine two") + + +def test_render_body_replaces_placeholders(): + assert render_body("Hi {{name}} ({{email}})", "Ana", "a@x.com") == "Hi Ana (a@x.com)" + + +def test_build_message_sets_headers_and_body(): + message = build_message("me@x.com", "you@x.com", "Subject line", "Body text") + + assert message["From"] == "me@x.com" + assert message["To"] == "you@x.com" + assert message["Subject"] == "Subject line" + assert "Body text" in message.get_payload()[0].get_payload(decode=True).decode("utf-8") + + +def test_send_email_uses_tls_and_credentials(smtp): + log, _ = smtp + + assert send_email("you@example.com", "Hi", "Body") is True + + assert log == [ + ("connect", "smtp.example.com", 2525), + ("starttls",), + ("login", "me@example.com", "secret"), + ("send", "you@example.com", "Hi"), + ] + + +def test_send_email_returns_false_without_configuration(monkeypatch, capsys): + monkeypatch.delenv("SMTP_HOST", raising=False) + + assert send_email("you@example.com", "Hi", "Body") is False + assert "Failed to send" in capsys.readouterr().err + + +def test_send_email_returns_false_on_smtp_error(smtp, capsys): + _, failing_recipients = smtp + failing_recipients.add("you@example.com") + + assert send_email("you@example.com", "Hi", "Body") is False + assert "you@example.com" in capsys.readouterr().err + + +def test_send_bulk_counts_successes_and_failures(smtp, campaign): + log, failing_recipients = smtp + failing_recipients.add("bad@example.com") + + sent, failed = send_bulk(*campaign) + + assert (sent, failed) == (1, 1) + assert ("send", "ana@example.com", "Hello") in log + + +def test_main_returns_error_code_when_any_delivery_fails(smtp, campaign, capsys): + _, failing_recipients = smtp + failing_recipients.add("bad@example.com") + + code = module.main(list(campaign)) + + assert code == 1 + assert "Sent: 1" in capsys.readouterr().out + + +def test_main_returns_zero_when_everything_is_sent(smtp, campaign, capsys): + assert module.main(list(campaign)) == 1 or True From 55aa2ef68a5366220f3b0ab06060319043356138 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:10:48 -0300 Subject: [PATCH 21/29] test: assert the success exit code properly --- tests/test_email_sender.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test_email_sender.py b/tests/test_email_sender.py index fca1d76..b97289c 100644 --- a/tests/test_email_sender.py +++ b/tests/test_email_sender.py @@ -133,4 +133,7 @@ def test_main_returns_error_code_when_any_delivery_fails(smtp, campaign, capsys) def test_main_returns_zero_when_everything_is_sent(smtp, campaign, capsys): - assert module.main(list(campaign)) == 1 or True + code = module.main(list(campaign)) + + assert code == 0 + assert "Sent: 2" in capsys.readouterr().out From 4a5060666b56c370aed5612770766765b27e8b38 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:11:00 -0300 Subject: [PATCH 22/29] test: cover Excel merger --- tests/test_excel_merger.py | 75 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 75 insertions(+) create mode 100644 tests/test_excel_merger.py diff --git a/tests/test_excel_merger.py b/tests/test_excel_merger.py new file mode 100644 index 0000000..7d46089 --- /dev/null +++ b/tests/test_excel_merger.py @@ -0,0 +1,75 @@ +import pytest +from openpyxl import Workbook, load_workbook + +from scripts import excel_merger as module +from scripts.excel_merger import merge_excels + + +def make_workbook(path, sheets): + workbook = Workbook() + workbook.remove(workbook.active) + for title, rows in sheets.items(): + sheet = workbook.create_sheet(title) + for row in rows: + sheet.append(row) + workbook.save(path) + + +@pytest.fixture +def workbooks(tmp_path): + source = tmp_path / "source" + source.mkdir() + make_workbook(source / "sales.xlsx", {"Q1": [["id", "total"], [1, 10]], "Q2": [["id"], [2]]}) + make_workbook(source / "costs.xlsx", {"Main": [["cost"], [5]]}) + return source + + +def test_merge_combines_every_sheet_with_prefixed_names(workbooks, tmp_path): + output = tmp_path / "out" / "merged.xlsx" + + summary = merge_excels(str(workbooks), str(output)) + + merged = load_workbook(output) + assert merged.sheetnames == ["costs_Main", "sales_Q1", "sales_Q2"] + assert [list(row) for row in merged["sales_Q1"].values] == [["id", "total"], [1, 10]] + assert summary["files"] == 2 + assert summary["sheets"] == 3 + assert summary["rows"] == 7 + assert summary["output"] == output.resolve() + + +def test_merge_truncates_long_sheet_names(tmp_path): + source = tmp_path / "source" + source.mkdir() + make_workbook(source / "a_very_long_workbook_file_name.xlsx", {"AVeryLongSheetName": [["x"]]}) + + merge_excels(str(source), str(tmp_path / "merged.xlsx")) + + [title] = load_workbook(tmp_path / "merged.xlsx").sheetnames + assert len(title) == 31 + + +def test_merge_rejects_invalid_folder(tmp_path): + with pytest.raises(NotADirectoryError): + merge_excels(str(tmp_path / "missing"), str(tmp_path / "out.xlsx")) + + +def test_merge_rejects_folder_without_workbooks(tmp_path): + with pytest.raises(FileNotFoundError): + merge_excels(str(tmp_path), str(tmp_path / "out.xlsx")) + + +def test_main_prints_summary(workbooks, tmp_path, capsys): + code = module.main([str(workbooks), str(tmp_path / "merged.xlsx")]) + + output = capsys.readouterr().out + assert code == 0 + assert "Files processed: 2" in output + assert "Sheets created: 3" in output + + +def test_main_reports_errors(tmp_path, capsys): + code = module.main([str(tmp_path / "missing"), str(tmp_path / "out.xlsx")]) + + assert code == 1 + assert "Error:" in capsys.readouterr().err From 79dca543ece331781a4c27145966444395aa9957 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:11:07 -0300 Subject: [PATCH 23/29] test: correct expected merged row count --- tests/test_excel_merger.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_excel_merger.py b/tests/test_excel_merger.py index 7d46089..43d37ff 100644 --- a/tests/test_excel_merger.py +++ b/tests/test_excel_merger.py @@ -34,7 +34,7 @@ def test_merge_combines_every_sheet_with_prefixed_names(workbooks, tmp_path): assert [list(row) for row in merged["sales_Q1"].values] == [["id", "total"], [1, 10]] assert summary["files"] == 2 assert summary["sheets"] == 3 - assert summary["rows"] == 7 + assert summary["rows"] == 6 assert summary["output"] == output.resolve() From ff0e54d2737204851293409de1ac373e57e7cfc5 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:11:17 -0300 Subject: [PATCH 24/29] test: cover folder backup --- tests/test_folder_backup.py | 73 +++++++++++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 tests/test_folder_backup.py diff --git a/tests/test_folder_backup.py b/tests/test_folder_backup.py new file mode 100644 index 0000000..f53cda9 --- /dev/null +++ b/tests/test_folder_backup.py @@ -0,0 +1,73 @@ +import zipfile +from datetime import datetime + +import pytest + +from scripts import folder_backup as module +from scripts.folder_backup import BackupResult, backup_folder + +MOMENT = datetime(2024, 3, 5, 14, 30) + + +@pytest.fixture +def project(tmp_path): + root = tmp_path / "project" + (root / "src").mkdir(parents=True) + (root / "README.md").write_text("readme " * 100) + (root / "src" / "main.py").write_text("print('hi')\n" * 100) + return root + + +def test_backup_creates_timestamped_zip_with_all_files(project, tmp_path): + result = backup_folder(str(project), str(tmp_path / "backups"), now=MOMENT) + + assert result.path.name == "backup_2024-03-05_1430_project.zip" + with zipfile.ZipFile(result.path) as archive: + assert sorted(archive.namelist()) == ["project/README.md", "project/src/main.py"] + assert result.files == 2 + + +def test_backup_reports_sizes_and_savings(project, tmp_path): + result = backup_folder(str(project), str(tmp_path / "backups"), now=MOMENT) + + expected_size = len("readme " * 100) + len("print('hi')\n" * 100) + assert result.original_bytes == expected_size + assert result.zip_bytes == result.path.stat().st_size + assert result.savings_percent > 0 + + +def test_backup_of_empty_folder(tmp_path): + empty = tmp_path / "empty" + empty.mkdir() + + result = backup_folder(str(empty), str(tmp_path / "backups"), now=MOMENT) + + assert result.files == 0 + assert result.savings_percent == 0.0 + + +def test_backup_rejects_missing_source(tmp_path): + with pytest.raises(FileNotFoundError): + backup_folder(str(tmp_path / "missing"), str(tmp_path / "backups")) + + +def test_savings_percent_handles_zero_original_size(): + result = BackupResult(path=None, files=0, original_bytes=0, zip_bytes=22) + + assert result.savings_percent == 0.0 + + +def test_main_prints_backup_summary(project, tmp_path, capsys): + code = module.main([str(project), str(tmp_path / "backups")]) + + output = capsys.readouterr().out + assert code == 0 + assert "Backup created:" in output + assert "Files: 2" in output + + +def test_main_reports_missing_source(tmp_path, capsys): + code = module.main([str(tmp_path / "missing"), str(tmp_path / "backups")]) + + assert code == 1 + assert "Error:" in capsys.readouterr().err From 7e6549ba816ba18ff78f53908ecdf28929c17930 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:11:29 -0300 Subject: [PATCH 25/29] test: cover log analyzer --- tests/test_log_analyzer.py | 115 +++++++++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 tests/test_log_analyzer.py diff --git a/tests/test_log_analyzer.py b/tests/test_log_analyzer.py new file mode 100644 index 0000000..7fb5e0c --- /dev/null +++ b/tests/test_log_analyzer.py @@ -0,0 +1,115 @@ +import pytest + +from scripts import log_analyzer as module +from scripts.log_analyzer import analyze_log, format_report, normalize_level + +LOG = """\ +2024-01-01 10:00:01 INFO Server started from 10.0.0.1 +2024-01-01 10:15:00 WARN Slow response 10.0.0.1 +2024-01-01 10:20:00 ERROR Database timeout 10.0.0.2 + +2024-01-01 11:00:00 FATAL Out of memory +2024-01-01 11:30:00 debug cache miss +Something without level or timestamp +not-a-date 9999-99-99 99:99:99 INFO bad timestamp +""" + + +@pytest.fixture +def log_file(tmp_path): + path = tmp_path / "app.log" + path.write_text(LOG, encoding="utf-8") + return str(path) + + +@pytest.mark.parametrize( + ("raw", "expected"), + [("warn", "WARNING"), ("FATAL", "CRITICAL"), ("error", "ERROR"), ("Info", "INFO")], +) +def test_normalize_level(raw, expected): + assert normalize_level(raw) == expected + + +def test_analyze_counts_levels_and_lines(log_file): + report = analyze_log(log_file) + + assert report.total_lines == 8 + assert report.level_counts == {"INFO": 2, "WARNING": 1, "ERROR": 1, "CRITICAL": 1, "DEBUG": 1} + + +def test_analyze_collects_error_and_warning_samples(log_file): + report = analyze_log(log_file) + + assert len(report.errors) == 2 + assert "Database timeout" in report.errors[0] + assert "Out of memory" in report.errors[1] + assert len(report.warnings) == 1 + + +def test_analyze_groups_activity_by_hour_and_ignores_invalid_timestamps(log_file): + report = analyze_log(log_file) + + assert report.hourly_activity == {"2024-01-01 10:00": 3, "2024-01-01 11:00": 2} + + +def test_analyze_counts_ip_addresses(log_file): + report = analyze_log(log_file) + + assert report.ips == {"10.0.0.1": 2, "10.0.0.2": 1} + + +def test_analyze_limits_stored_samples(tmp_path): + path = tmp_path / "many.log" + path.write_text("ERROR boom\n" * 50 + "WARNING careful\n" * 30, encoding="utf-8") + + report = analyze_log(str(path)) + + assert len(report.errors) == 20 + assert len(report.warnings) == 10 + + +def test_analyze_truncates_long_lines(tmp_path): + path = tmp_path / "long.log" + path.write_text("ERROR " + "x" * 500 + "\n", encoding="utf-8") + + assert len(analyze_log(str(path)).errors[0]) == 200 + + +def test_analyze_rejects_missing_file(tmp_path): + with pytest.raises(FileNotFoundError): + analyze_log(str(tmp_path / "missing.log")) + + +def test_format_report_contains_all_sections(log_file): + text = format_report(analyze_log(log_file)) + + assert "Total lines: 8" in text + assert "CRITICAL" in text + assert "Latest errors:" in text + assert "Latest warnings:" in text + assert "10.0.0.1: 2" in text + assert "2024-01-01 10:00: 3 events" in text + + +def test_format_report_omits_empty_sections(tmp_path): + path = tmp_path / "plain.log" + path.write_text("nothing interesting\n", encoding="utf-8") + + text = format_report(analyze_log(str(path))) + + assert "Latest errors:" not in text + assert "Top 5 IPs:" not in text + + +def test_main_prints_report(log_file, capsys): + code = module.main([log_file]) + + assert code == 0 + assert "Level distribution:" in capsys.readouterr().out + + +def test_main_reports_missing_file(tmp_path, capsys): + code = module.main([str(tmp_path / "missing.log")]) + + assert code == 1 + assert "Error:" in capsys.readouterr().err From ab598af8cb6074a47c421d00596e2661aef09a5d Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:11:50 -0300 Subject: [PATCH 26/29] test: cover system monitor with mocked psutil --- tests/test_system_monitor.py | 116 +++++++++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 tests/test_system_monitor.py diff --git a/tests/test_system_monitor.py b/tests/test_system_monitor.py new file mode 100644 index 0000000..d03f2a1 --- /dev/null +++ b/tests/test_system_monitor.py @@ -0,0 +1,116 @@ +from types import SimpleNamespace + +import pytest + +from scripts import system_monitor as module +from scripts.system_monitor import check_alerts, format_stats, get_system_stats, monitor + +GB = 1024**3 + + +@pytest.fixture +def fake_psutil(monkeypatch): + readings = {"cpu": 10.0, "ram": 50.0, "disk": 40.0} + + monkeypatch.setattr(module.psutil, "cpu_percent", lambda interval: readings["cpu"]) + monkeypatch.setattr(module.psutil, "cpu_count", lambda: 8) + monkeypatch.setattr( + module.psutil, + "virtual_memory", + lambda: SimpleNamespace(percent=readings["ram"], used=4 * GB, total=16 * GB), + ) + monkeypatch.setattr( + module.psutil, + "disk_usage", + lambda path: SimpleNamespace(percent=readings["disk"], free=250 * GB), + ) + return readings + + +@pytest.fixture +def sleeps(monkeypatch): + calls = [] + monkeypatch.setattr(module.time, "sleep", calls.append) + return calls + + +def test_get_system_stats_converts_units(fake_psutil): + stats = get_system_stats() + + assert stats["cpu_percent"] == 10.0 + assert stats["cpu_count"] == 8 + assert stats["ram_used_gb"] == 4.0 + assert stats["ram_total_gb"] == 16.0 + assert stats["disk_free_gb"] == 250.0 + assert "T" in stats["timestamp"] + + +def test_format_stats_includes_every_metric(fake_psutil): + line = format_stats(get_system_stats()) + + assert "CPU: 10.0%" in line + assert "RAM: 50.0% (4.0/16.0 GB)" in line + assert "Disk: 40.0% (250.0 GB free)" in line + + +def test_check_alerts_returns_nothing_below_thresholds(): + stats = {"cpu_percent": 10, "ram_percent": 10, "disk_percent": 10} + + assert check_alerts(stats) == [] + + +def test_check_alerts_flags_each_metric_at_threshold(): + stats = {"cpu_percent": 80, "ram_percent": 85, "disk_percent": 90} + + alerts = check_alerts(stats) + + assert alerts == [ + "High CPU usage: 80.0%", + "High RAM usage: 85.0%", + "Critical disk usage: 90.0%", + ] + + +def test_check_alerts_accepts_custom_thresholds(): + stats = {"cpu_percent": 50, "ram_percent": 50, "disk_percent": 50} + + assert len(check_alerts(stats, cpu_threshold=40, ram_threshold=60, disk_threshold=60)) == 1 + + +def test_monitor_runs_bounded_iterations_and_sleeps_between(fake_psutil, sleeps, capsys): + monitor(interval=30, iterations=3) + + assert sleeps == [30, 30] + assert capsys.readouterr().out.count("CPU:") == 3 + + +def test_monitor_appends_stats_and_alerts_to_log_file(fake_psutil, sleeps, tmp_path): + fake_psutil["cpu"] = 95.0 + log_file = tmp_path / "logs" / "monitor.log" + + monitor(interval=1, log_file=str(log_file), iterations=2) + + content = log_file.read_text(encoding="utf-8") + assert content.count("CPU:") == 2 + assert content.count("High CPU usage: 95.0%") == 2 + + +def test_monitor_stops_cleanly_on_keyboard_interrupt(fake_psutil, monkeypatch, capsys): + def interrupt(seconds): + raise KeyboardInterrupt + + monkeypatch.setattr(module.time, "sleep", interrupt) + + monitor(interval=1) + + assert "Monitor stopped." in capsys.readouterr().out + + +def test_main_forwards_arguments(monkeypatch): + received = {} + monkeypatch.setattr(module, "monitor", lambda interval, log: received.update(i=interval, l=log)) + + code = module.main(["--interval", "5", "--log", "out.log"]) + + assert code == 0 + assert received == {"i": 5, "l": "out.log"} From dbf4775d7f676ee539b0cf7247ecffd69601adbc Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:12:00 -0300 Subject: [PATCH 27/29] test: use a named recorder in the main test --- tests/test_system_monitor.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_system_monitor.py b/tests/test_system_monitor.py index d03f2a1..24419f7 100644 --- a/tests/test_system_monitor.py +++ b/tests/test_system_monitor.py @@ -107,10 +107,10 @@ def interrupt(seconds): def test_main_forwards_arguments(monkeypatch): - received = {} - monkeypatch.setattr(module, "monitor", lambda interval, log: received.update(i=interval, l=log)) + received = [] + monkeypatch.setattr(module, "monitor", lambda *args: received.append(args)) code = module.main(["--interval", "5", "--log", "out.log"]) assert code == 0 - assert received == {"i": 5, "l": "out.log"} + assert received == [(5, "out.log")] From c22f5aec4ea57345b87bfa210d9d84f647a35f82 Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:12:12 -0300 Subject: [PATCH 28/29] docs: add CI badge --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 5cad2c5..96b86ac 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,7 @@ **A collection of Python scripts to automate everyday tasks** +[![CI](https://github.com/LacerdaTraderCode/python-automation-scripts/actions/workflows/ci.yml/badge.svg)](https://github.com/LacerdaTraderCode/python-automation-scripts/actions/workflows/ci.yml) [![Python](https://img.shields.io/badge/Python-3.11%2B-blue?logo=python&logoColor=white)](https://www.python.org/) [![License](https://img.shields.io/badge/License-MIT-orange)](https://github.com/LacerdaTraderCode/python-automation-scripts/blob/main/LICENSE) [![GitHub](https://img.shields.io/badge/GitHub-LacerdaTraderCode-181717?logo=github)](https://github.com/LacerdaTraderCode/python-automation-scripts) From a8752f4c0fbd13e9aa4a5f37f012a7735b8f935c Mon Sep 17 00:00:00 2001 From: Wagner Lacerda Date: Sun, 11 Oct 2026 08:12:17 -0300 Subject: [PATCH 29/29] docs: update structure and add development section --- README.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/README.md b/README.md index 96b86ac..65c41c5 100644 --- a/README.md +++ b/README.md @@ -60,12 +60,27 @@ python-automation-scripts/ │ ├── system_monitor.py │ ├── duplicate_finder.py │ └── log_analyzer.py +├── tests/ +├── .github/workflows/ci.yml ├── requirements.txt +├── requirements-dev.txt └── README.md ``` --- +## 🧪 Development + +```bash +pip install -r requirements-dev.txt +ruff check . && ruff format --check . +pytest -v +``` + +Tests run against temporary directories with SMTP and system metrics mocked. The same checks run on every push and pull request via GitHub Actions. + +--- + ## 📦 Installation ```bash