#!/usr/bin/env python # coding: utf-8 """ FastAPI endpoint para el proyecto lanc-ner Recibe un archivo .txt y devuelve las anotaciones NER en formato JSON por línea, con el esquema: [ { "Anotacion": "", "entities": [ { "start": int, "end": int, "entity": str, "fields": [str] } ]}, ... ] Autenticación: Bearer token via variable de entorno API_TOKEN. """ from fastapi import FastAPI, File, UploadFile, HTTPException, Depends, Security, Query from fastapi.responses import JSONResponse, HTMLResponse from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.openapi.utils import get_openapi import spacy from spacy import displacy import os import logging # ── Logging ─────────────────────────────────────────────────────────────────── logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ── Configuración ───────────────────────────────────────────────────────────── API_TOKEN = os.getenv("API_TOKEN", "") if not API_TOKEN: logger.warning("⚠️ API_TOKEN no definido — la API está desprotegida.") MODELS = { "ruler": "./models/model_entities_ruler/model-last/", "toponyms": "./models/model_toponyms_patterns/model-last/", } # ── Carga de modelos ────────────────────────────────────────────────────────── loaded_models = {} for name, path in MODELS.items(): try: loaded_models[name] = spacy.load(path) logger.info(f"Modelo '{name}' cargado desde {path}") except OSError: logger.error(f"Modelo '{name}' no encontrado en {path}") # ── App ─────────────────────────────────────────────────────────────────────── app = FastAPI( title="Lancaster NER API", description=( "API REST para detección de entidades nombradas (NER) basada en spaCy.\n\n" "**Modelos disponibles:**\n" "- `ruler` — Spacy Ruler (Sentences): entidades generales\n" "- `toponyms` — DECM annotations with Paragraphs: topónimos\n\n" "**Autenticación:** todos los endpoints de anotación requieren " "`Authorization: Bearer ` en la cabecera.\n\n" "**Formato de salida:** lista de objetos por línea del documento, " "con campo `Anotacion` (texto de la línea) y `entities` (entidades detectadas)." ), version="1.3.0", docs_url="/docs", redoc_url="/redoc", ) # ── Seguridad: Bearer token ─────────────────────────────────────────────────── bearer_scheme = HTTPBearer(auto_error=True) def verify_token(credentials: HTTPAuthorizationCredentials = Security(bearer_scheme)): if API_TOKEN and credentials.credentials != API_TOKEN: raise HTTPException( status_code=401, detail="Token inválido o expirado.", headers={"WWW-Authenticate": "Bearer"}, ) return credentials.credentials # ── Función NER ─────────────────────────────────────────────────────────────── def run_ner(text: str, model_name: str) -> list: """ Procesa el texto línea a línea y devuelve una lista con el formato: [ { "Anotacion": "", "entities": [ { "start": int, "end": int, "entity": str, "fields": [str] } ] }, ... ] """ if model_name not in loaded_models: raise HTTPException( status_code=400, detail=f"Modelo '{model_name}' no disponible. Opciones: {list(loaded_models.keys())}", ) nlp = loaded_models[model_name] lines = text.splitlines() result = [] for line in lines: doc = nlp(line) entities = [] for ent in doc.ents: entities.append({ "start": ent.start_char, "end": ent.end_char, "entity": ent.label_, "fields": [], }) result.append({ "Anotacion": line, "entities": entities, }) return result # ── Endpoints ───────────────────────────────────────────────────────────────── @app.get("/", tags=["Health"], summary="Health check") def health_check(): """Estado de la API y modelos cargados. **No requiere autenticación.**""" return { "status": "ok", "auth_enabled": bool(API_TOKEN), "models_available": list(loaded_models.keys()), "models_missing": [k for k in MODELS if k not in loaded_models], } @app.post( "/annotate", tags=["NER"], summary="Anotar documento .txt", response_description="Lista de anotaciones NER por línea del documento.", ) async def annotate( file: UploadFile = File(..., description="Archivo de texto plano (.txt) a anotar."), model: str = Query( default="ruler", description="Modelo NER a usar: `ruler` o `toponyms`.", enum=["ruler", "toponyms"], ), token: str = Depends(verify_token), ): """ Recibe un archivo `.txt` y devuelve una lista de anotaciones NER por línea. **Cabecera requerida:** ``` Authorization: Bearer ``` **Formato de respuesta:** ```json [ { "Anotacion": "Registraron Baltasar Perez y Ernando de Sanctiago vesinos de Sevilla", "entities": [ { "start": 12, "end": 26, "entity": "Persona", "fields": ["Nombre", "Hombre"] }, { "start": 61, "end": 68, "entity": "Localización", "fields": ["Topónimo"] } ] } ] ``` """ if not file.filename.lower().endswith(".txt"): raise HTTPException(status_code=415, detail="Solo se aceptan archivos .txt.") try: text = (await file.read()).decode("utf-8") except UnicodeDecodeError: raise HTTPException(status_code=422, detail="El archivo no es UTF-8 válido.") if not text.strip(): raise HTTPException(status_code=422, detail="El archivo está vacío.") result = run_ner(text, model) return JSONResponse(content=result) @app.post( "/annotate/text", tags=["NER"], summary="Anotar texto plano (JSON)", response_description="Lista de anotaciones NER por línea.", ) async def annotate_text( payload: dict, model: str = Query( default="ruler", description="Modelo NER a usar: `ruler` o `toponyms`.", enum=["ruler", "toponyms"], ), token: str = Depends(verify_token), ): """ Envía el texto directamente en el body JSON. **Body:** ```json { "text": "El municipio de Atengo se encuentra en Jalisco." } ``` """ text = payload.get("text", "").strip() if not text: raise HTTPException(status_code=422, detail="El campo 'text' está vacío.") result = run_ner(text, model) return JSONResponse(content=result) @app.get( "/docu", tags=["Documentación"], summary="Documentación HTML del servicio", response_class=HTMLResponse, include_in_schema=False, ) def docu(): """Página HTML autodocumentada. **No requiere autenticación.**""" schema = get_openapi( title=app.title, version=app.version, description=app.description, routes=app.routes, ) endpoints_html = "" for path, methods in schema.get("paths", {}).items(): for method, info in methods.items(): badge_color = { "get": "#61affe", "post": "#49cc90", "put": "#fca130", "delete": "#f93e3e", }.get(method.lower(), "#888") params_html = "" for p in info.get("parameters", []): params_html += f""" {p.get('name')} {p.get('in')} {'✅' if p.get('required') else '—'} {p.get('description','')} """ params_section = ( f"""{params_html}
NombreEnReq.Descripción
""" if params_html else "" ) endpoints_html += f"""
{method.upper()} {path} {info.get('summary','')}

