feat(mcp): add list_print_jobs tool

bambu_run/mcp/jobs.py

@@ -0,0 +1,74 @@
+"""list_print_jobs, get_print_job, get_print_statistics -- the
+read_print_history capability tools."""
+from __future__ import annotations
+
+import statistics
+from datetime import timedelta
+
+from django.db.models import Avg, Count, ExpressionWrapper, F, FloatField, Q, Sum
+from django.utils import timezone
+
+from bambu_run.mcp.shapes import display_name, normalize_consumed, resolve_printer
+from bambu_run.models import BambuCloudTask, FilamentUsage, PrintJob
+
+
+def list_print_jobs(
+    status: str | None = None,
+    days: int = 30,
+    name_contains: str | None = None,
+    printer: str | int | None = None,
+    limit: int = 20,
+) -> dict:
+    """Print job history, most recent first.
+
+    status filters on FINISH/FAILED/CANCELLED, or "running" for the
+    in-flight job (final_status is null while printing). filament_grams
+    prefers Bambu Cloud's reported weight over the noisier AMS percentage
+    estimate -- check filament_grams_source ("cloud" vs None, meaning no
+    cloud data) before trusting precision. cost is always null until
+    filament purchase prices are recorded. limit has a hard ceiling of 100,
+    independent of days -- a busy printer's older jobs simply won't be
+    returned by one call; narrow days or increase limit deliberately.
+    """
+    limit = min(limit, 100)
+    qs = PrintJob.objects.select_related("cloud_task", "device")
+    if printer is not None:
+        qs = qs.filter(device__in=resolve_printer(printer))
+    since = timezone.now() - timedelta(days=days)
+    qs = qs.filter(start_time__gte=since)
+    if status == "running":
+        qs = qs.filter(final_status__isnull=True)
+    elif status:
+        qs = qs.filter(final_status=status)
+    if name_contains:
+        qs = qs.filter(project_name__icontains=name_contains)
+
+    jobs = list(qs.order_by("-start_time")[:limit])
+    return {"jobs": [_job_summary(j) for j in jobs]}
+
+
+def _job_summary(job):
+    cloud = job.cloud_task
+    est_min = int(cloud.cost_time_seconds / 60) if cloud and cloud.cost_time_seconds else None
+    grams = None
+    grams_source = None
+    if cloud and cloud.weight_grams is not None:
+        grams = float(cloud.weight_grams)
+        grams_source = "cloud"
+    return {
+        "id": job.id,
+        "display_name": display_name(
+            cloud.design_title if cloud else "", cloud.plate_title if cloud else "", job.project_name
+        ),
+        "status": job.final_status,
+        "printer": job.device.name,
+        "started_at": job.start_time.isoformat(),
+        "ended_at": job.end_time.isoformat() if job.end_time else None,
+        "duration_min": job.duration_minutes,
+        "estimated_duration_min": est_min,
+        "layers_total": job.total_layers,
+        "filament_grams": grams,
+        "filament_grams_source": grams_source,
+        "cost": None,
+        "has_cover_image": bool(cloud and cloud.cover_url),
+    }

tests/test_mcp_jobs.py

@@ -0,0 +1,71 @@
+import pytest
+from datetime import timedelta
+from django.utils import timezone
+
+from bambu_run.models import Printer, PrintJob, BambuCloudTask
+from bambu_run.mcp.jobs import list_print_jobs
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_orders_most_recent_first():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    older = PrintJob.objects.create(device=p, project_name="Old", start_time=now - timedelta(days=1), final_status="FINISH")
+    newer = PrintJob.objects.create(device=p, project_name="New", start_time=now, final_status="FINISH")
+
+    result = list_print_jobs()
+
+    assert [j["id"] for j in result["jobs"]] == [newer.id, older.id]
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_filters_by_status_running():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    running = PrintJob.objects.create(device=p, project_name="Live", start_time=timezone.now(), final_status=None)
+    PrintJob.objects.create(device=p, project_name="Done", start_time=timezone.now(), final_status="FINISH")
+
+    result = list_print_jobs(status="running")
+
+    assert [j["id"] for j in result["jobs"]] == [running.id]
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_limit_hard_ceiling_100():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    PrintJob.objects.bulk_create([
+        PrintJob(device=p, project_name=f"Job {i}", start_time=now - timedelta(hours=i), final_status="FINISH")
+        for i in range(150)
+    ])
+
+    result = list_print_jobs(limit=1000)
+
+    assert len(result["jobs"]) == 100
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_prefers_cloud_weight_and_tags_source():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    cloud = BambuCloudTask.objects.create(task_id=1, design_title="Cool Bracket", weight_grams=70.10)
+    PrintJob.objects.create(
+        device=p, project_name="raw_name", start_time=timezone.now(),
+        final_status="FINISH", cloud_task=cloud,
+    )
+
+    result = list_print_jobs()
+
+    job = result["jobs"][0]
+    assert job["display_name"] == "Cool Bracket"
+    assert job["filament_grams"] == 70.10
+    assert job["filament_grams_source"] == "cloud"
+    assert job["cost"] is None
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_display_name_falls_back_to_project_name():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrintJob.objects.create(device=p, project_name="raw_project", start_time=timezone.now(), final_status="FINISH")
+
+    result = list_print_jobs()
+
+    assert result["jobs"][0]["display_name"] == "raw_project"