app.py

MAX_API_RETRIES=3
from pydantic import BaseModel
import pydantic_core._pydantic_core as pydantic_core
from fastapi import FastAPI
import json,requests
app = FastAPI()
def morph(valor,tipo):
    if isinstance(valor,dict):
        return tipo(valor)
    elif isinstance(valor,list):
        return list(map(lambda v:morph(v,tipo),valor))
    else:
        return valor 
class ResponseData(BaseModel):
    error:bool
    message:str
class Diccionario:
    def __init__(self,diccionario:dict):
        self.d = diccionario
    def __str__(self):
        return json.dumps(self.d)
    def __repr__(self):
        return self.__str__()
    def __getattr__(self,data:str):
        attr = self.d.get(data,None)
        return morph(attr,Diccionario)
class Mensaje(BaseModel):
    content:str = ""
    role:str = "user"
class persona(BaseModel):
    nombres:str 
    apellidos:str
    edad:int
    personalidad:str
    biografia:str
    pais:str
    sexo:str
    hobbies:list[str]
class personaje(BaseModel):
    interpretador:persona
    nombre:str
    personalidad:str
    apariencia:str
    papel:str
    edad:int
class pelicula(BaseModel):
    personajes: list[personaje]
    titulo:str
    traduccion:dict[str,str]
    sinopsis:str
    duracion:int
    detailed_image_token_prompt_english:list[str]

class trabajador(BaseModel):
    identidad:persona
    trabajo:str
    sueldo:int
    remoto:bool=False
    historial_sueldo:list[str]
class comentario(BaseModel):
    mensaje:str
    autor:persona
    likes:int
    dislikes:int
class publicacion(BaseModel):
    mensaje:str
    likes:int
    dislikes:int
    visitas:int
    comentarios:list[comentario]
class perfil(BaseModel):
    data:persona
    amigos:list[persona]
    publicaciones:list[publicacion]
class steamgame(BaseModel):
    nombre:str
    descripcion_corta:str
    descripcion_larga:str
    reviews:list[comentario]
    calificacion:int
    descargas:int

class episodio(BaseModel):
    personajes_involucrados:list[personaje]
    dialogos:list[dict[int,dict[str,str]]]
    descripcion:str
class serie(BaseModel):
    nombre:str
    personajes:list[personaje]
    clasificacion:str
    rating:int
    episodios:int
    lista_episodios:list[episodio]
def ask(query:str,system_prompt:str)->ResponseData:
    msgs:list[dict[str,str]] = [
        Mensaje(role="system",content=system_prompt).model_dump(),
        Mensaje(content=query).model_dump()
    ]
    response = requests.post("https://text.pollinations.ai/openai",json={
        "messages":msgs,
        "private":True,
        "jsonMode":True,
        "model":"openai",
        "seed":__import__("random").randint(10000,999999)
    })
    data=response.json()

    error_msg = Diccionario(data).error
    structured_data = ResponseData(error=True if error_msg else False,message=error_msg or Diccionario(data).choices[0].message.content)
    return structured_data

def generate_model(model:BaseModel,gived_data:BaseModel=None,retries=0)->BaseModel | None:
    system = f"""Eres un generador de contenido avanzado(evita contenido repetido) capaz de comprender cualquier contexto, debes generar un json valido con estas propiedades, las claves son los valores de title y los valores deben ser del mismo tipo que el valor de type, el title que esta por fuera de propierties debe ser el tipo general, tipo el que contiene todas las propiedades,no realices ningun comentario o acotacion,otorga un json valido,las claves deben estar en minuscula pero los valores no necesariamente"""
    if gived_data:
        system+=f"\n Ademas tengo esta informacion: {gived_data.model_dump_json()}"
    resultado = ask(json.dumps(model.model_json_schema()),system)
    if resultado.error==True: 
        return {"error":resultado.message}
    try:
        data = list(json.loads(resultado.message).values())[0]
        if retries>MAX_API_RETRIES: return {"error":"Se ha intentado, pero no hubo caso, prueba nuevamente","failed_data":data}
        if isinstance(data,str):
            retries+=1
            return generate_model(model,gived_data,retries)
        

        return model(**(data))
    except (pydantic_core.ValidationError, json.JSONDecodeError) as e:
        retries+=1
        return generate_model(model,gived_data,retries) 

models:BaseModel= [
    persona,
    pelicula,
    trabajador,
    publicacion,
    perfil,
    steamgame,
    serie
]
def create_handler(model_class: BaseModel):
    def handler() -> dict:
        generated=generate_model(model_class)
        
        return generated if isinstance(generated,dict) else json.loads(generated.model_dump_json())
    handler.__name__=f"Crear {model_class.__name__}"
    return handler
for model in models:
    app.add_api_route(f"/{model.__name__}",create_handler(model))
@app.post("/generate_movie")
def gen_movie(user:persona):
    return generate_model(pelicula,user)