app.py

import os
import pandas as pd
import json
from datetime import datetime
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
import requests

app = FastAPI(title="HTS to HSN Classifier", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class ClassificationRequest(BaseModel):
    hts_code_or_desc: str

class ClassificationResponse(BaseModel):
    HSN_Code: str | None
    HSN_Description: str | None
    Confidence: str
    Reasoning: str

class HuggingFaceInferenceClient:
    def __init__(self, model_name: str = "meta-llama/Meta-Llama-3-8B-Instruct", api_token: str = None):
        self.model_name = model_name
        self.api_token = api_token or ""
        if not self.api_token:
            raise ValueError("Hugging Face API token not provided")
        self.headers = {
            "Authorization": f"Bearer {self.api_token}",
            "Content-Type": "application/json"
        }
        # Fixed API URL - use the correct inference endpoint
        self.api_url = f"https://api-inference.huggingface.co/models/{self.model_name}"

    def invoke(self, prompt: str) -> str:
        # Fixed payload structure for Hugging Face Inference API
        payload = {
            "inputs": prompt,
            "parameters": {
                "max_new_tokens": 500,
                "temperature": 0.6,
                "top_p": 0.95,
                "return_full_text": False
            }
        }
        
        try:
            response = requests.post(self.api_url, json=payload, headers=self.headers, timeout=60)
            response.raise_for_status()
            data = response.json()
            
            # Handle different response formats
            if isinstance(data, list) and len(data) > 0:
                if "generated_text" in data[0]:
                    return data[0]["generated_text"]
                elif "text" in data[0]:
                    return data[0]["text"]
            elif isinstance(data, dict):
                if "generated_text" in data:
                    return data["generated_text"]
                elif "text" in data:
                    return data["text"]
            
            return str(data)
            
        except requests.exceptions.RequestException as e:
            print(f"API request failed: {e}")
            raise Exception(f"Hugging Face API error: {e}")
        except json.JSONDecodeError as e:
            print(f"JSON decode error: {e}")
            raise Exception(f"Invalid JSON response from API: {e}")


@app.on_event("startup")
async def startup_event():
    try:
        global vs_hts, vs_hsn, df_hts, df_hsn, llm_client, hts_code_col, hts_desc_col, hsn_code_col, hsn_desc_col
        
        hts_path = "data/Htsdata.xlsx"
        hsn_path = "data/HSN_SAC.xlsx"
        cached_hts_vector_path = "data/faiss_hts_store"
        cached_hsn_vector_path = "data/faiss_hsn_store"

        print("Loading HSN data from:", hsn_path)
        
        hts_code_col = "HTS Number"
        hts_desc_col = "Description"
        hsn_code_col = "HSN_CD"
        hsn_desc_col = "HSN_Description"

        df_hts = pd.read_excel(hts_path)
        df_hts.columns = df_hts.columns.str.strip()
        df_hsn = pd.read_excel(hsn_path)
        df_hsn.columns = df_hsn.columns.str.strip()
        df_hsn[hsn_code_col] = df_hsn[hsn_code_col].astype(str)

        # Initialize with correct model name
        llm_client = HuggingFaceInferenceClient(model_name="meta-llama/Meta-Llama-3-8B-Instruct")
        
        print("✅ Application started successfully!")
        
    except Exception as e:
        print(f"❌ Startup error: {e}")
        raise e

@app.get("/health")
async def health_check():
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}

def extract_structure(code: str):
    code = "".join(filter(str.isdigit, str(code)))
    return {
        "chapter": code[:4] if len(code) >= 4 else None,
        "heading": code[:6] if len(code) >= 6 else None,
        "hsn8": code[:8] if len(code) >= 8 else code,
        "full": code
    }

def extract_json_from_text(text: str) -> dict:
    """Extract JSON from text response, handling various formats."""
    text = text.strip()
    
    # Find JSON content between braces
    start_idx = text.find('{')
    end_idx = text.rfind('}')
    
    if start_idx != -1 and end_idx != -1 and start_idx < end_idx:
        json_str = text[start_idx:end_idx + 1]
        try:
            return json.loads(json_str)
        except json.JSONDecodeError:
            pass
    
    # If JSON extraction fails, try to parse key-value pairs
    try:
        lines = text.split('\n')
        result = {}
        for line in lines:
            if ':' in line:
                key, value = line.split(':', 1)
                key = key.strip().strip('"\'')
                value = value.strip().strip('",\'')
                if key in ['HSN_Code', 'HSN_Description', 'Confidence', 'Reasoning']:
                    result[key] = value
        
        if len(result) >= 2:  # At least some keys found
            return result
    except:
        pass
    
    return None

