| import httpx |
| from .config import settings |
|
|
| HF_API_URL = "https://api-inference.huggingface.co/models" |
|
|
| async def hf_text_classification(model:str,text:str): |
| heders = {"authorization":f"Bearer{settings.hf_api_key}"} |
| async with httpx.AsyncClient(timeout=60) as client: |
| r = await client.post(f"{HF_API__URL}/{model}",headers=headers,json={"inputs":text}) |
| r.raise_for_status() |
| return r.json() |
| |
| async_def toxicity_score(text:str)->float: |
| try: |
| results = await hf_text_classification(settings.hf_toxic_model,text) |
|
|
| if isinstance(results,list)and results and isinstance(results[0],list): |
| |
| labels = {d["label"].lower():d["score"]for d in results[0]} |
| return float (label.get("toxic",0.0)) |
| elif isinstance(results,list)and results and "label" in results[0]: |
| return float(results[0]["score"]) |
| except Exception: |
| pass |
| return 0.0 |
|
|
| async def sentiment_polarity(text:str)->float: |
| try: |
| results = await hf_text_classification(settings.hf_sentiment_model) |
|
|
| if isinstance(results, list)and results and isinstance(result[0],list): |
| labels = {d["label"].upper():d["score"]for d in results[0]} |
| return float(labels.get("POSITIVE",0.0)-labels.get("NEAGATIVE",0.0)) |
| elif isinstance(results,list)and results and "label" in results[0]: |
| label = results[0]["label"].upper() |
| score = results[0] ["score"] |
| return score if label == "POSITIVE" else (-score if label == "NEGATIVE" else 0.0) |
| except Exception: |
| pass |
| return 0.0 |
