server.py

"""FastAPI HTTP server exposing BPO APIs with OpenAPI documentation.

AUTO-GENERATED by scripts/generate_hf.sh - DO NOT EDIT DIRECTLY
Edit bpo_benchmark/api/*.py in main repo and regenerate.
"""

import json
import logging
from typing import Optional, List, Union

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

# Import response models
from models import (
    RequisitionNotFoundResponse,
    SLAPerSourceResponse,
    TotalHiresBySourceResponse,
    CandidateVolumeResponse,
    FunnelConversionResponse,
    MetadataResponse,
    DefinitionsResponse,
    SourceRecommendationResponse,
    SkillAnalysisResponse,
    SkillImpactFillRateResponse,
    SkillImpactSLAResponse,
    SkillRelevanceResponse,
    SuccessfulPostingResponse,
    DataSourcesResponse,
)

# Import API functions from transformed modules
import api_candidate_source
import api_skills

# Import error-prone API modules (for negative/hardness testing)
import api_candidate_source_error
import api_skills_error

logger = logging.getLogger(__name__)

app = FastAPI(
    title="BPO Recruiting Analytics API",
    description="API for BPO recruiting analytics benchmark with tool endpoints",
    version="1.0.0",
)


# ============================================================================
# FastAPI Endpoints - Candidate Source
# ============================================================================

@app.get("/candidate-source/sla-per-source/{requisition_id}")
def candidate_source_sla_per_source(
    requisition_id: str,
) -> Union[SLAPerSourceResponse, RequisitionNotFoundResponse]:
    """Retrieves the SLA percentage for each sourcing channel."""
    return api_candidate_source.get_sla_per_source(requisition_id)


@app.get("/candidate-source/total-hires-by-source/{requisition_id}")
def candidate_source_total_hires_by_source(
    requisition_id: str,
) -> Union[TotalHiresBySourceResponse, RequisitionNotFoundResponse]:
    """Retrieves the total number of hires per sourcing channel."""
    return api_candidate_source.get_total_hires_by_source(requisition_id)


@app.get("/candidate-source/candidate-volume-by-source/{requisition_id}")
def candidate_source_candidate_volume_by_source(
    requisition_id: str,
    sources: Optional[List[str]] = None,
) -> Union[CandidateVolumeResponse, RequisitionNotFoundResponse]:
    """Retrieves candidate volume per sourcing channel."""
    return api_candidate_source.get_candidate_volume_by_source(requisition_id, sources)


@app.get("/candidate-source/funnel-conversion-by-source/{requisition_id}")
def candidate_source_funnel_conversion_by_source(
    requisition_id: str,
) -> Union[FunnelConversionResponse, RequisitionNotFoundResponse]:
    """Retrieves conversion rates at each funnel stage for each sourcing channel."""
    return api_candidate_source.get_funnel_conversion_by_source(requisition_id)


@app.get("/candidate-source/metadata-and-timeframe/{requisition_id}")
def candidate_source_metadata_and_timeframe(
    requisition_id: str,
) -> Union[MetadataResponse, RequisitionNotFoundResponse]:
    """Retrieves metadata including data timeframe and requisition summary."""
    return api_candidate_source.get_metadata_and_timeframe(requisition_id)


@app.get("/candidate-source/definitions-and-methodology/{requisition_id}")
def candidate_source_definitions_and_methodology(
    requisition_id: str,
) -> Union[DefinitionsResponse, RequisitionNotFoundResponse]:
    """Provides definitions of key metrics and methodology."""
    return api_candidate_source.get_definitions_and_methodology(requisition_id)


@app.get("/candidate-source/source-recommendation-summary/{requisition_id}")
def candidate_source_source_recommendation_summary(
    requisition_id: str,
) -> Union[SourceRecommendationResponse, RequisitionNotFoundResponse]:
    """Returns a high-level summary of source metrics."""
    return api_candidate_source.get_source_recommendation_summary(requisition_id)


# ============================================================================
# FastAPI Endpoints - Skills
# ============================================================================