{info.get('description','').strip().split(chr(10))[0]}

{params_section}
""" html = f""" {app.title} — Documentación

{app.title} v{app.version}

Detección de entidades nombradas (NER) sobre documentos de texto históricos.

Documentación interactiva

Formato de respuesta

El documento se procesa línea a línea. Cada línea genera un objeto con:

Anotacion — texto original de la línea
entities — lista de entidades detectadas con start, end, entity y fields
[
  {{
    "Anotacion": "Registraron Baltasar Perez y Ernando de Sanctiago vesinos de Sevilla",
    "entities": [
      {{ "start": 12, "end": 26, "entity": "Persona", "fields": ["Nombre", "Hombre"] }},
      {{ "start": 61, "end": 68, "entity": "Localización", "fields": ["Topónimo"] }}
    ]
  }}
]

Modelos disponibles

ruler — Spacy Ruler (Sentences)
Detecta entidades generales con reglas sobre oraciones.


toponyms — DECM annotations with Paragraphs
Especializado en topónimos y anotaciones geográficas por párrafos.

Autenticación

Todos los endpoints de anotación requieren un Bearer token:

Authorization: Bearer <tu-token>

Endpoints

{endpoints_html}

Ejemplos de uso

Subir archivo .txt:

curl -X POST "https://alezsd-lancaster-ner-api.hf.space/annotate?model=ruler" \\
     -H "Authorization: Bearer <token>" \\
     -F "file=@documento.txt"

Enviar texto como JSON:

curl -X POST "https://alezsd-lancaster-ner-api.hf.space/annotate/text?model=toponyms" \\
     -H "Authorization: Bearer <token>" \\
     -H "Content-Type: application/json" \\
     -d '{{"text": "El municipio de Atengo se encuentra en Jalisco."}}'
""" return HTMLResponse(content=html)