Spaces:
Paused
Paused
File size: 2,379 Bytes
e0cb8aa 2bc4c3f e0cb8aa 2bc4c3f e0cb8aa 2bc4c3f e0cb8aa 0c85842 e0cb8aa 2bc4c3f e0cb8aa 2bc4c3f 0c85842 2bc4c3f e0cb8aa 2bc4c3f e0cb8aa 2bc4c3f 0c85842 e0cb8aa 0c85842 2bc4c3f e0cb8aa |
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 |
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import httpx
from bs4 import BeautifulSoup
import asyncio
from datetime import datetime, timedelta
app = FastAPI()
class Notice(BaseModel):
name: str
link: str
# URL of the notice list page
url = 'https://www.durguniversity.ac.in/index.php/Home/Noticelist'
# Cache to store notices and the last updated timestamp
cache = {
"notices": [],
"last_updated": None
}
CACHE_TTL = timedelta(minutes=5) # Cache Time-to-Live (5 minutes)
async def fetch_notices():
"""Fetch notices from the website and update the cache."""
async with httpx.AsyncClient() as client:
response = await client.get(url)
if response.status_code != 200:
print(f"Failed to retrieve content. Status code: {response.status_code}")
return
# Parse the HTML content using a faster parser like 'lxml'
soup = BeautifulSoup(response.text, 'lxml')
rows = soup.select('td a') # Directly target <a> tags inside <td>
# Process notices
notice_list = []
seen_links = set()
for link_tag in rows:
notice_link = link_tag.get('href', '').strip()
notice_name = link_tag.text.strip()
if notice_link and notice_link not in seen_links:
seen_links.add(notice_link)
notice_name = notice_name or f"Notice {len(notice_list) + 1}"
notice_list.append({'name': notice_name, 'link': notice_link})
# Update the cache
cache["notices"] = notice_list[:50] # Store top 50 notices
cache["last_updated"] = datetime.utcnow()
print("Cache updated.")
@app.on_event("startup")
async def startup_event():
"""Schedule the cache refresh task on application startup."""
asyncio.create_task(cache_refresh_task())
async def cache_refresh_task():
"""Background task to refresh cache periodically."""
while True:
await fetch_notices()
await asyncio.sleep(CACHE_TTL.total_seconds())
@app.get("/notices", response_model=list[Notice])
async def get_notices():
"""Serve notices from the cache."""
if cache["last_updated"] and datetime.utcnow() - cache["last_updated"] < CACHE_TTL:
# Serve cached data if it's still valid
return cache["notices"]
# If cache is expired or unavailable, fetch notices live
await fetch_notices()
return cache["notices"]
|