diff --git a/src/aipass/api/apps/handlers/auth/env.py b/src/aipass/api/apps/handlers/auth/env.py index a3cc2405..7c0d077f 100644 --- a/src/aipass/api/apps/handlers/auth/env.py +++ b/src/aipass/api/apps/handlers/auth/env.py @@ -9,14 +9,10 @@ """ .env File Handler -Manages .env file reading, creation, and parsing. -Searches multiple paths, creates templates, handles env variables. +Manages .env file creation for API credential setup. Functions: - read_env_file() - Read environment variable from .env files (multi-path search) - read_env_file_dict() - Read all variables from .env file as dictionary create_env_template() - Create .env template for provider - validate_env_exists() - Check if .env file exists at any search path """ # Infrastructure @@ -24,7 +20,7 @@ from pathlib import Path import sys # Standard library -from typing import Optional, Dict, List +from typing import Optional, Dict # Logging from aipass.prax import logger @@ -33,114 +29,6 @@ from aipass.prax import logger from aipass.api.apps.handlers.json import json_handler -# ============================================== -# CONSTANTS -# ============================================== - -# Default .env search paths (in order of priority) -# Navigate: env.py -> auth/ -> handlers/ -> apps/ -> api/ -API_ROOT = Path(__file__).resolve().parent.parent.parent.parent -DEFAULT_ENV_PATHS = [ - Path.home() / ".secrets" / "aipass" / ".env", # ~/.secrets/aipass/.env (cross-platform standard) - API_ROOT / ".env", # /.env - Path.cwd() / ".env", # /.env -] - - -# ============================================== -# ENV FILE READING -# ============================================== - -def read_env_file(env_var: str, search_paths: Optional[List[Path]] = None) -> Optional[str]: - """ - Read environment variable from .env files with multi-path search. - - Searches multiple .env file locations in order: - 1. /.env (package-relative) - 2. /.env (current working directory) - - Args: - env_var: Environment variable name to read (e.g., 'OPENROUTER_API_KEY') - search_paths: Optional custom search paths (defaults to DEFAULT_ENV_PATHS) - - Returns: - str: Variable value if found, None otherwise - - Example: - >>> api_key = read_env_file('OPENROUTER_API_KEY') - >>> if api_key: - ... print(f"Found key: {api_key[:20]}...") - """ - paths = search_paths or DEFAULT_ENV_PATHS - - for env_file in paths: - if not env_file.exists(): - continue - - try: - with open(env_file, 'r', encoding='utf-8') as f: - for line in f: - line = line.strip() - # Skip empty lines and comments - if not line or line.startswith('#'): - continue - # Parse key=value - if '=' in line: - key, value = line.split('=', 1) - if key.strip() == env_var: - # Found env_var in env_file - json_handler.log_operation("env_loaded", {"variable": env_var, "source": str(env_file)}) - return value.strip() - except Exception as e: - # Error reading env_file - logger.warning(f"Error reading env file '{env_file}': {e}") - continue - - # Variable not found in any .env file - return None - - -def read_env_file_dict(env_path: Path) -> Dict[str, str]: - """ - Read all environment variables from a .env file as dictionary. - - Args: - env_path: Path to specific .env file to read - - Returns: - dict: Dictionary of key-value pairs from .env file - - Example: - >>> env_vars = read_env_file_dict(Path('api/.env')) - >>> print(env_vars.get('OPENROUTER_API_KEY')) - """ - env_dict = {} - - if not env_path.exists(): - # .env file not found - return env_dict - - try: - with open(env_path, 'r', encoding='utf-8') as f: - for line in f: - line = line.strip() - # Skip empty lines and comments - if not line or line.startswith('#'): - continue - # Parse key=value - if '=' in line: - key, value = line.split('=', 1) - env_dict[key.strip()] = value.strip() - - # Read variables from env_path - return env_dict - - except Exception as e: - # Error reading env_path - logger.error(f"Error reading env file '{env_path}': {e}") - return env_dict - - # ============================================== # ENV FILE CREATION # ============================================== @@ -206,6 +94,7 @@ OPENAI_API_KEY=sk-your-openai-key-here # Created .env template logger.info(f"Created .env template at {env_path}") + json_handler.log_operation("env_template_created", {"path": str(env_path), "provider": provider}) return True except Exception as e: @@ -214,105 +103,3 @@ OPENAI_API_KEY=sk-your-openai-key-here return False -def create_custom_env_template(variables: Dict[str, str], target_path: Path, - header: Optional[str] = None) -> bool: - """ - Create custom .env template with specific variables. - - Args: - variables: Dictionary of variable names to placeholder values - target_path: Path where .env file should be created - header: Optional custom header comment - - Returns: - bool: True if successful - - Example: - >>> vars = { - ... 'DATABASE_URL': 'postgresql://localhost/mydb', - ... 'SECRET_KEY': 'your-secret-key-here' - ... } - >>> create_custom_env_template(vars, Path('/path/to/.env')) - """ - # Don't overwrite existing file - if target_path.exists(): - # .env file already exists - return True - - try: - # Ensure parent directory exists - target_path.parent.mkdir(parents=True, exist_ok=True) - - # Build template content - content_lines = [] - - # Add header - if header: - content_lines.append(f"# {header}") - else: - content_lines.append("# Environment Variables") - content_lines.append("") - - # Add variables - for key, value in variables.items(): - content_lines.append(f"{key}={value}") - - content = "\n".join(content_lines) + "\n" - - # Write file - with open(target_path, 'w', encoding='utf-8') as f: - f.write(content) - - # Created custom .env template - logger.info(f"Created custom .env template at {target_path}") - return True - - except Exception as e: - # Failed to create custom .env template - logger.error(f"Failed to create custom .env template: {e}") - return False - - -# ============================================== -# VALIDATION -# ============================================== - -def validate_env_exists(search_paths: Optional[List[Path]] = None) -> Optional[Path]: - """ - Check if .env file exists at any search path. - - Args: - search_paths: Optional custom search paths (defaults to DEFAULT_ENV_PATHS) - - Returns: - Path: First found .env file path, or None if none exist - - Example: - >>> env_path = validate_env_exists() - >>> if env_path: - ... print(f"Found .env at {env_path}") - """ - paths = search_paths or DEFAULT_ENV_PATHS - - for env_path in paths: - if env_path.exists(): - # Found .env file - return env_path - - # No .env file found in search paths - return None - - -def get_env_search_paths() -> List[Path]: - """ - Get list of default .env search paths. - - Returns: - list: List of Path objects for .env search locations - - Example: - >>> paths = get_env_search_paths() - >>> for p in paths: - ... print(p) - """ - return DEFAULT_ENV_PATHS.copy() diff --git a/src/aipass/api/apps/handlers/auth/keys.py b/src/aipass/api/apps/handlers/auth/keys.py index cf160f1f..c9ef9031 100644 --- a/src/aipass/api/apps/handlers/auth/keys.py +++ b/src/aipass/api/apps/handlers/auth/keys.py @@ -10,13 +10,12 @@ API Key Management Handler Handles API key retrieval and validation for multiple providers. -Uses fallback chain: config → env → .env files. +Keys are read from config JSON or directly from ~/.secrets/aipass/.env. Functions: - get_api_key() - Get validated API key with fallback chain + get_api_key() - Get validated API key validate_key() - Validate key format for provider get_key_from_config() - Retrieve key from config JSON - get_key_from_env() - Retrieve key from environment variable get_validation_rules() - Get provider-specific validation rules """ @@ -25,15 +24,11 @@ from pathlib import Path import sys # Standard library -import os from typing import Optional, Dict, Any # Logging from aipass.prax import logger -# Internal handlers -from aipass.api.apps.handlers.auth.env import read_env_file - # JSON handler from aipass.api.apps.handlers.json import json_handler @@ -73,12 +68,11 @@ VALIDATION_RULES = { def get_api_key(provider: str = "openrouter") -> Optional[str]: """ - Get validated API key for provider with fallback chain. + Get validated API key for provider. - Fallback order: + Sources (in order): 1. Config JSON file (api_json/api_connect_config.json) - 2. Environment variable - 3. .env file (multi-path search) + 2. Secrets file (~/.secrets/aipass/.env) Args: provider: Provider name (default: 'openrouter') @@ -97,23 +91,13 @@ def get_api_key(provider: str = "openrouter") -> Optional[str]: # 1. Try config file key = get_key_from_config(provider) if key and validate_key(key, provider): - # Using key from config source = "config" - # 2. Try environment variable + # 2. Try secrets file if not source: - key = get_key_from_env(provider) + key = _read_key_from_secrets(provider) if key and validate_key(key, provider): - # Using key from environment - source = "env" - - # 3. Try .env file - if not source: - env_var = f"{provider.upper()}_API_KEY" - key = read_env_file(env_var) - if key and validate_key(key, provider): - # Using key from .env file - source = "dotenv" + source = "secrets" if source: json_handler.log_operation("key_retrieved", {"provider": provider, "source": source}) @@ -123,11 +107,41 @@ def get_api_key(provider: str = "openrouter") -> Optional[str]: return None except Exception as e: - # Failed to get key logger.error(f"Failed to get API key for provider '{provider}': {e}") return None +def _read_key_from_secrets(provider: str) -> Optional[str]: + """ + Read API key directly from ~/.secrets/aipass/.env. + + Args: + provider: Provider name (e.g., 'openrouter') + + Returns: + str: API key value or None if not found + """ + secrets_path = Path.home() / ".secrets" / "aipass" / ".env" + if not secrets_path.exists(): + return None + + try: + env_var = f"{provider.upper()}_API_KEY" + with open(secrets_path, 'r', encoding='utf-8') as f: + for line in f: + line = line.strip() + if not line or line.startswith('#'): + continue + if '=' in line: + key, value = line.split('=', 1) + if key.strip() == env_var: + return value.strip() + return None + except Exception as e: + logger.warning(f"Error reading secrets file: {e}") + return None + + def get_key_from_config(provider: str) -> Optional[str]: """ Retrieve API key from config JSON file. @@ -171,31 +185,6 @@ def get_key_from_config(provider: str) -> Optional[str]: return None -def get_key_from_env(provider: str) -> Optional[str]: - """ - Retrieve API key from environment variable. - - Checks os.environ for {PROVIDER}_API_KEY. - - Args: - provider: Provider name (e.g., 'openrouter') - - Returns: - str: API key from environment or None if not found - - Example: - >>> key = get_key_from_env('openrouter') - >>> # Checks OPENROUTER_API_KEY env variable - """ - env_var = f"{provider.upper()}_API_KEY" - key = os.getenv(env_var) - - if key: - # Found key in environment variable - return key - - return None - # ============================================== # KEY VALIDATION @@ -295,16 +284,12 @@ def diagnose_key(provider: str = "openrouter") -> str: source = "config" if not key: - key = get_key_from_env(provider) - source = "env" + key = _read_key_from_secrets(provider) + source = "secrets" if not key: - env_var = f"{provider.upper()}_API_KEY" - key = read_env_file(env_var) - source = "dotenv" - - if not key: - return "No API key found in any source (config, environment, .env file)" + secrets_path = Path.home() / ".secrets" / "aipass" / ".env" + return f"API key for {provider} not found. Expected at {secrets_path}. Run drone @api setup to configure." # Key exists but failed validation — explain why key = key.strip() diff --git a/src/aipass/api/apps/handlers/openrouter/provision.py b/src/aipass/api/apps/handlers/openrouter/provision.py index 78eb4f89..4397c75e 100644 --- a/src/aipass/api/apps/handlers/openrouter/provision.py +++ b/src/aipass/api/apps/handlers/openrouter/provision.py @@ -167,7 +167,6 @@ def provision_json_folder(json_folder: Path) -> bool: json_folder.mkdir(parents=True, exist_ok=True) logger.info(f"Created JSON folder: {json_folder}") - logger.info(f"Created JSON folder: {json_folder}") return True except Exception as e: diff --git a/src/aipass/api/apps/handlers/usage/aggregation.py b/src/aipass/api/apps/handlers/usage/aggregation.py index e5f2b8b6..0ca65f0b 100644 --- a/src/aipass/api/apps/handlers/usage/aggregation.py +++ b/src/aipass/api/apps/handlers/usage/aggregation.py @@ -242,51 +242,3 @@ def calculate_totals(usage_data: List[Dict]) -> Dict[str, float]: } -def get_model_breakdown(caller: Optional[str] = None) -> Dict[str, Dict[str, int]]: - """ - Calculate model usage breakdown by caller or globally - - Args: - caller: Optional caller name to filter by (None = all callers) - - Returns: - Dict of {model_name: {"requests": count}} - Returns empty dict {} if no data found - """ - try: - data_path = API_JSON_DIR / DATA_FILE - if not data_path.exists(): - logger.info(f"[{MODULE_NAME}] No model breakdown data file found") - return {} - - with open(data_path, 'r', encoding='utf-8') as f: - data = json.load(f) - - if not data or "data" not in data: - logger.info(f"[{MODULE_NAME}] No model breakdown data available") - return {} - - model_stats = {} - usage_by_caller = data["data"].get("usage_by_caller", {}) - - if caller: - # Single caller breakdown - caller_data = usage_by_caller.get(caller, {}) - models_used = caller_data.get("models_used", {}) - for model, count in models_used.items(): - model_stats[model] = {"requests": count} - else: - # Global breakdown across all callers - for caller_name, caller_data in usage_by_caller.items(): - models_used = caller_data.get("models_used", {}) - for model, count in models_used.items(): - if model not in model_stats: - model_stats[model] = {"requests": 0} - model_stats[model]["requests"] += count - - logger.info(f"[{MODULE_NAME}] Retrieved model breakdown: {len(model_stats)} models") - return model_stats - - except Exception as e: - logger.error(f"[{MODULE_NAME}] Failed to get model breakdown: {e}") - return {} diff --git a/src/aipass/api/apps/handlers/usage/cleanup.py b/src/aipass/api/apps/handlers/usage/cleanup.py index 26f9f2de..c6da1aae 100644 --- a/src/aipass/api/apps/handlers/usage/cleanup.py +++ b/src/aipass/api/apps/handlers/usage/cleanup.py @@ -125,70 +125,6 @@ def _identify_old_generations(generation_tracking: Dict, cutoff_date: datetime) return old_generations -def cleanup_daily_totals(data_file_path: Path, retention_days: int = 90) -> int: - """Remove daily total entries older than retention period.""" - try: - cutoff_date = (datetime.now() - timedelta(days=retention_days)).date() - data = _read_json(data_file_path) - if not data: - return 0 - - data_content = data.get("data", data) - old_dates = [] - daily_totals = data_content.get("daily_totals", {}) - - for date_str in daily_totals.keys(): - try: - date_obj = datetime.fromisoformat(date_str).date() - if date_obj < cutoff_date: - old_dates.append(date_str) - except (ValueError, TypeError) as e: - logger.warning(f"Invalid date format '{date_str}', marking for cleanup: {e}") - old_dates.append(date_str) - - if not old_dates: - return 0 - - for date_str in old_dates: - del data_content["daily_totals"][date_str] - - if "data" in data: - data["data"] = data_content - data["timestamp"] = datetime.now().isoformat() - - _write_json(data_file_path, data) - logger.info(f"Cleaned up {len(old_dates)} daily total entries") - logger.info(f"Cleaned up {len(old_dates)} daily total entries older than {retention_days} days") - - return len(old_dates) - - except Exception as e: - # logger.error(f"Daily totals cleanup failed: {e}") - logger.error(f"Daily totals cleanup failed: {e}") - raise - - -def auto_cleanup(data_file_path: Path, config: Optional[Dict] = None) -> Dict[str, int]: - """Perform automatic cleanup based on configuration.""" - try: - gen_retention = config.get("cleanup_old_data_days", 30) if config else 30 - daily_retention = config.get("cleanup_daily_totals_days", 90) if config else 90 - - generations_removed = cleanup_old_data(data_file_path, gen_retention) - daily_totals_removed = cleanup_daily_totals(data_file_path, daily_retention) - - logger.info(f"Auto cleanup: {generations_removed} generations, {daily_totals_removed} daily totals removed") - - return { - "generations_removed": generations_removed, - "daily_totals_removed": daily_totals_removed - } - - except Exception as e: - # logger.error(f"Auto cleanup failed: {e}") - logger.error(f"Auto cleanup failed: {e}") - raise - def get_cleanup_stats(data_file_path: Path) -> Dict[str, int]: """Get statistics about data that could be cleaned up.""" diff --git a/src/aipass/api/apps/modules/api_key.py b/src/aipass/api/apps/modules/api_key.py index 3600e151..1fda9c01 100644 --- a/src/aipass/api/apps/modules/api_key.py +++ b/src/aipass/api/apps/modules/api_key.py @@ -37,7 +37,7 @@ def print_introspection(): console.print("[cyan]Connected Handlers:[/cyan]") console.print(" • api.apps.handlers.auth.keys") - console.print(" • api.apps.handlers.auth.env") + console.print(" • api.apps.handlers.auth.env (template creation)") console.print(" • api.apps.handlers.config.provider") console.print(" • api.apps.handlers.json.json_handler") console.print()