f5b627256f
Full backend implementation with custom Odoo modules: - encoach_api: Core API, user management, JWT auth - encoach_exam: Exam generation (reading, writing, listening, speaking) - encoach_evaluate: AI-powered evaluation (writing, speaking) - encoach_training: Training tips and walkthrough - encoach_storage: File storage management - encoach_payment: Stripe, PayPal, Paymob integration - encoach_mail: Email notifications Made-with: Cursor
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
import json
|
|
import logging
|
|
|
|
import jwt as pyjwt
|
|
|
|
from odoo.http import request
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EncoachMixin:
|
|
"""Shared authentication and response helpers for all EnCoach API controllers."""
|
|
|
|
def _authenticate(self):
|
|
"""Decode JWT Bearer token and return the corresponding ``res.users`` record.
|
|
|
|
Returns ``None`` when the token is missing, expired or invalid.
|
|
"""
|
|
auth_header = request.httprequest.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Bearer "):
|
|
return None
|
|
token = auth_header[7:]
|
|
secret = (
|
|
request.env["ir.config_parameter"]
|
|
.sudo()
|
|
.get_param("encoach.jwt_secret")
|
|
)
|
|
if not secret:
|
|
_logger.error("System parameter 'encoach.jwt_secret' is not configured")
|
|
return None
|
|
try:
|
|
payload = pyjwt.decode(token, secret, algorithms=["HS256"])
|
|
except pyjwt.ExpiredSignatureError:
|
|
return None
|
|
except pyjwt.InvalidTokenError:
|
|
return None
|
|
user_id = payload.get("user_id")
|
|
if not user_id:
|
|
return None
|
|
user = request.env["res.users"].sudo().browse(int(user_id))
|
|
if not user.exists():
|
|
return None
|
|
return user
|
|
|
|
def _json_response(self, data, status=200):
|
|
return request.make_json_response(data, status=status)
|
|
|
|
def _error_response(self, message, status=400, code=None):
|
|
body = {"error": message}
|
|
if code:
|
|
body["code"] = code
|
|
return request.make_json_response(body, status=status)
|
|
|
|
def _get_json_body(self):
|
|
try:
|
|
return json.loads(request.httprequest.get_data(as_text=True))
|
|
except (ValueError, TypeError):
|
|
return {}
|
|
|
|
def _paginate_params(self, kwargs):
|
|
page = max(int(kwargs.get("page", 1)), 1)
|
|
limit = min(max(int(kwargs.get("limit", 20)), 1), 100)
|
|
offset = (page - 1) * limit
|
|
return offset, limit, page
|
|
|
|
def _serialize(self, record):
|
|
"""Delegate to the record's own to_encoach_dict() if available."""
|
|
if hasattr(record, "to_encoach_dict"):
|
|
return record.to_encoach_dict()
|
|
return {"id": record.id}
|
|
|
|
def _serialize_list(self, records):
|
|
return [self._serialize(r) for r in records]
|