wip

main.py

@@ -1,109 +0,0 @@
-import click
-import random
-import odoo_client
-import json
-from tasks import fetch_tasks
-from rich import print, console
-from datetime import datetime
-from collections import defaultdict
-from typing import Dict, List, Optional
-from config import vfyd_tags, team
-
-
-console = console.Console()
-
-
-def load_dispatch(date: Optional[str] = None) -> Dict[str, List[int]]:
-    if date is None:
-        date = datetime.now().strftime("%Y-%m-%d")
-    with open(f"out/{date}_dispatch.json", "r") as f:
-        return json.load(f)
-
-
-def save_dispatch(dispatch: Dict[str, List[int]], date: Optional[str] = None) -> None:
-    if date is None:
-        date = datetime.now().strftime("%Y-%m-%d")
-    with open(f"out/{date}_dispatch.json", "w") as f:
-        json.dump(dispatch, f)
-
-
-@click.group()
-def cli() -> None:
-    pass
-
-
-@cli.command()
-def dispatch():
-    """Randomly distribute unassigned tasks among team members."""
-
-    domain = [
-        ("stage_id", "=", 194),
-        ("user_ids", "=", False),
-        ("project_id", "=", 49),
-        ("tag_ids", "not in", vfyd_tags),
-    ]
-
-    client = odoo_client.OdooClient()
-    records = fetch_tasks(client, domain)
-
-    random.shuffle(records)
-    dispatch = defaultdict(list)
-    out = defaultdict(list)
-    for idx, task in enumerate(records):
-        dispatch[team[idx % len(team)]].append(task.id)
-        out[team[idx % len(team)]].append(task)
-
-    save_dispatch(dispatch)
-    for key, tasks in out.items():
-        print(f"**{key}:**")
-        for task in tasks:
-            print(task.url)
-
-
-@cli.command()
-def status():
-    """Show all unassigned tasks in the backlog."""
-    client = odoo_client.OdooClient()
-
-    domain = [
-        ("stage_id", "=", 194),
-        ("user_ids", "=", False),
-        ("project_id", "=", 49),
-        ("tag_ids", "not in", vfyd_tags),
-    ]
-
-    tasks = fetch_tasks(client, domain)
-    for task in tasks:
-        print(task)
-
-
-@cli.command()
-def check():
-    """Check which previously dispatched tasks are still not completed."""
-    dispatch = load_dispatch()
-
-    reverse_dispatch = {v: k for k, vs in dispatch.items() for v in vs}
-
-    client = odoo_client.OdooClient()
-
-    domain = [
-        ("id", "in", list(reverse_dispatch.keys())),
-        ("tag_ids", "not in", vfyd_tags),
-        ("stage_id", "=", 194),
-        ("project_id", "=", 49),
-    ]
-
-    tasks = fetch_tasks(client, domain)
-    not_done = defaultdict(list)
-    for task in tasks:
-        not_done[reverse_dispatch[task.id]].append(task)
-
-    for key, value in not_done.items():
-        if value:
-            console.rule(f"[red]{key}[/red]")
-            for task in value:
-                print(task)
-
-
-if __name__ == "__main__":
-    cli()

src/config.py

