mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Fable 5
40e553069d
Move files that share a delimiter-bounded name prefix within a folder (threshold 2) into a subfolder named after the prefix, dropping the prefix from the filename. Covers plugins/filter (docker, generate, has, merge, get/all), env handler subpackages, utils/roles/entity, utils/tests/swarm/write, role SQL/scripts/templates, keycloak/mailu task-includes, shared tasks/utils/load, and CI scripts. Update every import, query_file/script/include_tasks/src/lookup target, compose bind mount, and python -m path to the new locations. Extend the prefix-grouped-files lint to threshold 2 with dir-level .nocheck and file-head nocheck opt-outs, and skip groups whose shortest member's stem equals the shared prefix. Keep flat only where a loader forbids nesting: WordPress mu-plugins and the docs/requirements namespace. Ansible filter discovery is recursive, so nested filters stay Jinja-resolvable. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
"""Strip databases.csv rows for purged application entities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from utils.paths import FILE_DATABASE_SECRETS
|
|
from utils.roles.entity.name import get_entity_name
|
|
|
|
DEFAULT_CSV = FILE_DATABASE_SECRETS
|
|
|
|
_CSV_DELIMITER = ";"
|
|
_HEADER_COLUMNS = ("instance", "database", "username", "password")
|
|
|
|
|
|
def _resolve_csv_file() -> Path:
|
|
env_path = os.environ.get("FILE_DATABASE_SECRETS")
|
|
return Path(env_path) if env_path else DEFAULT_CSV
|
|
|
|
|
|
def _row_matches(row: list[str], targets: set[str]) -> bool:
|
|
if len(row) < 3:
|
|
return False
|
|
database = row[1].strip()
|
|
username = row[2].strip()
|
|
return database in targets or username in targets
|
|
|
|
|
|
def _is_header(row: list[str]) -> bool:
|
|
if len(row) < len(_HEADER_COLUMNS):
|
|
return False
|
|
return all(
|
|
row[i].strip().lower() == _HEADER_COLUMNS[i]
|
|
for i in range(len(_HEADER_COLUMNS))
|
|
)
|
|
|
|
|
|
def wipe_database_entries(
|
|
app_ids: list[str], csv_file: Path | None = None
|
|
) -> list[str]:
|
|
"""Remove rows whose database or username column matches an entity
|
|
derived from *app_ids* via :func:`get_entity_name`."""
|
|
path = csv_file or _resolve_csv_file()
|
|
if not path.exists():
|
|
return []
|
|
|
|
targets: set[str] = set()
|
|
for app_id in app_ids:
|
|
entity = get_entity_name(app_id)
|
|
if entity:
|
|
targets.add(entity)
|
|
if not targets:
|
|
return []
|
|
|
|
with path.open(newline="", encoding="utf-8") as fh:
|
|
rows = list(csv.reader(fh, delimiter=_CSV_DELIMITER))
|
|
|
|
kept: list[list[str]] = []
|
|
removed: list[str] = []
|
|
for row in rows:
|
|
if not row:
|
|
kept.append(row)
|
|
continue
|
|
if _is_header(row):
|
|
kept.append(row)
|
|
continue
|
|
if _row_matches(row, targets):
|
|
removed.append(f"{row[1].strip()}:{row[2].strip()}")
|
|
continue
|
|
kept.append(row)
|
|
|
|
if removed:
|
|
with path.open("w", newline="", encoding="utf-8") as fh:
|
|
csv.writer(fh, delimiter=_CSV_DELIMITER).writerows(kept)
|
|
|
|
return removed
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
if not argv:
|
|
print(
|
|
"usage: python -m utils.cleanup.databases_csv <APP_ID> [APP_ID ...]",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
|
|
removed = wipe_database_entries(argv)
|
|
if removed:
|
|
print(f">>> Wiped databases.csv entries: {', '.join(removed)}")
|
|
else:
|
|
print(">>> No databases.csv entries to wipe")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|