app.py

import time
from datetime import datetime
import os, warnings, nltk, json, subprocess
import numpy as np
from nltk.stem import WordNetLemmatizer
from dotenv import load_dotenv
from sklearn.preprocessing import MinMaxScaler

from bs4 import BeautifulSoup
import requests
from urllib.parse import parse_qs, urlparse

warnings.filterwarnings('ignore')
nltk.download('wordnet')
load_dotenv()
os.environ['CURL_CA_BUNDLE'] = ""

from huggingface_hub import configure_http_backend
def backend_factory() -> requests.Session:
    session = requests.Session()
    session.verify = False
    return session

configure_http_backend(backend_factory=backend_factory)

from datasets import load_dataset
import bm25s
from bm25s.hf import BM25HF

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

from schemas import *
from classes import *

lemmatizer = WordNetLemmatizer()

spec_metadatas_3gpp = load_dataset("OrganizedProgrammers/3GPPSpecMetadata")
spec_contents_3gpp = load_dataset("OrganizedProgrammers/3GPPSpecContent")
tdoc_locations_3gpp = load_dataset("OrganizedProgrammers/3GPPTDocLocation")

spec_metadatas_etsi = load_dataset("OrganizedProgrammers/ETSISpecMetadata")
spec_contents_etsi = load_dataset("OrganizedProgrammers/ETSISpecContent")

spec_contents_3gpp = spec_contents_3gpp["train"].to_list()
spec_metadatas_3gpp = spec_metadatas_3gpp["train"].to_list()
spec_contents_etsi = spec_contents_etsi["train"].to_list()
spec_metadatas_etsi = spec_metadatas_etsi["train"].to_list()
tdoc_locations = tdoc_locations_3gpp["train"].to_list()

bm25_index_3gpp = BM25HF.load_from_hub("OrganizedProgrammers/3GPPBM25IndexSingle", load_corpus=True, token=os.environ["HF_TOKEN"], )
bm25_index_etsi = BM25HF.load_from_hub("OrganizedProgrammers/ETSIBM25IndexSingle", load_corpus=True, token=os.environ["HF_TOKEN"])

def extract_args_and_map(href):
    if not href or not href.lower().startswith('javascript:'):
        return None
    js = href[len('javascript:'):].strip()
    m = re.match(r'\w+\((.*)\)', js)
    if not m:
        return None
    args_str = m.group(1).strip()
    parts = [part.strip() for part in args_str.split(',', 1)]
    if len(parts) != 2:
        return None
    try:
        media_id = int(parts[0])
    except ValueError:
        return None
    spec_type = parts[1].strip()
    if (spec_type.startswith("'") and spec_type.endswith("'")) or (spec_type.startswith('"') and spec_type.endswith('"')):
        spec_type = spec_type[1:-1]

    return media_id, spec_type

url = "https://globalplatform.org/wp-content/themes/globalplatform/ajax/specs-library.php"
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"}
resp = requests.post(url, verify=False, headers=headers)
soup = BeautifulSoup(resp.text, 'html.parser')

panels = soup.find_all('div', class_='panel panel-default')
gp_spec_locations = {}
for panel in panels:
  header = ''.join([t for t in panel.find('a').children if t.name is None]).strip()
  try:
    title, doc_id = header.split(' | ')
    panel_body = panel.find('div', class_='panel-body')

    download_btn_href = panel_body.find_all('a', href=lambda href: href and href.strip().lower().startswith('javascript:'))[0]
    media_id, spec_type = extract_args_and_map(download_btn_href['href'])

    changes_history = panel.find_all('div', class_="row")
    paragraphs_ch = [version.find('p').text for version in changes_history][::-1]
    document_commits = []
    for version in range(len(paragraphs_ch)):
       document_commits.append(f"Version {version + 1} : {paragraphs_ch[version]}")
    gp_spec_locations[doc_id] = {"title": title, "file_id": media_id, "committee": spec_type, "summary": "\n".join(document_commits)}
  except:
    continue

def get_docs_from_url(url):
    """Get list of documents/directories from a URL"""
    try:
        response = requests.get(url, verify=False, timeout=10)
        soup = BeautifulSoup(response.text, "html.parser")
        return [item.get_text() for item in soup.select("tr td a")]
    except Exception as e:
        print(f"Error accessing {url}: {e}")
        return []

def get_tdoc_url(doc_id):
    for tdoc in tdoc_locations:
        if tdoc["doc_id"] == doc_id:
            return tdoc["url"]
    return "Document not indexed (re-indexing documents ?)"
        
