#!/usr/bin/env python3 """Create persistent Manual Invoice configuration outside the plugin cache.""" from __future__ import annotations import argparse import getpass import json import os import subprocess import sys from pathlib import Path from typing import Dict, Optional from runtime_config import PLUGIN_ROOT, PROJECT_CONFIG_NAME, read_env_file, user_config_path DEFAULT_MAX_BYTES = 10 * 1024**3 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Configure Manual Invoice.") parser.add_argument("--scope", choices=("project", "user"), default="project") parser.add_argument("--project-root", type=Path, default=Path.cwd()) parser.add_argument("--config-path", type=Path) parser.add_argument("--credentials", help="Path to an ADC-compatible credential JSON file.") parser.add_argument("--job-project", help="Optional BigQuery query Job Project.") parser.add_argument("--location", help="Optional default BigQuery location.") parser.add_argument("--max-bytes-billed", type=int) parser.add_argument( "--github-token", help=( "GitHub PAT used by O11Y authentication and private Skill installation. " "Prefer the hidden interactive prompt or GITHUB_PAT environment variable." ), ) parser.add_argument( "--clinic-api-key", help=argparse.SUPPRESS, ) parser.add_argument("--o11y-skills-ref", default="main") parser.add_argument( "--install-system-tools", nargs="?", const="all", metavar="TOOLS", help=( "Install missing system CLIs without prompting. Use a comma-separated list " "such as aws,gh or omit the value for all installable missing tools." ), ) parser.add_argument("--skip-system-tools", action="store_true") parser.add_argument("--skip-o11y-skills", action="store_true") parser.add_argument("--skip-o11y-auth-check", action="store_true") parser.add_argument("--non-interactive", action="store_true") parser.add_argument("--force", action="store_true") return parser.parse_args() def target_path(args: argparse.Namespace) -> Path: if args.config_path: return args.config_path.expanduser().resolve() if args.scope == "user": return user_config_path().resolve() return args.project_root.expanduser().resolve() / PROJECT_CONFIG_NAME def prompt(label: str, default: str = "") -> str: suffix = f" [{default}]" if default else "" entered = input(f"{label}{suffix}: ").strip() return entered if entered else default def prompt_secret(label: str, has_existing: bool = False) -> str: suffix = " [configured; Enter keeps current]" if has_existing else "" return getpass.getpass(f"{label}{suffix}: ").strip() def validate_github_token(value: str) -> str: token = value.strip() if not token: return "" if any(character.isspace() for character in token): raise ValueError("GitHub token must not contain whitespace.") return token def validate_clinic_api_key(value: str) -> str: key = value.strip() if not key: return "" if any(character.isspace() for character in key): raise ValueError("Clinic API key must not contain whitespace.") return key def validate_credentials(value: str) -> tuple[str, Optional[str], Optional[str]]: if not value: return "", None, None path = Path(value).expanduser().resolve() if not path.is_file(): raise ValueError(f"Credential file does not exist: {path}") try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ValueError(f"Credential file is not valid JSON: {path}") from exc if not isinstance(data, dict): raise ValueError("Credential JSON must contain an object.") return str(path), data.get("project_id"), data.get("client_email") def dotenv_value(value: str) -> str: return json.dumps(value, ensure_ascii=False) def update_local_git_exclude(project_root: Path, config_path: Path) -> bool: try: git_dir_result = subprocess.run( ["git", "rev-parse", "--git-dir"], cwd=project_root, check=True, capture_output=True, text=True, ) top_level_result = subprocess.run( ["git", "rev-parse", "--show-toplevel"], cwd=project_root, check=True, capture_output=True, text=True, ) except (FileNotFoundError, subprocess.CalledProcessError): return False top_level = Path(top_level_result.stdout.strip()).resolve() try: relative = config_path.resolve().relative_to(top_level) except ValueError: return False git_dir = Path(git_dir_result.stdout.strip()) if not git_dir.is_absolute(): git_dir = (project_root / git_dir).resolve() exclude = git_dir / "info" / "exclude" exclude.parent.mkdir(parents=True, exist_ok=True) entry = relative.as_posix() existing = exclude.read_text(encoding="utf-8") if exclude.exists() else "" if entry in {line.strip() for line in existing.splitlines()}: return False with exclude.open("a", encoding="utf-8") as handle: if existing and not existing.endswith("\n"): handle.write("\n") handle.write(f"{entry}\n") return True def main() -> int: args = parse_args() path = target_path(args) existing: Dict[str, str] = read_env_file(path) credentials = args.credentials job_project = args.job_project location = args.location max_bytes = args.max_bytes_billed github_token = args.github_token or os.environ.get("GITHUB_PAT") or os.environ.get( "GITHUB_TOKEN" ) clinic_api_key = args.clinic_api_key or os.environ.get("CLINIC_API_KEY") if not args.non_interactive: if credentials is None: credentials = prompt( "GOOGLE_APPLICATION_CREDENTIALS (leave empty for ambient ADC)", existing.get("GOOGLE_APPLICATION_CREDENTIALS", ""), ) if job_project is None: job_project = prompt( "BQ_JOB_PROJECT (leave empty for ADC inference)", existing.get("BQ_JOB_PROJECT", ""), ) if location is None: location = prompt( "BQ_LOCATION (leave empty for table inference)", existing.get("BQ_LOCATION", ""), ) if max_bytes is None: default_bytes = existing.get("BQ_MAX_BYTES_BILLED", str(DEFAULT_MAX_BYTES)) max_bytes = int(prompt("BQ_MAX_BYTES_BILLED", default_bytes)) if github_token is None: entered_token = prompt_secret("GITHUB_PAT for O11Y", "GITHUB_PAT" in existing) github_token = entered_token or existing.get("GITHUB_PAT", "") credentials = credentials if credentials is not None else existing.get( "GOOGLE_APPLICATION_CREDENTIALS", "" ) job_project = job_project if job_project is not None else existing.get("BQ_JOB_PROJECT", "") location = location if location is not None else existing.get("BQ_LOCATION", "") max_bytes = max_bytes if max_bytes is not None else int( existing.get("BQ_MAX_BYTES_BILLED", str(DEFAULT_MAX_BYTES)) ) github_token = github_token if github_token is not None else existing.get("GITHUB_PAT", "") clinic_api_key = clinic_api_key if clinic_api_key is not None else existing.get( "CLINIC_API_KEY", "" ) credentials, credential_project, client_email = validate_credentials(credentials) github_token = validate_github_token(github_token) clinic_api_key = validate_clinic_api_key(clinic_api_key) if not github_token and not args.skip_o11y_skills: raise ValueError( "GitHub token is required to initialize O11Y Skills. " "Use the hidden prompt, --github-token, GITHUB_PAT, or --skip-o11y-skills." ) if path.exists() and not args.force and not existing: raise ValueError(f"Refusing to replace unrecognized config file without --force: {path}") lines = [ "# Generated by $manual-invoice configure. Keep this file private.", f"BQ_JOB_PROJECT={dotenv_value(job_project)}", f"BQ_LOCATION={dotenv_value(location)}", f"BQ_MAX_BYTES_BILLED={max_bytes}", ] if credentials: lines.append(f"GOOGLE_APPLICATION_CREDENTIALS={dotenv_value(credentials)}") else: lines.append("# GOOGLE_APPLICATION_CREDENTIALS= # Use ambient ADC") if github_token: lines.append(f"GITHUB_PAT={dotenv_value(github_token)}") else: lines.append("# GITHUB_PAT= # Required for O11Y metrics/log access") if clinic_api_key: # Backward compatibility for an existing config or an explicitly supplied legacy value. # Interactive configure no longer requests a separate Clinic credential. lines.append(f"CLINIC_API_KEY={dotenv_value(clinic_api_key)}") managed_keys = { "BQ_JOB_PROJECT", "BQ_LOCATION", "BQ_MAX_BYTES_BILLED", "GOOGLE_APPLICATION_CREDENTIALS", "GITHUB_PAT", "CLINIC_API_KEY", } for key, value in existing.items(): if key not in managed_keys: lines.append(f"{key}={dotenv_value(value)}") path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(f".{path.name}.tmp") temporary.write_text("\n".join(lines) + "\n", encoding="utf-8") os.chmod(temporary, 0o600) os.replace(temporary, path) os.chmod(path, 0o600) git_exclude_updated = False if args.scope == "project" and not args.config_path: git_exclude_updated = update_local_git_exclude(args.project_root.resolve(), path) system_tools = None if not args.skip_system_tools: tool_bootstrap = ( PLUGIN_ROOT / "skills" / "manual-invoice" / "scripts" / "bootstrap_system_tools.py" ) if args.install_system_tools is not None: install_result = subprocess.run( [ sys.executable, str(tool_bootstrap), "--yes", "--tools", args.install_system_tools, ], check=True, ) if install_result.returncode != 0: raise RuntimeError("System tool installation failed.") elif not args.non_interactive: subprocess.run([sys.executable, str(tool_bootstrap)], check=True) check_result = subprocess.run( [sys.executable, str(tool_bootstrap), "--check"], check=True, capture_output=True, text=True, ) system_tools = json.loads(check_result.stdout) o11y_skills = None o11y_auth = None if not args.skip_o11y_skills: child_env = os.environ.copy() child_env["GITHUB_PAT"] = github_token child_env["GITHUB_TOKEN"] = github_token installer = PLUGIN_ROOT / "skills" / "manual-invoice" / "scripts" / "install_o11y_skills.py" install_result = subprocess.run( [sys.executable, str(installer), "--ref", args.o11y_skills_ref], check=True, capture_output=True, text=True, env=child_env, ) o11y_skills = json.loads(install_result.stdout) if not args.skip_o11y_auth_check: checker = PLUGIN_ROOT / "skills" / "manual-invoice" / "scripts" / "test_o11y.py" check_env = child_env.copy() check_env["MANUAL_INVOICE_CONFIG"] = str(path) auth_result = subprocess.run( [sys.executable, str(checker), "--auth", "--provider", "aws"], check=False, capture_output=True, text=True, env=check_env, ) if auth_result.stdout.strip(): o11y_auth = json.loads(auth_result.stdout) else: o11y_auth = { "status": "error", "auth": {"ok": False, "error": "O11Y authentication check failed."}, } result = { "status": "configured", "scope": args.scope if not args.config_path else "custom", "config_path": str(path), "credential_mode": "file" if credentials else "ambient_adc", "credential_project": credential_project, "client_email": client_email, "job_project": job_project or None, "location": location or None, "max_bytes_billed": max_bytes, "github_token_configured": bool(github_token), "clinic_api_skill_installed": ( Path(os.environ.get("CODEX_HOME", Path.home() / ".codex")) / "skills" / "clinic-api" / "SKILL.md" ).is_file(), "clinic_api_auth_configured": bool(clinic_api_key), "system_tools": system_tools, "o11y_skills": o11y_skills, "o11y_auth": o11y_auth, "git_exclude_updated": git_exclude_updated, } print(json.dumps(result, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())