Spaces:
Running
Running
File size: 3,418 Bytes
7c7ef49 9183203 7c7ef49 9183203 7c7ef49 9183203 7c7ef49 baa22b7 9183203 baa22b7 9183203 7c7ef49 9183203 baa22b7 9183203 baa22b7 9183203 baa22b7 9183203 7c7ef49 9183203 7c7ef49 9183203 7c7ef49 9183203 7c7ef49 9183203 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
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 {}
|