core/moderation.py

#coding: utf-8

from typing import Any
from typing import Dict
#from typing import List
#from typing import Tuple

from typing import Optional
#from typing import Union
from os import getenv 
import json

from openai import OpenAI
from dotenv import load_dotenv

def moderate_text(input_text: str, 
                  model: Optional[str] = "omni-moderation-latest") -> Dict[str, Any]:
    
    # Load environment variables from the .env file
    load_dotenv()

    # correct values for `model` is "omni-moderation-latest" and "text-moderation-latest"
    try:
        client = OpenAI(api_key=getenv("OPENAI_API_KEY"))
        response = client.moderations.create(
            model=f"{model}",
            input=f"{input_text}"
        )
        return response
    except Exception as e:
        print(f"An error occurred: {e}")
        return None

def api_moderation_openai_text(text_to_moderate: str) -> Dict[str, Any]:
    response = moderate_text(text_to_moderate)
    if response:
        moderation_result = json.loads(response.to_json())
        
        result_dict = {
            "flagged": False,
            "results": []
        }
        
        for result in moderation_result["results"]:
            result_dict["flagged"] = result_dict["flagged"] or result["flagged"]
            result_dict["results"].append({
                "categories": result["categories"],
                "category_scores": result["category_scores"],
                "category_applied_input_types": result["category_applied_input_types"]
            })
        
        return result_dict
    else:
        return {"error": "La modération a échoué"}


#if __name__=='__main__':
#    text_to_moderate = "Ceci est un exemple de texte à modérer."
#    result = api_moderation_openai_text(text_to_moderate)
#    print("Résultat de la modération :", result)