def get_spec_url(document):
    series = document.split(".")[0].zfill(2)
    url = f"https://www.3gpp.org/ftp/Specs/archive/{series}_series/{document}"
    versions = get_docs_from_url(url)
    return url + "/" + versions[-1] if versions != [] else f"Specification {document} not found"

def get_document(spec_id: str, spec_title: str, source: str):
    text = [f"{spec_id} - {spec_title}"]
    spec_contents = spec_contents_3gpp if source == "3GPP" else spec_contents_etsi if source == "ETSI" else spec_contents_3gpp + spec_contents_etsi
    for section in spec_contents:
        if not isinstance(section, str) and spec_id == section["doc_id"]:
            text.extend([section['section'], section['content']])
    return text

def get_gp_spec_url(data):
    file_id = data['file_id']
    spec_type = data['committee']

    url = "https://globalplatform.org/wp-content/themes/globalplatform/ajax/download-spec-submit.php"
    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"}
    resp = requests.post(url, verify=False, headers=headers, data={"first_name": "", "last_name": "", "company": "", "email": "", "media_id": file_id, "spec_type": spec_type, "agree": "true"})

    r = resp.text
    mat = re.search(r"window\.location\.href\s*=\s*'([^']+)'", r)
    if mat:
        full_url = mat.group(1)
        parsed_url = urlparse(full_url)
        query_params = parse_qs(parsed_url.query)
        return query_params.get('f')[0]

tags_metadata = [
    {
        "name": "Document Retrieval",
        "description": """
        Direct document lookup operations for retrieving specific documents by their unique identifiers.
        
        These endpoints provide fast access to document URLs, versions, and metadata without requiring keyword searches.
        Perfect for when you know the exact document ID you're looking for.
        """,
    },
    {
        "name": "Content Search",
        "description": """
        Advanced search operations for finding documents based on keywords and content matching.
        
        Includes both quick metadata-based searches and deep content analysis with flexible filtering options.
        Supports different search modes and logical operators for precise results.
        """,
    },
]

app = FastAPI(
    title="3GPP & ETSI Document Finder API",
    description=open('documentation.md').read(),
    openapi_tags=tags_metadata
)

app.mount("/static", StaticFiles(directory="static"), name="static")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

etsi_doc_finder = ETSIDocFinder()
etsi_spec_finder = ETSISpecFinder()

valid_3gpp_doc_format = re.compile(r'^(S[1-6P]|C[1-6P]|R[1-6P])-\d+', flags=re.IGNORECASE)
valid_3gpp_spec_format = re.compile(r'^\d{2}\.\d{3}(?:-\d+)?')

valid_etsi_doc_format = re.compile(r'^(?:SET|SCP|SETTEC|SETREQ|SCPTEC|SCPREQ)\(\d+\)\d+(?:r\d+)?', flags=re.IGNORECASE)
valid_etsi_spec_format = re.compile(r'^\d{3} \d{3}(?:-\d+)?')

@app.get("/", tags=["Misc"], summary="Returns index.html file")
def frontend():
    return FileResponse(os.path.join('templates', 'index.html'))

@app.get("/reconnect", tags=["Misc"], summary="Reconnects to ETSI portal for document access", include_in_schema=False)
def reconnect():
    data = etsi_doc_finder.connect()
    if data.get('error', None) and not data.get('error'):
        return data['message']
    raise HTTPException(status_code=400, detail=data['message'])
@app.post("/find/single", response_model=DocResponse, tags=["Document Retrieval"], summary="Retrieve a single document by ID", responses={
             200: {
                 "description": "Document found successfully",
                 "content": {
                     "application/json": {
                         "example": {
                             "doc_id": "23.401",
                             "url": "https://www.3gpp.org/ftp/Specs/archive/23_series/23.401/23401-h20.zip",
                             "version": "h20",
                             "scope": "General Packet Radio Service (GPRS) enhancements for Evolved Universal Terrestrial Radio Access Network (E-UTRAN) access",
                             "search_time": 0.0234
                         }
                     }
                 }
             },
             404: {
                 "description": "Document not found or not indexed",
                 "content": {
                     "application/json": {
                         "example": {
                             "detail": "Specification 99.999 not found"
                         }
                     }
                 }
             }
         })