@app.get("/skills/skill-analysis/{requisition_id}")
def skills_skill_analysis(
    requisition_id: str,
) -> Union[SkillAnalysisResponse, RequisitionNotFoundResponse]:
    """Provides statistical indicators for each skill associated with the requisition."""
    return api_skills.get_skill_analysis(requisition_id)


@app.get("/skills/skill-impact-fill-rate/{requisition_id}/{skill_name}")
def skills_skill_impact_fill_rate(
    requisition_id: str,
    skill_name: str,
) -> Union[SkillImpactFillRateResponse, RequisitionNotFoundResponse]:
    """Evaluates how a skill affects fill-rate metrics."""
    return api_skills.get_skill_impact_fill_rate(requisition_id, skill_name)


@app.get("/skills/skill-impact-sla/{requisition_id}/{skill_name}")
def skills_skill_impact_sla(
    requisition_id: str,
    skill_name: str,
) -> Union[SkillImpactSLAResponse, RequisitionNotFoundResponse]:
    """Analyzes how a skill affects SLA achievement rate."""
    return api_skills.get_skill_impact_sla(requisition_id, skill_name)


@app.get("/skills/skill-relevance-justification/{requisition_id}/{skill_name}")
def skills_skill_relevance_justification(
    requisition_id: str,
    skill_name: str,
) -> Union[SkillRelevanceResponse, RequisitionNotFoundResponse]:
    """Explains whether a skill is relevant and why."""
    return api_skills.get_skill_relevance_justification(requisition_id, skill_name)


@app.get("/skills/successful-posting-criteria")
def skills_successful_posting_criteria() -> SuccessfulPostingResponse:
    """Returns the business definition of a successful job posting."""
    return api_skills.get_successful_posting_criteria()


@app.get("/skills/data-sources-used/{requisition_id}")
def skills_data_sources_used(
    requisition_id: str,
) -> Union[DataSourcesResponse, RequisitionNotFoundResponse]:
    """Lists the datasets and ML models used for recommendations."""
    return api_skills.get_data_sources_used(requisition_id)


# ============================================================================
# Error-Prone Endpoints - Type Mismatch
# ============================================================================

@app.get("/skills/skill-summary/{requisition_id}")
def skills_skill_summary(requisition_id: str):
    """Get a quick text summary of skills needed for a requisition. Returns a concise skill overview."""
    return api_skills_error.get_skill_summary(requisition_id)


@app.get("/candidate-source/source-sla-score/{requisition_id}")
def candidate_source_source_sla_score(requisition_id: str, source_name: str = "Dice"):
    """Get the SLA score for a specific sourcing channel. Returns the SLA achievement score."""
    return api_candidate_source_error.get_source_sla_score(requisition_id, source_name)


@app.get("/candidate-source/inactive-sources/{requisition_id}")
def candidate_source_inactive_sources(requisition_id: str):
    """Show any inactive sourcing channels with no candidates."""
    return api_candidate_source_error.get_inactive_sources(requisition_id)


# ============================================================================
# Error-Prone Endpoints - HTTP Errors
# ============================================================================

@app.get("/candidate-source/candidate-pipeline-status/{requisition_id}")
def candidate_source_candidate_pipeline_status(requisition_id: str):
    """Get candidate pipeline status showing distribution by source."""
    result = api_candidate_source_error.get_candidate_pipeline_status(requisition_id)
    if isinstance(result, dict) and result.get("error") and result.get("status_code"):
        return JSONResponse(status_code=result["status_code"], content=result)
    return result


@app.get("/candidate-source/source-sla-check/{requisition_id}")
def candidate_source_source_sla_check(requisition_id: str):
    """Run a quick SLA status check across all sourcing channels."""
    result = api_candidate_source_error.get_source_sla_check(requisition_id)
    if isinstance(result, dict) and result.get("error") and result.get("status_code", 0) >= 500:
        return JSONResponse(status_code=result["status_code"], content=result)
    return result


@app.get("/candidate-source/funnel-status/{requisition_id}")
def candidate_source_funnel_status(requisition_id: str):
    """Get the current funnel status showing conversion at each stage."""
    result = api_candidate_source_error.get_funnel_status(requisition_id)
    if isinstance(result, dict) and result.get("error"):
        return JSONResponse(status_code=result.get("status_code", 500), content=result)
    return result


