Fintech Review Category Classifier

import os import json import joblib import torch import numpy as np from flask import Flask, request, render_template_string, jsonify from transformers import AutoTokenizer, AutoModelForSequenceClassification app = Flask(__name__) # ----------------------- # Load artifacts # ----------------------- SAVE_DIR = "./model" try: # Load model & tokenizer tokenizer = AutoTokenizer.from_pretrained(SAVE_DIR) model = AutoModelForSequenceClassification.from_pretrained(SAVE_DIR) model.eval() DEVICE = "cuda" if torch.cuda.is_available() else "cpu" model.to(DEVICE) # Load MultiLabelBinarizer and labels mlb = joblib.load(os.path.join(SAVE_DIR, "mlb.joblib")) with open(os.path.join(SAVE_DIR, "labels.json"), "r", encoding="utf-8") as f: labels = json.load(f) MODEL_LOADED = True print(f"Model loaded successfully on device: {DEVICE}") print(f"Available labels: {labels}") except Exception as e: MODEL_LOADED = False print(f"Error loading model: {e}") tokenizer = None model = None mlb = None labels = [] # Sigmoid for probabilities def sigmoid(x): return 1 / (1 + np.exp(-x)) # ----------------------- # Prediction function (single text only) # ----------------------- def predict_single(text, threshold=0.5): """Predict categories for a single text.""" if not MODEL_LOADED: return [], [] # Tokenize encodings = tokenizer( [text], # Wrap in list since model expects batch truncation=True, padding=True, max_length=256, return_tensors="pt" ).to(DEVICE) # Forward pass with torch.no_grad(): outputs = model(**encodings) logits = outputs.logits.cpu().numpy() # Convert to probabilities probs = sigmoid(logits) # Apply fixed threshold (0.5) pred_bin = (probs >= threshold).astype(int) # Decode to label names row_2d = np.array([pred_bin[0]]) categories = mlb.inverse_transform(row_2d)[0] return list(categories), probs[0] # HTML Template with updated colors HTML_TEMPLATE = """ Fintech Review Category Classifier

{% if model_loaded %}

Model loaded successfully! Available categories: {{ labels|length }}

{% else %}

❌ Model could not be loaded. Please check if the model files exist in './model' directory.

{% endif %}

🤖 Analyzing review...

""" @app.route('/') def index(): return render_template_string( HTML_TEMPLATE, model_loaded=MODEL_LOADED, labels=labels, sample_text="" ) @app.route('/predict', methods=['POST']) def predict_route(): if not MODEL_LOADED: return jsonify({'error': 'Model not loaded. Please check model files.'}), 500 try: review_text = request.form.get('review_text', '').strip() if not review_text: return jsonify({'error': 'Please enter a review.'}), 400 # Predict for SINGLE review only categories, _ = predict_single(review_text, threshold=0.5) # Format result (only one result object) result = { 'text': review_text, 'categories': categories } return jsonify({ 'success': True, 'results': [result], # Still wrapped in list for frontend compatibility 'threshold': 0.5 }) except Exception as e: return jsonify({'error': f'Prediction error: {str(e)}'}), 500 @app.route('/health') def health(): return jsonify({ 'status': 'healthy', 'model_loaded': MODEL_LOADED, 'device': DEVICE if MODEL_LOADED else 'N/A', 'labels_count': len(labels) if labels else 0 }) if __name__ == '__main__': print("="*50) print("🚀 Starting Fintech Review Classification App") print("="*50) if MODEL_LOADED: print(f"Model loaded successfully on {DEVICE}") print(f"Available categories: {len(labels)}") print(f"Categories: {', '.join(labels[:5])}{'...' if len(labels) > 5 else ''}") else: print("❌ Model failed to load - app will run in demo mode") # Get port from environment variable (Hugging Face uses 7860) port = int(os.environ.get("PORT", 7860)) print(f"🌐 Starting server on port: {port}") print("="*50) app.run(host='0.0.0.0', port=port)

Fintech Review Classifier