Spaces:
Running
Running
File size: 4,647 Bytes
4fee431 1140ed3 4fee431 1140ed3 4fee431 1140ed3 4fee431 1140ed3 4fee431 1140ed3 4fee431 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
import asyncio
from io import BytesIO
from fastapi import HTTPException, UploadFile, status, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import os
from features.nepali_text_classifier.inferencer import classify_text
from features.nepali_text_classifier.preprocess import *
import re
security = HTTPBearer()
def contains_english(text: str) -> bool:
# Remove escape characters
cleaned = text.replace("\n", "").replace("\t", "")
return bool(re.search(r'[a-zA-Z]', cleaned))
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
expected_token = os.getenv("MY_SECRET_TOKEN")
if token != expected_token:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid or expired token"
)
return token
async def nepali_text_analysis(text: str):
end_symbol_for_NP_text(text)
words = text.split()
if len(words) < 10:
raise HTTPException(status_code=400, detail="Text must contain at least 10 words")
if len(text) > 10000:
raise HTTPException(status_code=413, detail="Text must be less than 10,000 characters")
result = await asyncio.to_thread(classify_text, text)
return result
#Extract text form uploaded files(.docx,.pdf,.txt)
async def extract_file_contents(file:UploadFile)-> str:
content = await file.read()
file_stream = BytesIO(content)
if file.content_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
return parse_docx(file_stream)
elif file.content_type =="application/pdf":
return parse_pdf(file_stream)
elif file.content_type =="text/plain":
return parse_txt(file_stream)
else:
raise HTTPException(status_code=415,detail="Invalid file type. Only .docx,.pdf and .txt are allowed")
async def handle_file_upload(file: UploadFile):
try:
file_contents = await extract_file_contents(file)
end_symbol_for_NP_text(file_contents)
if len(file_contents) > 10000:
raise HTTPException(status_code=413, detail="Text must be less than 10,000 characters")
cleaned_text = file_contents.replace("\n", " ").replace("\t", " ").strip()
if not cleaned_text:
raise HTTPException(status_code=404, detail="The file is empty or only contains whitespace.")
result = await asyncio.to_thread(classify_text, cleaned_text)
return result
except Exception as e:
logging.error(f"Error processing file: {e}")
raise HTTPException(status_code=500, detail="Error processing the file")
async def handle_sentence_level_analysis(text: str):
text = text.strip()
if len(text) > 10000:
raise HTTPException(status_code=413, detail="Text must be less than 10,000 characters")
end_symbol_for_NP_text(text)
# Split text into sentences
sentences = [s.strip() + "।" for s in text.split("।") if s.strip()]
results = []
for sentence in sentences:
end_symbol_for_NP_text(sentence)
result = await asyncio.to_thread(classify_text, sentence)
results.append({
"text": sentence,
"result": result["label"],
"likelihood": result["confidence"]
})
return {"analysis": results}
async def handle_file_sentence(file:UploadFile):
try:
file_contents = await extract_file_contents(file)
if len(file_contents) > 10000:
raise HTTPException(status_code=413, detail="Text must be less than 10,000 characters")
cleaned_text = file_contents.replace("\n", " ").replace("\t", " ").strip()
if not cleaned_text:
raise HTTPException(status_code=404, detail="The file is empty or only contains whitespace.")
# Ensure text ends with danda so last sentence is included
# Split text into sentences
sentences = [s.strip() + "।" for s in cleaned_text.split("।") if s.strip()]
results = []
for sentence in sentences:
end_symbol_for_NP_text(sentence)
result = await asyncio.to_thread(classify_text, sentence)
results.append({
"text": sentence,
"result": result["label"],
"likelihood": result["confidence"]
})
return {"analysis": results}
except Exception as e:
logging.error(f"Error processing file: {e}")
raise HTTPException(status_code=500, detail="Error processing the file")
def classify(text: str):
return classify_text(text)
|