main.py

import shutil
import tempfile
import os
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import JSONResponse

import birdnet_analyzer
from birdnet_analyzer import analyze

app = FastAPI(title="SS BirdNet API")


@app.get("/")
def health():
    return {
        "status": "ok",
        "service": "birdnet-analyzer",
        "version": birdnet_analyzer.__version__,
    }


# Keys we expect on a real detection row — used to filter out the
# metadata block BirdNET appends to its CSV output ("File splitting
# duration", "Segment length", etc., which DictReader otherwise picks
# up as a phantom row).
DETECTION_KEYS = {"Start (s)", "End (s)", "Scientific name", "Common name", "Confidence"}


@app.post("/analyze")
async def analyze_audio(
    audio: UploadFile = File(...),
    min_conf: float = Form(0.25),
    lat: Optional[float] = Form(None),
    lon: Optional[float] = Form(None),
    week: Optional[int] = Form(None),
    # None lets min_conf gate the results. Pass an int to force "top N per
    # segment regardless of confidence" — useful for debugging silent files,
    # noisy for normal use.
    top_n: Optional[int] = Form(None),
):
    suffix = Path(audio.filename or "audio.wav").suffix or ".wav"
    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp_in:
        tmp_in.write(await audio.read())
        in_path = tmp_in.name

    out_dir = tempfile.mkdtemp()
    try:
        analyze(
            audio_input=in_path,
            output=out_dir,
            min_conf=min_conf,
            lat=lat,
            lon=lon,
            week=week,
            top_n=top_n,
            rtype="csv",
        )

        import csv as csv_module
        results = []
        for csv_file in Path(out_dir).glob("*.csv"):
            with open(csv_file, newline="") as f:
                for row in csv_module.DictReader(f):
                    # Skip the trailing metadata block — its keys don't
                    # overlap with detection keys.
                    if not DETECTION_KEYS.issubset(row.keys()):
                        continue
                    # Defence-in-depth: drop rows below threshold even
                    # when top_n is set, so the response is consistent.
                    try:
                        if float(row.get("Confidence", 0)) < min_conf:
                            continue
                    except (TypeError, ValueError):
                        continue
                    results.append(dict(row))

        return JSONResponse({"results": results})

    finally:
        try:
            os.unlink(in_path)
        except OSError:
            pass
        shutil.rmtree(out_dir, ignore_errors=True)