from __future__ import annotations import re from datetime import date, datetime from typing import Any from urllib.parse import urlparse import gradio as gr from fastapi import Body from fastapi.responses import HTMLResponse from pydantic import BaseModel, ValidationError from src.dataset_store import DatasetStore, DatasetStoreError from src.llm_client import LLMClient, LLMError from src.olostep_client import OlostepClient, OlostepError from src.schemas import Opportunity, ResearchResult, UserProfile class ResearchRequest(BaseModel): origin: str applying_for: str field_of_study: str preferences: str = "" class SaveRequest(BaseModel): session_id: str opportunity_title: str STORE = DatasetStore() OLOSTEP = OlostepClient() RANKER = LLMClient() SESSIONS: dict[str, ResearchResult] = {} def build_app() -> gr.Server: app = gr.Server() @app.api(name="research") def research_api(profile: dict[str, Any]) -> dict[str, Any]: return research_payload(ResearchRequest(**profile)) @app.api(name="save") def save_api(session_id: str, opportunity_title: str) -> dict[str, Any]: return save_payload(SaveRequest(session_id=session_id, opportunity_title=opportunity_title)) @app.get("/", response_class=HTMLResponse) async def homepage() -> str: return FRONTEND_HTML @app.get("/health") async def health() -> dict[str, str]: return {"status": "ok"} @app.get("/custom/config") async def config() -> dict[str, Any]: return config_status() @app.get("/custom/history") async def history(q: str = "") -> dict[str, Any]: if q and q.strip(): return {"records": search_history_records(q.strip()), "query": q.strip()} return {"records": history_records()} @app.post("/custom/research") async def research(request: ResearchRequest = Body(...)) -> dict[str, Any]: return research_payload(request) @app.post("/custom/save") async def save(request: SaveRequest = Body(...)) -> dict[str, Any]: return save_payload(request) return app def research_payload(request: ResearchRequest) -> dict[str, Any]: try: profile = UserProfile(**request.model_dump()) except ValidationError as exc: return { "ok": False, "message": f"Input error: {exc.errors()[0]['msg']}", "result": None, "history": history_records(), } errors: list[str] = [] live_research_enabled = OLOSTEP.enabled and RANKER.enabled if not live_research_enabled: return { "ok": False, "message": "Live research is unavailable. Configure OLOSTEP_API_KEY and HF_TOKEN to search real funding sources.", "result": None, "history": history_records(), } try: pages = OLOSTEP.search(profile, limit=15) except OlostepError as exc: errors.append(str(exc)) pages = [] opportunities: list[Opportunity] = [] if pages: base = opportunities_from_sources(profile, pages) ranked: list[Opportunity] = [] if RANKER.enabled: try: ranked = RANKER.rank_opportunities(profile, pages[:8]) except LLMError as exc: errors.append(str(exc)) opportunities = select_providers(base, ranked) if ranked else [] if not opportunities: opportunities = filter_direct_providers(base) opportunities = enrich_opportunities(opportunities, pages) opportunities = drop_expired_opportunities(opportunities) if not opportunities: message = "No direct funding-provider pages were found for this profile. Try broadening the origin, degree, field, or preferences." if errors: message += " Service note: " + " | ".join(errors[:2]) return { "ok": False, "message": message, "result": None, "history": history_records(), } else: result = ResearchResult( profile=profile, query=profile.search_query(), opportunities=opportunities, errors=errors, ) try: STORE.record_search(result) except DatasetStoreError as exc: result.errors.append(str(exc)) SESSIONS[result.session_id] = result return { "ok": True, "message": result_status(result), "result": result.model_dump(mode="json"), "history": history_records(), } def save_payload(request: SaveRequest) -> dict[str, Any]: result = SESSIONS.get(request.session_id) if not result: return {"ok": False, "message": "Research session was not found.", "history": history_records()} opportunity = find_opportunity(result, request.opportunity_title) if not opportunity: return {"ok": False, "message": "Selected opportunity was not found.", "history": history_records()} try: STORE.save_opportunity(result, opportunity) message = f"Saved: {opportunity.title}" except DatasetStoreError as exc: message = f"Saved locally, but dataset write failed: {exc}" return { "ok": True, "message": message, "opportunity": opportunity.model_dump(mode="json"), "history": history_records(), } def find_opportunity(result: ResearchResult, title: str) -> Opportunity | None: return next((opportunity for opportunity in result.opportunities if opportunity.title == title), None) _MONTHS = ( r"(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|" r"Jul(?:y)?|Aug(?:ust)?|Sep(?:t(?:ember)?)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)" ) _DATE_PATTERN = ( r"\d{1,2}\s+" + _MONTHS + r"\.?\s+\d{4}" r"|" + _MONTHS + r"\.?\s+\d{1,2},?\s+\d{4}" r"|\d{4}-\d{2}-\d{2}" r"|\d{1,2}/\d{1,2}/\d{2,4}" r"|\d{1,2}[\s-]" + _MONTHS + r"[\s-]\d{4}" r"|(?:Spring|Summer|Fall|Autumn|Winter)\s+\d{4}" ) _MONEY_PATTERN = re.compile( r"[$€£]\s?\d[\d,]*(?:\.\d+)?\s?(?:thousand|million|mill|k|m|bn)?" r"(?:\s?/\s?(?:year|yr|month|mo|annum|day|week))?" r"|(?:USD|EUR|GBP|CA\$|AU\$)\s?\d[\d,]*(?:\.\d+)?\s?(?:thousand|million)?" r"(?:\s?(?:per|/)\s?(?:year|month|annum))?" r"|(?:up to|worth|award(?:ed)? of|prize of|stipend of|value of|grant(?:s)? of|" r"scholarship of|fellowship of|totaling|totalling|maximum of|amount of)\s+[$€£]?\s?\d[\d,]*(?:\.\d+)?" r"\s?(?:thousand|million|k|m)?" r"|[$€£]\s?\d[\d,]*(?:\.\d+)?\s?(?:to|[\-–—])\s?[$€£]?\s?\d[\d,]*" r"|full[\s-]?(?:tuition|ride|scholarship|funding|coverage)", re.IGNORECASE, ) _DEADLINE_KEYWORDS = ( r"(?:deadline|apply by|application(?:s)?\s+(?:close|end|due|deadline|open)|" r"closes?\s+(?:on|date)?|due(?:\s+date|s)?|submit(?:ted)?\s+by|submission\s+deadline|" r"closing\s+date|last\s+date|expires?|next\s+deadline|final\s+deadline|deadline\s+for|" r"applications?\s+(?:will\s+close|must\s+be\s+received))" ) _DEADLINE_PATTERN = re.compile( _DEADLINE_KEYWORDS + r"[^A-Za-z0-9]{0,28}(" + _DATE_PATTERN + r")", re.IGNORECASE, ) _ROLLING_PATTERN = re.compile( r"\brolling\s+(?:deadline|admissions?|applications?|basis|review)\b|\brolling\b", re.IGNORECASE, ) _FUTURE_YEAR = re.compile(r"20(?:2[4-9]|3[0-1])") _ANY_DATE = re.compile(_DATE_PATTERN, re.IGNORECASE) def _clean(value: str) -> str: return re.sub(r"\s+", " ", value).strip().rstrip(".,;:") def _extract_funding(text: str) -> str | None: if not text: return None match = _MONEY_PATTERN.search(text) return _clean(match.group(0)) if match else None def _extract_deadline(text: str) -> str | None: if not text: return None if _ROLLING_PATTERN.search(text): return "Rolling" match = _DEADLINE_PATTERN.search(text) if match: return _clean(match.group(1)) for date_match in _ANY_DATE.finditer(text): candidate = date_match.group(0) if _FUTURE_YEAR.search(candidate): return _clean(candidate) return None def _url_key(url: Any) -> str: parsed = urlparse(str(url)) host = (parsed.netloc or "").lower() host = host[4:] if host.startswith("www.") else host return f"{host}{(parsed.path or '').rstrip('/')}".lower() def _parse_deadline_date(deadline: str) -> date | None: if not deadline or deadline.strip().lower() in ("rolling", "not listed", "open", "ongoing", ""): return None candidates = [deadline.strip(), *(match.group(0) for match in _ANY_DATE.finditer(deadline))] for candidate in candidates: normalized = re.sub(r"(?<=\d)(st|nd|rd|th)\b", "", candidate, flags=re.IGNORECASE) for fmt in ( "%B %d, %Y", "%b %d, %Y", "%Y-%m-%d", "%d %B %Y", "%d %b %Y", "%m/%d/%Y", "%d/%m/%Y", "%m/%d/%y", "%d/%m/%y", ): try: return datetime.strptime(normalized, fmt).date() except ValueError: continue return None def drop_expired_opportunities(opportunities: list[Opportunity]) -> list[Opportunity]: """Drop clearly closed or expired opportunities; retain unclear deadlines.""" today = date.today() alive: list[Opportunity] = [] for opp in opportunities: status = opp.deadline.strip().lower() if re.search(r"\b(closed|expired|deadline passed|applications? ended)\b", status): continue d = _parse_deadline_date(opp.deadline) if d is None or d >= today: alive.append(opp) return alive def _host_of(url: Any) -> str: host = (urlparse(str(url)).netloc or "").lower() return host[4:] if host.startswith("www.") else host def enrich_opportunities(opportunities: list[Opportunity], pages: list[Any]) -> list[Opportunity]: """Fill in missing funding amount / deadline from the scraped source content.""" by_key = {_url_key(page.url): page for page in pages} by_host: dict[str, Any] = {} for page in pages: by_host.setdefault(_host_of(page.url), page) enriched: list[Opportunity] = [] for opportunity in opportunities: page = by_key.get(_url_key(opportunity.source_url)) or by_host.get(_host_of(opportunity.source_url)) text = "" if page: text = " ".join(part for part in (getattr(page, "snippet", ""), getattr(page, "content", "")) if part) updates: dict[str, Any] = {} if opportunity.funding_amount == "Not listed": found = _extract_funding(text) if found: updates["funding_amount"] = found if opportunity.deadline == "Not listed": found = _extract_deadline(text) if found: updates["deadline"] = found enriched.append(opportunity.model_copy(update=updates) if updates else opportunity) return enriched def select_providers( base: list[Opportunity], ranked: list[Opportunity] ) -> list[Opportunity]: """Layer AI analysis onto sources, retaining credible sources the AI omitted.""" by_key = {_url_key(opp.source_url): opp for opp in ranked} by_host: dict[str, Opportunity] = {} for opp in ranked: by_host.setdefault(_host_of(opp.source_url), opp) selected: list[Opportunity] = [] for opp in base: if _is_aggregator(opp.source_url, opp.title): continue match = by_key.get(_url_key(opp.source_url)) or by_host.get(_host_of(opp.source_url)) if match is None: if _has_funding_evidence(opp): selected.append(opp) continue updates: dict[str, Any] = {} if isinstance(match.match_score, int) and match.match_score: updates["match_score"] = match.match_score for field in ("why_it_fits", "eligibility_summary"): value = getattr(match, field, "") if value and value.strip(): updates[field] = value if match.concerns: updates["concerns"] = list(match.concerns) if match.funding_amount != "Not listed": updates["funding_amount"] = match.funding_amount if match.deadline != "Not listed": updates["deadline"] = match.deadline selected.append(opp.model_copy(update=updates) if updates else opp) return selected def _has_funding_evidence(opportunity: Opportunity) -> bool: text = " ".join( ( opportunity.title, str(opportunity.source_url), opportunity.eligibility_summary, ) ).lower() return any( term in text for term in ("scholarship", "grant", "fellowship", "financial aid", "funding", "stipend") ) _AGGREGATOR_PARTS = ( "portal.com", "mastersportal", "scholarshipportal", "bachelorsportal", "phdportal", "studyportals", "shortcoursesportal", "distancelearningportal", "abroadin.com", "scholarships.com", "scholarship.com", "fastweb.com", "cappex.com", "niche.com", "goingmerry.com", "scholarshipdb", "scholars4dev", "scholarshipaz", "scholarshippositions", "indeed.com", "glassdoor.com", "jooble.org", "simplyhired.com", "ziprecruiter.com", "monster.com", "medium.com", "wordpress.com", "blogspot.com", "quora.com", "youtube.com", "wemakescholars.com", "computerscience.org", "timesconsultant.com", "tiktok.com", "facebook.com", "instagram.com", "linkedin.com", "reddit.com", "pinterest.com", "twitter.com", "x.com", "wikipedia.org", "wikihow.com", ) _GENERIC_SCHOLARSHIP_TITLES = ( "top scholarships", "best scholarships", "scholarships list", "study scholarships in", "scholarships for students", "scholarships for pakistani", ) def _is_aggregator(url: Any, title: str = "") -> bool: text = str(url).lower() if any(part in text for part in _AGGREGATOR_PARTS): return True parsed = urlparse(text) path = parsed.path or "" if "/search" in path and "scholarship" in path: return True if "/list-of-" in path or "/list_of" in path: return True host = parsed.netloc.removeprefix("www.") institutional = any(marker in host for marker in (".edu", ".gov", ".ac.")) if not institutional and any(phrase in title.lower() for phrase in _GENERIC_SCHOLARSHIP_TITLES): return True return False def filter_direct_providers(opportunities: list[Opportunity]) -> list[Opportunity]: """Heuristic fallback: drop obvious aggregators / boards / listing pages.""" return [opp for opp in opportunities if not _is_aggregator(opp.source_url, opp.title)] def opportunities_from_sources(profile: UserProfile, pages: list[Any]) -> list[Opportunity]: opportunities: list[Opportunity] = [] for index, page in enumerate(pages[:15]): hostname = urlparse(str(page.url)).hostname or "Funding provider" organization = hostname.removeprefix("www.").lower() summary_source = page.snippet or page.content[:260] eligibility_summary = ( summary_source.strip() if summary_source.strip() else "Review the provider page for eligibility requirements, deadlines, award amount, and application rules." ) try: opportunities.append( Opportunity( title=page.title or f"Funding opportunity from {organization}", organization=organization, source_url=page.url, deadline="Not listed", funding_amount="Not listed", eligibility_region="Not listed", match_score=max(55, 82 - (index * 5)), eligibility_summary=eligibility_summary[:500], why_it_fits=( "This is a real web result returned for the submitted origin, target degree, field, and funding search context. " "Open the source to verify exact fit before applying." ), concerns=[ "The source is relevant to the search, but deadline, amount, and eligibility must be verified on the provider page." ], ) ) except ValueError: continue return opportunities def result_status(result: ResearchResult) -> str: message = f"Live research: found {len(result.opportunities)} opportunities for {result.profile.field_of_study}." if result.errors: message += " Showing real source results; verify details on provider websites." return message def history_records() -> list[dict[str, Any]]: records = [ record for record in STORE.recent_records(limit=100) if not is_demo_history_record(record.source_url, record.opportunity_title) ] return [record.model_dump(mode="json") for record in _top_result_per_session(records)[-20:]] def search_history_records(query: str) -> list[dict[str, Any]]: records = [ record for record in STORE.search_records(query, limit=50) if not is_demo_history_record(record.source_url, record.opportunity_title) ] return [record.model_dump(mode="json") for record in _top_result_per_session(records)[-50:]] def _top_result_per_session(records: list[Any]) -> list[Any]: """Return one highest-scoring search result per session, preserving session order.""" top_by_session: dict[str, Any] = {} order: list[str] = [] for record in records: if record.session_id not in top_by_session: top_by_session[record.session_id] = record order.append(record.session_id) elif record.match_score > top_by_session[record.session_id].match_score: top_by_session[record.session_id] = record return [top_by_session[session_id] for session_id in order] def is_demo_history_record(source_url: str, title: str) -> bool: text = f"{source_url} {title}".lower() parsed = urlparse(source_url) host = parsed.netloc.lower() demo_hosts = ("example.org", "example.com", "example.edu", "demo.", "localhost") return "demo" in text or any(part in host for part in demo_hosts) def config_status() -> dict[str, Any]: missing: list[str] = [] if not OLOSTEP.enabled: missing.append("OLOSTEP_API_KEY") if not RANKER.enabled: missing.append("HF_TOKEN") if not STORE.enabled: missing.extend(["HF_TOKEN", "HF_DATASET_REPO"]) return { "live_research_enabled": OLOSTEP.enabled and RANKER.enabled, "dataset_storage_enabled": STORE.enabled, "missing": missing, } FRONTEND_HTML = r""" ScholarScope

ScholarScope

AI funding research

Ready

Research Profile

Quick start

Ranked Opportunities

"""