import requests import json import logging import calendar from datetime import datetime, timedelta, timezone from fastapi import FastAPI, Depends, HTTPException, status logger = logging.getLogger("uvicorn") class OdooAPIClient: def __init__(self, odoo_url: str, db_name: str, secret_key: str, algorithm: str): self.odoo_url = odoo_url self.db_name = db_name self.session = requests.Session() def login(self, username: str, password: str) -> dict: auth_url = f"{self.odoo_url}/web/session/authenticate" payload = { "jsonrpc": "2.0", "method": "call", "params": { "db": self.db_name, "login": username, "password": password }, "id": 1 } headers = {'Content-Type': 'application/json'} response = self.session.post(auth_url, json=payload, headers=headers) if response.status_code != 200: # logger.error( # f"Authentication failed with status code {response.status_code}") raise Exception( f"Authentication failed with status code {response.status_code}") # for cookie in response.cookies: # self._session_cookies[cookie.name] = cookie.value data = response.json() if "error" in data: raise HTTPException( status_code=502, detail=f"Odoo RPC error: {data['error'].get('message', 'Unknown error')}" ) return data["result"] def get_hr_attendance_data(self, month=None, year=None, limit=None, domain=None): if month is None: month = datetime.now(timezone.utc).month if year is None: year = datetime.now(timezone.utc).year if domain is None: domain = [] # Default to no domain filter url = f"{self.odoo_url}/hr.attendance" first_day_of_month = datetime(year, month, 1) last_day_of_month = datetime( year, month, calendar.monthrange(year, month)[1], 23, 59, 59) start_date_str = first_day_of_month.strftime('%Y-%m-%d %H:%M:%S') end_date_str = last_day_of_month.strftime('%Y-%m-%d %H:%M:%S') domain.extend([ ('check_in', '>=', start_date_str), ('check_in', '<=', end_date_str) ]) kwargs = { "domain": domain, "fields": ["id", "employee_id", "check_in", "check_out", "worked_hours", "attendance_reason_ids"], } if limit is not None: kwargs["limit"] = limit result = self.call_odoo_method( "hr.attendance", "web_search_read", kwargs=kwargs) if "records" not in result: # logger.error("Unexpected response format: 'records' not found") raise HTTPException( status_code=502, detail="Unexpected response format: 'records' not found" ) sorted_data = sorted(result["records"], key=lambda x: x["check_in"]) return sorted_data def call_odoo_method(self, model: str, method: str, args: list = None, kwargs: dict = None): if args is None: args = [] if kwargs is None: kwargs = {} url = f"{self.odoo_url}/web/dataset/call_kw/{model}/{method}" headers = {'Content-Type': 'application/json'} payload = { "jsonrpc": "2.0", "method": "call", "params": { "model": model, "method": method, "args": args, "kwargs": kwargs }, "id": 1 } response = self.session.post(url, json=payload, headers=headers) # response = requests.post(url, json=payload, headers=headers, cookies=self._session_cookies) if response.status_code != 200: # logger.error( # f"Failed to call Odoo method {method} on model {model}: {response.status_code}") raise HTTPException( status_code=response.status_code, detail=f"Failed to call Odoo method {method} on model {model}" ) data = response.json() if "error" in data: raise HTTPException( status_code=502, detail=f"Odoo RPC error: {data['error'].get('message', 'Unknown error')}" ) return data["result"]