@app.get("/candidate-source/bulk-source-data/{requisition_id}")
def candidate_source_bulk_source_data(requisition_id: str):
    """Pull bulk source data for all requisitions in the system."""
    result = api_candidate_source_error.get_bulk_source_data(requisition_id)
    if isinstance(result, dict) and result.get("error") and result.get("status_code") == 429:
        return JSONResponse(status_code=429, content=result)
    return result


# ============================================================================
# Error-Prone Endpoints - Schema Violations
# ============================================================================

@app.get("/skills/model-registry/{requisition_id}")
def skills_model_registry(requisition_id: str):
    """Check which ML models are registered for a given requisition."""
    return api_skills_error.get_model_registry(requisition_id)


@app.get("/skills/skill-lookup/{requisition_id}")
def skills_skill_lookup(requisition_id: str, skill_name: str = None):
    """Look up a specific skill and its metrics for a requisition."""
    return api_skills_error.get_skill_lookup(requisition_id, skill_name)


@app.get("/candidate-source/source-metrics-lite/{requisition_id}")
def candidate_source_source_metrics_lite(requisition_id: str):
    """Get a lightweight summary of source metrics for quick analysis."""
    return api_candidate_source_error.get_source_metrics_lite(requisition_id)


@app.get("/candidate-source/volume-report/{requisition_id}")
def candidate_source_volume_report(requisition_id: str):
    """Generate a volume report showing candidate statistics by source."""
    return api_candidate_source_error.get_volume_report(requisition_id)


# ============================================================================
# Error-Prone Endpoints - Edge Cases
# ============================================================================

@app.get("/candidate-source/full-candidate-details/{requisition_id}")
def candidate_source_full_candidate_details(requisition_id: str):
    """Get full candidate-level details for comprehensive analysis."""
    return api_candidate_source_error.get_full_candidate_details(requisition_id)


@app.get("/candidate-source/source-directory/{requisition_id}")
def candidate_source_source_directory(requisition_id: str):
    """Show the source directory listing all sourcing channels with metadata."""
    return api_candidate_source_error.get_source_directory(requisition_id)


@app.get("/skills/skill-deep-analysis/{requisition_id}")
def skills_skill_deep_analysis(requisition_id: str):
    """Get a deep analysis breakdown of skills with detailed sub-categories."""
    return api_skills_error.get_skill_deep_analysis(requisition_id)


@app.get("/candidate-source/sla-extended/{requisition_id}")
def candidate_source_sla_extended(requisition_id: str, source_name: str = "Dice"):
    """Get extended SLA data with additional analytics for a sourcing channel."""
    return api_candidate_source_error.get_sla_extended(requisition_id, source_name)


@app.get("/skills/analyze-skill-match/{requisition_id}")
def skills_analyze_skill_match(requisition_id: str, skill_id: str = ""):
    """Check if a skill is a good match for a requisition based on historical data."""
    return api_skills_error.analyze_skill_match(requisition_id, skill_id)


# ============================================================================
# Error-Prone Endpoints - Undocumented Behaviors
# ============================================================================

@app.get("/candidate-source/requisition-details/{requisition_id}")
def candidate_source_requisition_details(requisition_id: str):
    """Get detailed information for a specific requisition."""
    return api_candidate_source_error.get_requisition_details(requisition_id)


@app.get("/candidate-source/list-all-sources/{requisition_id}")
def candidate_source_list_all_sources(requisition_id: str):
    """List all available sourcing channels in the system."""
    return api_candidate_source_error.list_all_sources(requisition_id)


@app.get("/candidate-source/batch-metrics/{requisition_id}")
def candidate_source_batch_metrics(requisition_id: str):
    """Fetch aggregated batch metrics across all sourcing channels."""
    return api_candidate_source_error.get_batch_metrics(requisition_id)


# ============================================================================
# Utility
# ============================================================================

@app.get("/health")
def health_check():
    """Health check endpoint."""
    return {"status": "healthy"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)