File size: 3,209 Bytes
a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 cf42286 a0c24d5 | 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 | from __future__ import annotations
import re
from pathlib import Path
import fitz
import httpx
from bs4 import BeautifulSoup
from docx import Document as DocxDocument
from app.config import settings
ALLOWED_EXTENSIONS = {".pdf", ".docx", ".txt", ".md"}
USER_AGENT = "Mozilla/5.0 (compatible; StudysonBot/1.0; +https://github.com/)"
class DocumentProcessor:
@staticmethod
def validate_file_type(filename: str) -> bool:
return Path(filename).suffix.lower() in ALLOWED_EXTENSIONS
@staticmethod
async def extract_text(file_path: Path) -> str:
suffix = file_path.suffix.lower()
if suffix == ".pdf":
return DocumentProcessor._extract_pdf(file_path)
if suffix == ".docx":
return DocumentProcessor._extract_docx(file_path)
if suffix in {".txt", ".md"}:
return file_path.read_text(encoding="utf-8", errors="replace")
raise ValueError(f"Unsupported file type: {suffix}")
@staticmethod
def _extract_pdf(file_path: Path) -> str:
with fitz.open(str(file_path)) as doc:
return "\n\n".join(page.get_text() for page in doc)
@staticmethod
def _extract_docx(file_path: Path) -> str:
document = DocxDocument(str(file_path))
return "\n".join(p.text for p in document.paragraphs if p.text)
@staticmethod
async def scrape_url(url: str) -> tuple[str, str]:
timeout = httpx.Timeout(settings.scrape_timeout_seconds)
headers = {"User-Agent": USER_AGENT, "Accept": "text/html,*/*"}
async with httpx.AsyncClient(
timeout=timeout,
headers=headers,
follow_redirects=True,
max_redirects=5,
) as client:
async with client.stream("GET", url) as response:
response.raise_for_status()
content_type = response.headers.get("content-type", "")
if "html" not in content_type and "text" not in content_type:
raise ValueError(f"Unsupported content-type: {content_type}")
chunks: list[bytes] = []
total = 0
async for chunk in response.aiter_bytes():
total += len(chunk)
if total > settings.max_scrape_bytes:
raise ValueError(
f"Page exceeds {settings.max_scrape_bytes} byte limit"
)
chunks.append(chunk)
html = b"".join(chunks).decode(
response.encoding or "utf-8", errors="replace"
)
soup = BeautifulSoup(html, "html.parser")
for tag in soup(["script", "style", "nav", "footer", "header", "aside", "noscript"]):
tag.decompose()
title_tag = soup.find("title")
title = title_tag.get_text(strip=True) if title_tag else "Web Document"
text = soup.get_text(separator="\n", strip=True)
return title, text
@staticmethod
def clean_text(text: str) -> str:
text = re.sub(r"[ \t]+", " ", text)
lines = (line.strip() for line in text.splitlines())
return "\n".join(line for line in lines if line)
|