Tokens: {'vorhanden' if has_tokens else 'nicht vorhanden'}
" '', mimetype="text/html", ) @app.get("/login") def start_login(): state = secrets.token_urlsafe(24) STATE_COOKIE_PATH.write_text(state) params = { "client_id": CLIENT_ID, "redirect_uri": REDIRECT_URI, "response_type": "code", "scope": SCOPES, "state": state, "prompt": "login", } return redirect(f"{AUTH_AUTHORIZE}?{urlencode(params)}", code=302) @app.get("/callback") def oauth_callback(): code = request.args.get("code") state = request.args.get("state") saved_state = STATE_COOKIE_PATH.read_text() if STATE_COOKIE_PATH.exists() else "" if not state or state != saved_state: return Response("Ungültiger state.", status=400) data = { "grant_type": "authorization_code", "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "code": code, "audience": FLEET_API_BASE, "redirect_uri": REDIRECT_URI, "scope": SCOPES, } r = requests.post(AUTH_TOKEN, data=data, timeout=REQUEST_TIMEOUT) r.raise_for_status() tokens = r.json() tokens["created_at"] = int(time.time()) save_tokens(tokens) return Response("Login erfolgreich. Tokens gespeichert. Du kannst dieses Fenster schließen.", mimetype="text/plain") def list_vehicles(access_token: str): url = f"{FLEET_API_BASE}/api/1/vehicles" r = requests.get(url, headers=auth_headers(access_token), timeout=REQUEST_TIMEOUT) r.raise_for_status() return r.json().get("response", []) def wake_up(access_token: str, vehicle_tag: str): url = f"{FLEET_API_BASE}/api/1/vehicles/{vehicle_tag}/wake_up" requests.post(url, headers=auth_headers(access_token), timeout=REQUEST_TIMEOUT) def fetch_soc(access_token: str, vehicle_tag: str): url = f"{FLEET_API_BASE}/api/1/vehicles/{vehicle_tag}/vehicle_data" params = {"endpoints": "charge_state"} r = requests.get(url, headers=auth_headers(access_token), params=params, timeout=REQUEST_TIMEOUT) if r.status_code == 408: # Fahrzeug schläft → aufwecken und erneut wake_up(access_token, vehicle_tag) r = requests.get(url, headers=auth_headers(access_token), params=params, timeout=REQUEST_TIMEOUT) r.raise_for_status() charge = r.json()["response"]["charge_state"] soc = charge.get("usable_battery_level", charge.get("battery_level")) return soc, charge @app.get("/soc") def soc_endpoint(): access_token = ensure_access_token() if not access_token: return jsonify({"error": "nicht eingeloggt", "hint": "Bitte /login aufrufen"}), 401 try: vehicles = list_vehicles(access_token) except requests.HTTPError as e: body = "" if e.response is not None: try: body = e.response.json() except Exception: body = e.response.text msg = { "error": "fleet_api_precondition_failed" if getattr(e.response, "status_code", None) == 412 else "fleet_api_error", "hint": [ "Prüfe Region/Endpoint (EU vs. NA).", "Prüfe in der Tesla-App unter 'Manage Third-Party Apps' die Freigabe → ggf. Zugriff entfernen und neu autorisieren.", "Verwende Third-Party-Token + Scope 'vehicle_device_data'." ], "details": body } status = getattr(e.response, "status_code", 500) or 500 return jsonify(msg), status if not vehicles: return jsonify({"error": "kein Fahrzeug gefunden"}), 404 idx = int(request.args.get("index", "0")) v = vehicles[idx] vehicle_tag = v["id"] soc, charge_full = fetch_soc(access_token, vehicle_tag) return jsonify({ "vehicle_tag": vehicle_tag, "display_name": v.get("display_name"), "soc_percent": soc, "charging_state": charge_full.get("charging_state"), "battery_range_km": charge_full.get("battery_range"), "timestamp": charge_full.get("timestamp"), }) def get_partner_token(): data = { "grant_type": "client_credentials", "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "audience": FLEET_API_BASE, # EU-Base als Audience "scope": PARTNER_SCOPE, } r = requests.post(AUTH_TOKEN, data=data, timeout=REQUEST_TIMEOUT) r.raise_for_status() return r.json()["access_token"] @app.post("/register") def register_partner_account(): token = get_partner_token() payload = {"domain": "vehicle.pipelet.com"} url = f"{FLEET_API_BASE}/api/1/partner_accounts" r = requests.post(url, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, json=payload, timeout=REQUEST_TIMEOUT) try: r.raise_for_status() return jsonify({"ok": True, "response": r.json()}), 200 except requests.HTTPError: return jsonify({"ok": False, "status": r.status_code, "body": r.text}), r.status_code def _wk_full_path(): return str(Path(PUBLIC_ROOT).joinpath(TESLA_WK_REL).resolve()) # --- NEUE ROUTE: exakt die geforderte URL bedienen --- @app.get("/.well-known/appspecific/com.tesla.3p.public-key.pem") def serve_tesla_public_key(): full = _wk_full_path() # Sicherheitsnetz: nur genau diese Datei ausliefern, wenn vorhanden if not os.path.isfile(full): raise NotFound("com.tesla.3p.public-key.pem not found on server") # passender MIME-Typ für PEM (Tesla akzeptiert auch text/plain) return send_file(full, mimetype="application/x-pem-file", as_attachment=False, download_name="com.tesla.3p.public-key.pem", max_age=3600) # --- OPTIONAL: healthcheck, zeigt ob Datei vorhanden ist --- @app.get("/.well-known/appspecific/health") def well_known_health(): full = _wk_full_path() return jsonify({ "exists": os.path.isfile(full), "path": full, "size": os.path.getsize(full) if os.path.isfile(full) else None }) @app.get("/detailed_charge_state") def detailed_charge_state(): access_token = ensure_access_token() if not access_token: return jsonify({"error": "nicht eingeloggt", "hint": "Bitte /login aufrufen"}), 401 vehicles = list_vehicles(access_token) if not vehicles: return jsonify({"error": "kein Fahrzeug gefunden"}), 404 idx = int(request.args.get("index", "0")) v = vehicles[idx] vehicle_tag = v["id"] # Charge-State abrufen url = f"{FLEET_API_BASE}/api/1/vehicles/{vehicle_tag}/vehicle_data" params = {"endpoints": "charge_state"} r = requests.get(url, headers=auth_headers(access_token), params=params, timeout=REQUEST_TIMEOUT) if r.status_code == 408: # Fahrzeug schläft wake_up(access_token, vehicle_tag) r = requests.get(url, headers=auth_headers(access_token), params=params, timeout=REQUEST_TIMEOUT) r.raise_for_status() charge = r.json()["response"]["charge_state"] return jsonify({ "vehicle_tag": vehicle_tag, "display_name": v.get("display_name"), "charge_state": charge # komplettes charge_state-Objekt zurückgeben }) if __name__ == "__main__": # Flask nur lokal binden; Apache macht TLS + Reverse Proxy app.run(host=APP_HOST, port=APP_PORT)