File size: 5,619 Bytes
0117df3 eb5aac2 31fda96 4d6298c eb5aac2 31fda96 eb5aac2 31fda96 753c2d1 31fda96 f11f069 31fda96 f11f069 31fda96 753c2d1 eb5aac2 753c2d1 4d6298c eb5aac2 753c2d1 31fda96 753c2d1 31fda96 eb5aac2 0117df3 805e1e5 31fda96 eb5aac2 9afba1d f11f069 eb5aac2 31fda96 eb5aac2 88da32f 31fda96 eb5aac2 88da32f 31fda96 eb5aac2 31fda96 eb5aac2 31fda96 eb5aac2 0117df3 31fda96 eb5aac2 9afba1d 31fda96 6396e6b 31fda96 9afba1d 31fda96 9afba1d 0117df3 eb5aac2 0117df3 bc13edc 88da32f 31fda96 0117df3 6396e6b eb5aac2 31fda96 6396e6b 31fda96 eb5aac2 31fda96 eb5aac2 31fda96 eb5aac2 31fda96 0117df3 | 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 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | import asyncio
import logging
from io import BytesIO
from fastapi import Depends, HTTPException, UploadFile, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from config import Config
from .inferencer import analyze_text_with_sentences, classify_text
from .preprocess import parse_docx, parse_pdf, parse_txt
security = HTTPBearer()
# def build_bias_summary(ai_likelihood: float) -> dict[str, object]:
# """Convert an AI likelihood score into a human-readable bias summary."""
# if ai_likelihood > 50:
# overall_bias = "AI"
# bias_statement = f"The text is biased toward AI-generated writing ({ai_likelihood}% AI likelihood)."
# elif ai_likelihood < 50:
# overall_bias = "Human"
# bias_statement = f"The text is biased toward human writing ({100 - ai_likelihood}% human likelihood)."
# else:
# overall_bias = "Balanced"
# bias_statement = "The text is balanced between AI and human writing."
# return {
# "overall_bias": overall_bias,
# "bias_statement": bias_statement,
# }
# Verify Bearer token from Authorization header
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
expected_token = Config.SECRET_TOKEN
if token != expected_token:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail="Invalid or expired token"
)
return token
# Classify plain text input
async def handle_text_analysis(text: str):
text = text.strip()
if not text or len(text.split()) < 10:
raise HTTPException(
status_code=400, detail="Text must contain at least 10 words"
)
if len(text) > 50000:
raise HTTPException(
status_code=413, detail="Text must be less than 50,000 characters"
)
label, perplexity, ai_likelihood = await asyncio.to_thread(classify_text, text)
# bias_summary = build_bias_summary(ai_likelihood)
return {
"result": label,
"perplexity": round(perplexity, 2),
"ai_likelihood": ai_likelihood,
}
# Extract text from 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.",
)
# Classify text from uploaded file
async def handle_file_upload(file: UploadFile):
try:
file_contents = await extract_file_contents(file)
logging.info(f"Extracted text length: {len(file_contents)} characters")
if len(file_contents) > 50000:
return {
"status_code": 413,
"detail": "Text must be less than 50,000 characters",
}
cleaned_text = file_contents.replace("\n", " ").replace("\t", " ").strip()
if not cleaned_text:
raise HTTPException(
status_code=400,
detail="The uploaded file is empty or only contains whitespace.",
)
# print(f"Cleaned text: '{cleaned_text}'") # Debugging statement
label, perplexity, ai_likelihood = await asyncio.to_thread(
classify_text, cleaned_text
)
return {
"content": file_contents,
"result": label,
"perplexity": round(perplexity, 2),
"ai_likelihood": ai_likelihood,
}
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 not text or len(text.split()) < 10:
raise HTTPException(
status_code=400, detail="Text must contain at least 10 words"
)
if len(text) > 50000:
raise HTTPException(
status_code=413, detail="Text must be less than 50,000 characters"
)
result = await asyncio.to_thread(analyze_text_with_sentences, text)
return result
# Analyze each sentence from uploaded file
async def handle_file_sentence(file: UploadFile):
try:
file_contents = await extract_file_contents(file)
if len(file_contents) > 50000:
# raise HTTPException(status_code=413, detail="Text must be less than 10,000 characters")
return {
"status_code": 413,
"detail": "Text must be less than 50,000 characters",
}
cleaned_text = file_contents.replace("\n", " ").replace("\t", " ").strip()
if not cleaned_text:
raise HTTPException(
status_code=400,
detail="The uploaded file is empty or only contains whitespace.",
)
result = await handle_sentence_level_analysis(cleaned_text)
return {"content": file_contents, **result}
except HTTPException:
raise
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)
|