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 {}