"""
Timesheet management tools.
Provides CRUD operations for timesheets and summary reports.
"""
from __future__ import annotations
from collections import defaultdict
from datetime import date, datetime, timedelta
from typing import TYPE_CHECKING
from ..constants import OdooModel
from ..decorators import handle_odoo_errors
from ..formatters import MarkdownBuilder, format_hours
from ..validators import validate_date, validate_date_range, validate_hours, today
from .base import extract_name, get_odoo_client, normalize_pagination
if TYPE_CHECKING:
from mcp.server.fastmcp import FastMCP
def register_tools(mcp: "FastMCP") -> None:
"""Register timesheet tools with the MCP server."""
@mcp.tool()
@handle_odoo_errors
def list_timesheets(
date_from: str | None = None,
date_to: str | None = None,
project_id: int | None = None,
limit: int = 100,
offset: int = 0,
all_users: bool = False,
) -> str:
"""
List existing timesheet entries.
Args:
date_from: Start date (format YYYY-MM-DD, optional)
date_to: End date (format YYYY-MM-DD, optional)
project_id: Filter by project ID (optional)
limit: Maximum number of entries (default: 100)
offset: Offset for pagination (default: 0)
all_users: If False, filter on current user (default: False)
Returns:
List of timesheet entries
"""
date_from, date_to = validate_date_range(date_from, date_to)
limit, offset = normalize_pagination(limit, offset)
client = get_odoo_client()
domain: list = []
if not all_users:
domain.append(("user_id", "=", client.uid))
if date_from:
domain.append(("date", ">=", date_from))
if date_to:
domain.append(("date", "<=", date_to))
if project_id:
domain.append(("project_id", "=", project_id))
timesheets = client.search_read(
OdooModel.ANALYTIC_LINE,
domain,
["date", "name", "unit_amount", "project_id", "task_id", "employee_id"],
limit=limit,
offset=offset,
order="date desc",
)
if not timesheets:
return "No timesheet entries found."
builder = MarkdownBuilder("Timesheet Entries")
rows = []
total_hours = 0.0
for ts in timesheets:
hours = ts.get("unit_amount", 0)
total_hours += hours
rows.append([
ts["id"],
ts["date"],
extract_name(ts.get("project_id"), "-"),
extract_name(ts.get("task_id"), "-"),
ts.get("name", "-")[:30],
format_hours(hours),
])
builder.add_table(
["ID", "Date", "Project", "Task", "Description", "Hours"],
rows,
alignments=["right", "center", "left", "left", "left", "right"],
)
builder.add_line()
builder.add_text(f"**Total: {format_hours(total_hours)}**")
builder.add_pagination(len(timesheets), limit, offset)
return builder.build()
@mcp.tool()
@handle_odoo_errors
def create_timesheet(
project_id: int,
hours: float,
description: str,
task_id: int | None = None,
date_entry: str | None = None,
) -> str:
"""
Create a new timesheet entry.
Args:
project_id: Project ID
hours: Number of hours worked
description: Description of work done
task_id: Task ID (optional)
date_entry: Entry date in YYYY-MM-DD format (default: today)
Returns:
Confirmation with created entry ID
"""
hours = validate_hours(hours)
entry_date = validate_date(date_entry, "date_entry") if date_entry else today()
client = get_odoo_client()
values = {
"project_id": project_id,
"unit_amount": hours,
"name": description,
"date": entry_date,
}
if task_id:
values["task_id"] = task_id
record_id = client.create(OdooModel.ANALYTIC_LINE, values)
return f"Timesheet entry created successfully (ID: {record_id})"
@mcp.tool()
@handle_odoo_errors
def update_timesheet(
timesheet_id: int,
hours: float | None = None,
description: str | None = None,
date_entry: str | None = None,
) -> str:
"""
Update an existing timesheet entry.
Args:
timesheet_id: ID of entry to update
hours: New number of hours (optional)
description: New description (optional)
date_entry: New date in YYYY-MM-DD format (optional)
Returns:
Update confirmation
"""
client = get_odoo_client()
values = {}
if hours is not None:
values["unit_amount"] = validate_hours(hours)
if description is not None:
values["name"] = description
if date_entry is not None:
values["date"] = validate_date(date_entry, "date_entry")
if not values:
return "No fields to update specified."
client.write(OdooModel.ANALYTIC_LINE, [timesheet_id], values)
return f"Timesheet entry {timesheet_id} updated successfully."
@mcp.tool()
@handle_odoo_errors
def delete_timesheet(timesheet_id: int) -> str:
"""
Delete a timesheet entry.
Args:
timesheet_id: ID of entry to delete
Returns:
Deletion confirmation
"""
client = get_odoo_client()
client.unlink(OdooModel.ANALYTIC_LINE, [timesheet_id])
return f"Timesheet entry {timesheet_id} deleted successfully."
@mcp.tool()
@handle_odoo_errors
def get_timesheet_summary_by_employee(
date_from: str,
date_to: str,
expected_hours_per_day: float = 8.0,
) -> str:
"""
Get a summary of hours logged by employee for a given period.
Args:
date_from: Start date (format YYYY-MM-DD)
date_to: End date (format YYYY-MM-DD)
expected_hours_per_day: Expected hours per working day (default: 8.0)
Returns:
Summary of hours by employee with comparison to expected hours
"""
date_from = validate_date(date_from, "date_from")
date_to = validate_date(date_to, "date_to")
client = get_odoo_client()
# Get all timesheets in period
timesheets = client.search_read(
OdooModel.ANALYTIC_LINE,
[
("date", ">=", date_from),
("date", "<=", date_to),
],
["employee_id", "unit_amount", "date"],
limit=500,
)
if not timesheets:
return f"No timesheet entries found between {date_from} and {date_to}."
# Calculate working days (excluding weekends)
start = datetime.strptime(date_from, "%Y-%m-%d")
end = datetime.strptime(date_to, "%Y-%m-%d")
working_days = 0
current = start
while current <= end:
if current.weekday() < 5: # Mon-Fri
working_days += 1
current += timedelta(days=1)
expected_total = working_days * expected_hours_per_day
# Aggregate by employee
employee_hours: dict[str, float] = defaultdict(float)
employee_names: dict[str, str] = {}
for ts in timesheets:
emp = ts.get("employee_id")
if emp and isinstance(emp, (list, tuple)) and len(emp) >= 2:
emp_id = str(emp[0])
emp_name = emp[1]
employee_hours[emp_id] += ts.get("unit_amount", 0)
employee_names[emp_id] = emp_name
builder = MarkdownBuilder("Timesheet Summary by Employee")
builder.add_text(f"*Period: {date_from} to {date_to}*")
builder.add_text(f"*Working days: {working_days} | Expected hours: {format_hours(expected_total)}*\n")
rows = []
for emp_id, hours in sorted(employee_hours.items(), key=lambda x: -x[1]):
name = employee_names.get(emp_id, f"Employee {emp_id}")
diff = hours - expected_total
diff_str = f"+{format_hours(diff)}" if diff >= 0 else format_hours(diff)
percent = (hours / expected_total * 100) if expected_total > 0 else 0
rows.append([
name,
format_hours(hours),
format_hours(expected_total),
diff_str,
f"{percent:.0f}%",
])
builder.add_table(
["Employee", "Logged", "Expected", "Difference", "Completion"],
rows,
alignments=["left", "right", "right", "right", "right"],
)
total_logged = sum(employee_hours.values())
builder.add_line()
builder.add_text(f"**Total logged: {format_hours(total_logged)}**")
return builder.build()
