des trucs

Co-authored-by: sesn-odoo <sesn@odoo.com>
This commit is contained in:
Hubert Van De Walle 2025-07-22 11:47:33 +02:00
parent a49c5f012f
commit eebeed58a5
5 changed files with 42 additions and 34 deletions

View File

@ -34,9 +34,7 @@ def dispatch():
) )
leaves = client.web_search_read(Leave, domain) leaves = client.web_search_read(Leave, domain)
team_availability = dict() team_availability = {employee: (True, True) for employee in team}
for employee in team:
team_availability[employee] = (True, True)
for leave in leaves: for leave in leaves:
employee = re.search(r"\((.*?)\)", leave.employee_id).group(1).lower() employee = re.search(r"\((.*?)\)", leave.employee_id).group(1).lower()
@ -49,10 +47,14 @@ def dispatch():
available_employees = [ available_employees = [
employee employee
for employee, availability in team_availability.items() for employee, (morning, afternoon) in team_availability.items()
if availability[0] or availability[1] if morning or afternoon
] ]
if not available_employees:
print("[red]No available employees for dispatch.[/red]")
return
previous_dispatch = DispatchStorage.load_week_to_date() previous_dispatch = DispatchStorage.load_week_to_date()
previous_dispatch_ids = [ previous_dispatch_ids = [
task_id for tasks in previous_dispatch.values() for task_id in tasks task_id for tasks in previous_dispatch.values() for task_id in tasks

View File

@ -1,12 +1,12 @@
from typing import Dict, Self, Any, get_origin from typing import Self, Any, get_origin
from dataclasses import fields, Field from dataclasses import fields, Field
from datetime import datetime from datetime import datetime
class OdooModel: class OdooModel:
@classmethod @classmethod
def specification(cls) -> Dict[str, Any]: def specification(cls) -> dict[str, Any]:
def field_spec(f: Field) -> Dict[str, Any]: def field_spec(f: Field) -> dict[str, Any]:
field_type = f.type field_type = f.type
if f.name.endswith("_id") or f.name.endswith("_ids"): if f.name.endswith("_id") or f.name.endswith("_ids"):
return { return {
@ -29,7 +29,7 @@ class OdooModel:
return spec return spec
@classmethod @classmethod
def from_record(cls, record: Dict) -> Self: def from_record(cls, record: dict) -> Self:
init_args = {} init_args = {}
for f in fields(cls): for f in fields(cls):
if f.init and f.name in record: if f.init and f.name in record:

View File

@ -1,7 +1,7 @@
import os import os
import requests import requests
import random import random
from typing import Any, Dict, List, TypeVar, Type from typing import Any, TypeVar
from model import OdooModel from model import OdooModel
@ -22,10 +22,10 @@ class OdooClient:
def rpc( def rpc(
self, model: str, method: str, *args: Any, **kwargs: Any self, model: str, method: str, *args: Any, **kwargs: Any
) -> requests.Response: ) -> requests.Response:
body: Dict[str, Any] = { body: dict[str, Any] = {
"jsonrpc": "2.0", "jsonrpc": "2.0",
"method": method, "method": method,
"id": random.randint(1, 1000000), "id": random.randint(1, 1_000_000),
"params": { "params": {
"model": model, "model": model,
"method": method, "method": method,
@ -34,23 +34,27 @@ class OdooClient:
}, },
} }
return self.session.post( response = self.session.post(
f"{self.base_url}/web/dataset/call_kw", f"{self.base_url}/web/dataset/call_kw",
json=body, json=body,
) )
response.raise_for_status()
return response
def web_search_read( def web_search_read(
self, self,
model: Type[T], model: type[T],
domain: List[tuple], domain: list[tuple[Any, ...]],
*args: Any, *args: Any,
**kwargs: Any, **kwargs: Any,
) -> List[T]: ) -> list[T]:
args = [domain, model.specification(), *args] args = [domain, model.specification(), *args]
res = self.rpc(model._name, "web_search_read", *args, **kwargs) res = self.rpc(model._name, "web_search_read", *args, **kwargs)
res.raise_for_status() res.raise_for_status()
json = res.json() json = res.json()
if json.get("error"):
raise Exception(json.get("error")) if error := json.get("error"):
raise RuntimeError(f"Odoo RPC Error: {error}")
records = json.get("result", {}).get("records", []) records = json.get("result", {}).get("records", [])
return [model.from_record(record) for record in records] return [model.from_record(record) for record in records]

View File

@ -1,30 +1,33 @@
from typing import Dict, List, Optional from typing import Optional
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path
import json import json
class DispatchStorage: class DispatchStorage:
@classmethod @staticmethod
def path(cls, date: Optional[str] = None) -> str: def path(date: Optional[str] = None) -> Path:
if date is None: if date is None:
date = datetime.now().strftime("%Y-%m-%d") date = datetime.now().strftime("%Y-%m-%d")
return f"out/{date}_dispatch.json" return Path("out", f"{date}_dispatch.json")
@classmethod @staticmethod
def load(cls, date: Optional[str] = None) -> Dict[str, List[int]]: def load(date: Optional[str] = None) -> dict[str, list[int]]:
try: try:
with open(cls.path(date), "r") as f: with open(DispatchStorage.path(date), "r") as f:
return json.load(f) return json.load(f)
except FileNotFoundError: except FileNotFoundError:
return {} return {}
@classmethod @staticmethod
def save(cls, dispatch: Dict[str, List[int]], date: Optional[str] = None) -> None: def save(dispatch: dict[str, list[int]], date: Optional[str] = None) -> None:
with open(cls.path(date), "w") as f: path = DispatchStorage.path(date)
path.parent.mkdir(exist_ok=True)
with open(path, "w") as f:
json.dump(dispatch, f) json.dump(dispatch, f)
@classmethod @staticmethod
def load_week_to_date(cls) -> Dict[str, List[int]]: def load_week_to_date() -> dict[str, list[int]]:
""" """
Loads and combines dispatch data from the start of the current week up to today. Loads and combines dispatch data from the start of the current week up to today.
""" """
@ -32,10 +35,10 @@ class DispatchStorage:
days_since_monday = today.weekday() days_since_monday = today.weekday()
current_date = today - timedelta(days=days_since_monday) current_date = today - timedelta(days=days_since_monday)
combined_dispatch: Dict[str, List[int]] = {} combined_dispatch: dict[str, list[int]] = {}
while current_date.date() < today.date(): while current_date.date() < today.date():
date_str = current_date.strftime("%Y-%m-%d") date_str = current_date.strftime("%Y-%m-%d")
daily_dispatch = cls.load(date_str) daily_dispatch = DispatchStorage.load(date_str)
for key, values in daily_dispatch.items(): for key, values in daily_dispatch.items():
if key in combined_dispatch: if key in combined_dispatch:
combined_dispatch[key].extend(values) combined_dispatch[key].extend(values)

View File

@ -1,5 +1,4 @@
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import List
from model import OdooModel from model import OdooModel
@ -9,7 +8,7 @@ class Task(OdooModel):
id: int id: int
name: str name: str
tag_ids: List[str] tag_ids: list[str]
stage_id: str stage_id: str
priority: bool priority: bool
url: str = field(init=False) url: str = field(init=False)