app.py

from flask import Flask, request, jsonify
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import os


# Flask 앱 초기화
app = Flask(__name__)

# Load the BERTweet model and tokenizer
tokenizer = AutoTokenizer.from_pretrained("finiteautomata/bertweet-base-sentiment-analysis")
model = AutoModelForSequenceClassification.from_pretrained("finiteautomata/bertweet-base-sentiment-analysis")

LABELS = ['Negative', 'Neutral', 'Positive']

def analyze_sentiment(text):
    # Converts text into tokens that the model can process
    # return_tensors="pt": Returns PyTorch tensors
    # truncation=True: Cuts text if it's too long
    # padding=True: Adds padding to make sequences uniform length
    inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True)

    # Predict with the model
    # Passes tokenized input through the model
    # Extracts logits (raw prediction scores) from model output
    outputs = model(**inputs)
    logits = outputs.logits

    # Sentiment Classification:
    # torch.argmax(): Finds index of highest score 가장 높은 값의 인덱스를 추출하여 감성 레이블을 선택
    # Maps this index to a sentiment label using LABELS dictionary
    # dim=1: Operates along rows, .item(): Converts tensor to Python scalar
    predicted_class = torch.argmax(logits, dim=1).item()
    sentiment = LABELS[predicted_class]

    return sentiment

# API endpoint to analyze a text
@app.route('/analyze', methods=['POST'])
def analyze():
    data = request.get_json()

    if 'text' not in data:
        return jsonify({'error': 'No text provided'}), 400

    # call 감성 분석 function for 입력 텍스트
    text = data['text']
    sentiment = analyze_sentiment(text)

    return jsonify({'text': text, 'sentiment': sentiment})

# API endpoint to analyze a list of texts
@app.route('/analyze_texts', methods=['POST'])
def analyze_texts():
    data = request.get_json()

    if 'texts' not in data:
        return jsonify({'error': 'No texts provided'}), 400

    # Perform analysis on each text in the list, process each one, and return a list of sentiment responses
    texts = data['texts']
    results = [{'text': text, 'sentiment': analyze_sentiment(text)} for text in texts]

    # Count negative sentiments
    negative_count = sum(1 for result in results if result['sentiment'] == 'Negative')

    # Determine if it's a risk (more than half are negative)
    risk = negative_count > (len(texts) / 2)

    # Return the results as a JSON response
    # return jsonify(results)
    # Return the results and risk status as a JSON response
    return jsonify({
        'results': results,
        'Negative Sentiment?': risk
    })

# run

if __name__ == '__main__':
    # 환경 변수 PORT를 우선 사용, 없으면 기본값 5000 사용
    #port = int(os.getenv('PORT', 5001))
    app.run(host='0.0.0.0', port=7860)