def find_document(request: DocRequest):
    start_time = time.time()
    document = request.doc_id
    if valid_3gpp_doc_format.match(document):
        url = get_tdoc_url(document)
    elif valid_3gpp_spec_format.match(document):
        url = get_spec_url(document)
    elif valid_etsi_doc_format.match(document):
        url = etsi_doc_finder.search_document(document)
    elif valid_etsi_spec_format.match(document):
        url = etsi_spec_finder.search_document(document)
    elif document.startswith("GP"):
        for sp in gp_spec_locations:
            if document.lower() in sp.lower():
                url = get_gp_spec_url(gp_spec_locations[sp])
    else:
        url = "Document ID not supported"

    if "Specification" in url or "Document" in url:
        raise HTTPException(status_code=404, detail=url)

    version = None
    if valid_3gpp_spec_format.match(document):
        version = url.split("/")[-1].replace(".zip", "").split("-")[-1]
    scope = None
    spec_metadatas = spec_metadatas_3gpp if valid_3gpp_spec_format.match(document) else spec_metadatas_etsi
    for spec in spec_metadatas:
        if spec['id'] == document:
            scope = spec['scope']
            break

    return DocResponse(
        doc_id=document,
        version=version,
        url=url,
        search_time=time.time() - start_time,
        scope=scope
    )

@app.post("/find/batch", response_model=BatchDocResponse, summary="Retrieve multiple documents by IDs", tags=["Document Retrieval"], responses={
             200: {
                 "description": "Batch processing completed",
                 "content": {
                     "application/json": {
                         "example": {
                             "results": {
                                 "23.401": "https://www.3gpp.org/ftp/Specs/archive/23_series/23.401/23401-h20.zip",
                                 "S1-123456": "https://www.3gpp.org/ftp/tsg_sa/WG1_Serv/TSGSI_123/Docs/S1-123456.zip"
                             },
                             "missing": ["99.999", "INVALID-DOC"],
                             "search_time": 0.156
                         }
                     }
                 }
             }
         })
def find_document_batch(request: BatchDocRequest):
    start_time = time.time()
    documents = request.doc_ids
    results = {}
    missing = []

    for document in documents:
        if valid_3gpp_doc_format.match(document):
            url = get_tdoc_url(document)
        elif valid_3gpp_spec_format.match(document):
            url = get_spec_url(document)
        elif valid_etsi_doc_format.match(document):
            etsi_doc_finder.search_document(document)
        elif valid_etsi_spec_format.match(document):
            etsi_spec_finder.search_document(document)
        elif document.startswith("GP"):
            for sp in gp_spec_locations:
                if document.lower() in sp.lower():
                    url = get_gp_spec_url(gp_spec_locations[sp])
        else:
            url = "Document ID not supported"
        if "Specification" in url or "Document" in url:
            missing.append(document)
        else:
            results[document] = url
    
    return BatchDocResponse(
        results=results,
        missing=missing,
        search_time=time.time()-start_time
    )

@app.post('/search', response_model=KeywordResponse, tags=["Content Search"], summary="Search specifications by keywords", responses={
             200: {
                 "description": "Search completed successfully"
             },
             400: {
                 "description": "You must enter keywords in deep search mode"
             },
             404: {
                 "description": "No specifications found matching the criteria"
             }
         })
def search_specifications(request: KeywordRequest):
    start_time = time.time()
    boolSensitiveCase = request.case_sensitive
    search_mode = request.search_mode
    source = request.source
    spec_metadatas = spec_metadatas_3gpp if source == "3GPP" else spec_metadatas_etsi if source == "ETSI" else spec_metadatas_3gpp + spec_metadatas_etsi
    spec_type = request.spec_type
    keywords = [string.lower() if not boolSensitiveCase else string for string in request.keywords.split(",")]
    print(keywords)
    unique_specs = set()
    results = []
    
    if keywords == [""] and search_mode == "deep":
        raise HTTPException(status_code=400, detail="You must enter keywords in deep search mode !")
    
    for spec in spec_metadatas:
        valid = False
        if spec['id'] in unique_specs: continue
        if spec.get('type', None) is None or (spec_type is not None and spec["type"] != spec_type): continue
        if search_mode == "deep":
            contents = []
            doc = get_document(spec["id"], spec["title"], source)
            docValid = len(doc) > 1
        
        if request.mode == "and":
            string = f"{spec['id']}+-+{spec['title']}+-+{spec['type']}+-+{spec['version']}"
            if all(keyword in (string.lower() if not boolSensitiveCase else string) for keyword in keywords):
                valid = True
            if search_mode == "deep":
                if docValid:
                    for x in range(1, len(doc) - 1, 2):
                        section_title = doc[x]
                        section_content = doc[x+1]
                        if "reference" not in section_title.lower() and "void" not in section_title.lower() and "annex" not in section_content.lower():
                            if all(keyword in (section_content.lower() if not boolSensitiveCase else section_content) for keyword in keywords):
                                valid = True
                                contents.append({section_title: section_content})
        elif request.mode == "or":
            string = f"{spec['id']}+-+{spec['title']}+-+{spec['type']}+-+{spec['version']}"
            if any(keyword in (string.lower() if not boolSensitiveCase else string) for keyword in keywords):
                valid = True
            if search_mode == "deep":
                if docValid:
                    for x in range(1, len(doc) - 1, 2):
                        section_title = doc[x]
                        section_content = doc[x+1]
                        if "reference" not in section_title.lower() and "void" not in section_title.lower() and "annex" not in section_content.lower():
                            if any(keyword in (section_content.lower() if not boolSensitiveCase else section_content) for keyword in keywords):
                                valid = True
                                contents.append({section_title: section_content})
        if valid:
            spec_content = spec
            if search_mode == "deep":
                spec_content["contains"] = {k: v for d in contents for k, v in d.items()}
            results.append(spec_content)
        else:
            unique_specs.add(spec['id'])
    
    if len(results) > 0:
        return KeywordResponse(
            results=results,
            search_time=time.time() - start_time
        )
    else:
        raise HTTPException(status_code=404, detail="Specifications not found")
    
