import os from fastapi import FastAPI, Query, HTTPException from sqlalchemy import create_engine, text from typing import Optional DB_URI = os.getenv("DB_URI", "postgresql://postgres:password@localhost:5432/postgres") engine = create_engine(DB_URI) app = FastAPI( title="Shamela4 Library API", description="REST API to query the Shamela4 dataset including categories, authors, books, and pages.", version="1.0.0" ) def fetch_data(query: str, params: dict = {}): try: with engine.connect() as conn: result = conn.execute(text(query), params) return [dict(row._mapping) for row in result] except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def fetch_one(query: str, params: dict = {}): try: with engine.connect() as conn: result = conn.execute(text(query), params).fetchone() if result: return dict(result._mapping) return None except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/stats", tags=["Stats"]) def get_statistics(): """Get database statistics (counts).""" try: with engine.connect() as conn: books_count = conn.execute(text("SELECT COUNT(book_id) FROM books")).scalar() or 0 authors_count = conn.execute(text("SELECT COUNT(author_id) FROM authors")).scalar() or 0 categories_count = conn.execute(text("SELECT COUNT(category_id) FROM categories")).scalar() or 0 pages_count = conn.execute(text("SELECT COUNT(page_id) FROM pages")).scalar() or 0 synced_books = conn.execute(text("SELECT COUNT(DISTINCT book_id) FROM pages")).scalar() or 0 return { "total_categories": categories_count, "total_authors": authors_count, "total_books": books_count, "synced_books": synced_books, "total_pages_downloaded": pages_count, "sync_percentage": round((synced_books / books_count * 100), 2) if books_count > 0 else 0 } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/categories", tags=["Categories"]) def get_categories(limit: int = Query(50, le=1000), offset: int = 0): """Get list of categories.""" query = "SELECT category_id, category_name_ar, name_en, sort_order FROM categories ORDER BY sort_order LIMIT :limit OFFSET :offset" return fetch_data(query, {"limit": limit, "offset": offset}) @app.get("/api/categories/{category_id}", tags=["Categories"]) def get_category_detail(category_id: int): """Get details of a specific category.""" query = "SELECT category_id, category_name_ar, name_en, sort_order FROM categories WHERE category_id = :category_id" category = fetch_one(query, {"category_id": category_id}) if not category: raise HTTPException(status_code=404, detail="Category not found") return category @app.get("/api/authors", tags=["Authors"]) def get_authors(search: Optional[str] = None, limit: int = Query(50, le=1000), offset: int = 0): """Get list of authors. Optionally search by name.""" if search: query = "SELECT author_id, name_ar, death_hijri FROM authors WHERE name_ar LIKE :search ORDER BY death_hijri LIMIT :limit OFFSET :offset" return fetch_data(query, {"search": f"%{search}%", "limit": limit, "offset": offset}) else: query = "SELECT author_id, name_ar, death_hijri FROM authors ORDER BY death_hijri LIMIT :limit OFFSET :offset" return fetch_data(query, {"limit": limit, "offset": offset}) @app.get("/api/authors/{author_id}", tags=["Authors"]) def get_author_detail(author_id: int): """Get details of a specific author and their books.""" query = "SELECT * FROM authors WHERE author_id = :author_id" author = fetch_one(query, {"author_id": author_id}) if not author: raise HTTPException(status_code=404, detail="Author not found") books_query = "SELECT book_id, title_ar, category_id, volume_count_observed FROM books WHERE main_author_id = :author_id ORDER BY book_id" author['books'] = fetch_data(books_query, {"author_id": author_id}) return author @app.get("/api/books", tags=["Books"]) def get_books(search: Optional[str] = None, category_id: Optional[int] = None, author_id: Optional[int] = None, limit: int = Query(50, le=1000), offset: int = 0): """Get list of books. Filter by category, author, or search keyword.""" where_clauses = [] params = {"limit": limit, "offset": offset} if search: where_clauses.append("(title_ar LIKE :search OR main_author_name_ar LIKE :search)") params["search"] = f"%{search}%" if category_id: where_clauses.append("category_id = :category_id") params["category_id"] = category_id if author_id: where_clauses.append("main_author_id = :author_id") params["author_id"] = author_id where_str = " WHERE " + " AND ".join(where_clauses) if where_clauses else "" query = f"SELECT book_id, title_ar, category_id, main_author_id, main_author_name_ar, main_author_death_hijri, volume_count_observed, version_major FROM books {where_str} ORDER BY book_id LIMIT :limit OFFSET :offset" return fetch_data(query, params) @app.get("/api/books/{book_id}", tags=["Books"]) def get_book_detail(book_id: int): """Get details of a specific book, including synopsis and TOC.""" query = "SELECT * FROM books WHERE book_id = :book_id" book = fetch_one(query, {"book_id": book_id}) if not book: raise HTTPException(status_code=404, detail="Book not found") # Get TOC (Daftar Isi) if available toc_query = "SELECT title_id, title_text, page_id, parent_id FROM toc WHERE book_id = :book_id ORDER BY title_id" book['toc'] = fetch_data(toc_query, {"book_id": book_id}) # Get total pages pages_query = "SELECT COUNT(*) as total_pages FROM pages WHERE book_id = :book_id" pages_result = fetch_one(pages_query, {"book_id": book_id}) book['total_pages'] = pages_result['total_pages'] if pages_result else 0 return book @app.get("/api/books/{book_id}/pages", tags=["Pages"]) def get_book_pages(book_id: int, limit: int = Query(100, le=500), offset: int = 0): """Get pages of a specific book, ordered by sequence_num.""" query = "SELECT * FROM pages WHERE book_id = :book_id ORDER BY sequence_num LIMIT :limit OFFSET :offset" return fetch_data(query, {"book_id": book_id, "limit": limit, "offset": offset}) @app.get("/api/search", tags=["Search"]) def search_text(query: str, limit: int = Query(20, le=100), offset: int = 0): """Full-text search inside book pages.""" # We use PostgreSQL to_tsquery for full-text search. The body column has a GIN index on to_tsvector('arabic', body). # We format the query string to join words with & (AND) for tsquery words = [w for w in query.split() if w.strip()] if not words: return [] tsquery = " & ".join(words) sql = """ SELECT p.page_id, p.book_id, p.page_num, p.sequence_num, ts_headline('arabic', p.body, to_tsquery('arabic', :tsquery)) as highlight, b.title_ar as book_title FROM pages p JOIN books b ON p.book_id = b.book_id WHERE to_tsvector('arabic', p.body) @@ to_tsquery('arabic', :tsquery) ORDER BY p.book_id, p.sequence_num LIMIT :limit OFFSET :offset """ return fetch_data(sql, {"tsquery": tsquery, "limit": limit, "offset": offset})