#!/usr/bin/env python3 """ Seed FailDaily with demo users + fails (no npm/env copy-paste needed). This wrapper: 1) Auto-detects DB_HOST/DB_PORT/DB_USER/DB_NAME from repo config (OVH stack / compose / docs). 2) Asks you for DB_PASSWORD (or uses DB_PASSWORD env / /run/secrets/db_password). 3) Runs the Node seeder: backend-api/scripts/seed-fails.js Usage: python3 scripts/seed_fails.py --count 100 Options: --count N number of fails/users to create (default 100) --start N start index (default 1) --days N spread timestamps over last N days (default 30) --fail-anon PCT percent of anonymous fails (0-100, default 0) --prefix TEXT optional prefix for pseudo (default "") --yes non-interactive (won't prompt for confirmation) --source PATH force a config source file to parse Notes: - We never print the password. """ from __future__ import annotations import argparse import getpass import os import re import subprocess import sys from pathlib import Path from typing import Dict, Optional, Tuple def repo_root() -> Path: return Path(__file__).resolve().parents[1] def read_text(path: Path) -> str: return path.read_text(encoding="utf-8", errors="replace") def parse_env_from_yaml_like(text: str) -> Dict[str, str]: """ Very small parser for lines like: DB_HOST: something - DB_HOST=something - DB_HOST: something """ out: Dict[str, str] = {} for line in text.splitlines(): s = line.strip() if not s or s.startswith("#"): continue # "- KEY=VALUE" m = re.match(r"^-+\s*([A-Z0-9_]+)\s*=\s*(.+?)\s*$", s) if m: out[m.group(1)] = m.group(2).strip().strip("'\"") continue # "KEY: VALUE" or "- KEY: VALUE" m = re.match(r"^-*\s*([A-Z0-9_]+)\s*:\s*(.+?)\s*$", s) if m: out[m.group(1)] = m.group(2).strip().strip("'\"") continue return out def parse_ovh_md(text: str) -> Dict[str, str]: out: Dict[str, str] = {} # Examples in MIGRATION_OVH_CLOUD.md: # - Host: `gb9434-001.eu.clouddb.ovh.net` m = re.search(r"Host:\s*`([^`]+)`", text, flags=re.IGNORECASE) if m: out["DB_HOST"] = m.group(1).strip() # Example: "...clouddb.ovh.net:35670/faildaily_bdd" m = re.search(r"clouddb\.ovh\.net:(\d{2,6})/([A-Za-z0-9_]+)", text) if m: out.setdefault("DB_PORT", m.group(1)) out.setdefault("DB_NAME", m.group(2)) # Example commands: "-u taaazzz" m = re.search(r"\s-u\s+([A-Za-z0-9_]+)\s", text) if m: out.setdefault("DB_USER", m.group(1)) return out def detect_db_config(forced_source: Optional[Path] = None) -> Tuple[Dict[str, str], Optional[Path]]: root = repo_root() candidates = [] if forced_source: candidates.append(forced_source) else: candidates.extend( [ root / "docker" / "docker-stack.ovh-cloud.yml", root / "docker-compose.ssl-production.yml", root / "docker" / "docker-compose.ovh-v2.yml", root / "docker" / "docker-compose.ovh.yml", root / "MIGRATION_OVH_CLOUD.md", root / "README.md", ] ) for path in candidates: if not path.exists(): continue text = read_text(path) if path.suffix in (".yml", ".yaml"): env_map = parse_env_from_yaml_like(text) elif path.suffix.lower() == ".md": env_map = parse_ovh_md(text) env_map.update(parse_env_from_yaml_like(text)) else: env_map = parse_env_from_yaml_like(text) keys = ("DB_HOST", "DB_PORT", "DB_USER", "DB_NAME") found = {k: v for k, v in env_map.items() if k in keys and v} if found: return found, path return {}, None def read_secret_file(path: Path) -> Optional[str]: try: if path.exists(): v = path.read_text(encoding="utf-8", errors="replace").strip() return v or None except Exception: return None return None def build_final_env(detected: Dict[str, str]) -> Dict[str, str]: env = dict(os.environ) for key in ("DB_HOST", "DB_PORT", "DB_USER", "DB_NAME"): if env.get(key): continue if detected.get(key): env[key] = str(detected[key]) # Password: env > /run/secrets > prompt if not env.get("DB_PASSWORD"): secret = read_secret_file(Path("/run/secrets/db_password")) if secret: env["DB_PASSWORD"] = secret return env def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--count", type=int, default=100) parser.add_argument("--start", type=int, default=1) parser.add_argument("--days", type=int, default=30) parser.add_argument("--fail-anon", type=float, default=0.0, dest="fail_anon") parser.add_argument("--prefix", type=str, default="") parser.add_argument("--yes", action="store_true") parser.add_argument("--source", type=str, default=None, help="force config source path") args = parser.parse_args() if args.count < 1: print("Invalid --count", file=sys.stderr) return 2 if args.start < 1: print("Invalid --start", file=sys.stderr) return 2 if args.days < 1: print("Invalid --days", file=sys.stderr) return 2 if args.fail_anon < 0 or args.fail_anon > 100: print("Invalid --fail-anon (0-100)", file=sys.stderr) return 2 forced = Path(args.source).expanduser().resolve() if args.source else None detected, source_path = detect_db_config(forced_source=forced) env = build_final_env(detected) missing = [k for k in ("DB_HOST", "DB_PORT", "DB_USER", "DB_NAME") if not env.get(k)] if missing: print(f"Missing DB config keys: {', '.join(missing)}", file=sys.stderr) print("Provide them via env vars or use --source to point to your OVH config/documentation file.", file=sys.stderr) return 2 if not env.get("DB_PASSWORD"): env["DB_PASSWORD"] = getpass.getpass("DB_PASSWORD (OVH MySQL): ") if not env["DB_PASSWORD"].strip(): print("Empty password, abort.", file=sys.stderr) return 2 print("🧭 DB target:") print(f" host: {env['DB_HOST']}") print(f" port: {env['DB_PORT']}") print(f" user: {env['DB_USER']}") print(f" name: {env['DB_NAME']}") if source_path: print(f" detectedFrom: {source_path}") if not args.yes: answer = input(f"Seed {args.count} fails/users into this DB? Type 'oui' to continue: ").strip().lower() if answer != "oui": print("Canceled.") return 0 root = repo_root() node_script = root / "backend-api" / "scripts" / "seed-fails.mjs" if not node_script.exists(): print(f"Missing node seeder: {node_script}", file=sys.stderr) return 2 cmd = [ "node", str(node_script), "--count", str(args.count), "--start", str(args.start), "--days", str(args.days), "--fail-anon", str(args.fail_anon), "--prefix", args.prefix, "--yes", ] print("▶ Running:", " ".join(cmd)) proc = subprocess.run(cmd, cwd=str(root), env=env) return int(proc.returncode) if __name__ == "__main__": raise SystemExit(main())