@app.post("/search/bm25", response_model=KeywordResponse, tags=["Content Search"], summary="Advanced BM25 search with relevance scoring", responses={
             200: {
                 "description": "BM25 search completed successfully"
             },
             404: {
                 "description": "No specifications found above the relevance threshold"
             }
         })
def bm25_search_specification(request: BM25KeywordRequest):
    start_time = time.time()
    source = request.source
    spec_type = request.spec_type
    threshold = request.threshold
    query = request.keywords

    results_out = []
    query_tokens = bm25s.tokenize(query)
    if source == "3GPP":
        results, scores = bm25_index_3gpp.retrieve(query_tokens, k=len(bm25_index_3gpp.corpus))
    elif source == "ETSI":
        results, scores = bm25_index_etsi.retrieve(query_tokens, k=len(bm25_index_etsi.corpus))
    else:
        print(len(bm25_index_3gpp.corpus), len(bm25_index_etsi.corpus))
        results1, scores1 = bm25_index_3gpp.retrieve(query_tokens, k=len(bm25_index_3gpp.corpus))
        results2, scores2 = bm25_index_etsi.retrieve(query_tokens, k=len(bm25_index_etsi.corpus))
        results = np.concatenate([results1, results2], axis=1)
        scores = np.concatenate([scores1, scores2], axis=1)

    def calculate_boosted_score(metadata, score, query):
        title = set(metadata['title'].lower().split())
        q = set(query.lower().split())
        spec_id_presence = 0.5 if metadata['id'].lower() in q else 0
        booster = len(q & title) * 0.5
        return score + spec_id_presence + booster

    spec_scores = {}
    spec_indices = {}
    spec_details = {}

    for i in range(results.shape[1]):
        doc = results[0, i]
        score = scores[0, i]
        spec = doc["metadata"]["id"]

        boosted_score = calculate_boosted_score(doc['metadata'], score, query)

        if spec not in spec_scores or boosted_score > spec_scores[spec]:
            spec_scores[spec] = boosted_score
            spec_indices[spec] = i
            spec_details[spec] = {
                'original_score': score,
                'boosted_score': boosted_score,
                'doc': doc
            }

    def normalize_scores(scores_dict):
        if not scores_dict:
            return {}
        
        scores_array = np.array(list(scores_dict.values())).reshape(-1, 1)
        scaler = MinMaxScaler()
        normalized_scores = scaler.fit_transform(scores_array).flatten()
        
        normalized_dict = {}
        for i, spec in enumerate(scores_dict.keys()):
            normalized_dict[spec] = normalized_scores[i]
        
        return normalized_dict

    normalized_scores = normalize_scores(spec_scores)

    for spec in spec_details:
        spec_details[spec]["normalized_score"] = normalized_scores[spec]

    unique_specs = sorted(normalized_scores.keys(), key=lambda x: normalized_scores[x], reverse=True)
    
    for rank, spec in enumerate(unique_specs, 1):
        details = spec_details[spec]
        metadata = details['doc']['metadata']
        if metadata.get('type', None) is None or (spec_type is not None and metadata["type"] != spec_type):
            continue
        if details['normalized_score'] < threshold / 100:
            break
        results_out.append(metadata)
    
    if len(results_out) > 0:
        return KeywordResponse(
            results=results_out,
            search_time=time.time() - start_time
        )
    else:
        raise HTTPException(status_code=404, detail="Specifications not found")