Spaces:
Sleeping
Sleeping
| import logging | |
| from datetime import datetime | |
| import json | |
| from pathlib import Path | |
| import os | |
| from typing import Optional | |
| # Configure logging | |
| # Use /tmp for logs in production (e.g. Hugging Face Spaces) or local logs dir in development | |
| log_dir = Path("/tmp/schematic_ai_logs") if os.environ.get("SPACE_ID") else Path(__file__).parent.parent / "logs" | |
| log_dir.mkdir(exist_ok=True, parents=True) | |
| # Configure file handler for general logs | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', | |
| handlers=[ | |
| logging.StreamHandler() # Always log to console | |
| ] | |
| ) | |
| # Only add file handler if we can write to the directory | |
| try: | |
| if os.access(log_dir, os.W_OK): | |
| file_handler = logging.FileHandler(log_dir / "app.log") | |
| file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) | |
| logging.getLogger().addHandler(file_handler) | |
| except Exception as e: | |
| logging.warning(f"Could not set up file logging: {e}") | |
| def log_category_usage(category: Optional[str] = None, endpoint: Optional[str] = None, success: bool = True): | |
| """Log the usage of a category with endpoint and success information.""" | |
| if not os.access(log_dir, os.W_OK): | |
| logging.warning("Log directory is not writable, skipping category usage logging") | |
| return | |
| stats_file = log_dir / "category_stats.json" | |
| timestamp = datetime.now().isoformat() | |
| try: | |
| if stats_file.exists(): | |
| with open(stats_file, 'r') as f: | |
| stats = json.load(f) | |
| else: | |
| stats = {} | |
| # Initialize category if not exists | |
| category = category or "default" | |
| if category not in stats: | |
| stats[category] = { | |
| "total_requests": 0, | |
| "successful_requests": 0, | |
| "failed_requests": 0, | |
| "endpoints": {}, | |
| "last_used": None | |
| } | |
| # Update category stats | |
| stats[category]["total_requests"] += 1 | |
| if success: | |
| stats[category]["successful_requests"] += 1 | |
| else: | |
| stats[category]["failed_requests"] += 1 | |
| # Update endpoint stats | |
| if endpoint: | |
| if "endpoints" not in stats[category]: | |
| stats[category]["endpoints"] = {} | |
| if endpoint not in stats[category]["endpoints"]: | |
| stats[category]["endpoints"][endpoint] = 0 | |
| stats[category]["endpoints"][endpoint] += 1 | |
| # Update timestamp | |
| stats[category]["last_used"] = timestamp | |
| # Save updated stats | |
| with open(stats_file, 'w') as f: | |
| json.dump(stats, f, indent=4) | |
| except Exception as e: | |
| logging.error(f"Error logging category usage: {e}") | |
| def get_category_statistics(): | |
| """Get the usage statistics for all categories.""" | |
| if not os.access(log_dir, os.W_OK): | |
| logging.warning("Log directory is not writable, cannot read category statistics") | |
| return {} | |
| stats_file = log_dir / "category_stats.json" | |
| try: | |
| if stats_file.exists(): | |
| with open(stats_file, 'r') as f: | |
| return json.load(f) | |
| return {} | |
| except Exception as e: | |
| logging.error(f"Error reading category statistics: {e}") | |
| return {} | |