Tracker/Code/Tools/Service/main.py

"""
Generic Excel API Service

A FastAPI service for reading Excel files (.xlsx).
All file paths are relative to /data/ mount.

Endpoints:
- GET /health - Health check
- GET /excel/sheets - List all sheet names
- GET /excel/schema - Get column headers
- GET /excel/read - Read all rows
- GET /excel/row/{row_num} - Get specific row (1-indexed)
"""

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from pathlib import Path
from typing import Optional
from openpyxl import load_workbook
import os

app = FastAPI(
    title="JingTian Tools Service",
    description="Generic Excel API for reading xlsx files",
    version="0.2.0",
)

DATA_ROOT = Path(os.getenv("DATA_ROOT", "/data"))


def get_workbook(file_path: str):
    """Load workbook from file path relative to DATA_ROOT."""
    full_path = DATA_ROOT / file_path
    if not full_path.exists():
        raise HTTPException(status_code=404, detail=f"File not found: {file_path}")
    if not full_path.suffix.lower() == ".xlsx":
        raise HTTPException(status_code=400, detail="Only .xlsx files supported")
    try:
        return load_workbook(full_path, read_only=True, data_only=True)
    except Exception as e:
        raise HTTPException(
            status_code=500, detail=f"Failed to open workbook: {str(e)}"
        )


def get_sheet(wb, sheet_name: str):
    """Get sheet by name."""
    if sheet_name not in wb.sheetnames:
        raise HTTPException(
            status_code=404,
            detail=f"Sheet '{sheet_name}' not found. Available: {wb.sheetnames}",
        )
    return wb[sheet_name]


def cell_to_str(cell_value) -> str:
    """Convert cell value to string, empty string for None."""
    if cell_value is None:
        return ""
    return str(cell_value)


@app.get("/health")
async def health():
    return {"status": "healthy", "service": "tools", "version": "0.2.0"}


@app.get("/excel/sheets")
async def list_sheets(
    file_path: str = Query(..., description="Path to xlsx file relative to /data/"),
):
    """List all sheet names in the workbook."""
    wb = get_workbook(file_path)
    sheets = wb.sheetnames
    wb.close()
    return {"file_path": file_path, "sheets": sheets, "count": len(sheets)}


@app.get("/excel/schema")
async def get_schema(
    file_path: str = Query(..., description="Path to xlsx file relative to /data/"),
    sheet_name: str = Query(..., description="Sheet name to read"),
    header_row: int = Query(1, description="Row number containing headers (1-indexed)"),
):
    """Get column headers from the specified sheet."""
    wb = get_workbook(file_path)
    ws = get_sheet(wb, sheet_name)

    headers = []
    for cell in ws[header_row]:
        headers.append(cell_to_str(cell.value))

    wb.close()
    return {
        "file_path": file_path,
        "sheet_name": sheet_name,
        "header_row": header_row,
        "columns": headers,
        "column_count": len(headers),
    }


@app.get("/excel/read")
async def read_all(
    file_path: str = Query(..., description="Path to xlsx file relative to /data/"),
    sheet_name: str = Query(..., description="Sheet name to read"),
    header_row: int = Query(1, description="Row number containing headers (1-indexed)"),
    start_row: Optional[int] = Query(
        None,
        description="Start reading from this row (1-indexed, defaults to header_row + 1)",
    ),
    limit: Optional[int] = Query(None, description="Max rows to return"),
):
    """Read all rows from the specified sheet."""
    wb = get_workbook(file_path)
    ws = get_sheet(wb, sheet_name)

    # Get headers
    headers = [cell_to_str(cell.value) for cell in ws[header_row]]

    # Determine data start row
    data_start = start_row if start_row else header_row + 1

    # Read rows
    rows = []
    row_count = 0
    for row_num, row in enumerate(ws.iter_rows(min_row=data_start), start=data_start):
        # Stop if limit reached
        if limit and row_count >= limit:
            break

        # Skip completely empty rows
        values = [cell_to_str(cell.value) for cell in row]
        if all(v == "" for v in values):
            continue

        # Build row dict with headers as keys
        row_data = {"_row_num": row_num}
        for i, header in enumerate(headers):
            if i < len(values):
                key = header if header else f"_col_{i + 1}"
                row_data[key] = values[i]

        rows.append(row_data)
        row_count += 1

    wb.close()
    return {
        "file_path": file_path,
        "sheet_name": sheet_name,
        "columns": headers,
        "rows": rows,
        "row_count": len(rows),
    }


@app.get("/excel/row/{row_num}")
async def get_row(
    row_num: int,
    file_path: str = Query(..., description="Path to xlsx file relative to /data/"),
    sheet_name: str = Query(..., description="Sheet name to read"),
    header_row: int = Query(1, description="Row number containing headers (1-indexed)"),
):
    """Get a specific row by row number (1-indexed)."""
    if row_num < 1:
        raise HTTPException(status_code=400, detail="Row number must be >= 1")

    wb = get_workbook(file_path)
    ws = get_sheet(wb, sheet_name)

    # Get headers
    headers = [cell_to_str(cell.value) for cell in ws[header_row]]

    # Get the specific row
    try:
        row = ws[row_num]
    except Exception:
        wb.close()
        raise HTTPException(status_code=404, detail=f"Row {row_num} not found")

    values = [cell_to_str(cell.value) for cell in row]

    # Build row dict
    row_data = {"_row_num": row_num}
    for i, header in enumerate(headers):
        if i < len(values):
            key = header if header else f"_col_{i + 1}"
            row_data[key] = values[i]

    wb.close()
    return {"file_path": file_path, "sheet_name": sheet_name, "row": row_data}


# Future endpoints (Phase 2 & 3)
# POST /excel/row - Add new row
# PUT /excel/row/{row_num} - Update row
# DELETE /excel/row/{row_num} - Delete row
# POST /excel/column - Add column
# POST /excel/format - Apply formatting
# POST /excel/create - Create new workbook