spoolman2/spoolman/api/v1/other.py

"""Filament related endpoints."""

import logging
from datetime import datetime, timedelta
from typing import Annotated, Optional

import httpx
import sqlalchemy
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, Field, RootModel
from sqlalchemy import func
from sqlalchemy.ext.asyncio import AsyncSession

from spoolman.database import filament, models, spool
from spoolman.database.database import get_db_session

logger = logging.getLogger(__name__)

router = APIRouter(
    prefix="",
    tags=["other"],
)

# ruff: noqa: D103, B008


@router.get(
    "/material",
    name="Find materials",
    description="Get a list of all filament materials.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all filament materials.",
            "content": {
                "application/json": {
                    "example": [
                        "PLA",
                        "ABS",
                        "PETG",
                    ],
                },
            },
        },
    },
)
async def find_materials(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await filament.find_materials(db=db)


@router.get(
    "/color",
    name="Find colors",
    description="Get a list of all filament colors (hex codes).",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all filament color hex codes.",
            "content": {
                "application/json": {
                    "example": [
                        "FF0000",
                        "00FF00",
                        "0000FF",
                    ],
                },
            },
        },
    },
)
async def find_colors(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await filament.find_colors(db=db)


@router.get(
    "/article-number",
    name="Find article numbers",
    description="Get a list of all article numbers.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all article numbers.",
            "content": {
                "application/json": {
                    "example": [
                        "123456",
                        "987654",
                    ],
                },
            },
        },
    },
)
async def find_article_numbers(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await filament.find_article_numbers(db=db)


@router.get(
    "/lot-number",
    name="Find lot numbers",
    description="Get a list of all lot numbers.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all lot numbers.",
            "content": {
                "application/json": {
                    "example": [
                        "123456",
                        "987654",
                    ],
                },
            },
        },
    },
)
async def find_lot_numbers(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await spool.find_lot_numbers(db=db)


@router.get(
    "/location",
    name="Find locations",
    description="Get a list of all spool locations.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all spool locations.",
            "content": {
                "application/json": {
                    "example": [
                        "Printer 1",
                        "Printer 2",
                        "Storage Shelf A",
                    ],
                },
            },
        },
    },
)
async def find_locations(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await spool.find_locations(db=db)


class RenameLocationBody(BaseModel):
    name: str = Field(description="The new name of the location.", min_length=1)


@router.patch(
    "/location/{location}",
    name="Rename location",
    description="Rename a spool location. All spools in this location will be moved to the new location.",
    response_model_exclude_none=True,
    response_model=RootModel[str],
)
async def rename_location(
    location: str,
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
    body: RenameLocationBody,
) -> str:
    logger.info("Renaming location %s to %s", location, body.name)
    await spool.rename_location(db=db, current_name=location, new_name=body.name)
    return body.name


class UsageDataPoint(BaseModel):
    """A single data point for usage analytics."""

    date: str = Field(description="Date in YYYY-MM-DD format.")
    weight_used: float = Field(description="Total weight used in grams (positive value).")


@router.get(
    "/analytics/usage",
    name="Get usage analytics",
    description="Get filament usage data aggregated by day for the specified period.",
    response_model=list[UsageDataPoint],
)
async def get_usage_analytics(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
    days: Annotated[
        int,
        Query(description="Number of days to look back.", ge=1, le=365),
    ] = 30,
) -> list[UsageDataPoint]:
    """Get usage analytics aggregated by day."""
    # Calculate the start date
    start_date = datetime.utcnow() - timedelta(days=days)

    # Query adjustments grouped by date
    # Only include negative values (filament used, not added)
    stmt = (
        sqlalchemy.select(
            func.date(models.SpoolAdjustment.timestamp).label("date"),
            func.sum(
                sqlalchemy.case(
                    (models.SpoolAdjustment.value < 0, -models.SpoolAdjustment.value),
                    else_=0,
                )
            ).label("weight_used"),
        )
        .where(models.SpoolAdjustment.timestamp >= start_date)
        .where(models.SpoolAdjustment.adjustment_type == "weight")
        .group_by(func.date(models.SpoolAdjustment.timestamp))
        .order_by(func.date(models.SpoolAdjustment.timestamp))
    )

    result = await db.execute(stmt)
    rows = result.all()

    # Fill in missing dates with 0
    data_by_date = {str(row.date): row.weight_used for row in rows}
    all_dates = []
    current_date = start_date.date()
    end_date = datetime.utcnow().date()

    while current_date <= end_date:
        date_str = current_date.strftime("%Y-%m-%d")
        all_dates.append(
            UsageDataPoint(
                date=date_str,
                weight_used=data_by_date.get(date_str, 0),
            )
        )
        current_date += timedelta(days=1)

    return all_dates


class MaterialUsage(BaseModel):
    """Usage breakdown by material."""

    material: str = Field(description="Material name.")
    weight_used: float = Field(description="Total weight used in grams.")


@router.get(
    "/analytics/by-material",
    name="Get usage by material",
    description="Get filament usage breakdown by material type.",
    response_model=list[MaterialUsage],
)
async def get_usage_by_material(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
    days: Annotated[
        Optional[int],
        Query(description="Number of days to look back. If not set, returns all-time usage.", ge=1, le=365),
    ] = None,
) -> list[MaterialUsage]:
    """Get usage breakdown by material."""
    # Join adjustments with spools and filaments to get material
    stmt = (
        sqlalchemy.select(
            func.coalesce(models.Filament.material, "Unknown").label("material"),
            func.sum(
                sqlalchemy.case(
                    (models.SpoolAdjustment.value < 0, -models.SpoolAdjustment.value),
                    else_=0,
                )
            ).label("weight_used"),
        )
        .join(models.Spool, models.SpoolAdjustment.spool_id == models.Spool.id)
        .join(models.Filament, models.Spool.filament_id == models.Filament.id)
        .where(models.SpoolAdjustment.adjustment_type == "weight")
    )

    if days is not None:
        start_date = datetime.utcnow() - timedelta(days=days)
        stmt = stmt.where(models.SpoolAdjustment.timestamp >= start_date)

    stmt = stmt.group_by(models.Filament.material).order_by(func.sum(-models.SpoolAdjustment.value).desc())

    result = await db.execute(stmt)
    rows = result.all()

    return [MaterialUsage(material=row.material or "Unknown", weight_used=row.weight_used) for row in rows]


class CostByMaterial(BaseModel):
    """Cost breakdown by material."""

    material: str = Field(description="Material name.")
    total_cost: float = Field(description="Total cost spent on this material.")
    total_weight: float = Field(description="Total initial weight purchased in grams.")


class CostStats(BaseModel):
    """Overall cost statistics."""

    total_spent: float = Field(description="Sum of all spool purchase prices.")
    total_weight_purchased: float = Field(description="Total initial weight of all spools in grams.")
    avg_cost_per_kg: Optional[float] = Field(description="Average cost per kilogram of filament.")
    cost_by_material: list[CostByMaterial] = Field(description="Cost breakdown by material type.")


@router.get(
    "/analytics/cost",
    name="Get cost statistics",
    description="Get overall cost tracking statistics across all spools.",
    response_model=CostStats,
)
async def get_cost_stats(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> CostStats:
    """Get cost statistics including total spent and cost per kg."""
    # Use COALESCE(spool.price, filament.price) for effective price
    effective_price = func.coalesce(models.Spool.price, models.Filament.price)
    effective_weight = func.coalesce(models.Spool.initial_weight, models.Filament.weight)

    # Total spent and total weight across all spools (including archived)
    totals_stmt = (
        sqlalchemy.select(
            func.sum(effective_price).label("total_spent"),
            func.sum(effective_weight).label("total_weight"),
        )
        .join(models.Filament, models.Spool.filament_id == models.Filament.id)
        .where(effective_price.isnot(None))
    )
    totals_result = await db.execute(totals_stmt)
    totals_row = totals_result.one()

    total_spent = totals_row.total_spent or 0.0
    total_weight_purchased = totals_row.total_weight or 0.0
    avg_cost_per_kg = (total_spent / total_weight_purchased * 1000) if total_weight_purchased > 0 else None

    # Cost by material
    material_stmt = (
        sqlalchemy.select(
            func.coalesce(models.Filament.material, "Unknown").label("material"),
            func.sum(effective_price).label("total_cost"),
            func.sum(effective_weight).label("total_weight"),
        )
        .join(models.Filament, models.Spool.filament_id == models.Filament.id)
        .where(effective_price.isnot(None))
        .group_by(models.Filament.material)
        .order_by(func.sum(effective_price).desc())
    )
    material_result = await db.execute(material_stmt)
    material_rows = material_result.all()

    cost_by_material = [
        CostByMaterial(
            material=row.material or "Unknown",
            total_cost=row.total_cost or 0.0,
            total_weight=row.total_weight or 0.0,
        )
        for row in material_rows
    ]

    return CostStats(
        total_spent=total_spent,
        total_weight_purchased=total_weight_purchased,
        avg_cost_per_kg=avg_cost_per_kg,
        cost_by_material=cost_by_material,
    )


# --- filamentcolors.xyz integration ---

FILAMENTCOLORS_API = "https://filamentcolors.xyz/api"


class FilamentColorSwatch(BaseModel):
    """A color swatch from filamentcolors.xyz."""

    id: int
    color_name: str
    hex_color: Optional[str] = None
    manufacturer_name: str
    filament_type: str
    image_front: Optional[str] = None


class FilamentColorManufacturer(BaseModel):
    """A manufacturer from filamentcolors.xyz."""

    id: int
    name: str


class FilamentColorSearchResult(BaseModel):
    """Search results from filamentcolors.xyz."""

    count: int
    swatches: list[FilamentColorSwatch]


@router.get(
    "/filamentcolors/manufacturers",
    name="Get filamentcolors.xyz manufacturers",
    description="List all manufacturers from filamentcolors.xyz.",
    response_model=list[FilamentColorManufacturer],
)
async def get_filamentcolors_manufacturers() -> list[FilamentColorManufacturer]:
    """Proxy manufacturer list from filamentcolors.xyz."""
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(f"{FILAMENTCOLORS_API}/manufacturer/", params={"format": "json", "limit": 500})
        resp.raise_for_status()
        data = resp.json()

    return [
        FilamentColorManufacturer(id=m["id"], name=m["name"])
        for m in data.get("results", [])
    ]


@router.get(
    "/filamentcolors/search",
    name="Search filamentcolors.xyz swatches",
    description="Search color swatches from filamentcolors.xyz by manufacturer and/or color name.",
    response_model=FilamentColorSearchResult,
)
async def search_filamentcolors(
    *,
    manufacturer: Annotated[
        Optional[str],
        Query(description="Manufacturer name to filter by."),
    ] = None,
    color: Annotated[
        Optional[str],
        Query(description="Color name to search for."),
    ] = None,
    filament_type: Annotated[
        Optional[str],
        Query(description="Filament type to filter by (e.g., PLA, PETG)."),
    ] = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 50,
    page: Annotated[int, Query(ge=1)] = 1,
) -> FilamentColorSearchResult:
    """Proxy swatch search from filamentcolors.xyz."""
    params: dict = {"format": "json", "limit": limit, "page": page}
    if manufacturer:
        params["manufacturer__name"] = manufacturer
    if color:
        params["color_name__icontains"] = color
    if filament_type:
        params["filament_type__name__icontains"] = filament_type

    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(f"{FILAMENTCOLORS_API}/swatch/", params=params)
        resp.raise_for_status()
        data = resp.json()

    swatches = [
        FilamentColorSwatch(
            id=s["id"],
            color_name=s.get("color_name", ""),
            hex_color=s.get("hex_color"),
            manufacturer_name=s.get("manufacturer", {}).get("name", ""),
            filament_type=s.get("filament_type", {}).get("name", ""),
            image_front=s.get("image_front"),
        )
        for s in data.get("results", [])
    ]

    return FilamentColorSearchResult(count=data.get("count", 0), swatches=swatches)