添加api调用接口

Co-authored-by: Copilot <copilot@github.com>

app/utils/api_handle.py

@@ -0,0 +1,77 @@
+import functools
+from flask import Blueprint, request, jsonify
+from app.models import ApiKey
+from app.utils.octoprint_client import OctoPrintClient
+from app.models import SystemConfig
+
+api_bp = Blueprint('api_handle', __name__, url_prefix='/api/v1')
+
+def get_octo_client():
+    url = SystemConfig.query.filter_by(key='octoprint_url').first()
+    apikey = SystemConfig.query.filter_by(key='octoprint_apikey').first()
+    if url and url.value and apikey and apikey.value:
+        return OctoPrintClient(url.value, apikey.value)
+    return None
+
+def require_api_key(f):
+    @functools.wraps(f)
+    def decorated(*args, **kwargs):
+        api_key_header = request.headers.get('X-Api-Key')
+        if not api_key_header:
+            return jsonify({'error': 'Missing API Key in headers (X-Api-Key)'}), 401
+        
+        key_record = ApiKey.query.filter_by(key=api_key_header).first()
+        if not key_record:
+            return jsonify({'error': 'Invalid API Key'}), 401
+            
+        return f(*args, **kwargs)
+    return decorated
+
+@api_bp.route('/status', methods=['GET'])
+@require_api_key
+def get_status():
+    client = get_octo_client()
+    if not client:
+        return jsonify({'error': 'Printer not configured'}), 503
+    try:
+        status_data = client.get_printer_status()
+        job_data = client.get_job_info()
+        return jsonify({'status': status_data, 'job': job_data})
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+@api_bp.route('/octoprint_client', methods=['POST'])
+@require_api_key
+def invoke_octoprint_client():
+    """
+    Expects JSON payload like:
+    {
+      "method": "pause_print",
+      "kwargs": {"action": "pause"}
+    }
+    """
+    client = get_octo_client()
+    if not client:
+        return jsonify({'error': 'Printer not configured'}), 503
+    
+    data = request.get_json()
+    if not data or 'method' not in data:
+        return jsonify({'error': 'Missing method in JSON payload'}), 400
+        
+    method_name = data['method']
+    kwargs = data.get('kwargs', {})
+    args = data.get('args', [])
+    
+    if not hasattr(client, method_name):
+        return jsonify({'error': f'Method {method_name} not found on OctoPrintClient'}), 400
+        
+    func = getattr(client, method_name)
+    if not callable(func) or method_name.startswith('_'):
+        return jsonify({'error': f'Method {method_name} is not allowed'}), 403
+        
+    try:
+        result = func(*args, **kwargs)
+        return jsonify({'success': True, 'result': result})
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+

app/utils/gcode_parser.py

@@ -0,0 +1,32 @@
+import os
+
+def get_gcode_metadata(filepath):
+    metadata = {
+        'print_time': '-',
+        'first_layer_time': '-',
+        'filament_used': '-'
+    }
+    if not os.path.exists(filepath):
+        return metadata
+        
+    try:
+        # Read the last few KB to find estimated time and filament used
+        with open(filepath, 'rb') as f:
+            f.seek(0, 2)
+            file_size = f.tell()
+            chunk_size = min(65536, file_size)  # read last 64KB
+            f.seek(file_size - chunk_size)
+            chunk = f.read().decode('utf-8', errors='ignore')
+            
+            lines = chunk.splitlines()
+            for line in reversed(lines):
+                if line.startswith('; estimated printing time (normal mode) ='):
+                    metadata['print_time'] = line.split('=')[1].strip()
+                elif line.startswith('; estimated first layer printing time (normal mode) ='):
+                    metadata['first_layer_time'] = line.split('=')[1].strip()
+                elif line.startswith('; filament used [mm] ='):
+                    metadata['filament_used'] = line.split('=')[1].strip()
+    except Exception:
+        pass
+        
+    return metadata

app/utils/octoprint_client.py

@@ -141,6 +141,9 @@ class OctoPrintClient:
         """Convenience method to home the printer axes."""
         return self._request("POST", "/api/printer/printhead", json={"command": "home", "axes": axes})
 
+    def auto_leveling(self):
+        return self.send_gcode("G29")
+
     # -------------------------------------------------------------------------
     # Webcam / Video
     # -------------------------------------------------------------------------