|
|
import requests, os, zipfile, subprocess, re, warnings |
|
|
warnings.filterwarnings("ignore") |
|
|
os.environ["CURL_CA_BUNDLE"] = "" |
|
|
from io import BytesIO |
|
|
from dotenv import load_dotenv |
|
|
load_dotenv() |
|
|
from datasets import load_dataset |
|
|
import fitz |
|
|
from fastapi import FastAPI, HTTPException |
|
|
from fastapi.middleware.cors import CORSMiddleware |
|
|
from pydantic import BaseModel |
|
|
|
|
|
app = FastAPI(title="Specification Retriever/Splitter API", |
|
|
description=open('documentation.md').read(), |
|
|
docs_url="/") |
|
|
|
|
|
origins = [ |
|
|
"*", |
|
|
] |
|
|
|
|
|
app.add_middleware( |
|
|
CORSMiddleware, |
|
|
allow_origins=origins, |
|
|
allow_credentials=True, |
|
|
allow_methods=["*"], |
|
|
allow_headers=["*"], |
|
|
) |
|
|
|
|
|
spec_contents_3gpp = load_dataset("OrganizedProgrammers/3GPPSpecContent") |
|
|
spec_contents_3gpp = spec_contents_3gpp["train"].to_list() |
|
|
|
|
|
spec_contents_etsi = load_dataset("OrganizedProgrammers/ETSISpecContent") |
|
|
spec_contents_etsi = spec_contents_etsi["train"].to_list() |
|
|
|
|
|
spec_3gpp_format = re.compile(r'^\d{2}\.\d{3}(?:-\d+)?') |
|
|
spec_etsi_format = re.compile(r'^\d{,3} \d{,3}(?:-\d+)?') |
|
|
|
|
|
class SpecRequest(BaseModel): |
|
|
spec_id: str |
|
|
|
|
|
def is_doc_indexed(spec_id: str): |
|
|
return any([True if spec_id == s["doc_id"] else False for s in spec_contents_3gpp]) or any([True if spec_id == s["doc_id"] else False for s in spec_contents_etsi]) |
|
|
|
|
|
def get_doc(spec_id: str): |
|
|
doc = [] |
|
|
for spec in spec_contents_3gpp + spec_contents_etsi: |
|
|
if spec["doc_id"] == spec_id: |
|
|
doc.append(f"{spec['section']}\n{spec['content']}") |
|
|
return "\n\n".join(doc) |
|
|
|
|
|
def get_structured_doc(spec_id: str): |
|
|
doc = {} |
|
|
for spec in spec_contents_3gpp + spec_contents_etsi: |
|
|
if spec["doc_id"] == spec_id: |
|
|
doc[spec["section"]] = spec["content"] |
|
|
return doc |
|
|
|
|
|
def get_pdf_data(request: SpecRequest): |
|
|
specification = request.spec_id |
|
|
if is_doc_indexed(specification): |
|
|
return get_doc(specification) |
|
|
url = requests.post( |
|
|
"https://organizedprogrammers-docfinder.hf.space/find/single", |
|
|
verify=False, |
|
|
headers={"Content-Type": "application/json"}, |
|
|
json={"doc_id": specification} |
|
|
) |
|
|
|
|
|
if url.status_code != 200: |
|
|
raise HTTPException(404, detail="Not found") |
|
|
|
|
|
url = url.json()['url'] |
|
|
response = requests.get( |
|
|
url, |
|
|
verify=False, |
|
|
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"} |
|
|
|
|
|
) |
|
|
|
|
|
pdf = fitz.open(stream=response.content, filetype="pdf") |
|
|
return pdf, pdf.get_toc() |
|
|
|
|
|
@app.post("/extract_text/full") |
|
|
def extract_full_spec(request: SpecRequest): |
|
|
specification = request.spec_id |
|
|
if is_doc_indexed(specification): |
|
|
return get_doc(specification) |
|
|
print(f"[WARNING] Document no. {specification} not indexed or is a TDoc, if it's a specification, try to reindex") |
|
|
total_file = [] |
|
|
if spec_3gpp_format.match(specification): |
|
|
url = requests.post( |
|
|
"https://organizedprogrammers-docfinder.hf.space/find/single", |
|
|
verify=False, |
|
|
headers={"Content-Type": "application/json"}, |
|
|
json={"doc_id": specification} |
|
|
) |
|
|
|
|
|
if url.status_code != 200: |
|
|
raise HTTPException(404, detail="Not found") |
|
|
|
|
|
url = url.json()['url'] |
|
|
response = requests.get( |
|
|
url, |
|
|
verify=False, |
|
|
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"} |
|
|
|
|
|
) |
|
|
|
|
|
zip_bytes = BytesIO(response.content) |
|
|
current_zip_file = zipfile.ZipFile(zip_bytes) |
|
|
for file_info in current_zip_file.infolist(): |
|
|
if file_info.filename.endswith(".zip") and len(current_zip_file.namelist()) == 1: |
|
|
nested_zip_bytes = BytesIO(current_zip_file.read(file_info.filename)) |
|
|
current_zip_file = zipfile.ZipFile(nested_zip_bytes) |
|
|
break |
|
|
|
|
|
for file_info in current_zip_file.infolist(): |
|
|
filename = file_info.filename |
|
|
if (filename.endswith('.doc') or filename.endswith('.docx')) and ("cover" not in filename.lower() and "annex" not in filename.lower()): |
|
|
doc_bytes = current_zip_file.read(filename) |
|
|
ext = filename.split(".")[-1] |
|
|
input_path = f"/tmp/{specification}.{ext}" |
|
|
output_path = f"/tmp/{specification}.txt" |
|
|
with open(input_path, "wb") as f: |
|
|
f.write(doc_bytes) |
|
|
|
|
|
subprocess.run([ |
|
|
"libreoffice", |
|
|
"--headless", |
|
|
"--convert-to", "txt", |
|
|
"--outdir", "/tmp", |
|
|
input_path |
|
|
], check=True) |
|
|
|
|
|
with open(output_path, "r") as f: |
|
|
txt_data = [line.strip() for line in f if line.strip()] |
|
|
|
|
|
os.remove(input_path) |
|
|
os.remove(output_path) |
|
|
total_file.extend(txt_data) |
|
|
if total_file == []: |
|
|
raise HTTPException(status_code=404, detail="Not found !") |
|
|
else: |
|
|
return total_file |
|
|
elif spec_etsi_format.match(specification): |
|
|
print("\n[INFO] Tentative de récupération du texte", flush=True) |
|
|
pdf, doc_toc = get_pdf_data(request) |
|
|
text = [] |
|
|
first = 0 |
|
|
for level, title, page in doc_toc: |
|
|
if title[0].isnumeric(): |
|
|
first = page - 1 |
|
|
break |
|
|
for page in pdf[first:]: |
|
|
text.append("\n".join([line.strip() for line in page.get_text().splitlines()])) |
|
|
text = "\n".join(text) |
|
|
|
|
|
if not text or not doc_toc: |
|
|
print("\n[ERREUR] Pas de texte/table of contents trouvé !") |
|
|
return {} |
|
|
print(f"\n[INFO] Texte {request.spec_id} récupéré", flush=True) |
|
|
return text |
|
|
else: |
|
|
raise HTTPException(status_code=400, detail="Document ID format invalid !") |
|
|
|
|
|
@app.post("/extract_text/structured") |
|
|
def extract_full_spec_by_chapters(request: SpecRequest): |
|
|
specification = request.spec_id |
|
|
if is_doc_indexed(request.spec_id): |
|
|
return get_structured_doc(request.spec_id) |
|
|
print(f"[WARNING] Document no. {specification} not indexed or is a TDoc, if it's a specification, try to reindex") |
|
|
total_file = [] |
|
|
text = extract_full_spec(request) |
|
|
if spec_3gpp_format.match(specification): |
|
|
chapters = [] |
|
|
chapter_regex = re.compile(r"^(\d+[a-z]?(?:\.\d+)*)\t[A-Z0-9][\ \S]+$") |
|
|
|
|
|
for i, line in enumerate(text): |
|
|
if chapter_regex.fullmatch(line): |
|
|
chapters.append((i, line)) |
|
|
|
|
|
document = {} |
|
|
for i in range(len(chapters)): |
|
|
start_index, chapter_title = chapters[i] |
|
|
end_index = chapters[i+1][0] if i+1 < len(chapters) else len(text) |
|
|
content_lines = text[start_index + 1 : end_index] |
|
|
document[chapter_title.replace('\t', " ")] = "\n".join(content_lines) |
|
|
return document |
|
|
elif spec_etsi_format.match(specification): |
|
|
def extract_sections(text, titles): |
|
|
sections = {} |
|
|
|
|
|
sorted_titles = sorted(titles, key=lambda t: text.find(t)) |
|
|
for i, title in enumerate(sorted_titles): |
|
|
start = text.find(title) |
|
|
if i + 1 < len(sorted_titles): |
|
|
end = text.find(sorted_titles[i + 1]) |
|
|
sections[re.sub(r"\s+", " ", title)] = re.sub(r"\s+", " ", text[start:end].replace(title, "").strip().rstrip()) |
|
|
else: |
|
|
sections[re.sub(r"\s+", " ", title)] = re.sub(r"\s+", " ", text[start:].replace(title, "").strip().rstrip()) |
|
|
return sections |
|
|
pdf, toc = get_pdf_data(request) |
|
|
if not text or not toc: |
|
|
print("\n[ERREUR] Pas de texte/table of contents trouvé !") |
|
|
return {} |
|
|
print(f"\n[INFO] Texte {request.spec_id} récupéré", flush=True) |
|
|
titles = [] |
|
|
for level, title, page in toc: |
|
|
if title[0].isnumeric() and '\n'.join(title.strip().split(" ", 1)) in text: |
|
|
titles.append('\n'.join(title.strip().split(" ", 1))) |
|
|
|
|
|
return extract_sections(text, titles) |
|
|
else: |
|
|
raise HTTPException(status_code=400, detail="Document ID format invalid !") |