@@ -22,7 +22,7 @@ team = [
     "nle",
     "abz",
     "hael",
-    #"roen",
+    "roen",
     "sbel",
     "wasa",
     "huvw",

src/leave.py

@@ -0,0 +1,12 @@
+from dataclasses import dataclass
+from model import OdooModel
+from datetime import datetime
+
+
+@dataclass
+class Leave(OdooModel):
+    _name = "hr.leave.report.calendar"
+
+    start_datetime: datetime
+    stop_datetime: datetime
+    employee_id: str

src/main.py

@@ -0,0 +1,147 @@
+import click
+import random
+import odoo_client
+import re
+from rich import print, console
+from datetime import datetime
+from collections import defaultdict
+from config import vfyd_tags, team
+from tasks import Task
+from leave import Leave
+from storage import DispatchStorage
+
+console = console.Console()
+
+
+@click.group()
+def cli() -> None:
+    pass
+
+
+@cli.command()
+def dispatch():
+    """Randomly distribute unassigned tasks among team members."""
+
+    client = odoo_client.OdooClient()
+    today = datetime.now().strftime("%Y-%m-%d")
+    domain = [
+        ("start_datetime", "<=", today),
+        ("stop_datetime", ">=", today),
+    ]
+    domain.extend(["|"] * (len(team) - 1))
+    domain.extend(
+        [("employee_id.name", "=ilike", f"%({employee})") for employee in team]
+    )
+    leaves = client.web_search_read(Leave, domain)
+
+    team_availability = dict()
+    for employee in team:
+        team_availability[employee] = (True, True)
+
+    for leave in leaves:
+        employee = re.search(r"\((.*?)\)", leave.employee_id).group(1).lower()
+        is_morning = leave.start_datetime.hour < 11
+        is_afternoon = leave.stop_datetime.hour >= 14
+        if is_morning:
+            team_availability[employee] = (False, team_availability[employee][1])
+        if is_afternoon:
+            team_availability[employee] = (team_availability[employee][0], False)
+
+    available_employees = [
+        employee
+        for employee, availability in team_availability.items()
+        if availability[0] or availability[1]
+    ]
+
+    previous_dispatch = DispatchStorage.load_week_to_date()
+    previous_dispatch_ids = [
+        task_id for tasks in previous_dispatch.values() for task_id in tasks
+    ]
+
+    domain = [
+        ("stage_id", "=", 194),
+        ("user_ids", "=", False),
+        ("project_id", "=", 49),
+        ("tag_ids", "not in", vfyd_tags),
+        ("id", "not in", previous_dispatch_ids),
+    ]
+    records = client.web_search_read(Task, domain)
+
+    random.shuffle(records)
+    dispatch = defaultdict(list)
+    out = defaultdict(list)
+    for idx, task in enumerate(records):
+        dispatch[available_employees[idx % len(available_employees)]].append(task.id)
+        out[available_employees[idx % len(available_employees)]].append(task)
+
+    DispatchStorage.save(dispatch)
+
+    domain = [
+        ("stage_id", "=", 194),
+        ("user_ids", "=", False),
+        ("project_id", "=", 49),
+        ("tag_ids", "not in", vfyd_tags),
+        ("id", "in", previous_dispatch_ids),
+    ]
+    records = client.web_search_read(Task, domain)
+
+    merged = defaultdict(list)
+    for member, task_ids in previous_dispatch.items():
+        for task_id in task_ids:
+            task = next((t for t in records if t.id == task_id), None)
+            if task:
+                merged[member].append(task)
+
+    for member, tasks in out.items():
+        merged[member].extend(tasks)
+
+    for key, tasks in merged.items():
+        print(f"**{key}:**")
+        for task in tasks:
+            print(task.url)
+
+
+@cli.command()
+def status():
+    """Show all unassigned tasks in the backlog."""
+
+    client = odoo_client.OdooClient()
+    domain = [
+        ("stage_id", "=", 194),
+        ("user_ids", "=", False),
+        ("project_id", "=", 49),
+        ("tag_ids", "not in", vfyd_tags),
+    ]
+    tasks = client.web_search_read(Task, domain)
+    for task in tasks:
+        print(task)
+
+
+@cli.command()
+def check():
+    """Check which previously dispatched tasks are still not completed."""
+
+    dispatch = DispatchStorage.load()
+    reverse_dispatch = {v: k for k, vs in dispatch.items() for v in vs}
+    client = odoo_client.OdooClient()
+    domain = [
+        ("id", "in", list(reverse_dispatch.keys())),
+        ("tag_ids", "not in", vfyd_tags),
+        ("stage_id", "=", 194),
+        ("project_id", "=", 49),
+    ]
+    tasks = client.web_search_read(Task, domain)
+
+    not_done = defaultdict(list)
+    for task in tasks:
+        not_done[reverse_dispatch[task.id]].append(task)
+
+    for key, value in not_done.items():
+        if value:
+            console.rule(f"[red]{key}[/red]")
+            for task in value:
+                print(task)
+
+
+if __name__ == "__main__":
+    cli()

src/model.py

@@ -0,0 +1,55 @@
+from typing import Dict, Self, Any, get_origin
+from dataclasses import fields, Field
+from datetime import datetime
+
+
+class OdooModel:
+    @classmethod
+    def specification(cls) -> Dict[str, Any]:
+        def field_spec(f: Field) -> Dict[str, Any]:
+            field_type = f.type
+            if f.name.endswith("_id") or f.name.endswith("_ids"):
+                return {
+                    f.name: {
+                        "fields": {
+                            "display_name": {},
+                        },
+                    },
+                }
+            elif field_type in {str, int, bool, datetime}:
+                return {f.name: {}}
+            else:
+                raise NotImplementedError(f"Unsupported field type: {field_type}")
+
+        spec = dict()
+        for f in fields(cls):
+            if f.init:
+                spec.update(field_spec(f))
+
+        return spec
+
+    @classmethod
+    def from_record(cls, record: Dict) -> Self:
+        init_args = {}
+        for f in fields(cls):
+            if f.init and f.name in record:
+                value = record[f.name]
+
+                if f.type is datetime:
+                    value = datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
+
+                if f.name.endswith("_id"):
+                    if f.type is str:
+                        value = value["display_name"]
+                    else:
+                        raise NotImplementedError()
+
+                if f.name.endswith("_ids"):
+                    if get_origin(f.type) is list and f.type.__args__[0] is str:
+                        value = [v["display_name"] for v in value]
+                    else:
+                        raise NotImplementedError()
+
+                init_args[f.name] = value
+
+        return cls(**init_args)

src/odoo_client.py

@@ -1,7 +1,11 @@
 import os
 import requests
 import random
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TypeVar, Type
+from model import OdooModel
+
+
+T = TypeVar("T", bound=OdooModel)
 
 
 class OdooClient:
@@ -37,16 +41,16 @@ class OdooClient:
 
     def web_search_read(
         self,
-        model: str,
+        model: Type[T],
         domain: List[tuple],
-        spec: Dict[str, Any],
         *args: Any,
         **kwargs: Any,
-    ) -> List[Dict[str, Any]]:
-        args = [domain, spec, *args]
-        res = self.rpc(model, "web_search_read", *args, **kwargs)
+    ) -> List[T]:
+        args = [domain, model.specification(), *args]
+        res = self.rpc(model._name, "web_search_read", *args, **kwargs)
         res.raise_for_status()
         json = res.json()
         if json.get("error"):
             raise Exception(json.get("error"))
