Upload french_title_api.py

french_title_api.py

@@ -0,0 +1,246 @@
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+import torch
+import pandas as pd
+import numpy as np
+from transformers import MarianMTModel, MarianTokenizer
+import torch.nn as nn
+from torch.utils.data import Dataset, DataLoader
+from sklearn.model_selection import train_test_split
+import logging
+import os
+import sys
+import time
+from waitress import serve  #  new import for production server
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    handlers=[logging.StreamHandler()]
+)
+logger = logging.getLogger(__name__)
+
+# --------------------- MODEL CLASSES ---------------------
+class FrenchTitleGenerator(nn.Module):
+    def __init__(self, model_name='Helsinki-NLP/opus-mt-en-fr'):
+        super(FrenchTitleGenerator, self).__init__()
+        try:
+            logger.info("Loading tokenizer and model...")
+            self.tokenizer = MarianTokenizer.from_pretrained(model_name)
+            self.model = MarianMTModel.from_pretrained(model_name)
+            
+            special_tokens = ['[YEAR]', '[CITY]', '[DESC]', '[TITLE]']
+            self.tokenizer.add_tokens(special_tokens)
+            self.model.resize_token_embeddings(len(self.tokenizer))
+            logger.info("FrenchTitleGenerator initialized successfully")
+        except Exception as e:
+            logger.error(f"Failed to initialize FrenchTitleGenerator: {e}")
+            raise
+
+    def forward(self, input_ids, attention_mask, labels=None):
+        outputs = self.model(
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+            labels=labels
+        )
+        return outputs
+
+# --------------------- SERVICE CLASS ---------------------
+class FrenchTitleService:
+    def __init__(self, csv_path="Data_1.csv", model_path=None):
+        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+        logger.info(f"Using device: {self.device}")
+        
+        self.model = None
+        self.dataset_info = {}
+        self.csv_path = csv_path
+        
+        if not os.path.exists(csv_path):
+            logger.warning(f"Dataset file '{csv_path}' not found. Creating sample data...")
+            self.create_sample_data()
+        
+        self.load_model(model_path)
+        self.load_dataset_info(csv_path)
+    
+    def create_sample_data(self):
+        sample_data = {
+            'Year': [2025, 2025, 2024, 2024, 2025],
+            'City_name': ['Montpellier', 'Montpellier', 'Paris', 'Lyon', 'Marseille'],
+            'Original_title': [
+                'Conference about Compostela Way',
+                'The small papers - Last of the season!!!',
+                'Classical music festival',
+                'Modern art exhibition',
+                'Open air jazz concert'
+            ],
+            'Title': [
+                'CONFÉRENCE SUR LE CHEMIN DE COMPOSTELLE',
+                'LES PETITS PAPIERS - DERNIÈRE DE LA SAISON !!!',
+                'FESTIVAL DE MUSIQUE CLASSIQUE',
+                'EXPOSITION D\'ART MODERNE',
+                'CONCERT DE JAZZ EN PLEIN AIR'
+            ]
+        }
+        pd.DataFrame(sample_data).to_csv(self.csv_path, index=False)
+        logger.info(f"Sample data created: {self.csv_path}")
+    
+    def load_dataset_info(self, csv_path: str):
+        try:
+            logger.info(f"Loading dataset from {csv_path}")
+            df = pd.read_csv(csv_path)
+            
+            required_columns = ['Year', 'City_name', 'Original_title', 'Title']
+            missing_columns = [col for col in required_columns if col not in df.columns]
+            
+            if missing_columns:
+                logger.warning(f"Missing columns in dataset: {missing_columns}")
+            
+            self.dataset_info = {
+                'total_samples': len(df),
+                'columns': df.columns.tolist(),
+                'missing_columns': missing_columns,
+                'sample_data': df.head(3).to_dict('records')
+            }
+            
+            logger.info(f"Dataset loaded: {len(df)} samples")
+            return df
+            
+        except Exception as e:
+            logger.error(f"Dataset loading failed: {e}")
+            self.dataset_info = {'error': str(e)}
+            return None
+    
+    def load_model(self, model_path: str = None):
+        try:
+            if model_path and os.path.exists(model_path):
+                logger.info(f"Loading trained model from {model_path}")
+                self.model = FrenchTitleGenerator()
+                checkpoint = torch.load(model_path, map_location=self.device)
+                self.model.load_state_dict(checkpoint['model_state_dict'])
+                self.model.to(self.device)
+                self.model.eval()
+                logger.info("Trained model loaded successfully!")
+            else:
+                logger.info("Using base model...")
+                self.model = FrenchTitleGenerator()
+                self.model.to(self.device)
+                self.model.eval()
+        except Exception as e:
+            logger.error(f"Model loading failed: {e}")
+            logger.info("Trying to load base model as fallback...")
+            self.model = FrenchTitleGenerator()
+            self.model.to(self.device)
+            self.model.eval()
+    
+    def generate_title(self, year: str, city: str, original_title: str, 
+                      max_length: int = 50, temperature: float = 0.8) -> str:
+        if self.model is None:
+            return f"{original_title} - {city} {year}"
+        
+        try:
+            input_text = f"Generate title from: Year: {year} City: {city} Description: {original_title}"
+            inputs = self.model.tokenizer(
+                input_text,
+                return_tensors='pt',
+                max_length=128,
+                truncation=True,
+                padding=True
+            ).to(self.device)
+            
+            with torch.no_grad():
+                outputs = self.model.model.generate(
+                    inputs['input_ids'],
+                    attention_mask=inputs['attention_mask'],
+                    max_length=max_length,
+                    temperature=temperature,
+                    do_sample=True,
+                    num_beams=3,
+                    early_stopping=True
+                )
+            
+            return self.model.tokenizer.decode(outputs[0], skip_special_tokens=True)
+        except Exception as e:
+            logger.error(f"Title generation failed: {e}")
+            return f"{original_title} - {city} {year}"
+
+# --------------------- FLASK APP ---------------------
+app = Flask(__name__)
+CORS(app)
+title_service = None
+
+def initialize_service():
+    global title_service
+    try:
+        csv_path = "Data_1.csv"
+        model_path = "french_title_generator.pth"
+        title_service = FrenchTitleService(csv_path=csv_path, model_path=model_path)
+        logger.info("French Title Service initialized successfully")
+        return True
+    except Exception as e:
+        logger.error(f"Service initialization failed: {e}")
+        return False
+
+@app.route('/api/health', methods=['GET'])
+def health_check():
+    return jsonify({
+        'status': 'healthy',
+        'timestamp': time.strftime("%Y-%m-%d %H:%M:%S"),
+        'model_loaded': title_service.model is not None if title_service else False
+    })
+
+@app.route('/api/info', methods=['GET'])
+def service_info():
+    if not title_service:
+        return jsonify({'error': 'Service not initialized'}), 500
+    return jsonify({
+        'device': str(title_service.device),
+        'model_loaded': title_service.model is not None,
+        'dataset_info': title_service.dataset_info,
+        'csv_path': title_service.csv_path
+    })
+
+@app.route('/api/generate', methods=['POST'])
+def generate_title():
+    if not title_service:
+        return jsonify({'error': 'Service not initialized'}), 500
+    try:
+        data = request.get_json()
+        for field in ['year', 'city', 'original_title']:
+            if field not in data:
+                return jsonify({'error': f'Missing field: {field}'}), 400
+        generated = title_service.generate_title(
+            year=str(data['year']),
+            city=str(data['city']),
+            original_title=str(data['original_title']),
+            temperature=data.get('temperature', 0.8),
+            max_length=data.get('max_length', 50)
+        )
+        return jsonify({
+            'input': data,
+            'generated_title': generated,
+            'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
+        })
+    except Exception as e:
+        logger.error(f"Title generation error: {e}")
+        return jsonify({'error': str(e)}), 500
+
+@app.route('/')
+def home():
+    return jsonify({
+        'message': ' French Title Generator API is running',
+        'endpoints': {
+            '/api/health': 'Health check',
+            '/api/info': 'Service info',
+            '/api/generate': 'Generate a title'
+        }
+    })
+
+if __name__ == '__main__':
+    if initialize_service():
+        port = 5000
+        logger.info(f" Starting French Title Generator API on port {port} (via Waitress)")
+        serve(app, host="0.0.0.0", port=port)  #  Waitress replaces Flask’s dev server
+    else:
+        logger.error("Failed to initialize service. Exiting.")
+        sys.exit(1)