def map_hts_to_hsn(hts_code_or_desc: str):
    reasoning_parts = []

    if hts_code_or_desc.isdigit():
        struct = extract_structure(hts_code_or_desc)
        reasoning_parts.append(f"Input HTS code: {struct['full']}")

        hts_match = df_hts[df_hts[hts_code_col].astype(str).str.startswith(struct["chapter"])]
        hts_desc_list = hts_match[hts_desc_col].head(3).tolist() if not hts_match.empty else []
        hts_desc_text = "; ".join(hts_desc_list) if hts_desc_list else "No HTS description found."
        reasoning_parts.append(f"HTS Chapter {struct['chapter']}: {hts_desc_text}")

        hsn_match = df_hsn[df_hsn[hsn_code_col] == struct["hsn8"]]
        if not hsn_match.empty:
            best_match = hsn_match.iloc[0]
            reasoning_parts.append(f"Exact 8-digit HSN {struct['hsn8']} found.")
            return {
                "HSN_Code": best_match[hsn_code_col],
                "HSN_Description": best_match[hsn_desc_col],
                "Confidence": "High",
                "Reasoning": " ".join(reasoning_parts)
            }

        fallback_heading_match = df_hsn[df_hsn[hsn_code_col].str.startswith(struct["heading"])]
        if not fallback_heading_match.empty:
            fallback_heading = fallback_heading_match.iloc[0]
            reasoning_parts.append(f"No exact 8-digit HSN. Fallback heading {struct['heading']} found.")

            # Improved prompt with better formatting
            system_prompt = "You are an expert in Indian HSN classification. Respond only with valid JSON containing the keys: HSN_Code, HSN_Description, Confidence, Reasoning."
            
            user_prompt = f"""
Input HTS code: {struct['full']}
HTS Description: {hts_desc_text}
Fallback HSN heading: {fallback_heading[hsn_code_col]} - {fallback_heading[hsn_desc_col]}

Based on this information, provide the most appropriate 8-digit HSN code and description.

Required JSON format:
{{
    "HSN_Code": "XXXXXXXX",
    "HSN_Description": "description here",
    "Confidence": "High/Medium/Low",
    "Reasoning": "explanation here"
}}
"""

            full_prompt = f"<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n{system_prompt}<|eot_id|><|start_header_id|>user<|end_header_id|>\n{user_prompt}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n"

            try:
                llm_response = llm_client.invoke(full_prompt).strip()
                print(f"LLM Response: {llm_response}")  # Debug logging
                
                parsed_response = extract_json_from_text(llm_response)
                
                if parsed_response and all(key in parsed_response for key in ["HSN_Code", "HSN_Description"]):
                    # Ensure all required keys are present with defaults
                    return {
                        "HSN_Code": parsed_response.get("HSN_Code"),
                        "HSN_Description": parsed_response.get("HSN_Description"),
                        "Confidence": parsed_response.get("Confidence", "Medium"),
                        "Reasoning": parsed_response.get("Reasoning", "LLM classification")
                    }
                else:
                    print(f"Invalid LLM response format: {parsed_response}")
                    
            except Exception as e:
                print(f"LLM failed: {e}")
                
            # Fallback if LLM fails
            return {
                "HSN_Code": fallback_heading[hsn_code_col],
                "HSN_Description": fallback_heading[hsn_desc_col],
                "Confidence": "Medium",
                "Reasoning": " ".join(reasoning_parts) + " LLM failed, using fallback 6-digit heading."
            }

        chapter_match = df_hsn[df_hsn[hsn_code_col].str.startswith(struct["chapter"][:4])]
        if not chapter_match.empty:
            best_match = chapter_match.iloc[0]
            reasoning_parts.append(f"No heading match. Fallback to chapter {struct['chapter'][:4]}.")
            return {
                "HSN_Code": best_match[hsn_code_col],
                "HSN_Description": best_match[hsn_desc_col],
                "Confidence": "Low",
                "Reasoning": " ".join(reasoning_parts)
            }

        return {
            "HSN_Code": None,
            "HSN_Description": None,
            "Confidence": "Low",
            "Reasoning": " ".join(reasoning_parts) + " No HSN match found."
        }

    else:
        reasoning_parts.append("Input is description. Semantic search not implemented for Hugging Face deployment.")
        return {"HSN_Code": None, "HSN_Description": None, "Confidence": "Low",
                "Reasoning": "Description search not available in this deployment."}

@app.post("/classify", response_model=ClassificationResponse)
async def classify_hts(request: ClassificationRequest):
    try:
        result = map_hts_to_hsn(request.hts_code_or_desc)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Classification error: {str(e)}")

@app.get("/")
async def root():
    return {"message": "HTS to HSN Classification API", "status": "running"}

if __name__ == "__main__":
    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=True)