246 lines
9.2 KiB
Python
246 lines
9.2 KiB
Python
"""
|
|
Hadith endpoints — details, listing, search by keyword/narrator/topic/place.
|
|
"""
|
|
from fastapi import APIRouter, Query, HTTPException
|
|
from typing import Optional
|
|
|
|
from app.services.database import db
|
|
from app.models.schemas import (
|
|
HadithDetail, HadithSummary, NarratorInChain, TopicTag,
|
|
PaginatedResponse, PaginationMeta,
|
|
)
|
|
|
|
router = APIRouter(prefix="/hadiths", tags=["Hadiths"])
|
|
|
|
|
|
@router.get("/{hadith_id}", response_model=HadithDetail,
|
|
summary="Get hadith by ID",
|
|
description="Retrieve full hadith details including Arabic text, sanad/matn separation, "
|
|
"ordered narrator chain from the knowledge graph, and topic tags.")
|
|
async def get_hadith(hadith_id: str):
|
|
"""Get full hadith details by ID, including narrator chain and topics from Neo4j."""
|
|
|
|
# Base hadith from PostgreSQL
|
|
hadith = db.pg_query_one("""
|
|
SELECT h.id, c.name_english AS collection, h.hadith_number,
|
|
h.grade, h.arabic_text, h.sanad, h.matn
|
|
FROM hadiths h
|
|
JOIN collections c ON c.id = h.collection_id
|
|
WHERE h.id = %s
|
|
""", (hadith_id,))
|
|
|
|
if not hadith:
|
|
raise HTTPException(status_code=404, detail="Hadith not found")
|
|
|
|
# Enrich with chain + topics from Neo4j
|
|
chain = db.neo4j_query("""
|
|
MATCH (n:Narrator)-[r:APPEARS_IN]->(h:Hadith {id: $hid})
|
|
RETURN n.name_arabic AS name_arabic,
|
|
n.name_transliterated AS name_transliterated,
|
|
n.entity_type AS entity_type,
|
|
r.chain_order AS order,
|
|
r.transmission_verb AS transmission_verb
|
|
ORDER BY r.chain_order
|
|
""", {"hid": hadith_id})
|
|
|
|
topics = db.neo4j_query("""
|
|
MATCH (h:Hadith {id: $hid})-[:HAS_TOPIC]->(t:Topic)
|
|
RETURN t.topic_arabic AS topic_arabic,
|
|
t.topic_english AS topic_english,
|
|
t.category AS category
|
|
""", {"hid": hadith_id})
|
|
|
|
return HadithDetail(
|
|
id=str(hadith["id"]),
|
|
collection=hadith["collection"],
|
|
hadith_number=hadith["hadith_number"],
|
|
grade=hadith["grade"],
|
|
arabic_text=hadith["arabic_text"],
|
|
sanad_text=hadith.get("sanad"),
|
|
matn_text=hadith.get("matn"),
|
|
narrator_chain=[NarratorInChain(**c) for c in chain],
|
|
topics=[TopicTag(**t) for t in topics],
|
|
)
|
|
|
|
|
|
@router.get("/collection/{collection_name}", response_model=PaginatedResponse,
|
|
summary="List hadiths by collection",
|
|
description="Paginated listing of hadiths in a specific collection. "
|
|
"Collection names use partial matching (e.g. 'bukhari' matches 'Sahih Bukhari').")
|
|
async def list_by_collection(
|
|
collection_name: str = Field(description="Collection name (partial match). Examples: bukhari, muslim, tirmidhi, abudawud"),
|
|
page: int = Query(1, ge=1, description="Page number"),
|
|
per_page: int = Query(20, ge=1, le=100, description="Results per page"),
|
|
):
|
|
"""List hadiths in a collection with pagination."""
|
|
offset = (page - 1) * per_page
|
|
|
|
total_row = db.pg_query_one("""
|
|
SELECT COUNT(*) AS total
|
|
FROM hadiths h
|
|
JOIN collections c ON c.id = h.collection_id
|
|
WHERE c.name_english ILIKE %s
|
|
""", (f"%{collection_name}%",))
|
|
total = total_row["total"] if total_row else 0
|
|
|
|
rows = db.pg_query("""
|
|
SELECT h.id, c.name_english AS collection, h.hadith_number,
|
|
h.grade, LEFT(h.arabic_text, 300) AS arabic_text
|
|
FROM hadiths h
|
|
JOIN collections c ON c.id = h.collection_id
|
|
WHERE c.name_english ILIKE %s
|
|
ORDER BY h.hadith_number
|
|
LIMIT %s OFFSET %s
|
|
""", (f"%{collection_name}%", per_page, offset))
|
|
|
|
return PaginatedResponse(
|
|
meta=PaginationMeta(
|
|
total=total, page=page, per_page=per_page,
|
|
pages=(total + per_page - 1) // per_page,
|
|
),
|
|
data=[HadithSummary(
|
|
id=str(r["id"]), collection=r["collection"],
|
|
hadith_number=r["hadith_number"], grade=r["grade"],
|
|
arabic_text=r["arabic_text"],
|
|
) for r in rows],
|
|
)
|
|
|
|
|
|
@router.get("/number/{collection_name}/{number}", response_model=HadithDetail)
|
|
async def get_by_number(collection_name: str, number: int):
|
|
"""Get a hadith by collection name and number."""
|
|
hadith = db.pg_query_one("""
|
|
SELECT h.id
|
|
FROM hadiths h
|
|
JOIN collections c ON c.id = h.collection_id
|
|
WHERE c.name_english ILIKE %s AND h.hadith_number = %s
|
|
""", (f"%{collection_name}%", number))
|
|
|
|
if not hadith:
|
|
raise HTTPException(status_code=404, detail=f"Hadith #{number} not found in {collection_name}")
|
|
|
|
return await get_hadith(str(hadith["id"]))
|
|
|
|
|
|
@router.get("/search/keyword", response_model=PaginatedResponse,
|
|
summary="Search hadiths by Arabic keyword",
|
|
description="Full-text keyword search across all hadith Arabic text. "
|
|
"Supports both vocalized (مَكَّةَ) and unvocalized (مكة) Arabic.")
|
|
async def search_by_keyword(
|
|
q: str = Query(
|
|
..., min_length=2,
|
|
description="Arabic keyword to search. Examples: صلاة (prayer), زكاة (zakat), صيام (fasting), حج (hajj), نية (intention)",
|
|
examples=["صلاة", "الجنة", "رمضان"],
|
|
),
|
|
collection: Optional[str] = Query(
|
|
None,
|
|
description="Filter by collection name. Examples: Sahih Bukhari, Sahih Muslim, Sunan Abu Dawood",
|
|
examples=["Sahih Bukhari"],
|
|
),
|
|
grade: Optional[str] = Query(
|
|
None,
|
|
description="Filter by hadith grade. Examples: Sahih, Hasan, Da'if",
|
|
examples=["Sahih"],
|
|
),
|
|
page: int = Query(1, ge=1, description="Page number (1-indexed)"),
|
|
per_page: int = Query(20, ge=1, le=100, description="Results per page (max 100)"),
|
|
):
|
|
"""Search hadiths by Arabic keyword in text."""
|
|
offset = (page - 1) * per_page
|
|
|
|
conditions = ["h.arabic_text ILIKE %s"]
|
|
params = [f"%{q}%"]
|
|
|
|
if collection:
|
|
conditions.append("c.name_english ILIKE %s")
|
|
params.append(f"%{collection}%")
|
|
if grade:
|
|
conditions.append("h.grade ILIKE %s")
|
|
params.append(f"%{grade}%")
|
|
|
|
where = " AND ".join(conditions)
|
|
|
|
total_row = db.pg_query_one(f"""
|
|
SELECT COUNT(*) AS total
|
|
FROM hadiths h
|
|
JOIN collections c ON c.id = h.collection_id
|
|
WHERE {where}
|
|
""", tuple(params))
|
|
total = total_row["total"] if total_row else 0
|
|
|
|
params.extend([per_page, offset])
|
|
rows = db.pg_query(f"""
|
|
SELECT h.id, c.name_english AS collection, h.hadith_number,
|
|
h.grade, LEFT(h.arabic_text, 300) AS arabic_text
|
|
FROM hadiths h
|
|
JOIN collections c ON c.id = h.collection_id
|
|
WHERE {where}
|
|
ORDER BY c.name_english, h.hadith_number
|
|
LIMIT %s OFFSET %s
|
|
""", tuple(params))
|
|
|
|
return PaginatedResponse(
|
|
meta=PaginationMeta(
|
|
total=total, page=page, per_page=per_page,
|
|
pages=(total + per_page - 1) // per_page,
|
|
),
|
|
data=[HadithSummary(
|
|
id=str(r["id"]), collection=r["collection"],
|
|
hadith_number=r["hadith_number"], grade=r["grade"],
|
|
arabic_text=r["arabic_text"],
|
|
) for r in rows],
|
|
)
|
|
|
|
|
|
@router.get("/search/topic/{topic}", response_model=list[HadithSummary])
|
|
async def search_by_topic(topic: str, limit: int = Query(20, ge=1, le=100)):
|
|
"""Search hadiths by topic tag (from Neo4j)."""
|
|
rows = db.neo4j_query("""
|
|
CALL db.index.fulltext.queryNodes('hadith_arabic_text', $topic)
|
|
YIELD node, score
|
|
RETURN node.id AS id,
|
|
node.collection AS collection,
|
|
node.hadith_number AS hadith_number,
|
|
node.grade AS grade,
|
|
left(node.matn_text, 300) AS matn_text,
|
|
score
|
|
ORDER BY score DESC
|
|
LIMIT $limit
|
|
""", {"topic": topic, "limit": limit})
|
|
|
|
return [HadithSummary(
|
|
id=str(r["id"]), collection=r["collection"] or "",
|
|
hadith_number=r["hadith_number"] or 0, grade=r["grade"],
|
|
matn_text=r["matn_text"],
|
|
) for r in rows]
|
|
|
|
|
|
@router.get("/search/narrator/{narrator_name}", response_model=list[HadithSummary],
|
|
summary="Find hadiths by narrator",
|
|
description="Find all hadiths where a specific narrator appears in the chain. "
|
|
"Searches both Arabic name and transliteration. "
|
|
"Example: `/hadiths/search/narrator/أبو هريرة`")
|
|
async def search_by_narrator(
|
|
narrator_name: str,
|
|
limit: int = Query(50, ge=1, le=200, description="Maximum results"),
|
|
):
|
|
"""Find all hadiths narrated by a specific person."""
|
|
rows = db.neo4j_query("""
|
|
MATCH (n:Narrator)-[r:APPEARS_IN]->(h:Hadith)
|
|
WHERE n.name_arabic CONTAINS $name
|
|
OR n.name_transliterated CONTAINS $name
|
|
RETURN h.id AS id,
|
|
h.collection AS collection,
|
|
h.hadith_number AS hadith_number,
|
|
h.grade AS grade,
|
|
left(h.matn_text, 300) AS matn_text
|
|
ORDER BY h.collection, h.hadith_number
|
|
LIMIT $limit
|
|
""", {"name": narrator_name, "limit": limit})
|
|
|
|
return [HadithSummary(
|
|
id=str(r["id"]), collection=r["collection"] or "",
|
|
hadith_number=r["hadith_number"] or 0, grade=r["grade"],
|
|
matn_text=r["matn_text"],
|
|
) for r in rows]
|