-        return json.get("result", {}).get("records", [])
+        records = json.get("result", {}).get("records", [])
+        return [model.from_record(record) for record in records]

src/storage.py

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

src/tasks.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass, field
+from typing import List
+from model import OdooModel
+
+
+@dataclass
+class Task(OdooModel):
+    _name = "project.task"
+
+    id: int
+    name: str
+    tag_ids: List[str]
+    stage_id: str
+    priority: bool
+    url: str = field(init=False)
+
+    def __post_init__(self) -> None:
+        self.url = f"https://www.odoo.com/odoo/49/tasks/{self.id}"

tasks.py

@@ -1,60 +0,0 @@
-from dataclasses import dataclass, field
-from typing import Dict, List
-from odoo_client import OdooClient
-from rich.table import Table
-
-@dataclass
-class Task:
-    id: int
-    name: str
-    tags: List[str]
-    stage: str
-    priority: bool
-    url: str = field(init=False)
-
-    def __post_init__(self) -> None:
-        self.url = f"https://www.odoo.com/odoo/49/tasks/{self.id}"
-
-    def __rich__(self):
-        table = Table(show_header=True, header_style="bold magenta", box=None)
-        table.add_column("Field", style="cyan", justify="right")
-        table.add_column("Value", style="bold green")
-        table.add_row("ID", str(self.id))
-        table.add_row("Name", self.name)
-        table.add_row("Tags", ", ".join(self.tags) if self.tags else "[dim]No tags[/dim]")
-        table.add_row("Stage", f"[yellow]{self.stage}[/yellow]")
-        table.add_row("Priority", "[bold red]High[/bold red]" if self.priority else "[dim]Low[/dim]")
-        table.add_row("URL", f"[blue underline]{self.url}[/blue underline]")
-        return table
-
-
-def map_record_to_task(record: Dict) -> Task:
-    return Task(
-        id=record["id"],
-        name=record["name"],
-        tags=[tag["display_name"] for tag in record["tag_ids"]],
-        stage=record["stage_id"]["display_name"],
-        priority=record["priority"] == "1",
-    )
-
-
-def fetch_tasks(client: OdooClient, domain: List[tuple[str, str, str]]) -> List[Task]:
-    records = client.web_search_read(
-        "project.task",
-        domain,
-        spec={
-            "name": {},
-            "priority": {},
-            "tag_ids": {
-                "fields": {
-                    "display_name": {},
-                },
-            },
-            "stage_id": {
-                "fields": {
-                    "display_name": {},
-                },
-            },
-        },
-    )
-    return [map_record_to_task(record) for record in records]