""" AISHE College Data Scraper — Gradio App ========================================= Automates downloading and building the Master Educational Database. Run locally: python app.py """ import sys import subprocess from datetime import datetime from pathlib import Path import gradio as gr import pandas as pd import os from scraper.aishe_merger import merge_and_filter from hf_store import upload_combined_excel, is_configured, list_combined_files, download_combined_file # Global state for stopping process CURRENT_PROCESS = None # Install Playwright chromium automatically if running on Hugging Face if "SPACE_ID" in os.environ: os.system("playwright install chromium") # ── Paths ───────────────────────────────────────────────────────────────────── PROJECT_ROOT = Path(__file__).parent DOWNLOADS_DIR = PROJECT_ROOT / "output" / "downloads" OUTPUT_DIR = PROJECT_ROOT / "output_excel" def _has_cached_downloads() -> bool: return DOWNLOADS_DIR.exists() and any(DOWNLOADS_DIR.glob("*.xlsx")) # ── Core pipeline ───────────────────────────────────────────────────────────── def get_output_filename() -> str: timestamp = datetime.now().strftime('%b_%Y_%H%M%S').lower() return f"aishe_colleges_{timestamp}.xlsx" def _run_download_and_build(): global CURRENT_PROCESS log = "🚀 Starting AISHE Download...\n\n" yield log, gr.update(interactive=False), gr.update(visible=True, interactive=True), gr.update(visible=False, value=None) downloader = str(PROJECT_ROOT / "scraper" / "aishe_downloader.py") process = subprocess.Popen( [sys.executable, downloader, "--output-dir", str(DOWNLOADS_DIR)], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, cwd=str(PROJECT_ROOT), ) CURRENT_PROCESS = process success_count = 0 for raw in process.stdout: line = raw.strip() if not line: continue if line.startswith("STEP:"): _, step, name = line.split(":", 2) log += f"🔄 [{step}/5] Downloading {name}...\n" elif line.startswith("DONE:"): _, step, name = line.split(":", 2) log += f"✅ [{step}/5] {name} saved.\n" success_count += 1 elif line.startswith("ERROR:"): parts = line.split(":", 3) log += f"❌ [{parts[1]}/5] {parts[2]} error: {parts[3]}\n" else: log += f" {line}\n" yield log, gr.update(interactive=False), gr.update(visible=True, interactive=True), gr.update(visible=False, value=None) process.wait() CURRENT_PROCESS = None if success_count == 0: log += "\n❌ Process stopped or all downloads failed.\n" yield log, gr.update(interactive=True), gr.update(visible=False), gr.update(visible=False, value=None) return log += f"\n📦 Successfully downloaded {success_count} out of 5 files.\n" log += "\n🔄 Formatting and grouping data into an Excel file...\n" yield log, gr.update(interactive=False), gr.update(visible=False), gr.update(visible=False, value=None) output_path = OUTPUT_DIR / get_output_filename() try: result = merge_and_filter(DOWNLOADS_DIR, output_path) if result is None: log += "⚠️ No data available.\n" yield log, gr.update(interactive=True), gr.update(visible=False), gr.update(visible=False, value=None) return log += f"✅ Combined Excel is ready: {result.name}\n" if is_configured(): log += "☁️ Saving to Hugging Face...\n" yield log, gr.update(interactive=False), gr.update(visible=False), gr.update(visible=False, value=None) hf_url = upload_combined_excel(result) log += f"☁️ Saved online -> {hf_url}\n" if hf_url else "⚠️ Failed to save online.\n" log += "\n🎉 Finished!\n" yield log, gr.update(interactive=True), gr.update(visible=False), gr.update(visible=True, value=str(result)) except Exception as exc: log += f"❌ Failed to combine data: {exc}\n" yield log, gr.update(interactive=True), gr.update(visible=False), gr.update(visible=False, value=None) def _run_naac_download(): output_lines = [] def get_ui_update(done=False): if done: return "\n".join(output_lines), gr.update(interactive=True), gr.update(visible=False) return "\n".join(output_lines), gr.update(interactive=False), gr.update(visible=True, interactive=True) output_lines.append("🚀 Starting NAAC Download...") yield get_ui_update() base_dir = Path.cwd() / "output" downloads_dir = base_dir / "downloads" try: from scraper.naac_downloader import download_naac def log_naac(msg: str): output_lines.append(msg) output_lines.append("\n⏳ Downloading NAAC Accredited Institutions...") yield get_ui_update() raw_naac_file = download_naac(output_dir=downloads_dir, headless=True, log_fn=log_naac) except Exception as e: output_lines.append(f"❌ NAAC Download error: {e}") output_lines.append("\n❌ Process stopped or download failed.") yield get_ui_update(done=True) return output_lines.append("\n⏳ Formatting NAAC data...") yield get_ui_update() try: timestamp = datetime.now().strftime("%b_%Y_%H%M%S").lower() final_filename = f"naac_colleges_{timestamp}.xlsx" final_output_path = base_dir / final_filename import shutil shutil.copy2(raw_naac_file, final_output_path) output_lines.append(f"✅ Successfully prepared NAAC data!") output_lines.append(f"💾 Saved locally as: {final_filename}") yield get_ui_update() if is_configured(): output_lines.append("\n⏳ Uploading NAAC file to Hugging Face dataset...") yield get_ui_update() try: upload_combined_excel(final_output_path) output_lines.append("✅ Upload to Hugging Face successful!") output_lines.append("🎉 All NAAC steps completed successfully!") except Exception as e: output_lines.append(f"❌ Failed to upload to Hugging Face: {e}") else: output_lines.append("\n⚠️ Hugging Face upload skipped (HF_TOKEN not configured).") output_lines.append("🎉 All NAAC steps completed successfully!") yield get_ui_update(done=True) except Exception as e: output_lines.append(f"❌ Error preparing NAAC data: {e}") yield get_ui_update(done=True) return def _run_urise_iti_download(): output_lines = [] output_lines.append("🚀 Initializing URISE UP Scraper...") yield "\n".join(output_lines), gr.update(visible=False, value=None) try: from scraper.urise_iti_scraper import get_urise_iti_data out_file = None for msg in get_urise_iti_data(OUTPUT_DIR): if isinstance(msg, Path): out_file = msg else: output_lines.append(msg) yield "\n".join(output_lines), gr.update(visible=False, value=None) if out_file: if is_configured(): output_lines.append("☁️ Saving to Hugging Face...") yield "\n".join(output_lines), gr.update(visible=False, value=None) hf_url = upload_combined_excel(out_file) output_lines.append(f"☁️ Saved online -> {hf_url}" if hf_url else "⚠️ Failed to save online.") yield "\n".join(output_lines), gr.update(visible=True, value=str(out_file)) except Exception as e: output_lines.append(f"❌ Critical Error: {e}") yield "\n".join(output_lines), gr.update(visible=False, value=None) def _run_ncvt_iti_download(): output_lines = [] output_lines.append("🚀 Initializing NCVT MIS Scraper...") yield "\n".join(output_lines), gr.update(visible=False, value=None) try: from scraper.ncvt_iti_scraper import get_ncvt_iti_data out_file = None for msg in get_ncvt_iti_data(OUTPUT_DIR): if isinstance(msg, Path): out_file = msg else: output_lines.append(msg) yield "\n".join(output_lines), gr.update(visible=False, value=None) if out_file: if is_configured(): output_lines.append("☁️ Saving to Hugging Face...") yield "\n".join(output_lines), gr.update(visible=False, value=None) hf_url = upload_combined_excel(out_file) output_lines.append(f"☁️ Saved online -> {hf_url}" if hf_url else "⚠️ Failed to save online.") yield "\n".join(output_lines), gr.update(visible=True, value=str(out_file)) except Exception as e: output_lines.append(f"❌ Critical Error: {e}") yield "\n".join(output_lines), gr.update(visible=False, value=None) try: import spaces @spaces.GPU def dummy_gpu_func(): return "OK" except ImportError: def dummy_gpu_func(): return "OK" def stop_scrape_process(): global CURRENT_PROCESS if CURRENT_PROCESS is not None: try: CURRENT_PROCESS.terminate() CURRENT_PROCESS = None except Exception: pass return "🛑 Scrape stopped by user. You can start a new download.", gr.update(visible=False), gr.update(interactive=True) def fetch_hf_files_list(): files = list_combined_files() return gr.Dropdown(choices=files, value=files[0]) def download_file_from_hf(filename: str): if not filename: return None local_path = download_combined_file(filename, OUTPUT_DIR) if not local_path: return None return str(local_path) # ── CSS ─────────────────────────────────────────────────────────────────────── CSS = """ #log_box textarea { font-family: monospace; font-size: 14px; background: #f8f9fa; color: #333; border-radius: 8px; padding: 12px; } #download_btn { font-size: 16px; font-weight: bold; padding: 15px; } """ # ── App builder ─────────────────────────────────────────────────────────────── def build_app(): with gr.Blocks() as demo: with gr.Row(): gr.Markdown(""" # 🎓 Indian College Directory Engine Easily download and combine official verified higher education institutional data from across India. """) with gr.Tabs(): # ── Tab 1: AISHE ────────────────────────────────────────────── with gr.Tab("📥 AISHE"): with gr.Row(): with gr.Column(): gr.Markdown(""" ### All India Survey on Higher Education (AISHE) Downloads the primary government registry covering Universities, Colleges, and Standalone Institutions. """) with gr.Row(): download_btn = gr.Button( "▶ Download AISHE Data Now", variant="primary", elem_id="download_btn", ) stop_btn = gr.Button( "🛑 Stop Download", variant="stop", visible=False, ) dummy_btn = gr.Button("Dummy", visible=False) with gr.Column(): log_output = gr.Textbox( label="Download Progress", lines=12, max_lines=12, interactive=False, placeholder="Press the download button to begin...", elem_id="log_box", ) output_file = gr.File( label="📥 Download Excel Workbook Here", visible=False, interactive=False, ) # ── Tab 2: NAAC ────────────────────────────────────────────── with gr.Tab("🏅 NAAC"): with gr.Row(): with gr.Column(scale=1): gr.Markdown(""" ### National Assessment and Accreditation Council (NAAC) Downloads the official accreditation grades (A++, A+, B, etc.) and quality inspection CGPA scores for colleges and universities nationwide. """) with gr.Row(): btn_download_naac = gr.Button("▶ Download NAAC Data Now", variant="primary") stop_naac_btn = gr.Button("🛑 Stop Download", variant="stop", visible=False) with gr.Column(scale=1): naac_output = gr.Textbox( label="Download Progress", lines=12, interactive=False, placeholder="Press the download button to begin..." ) btn_download_naac.click( fn=_run_naac_download, outputs=[naac_output, btn_download_naac, stop_naac_btn] ) stop_naac_btn.click( fn=stop_scrape_process, inputs=[], outputs=[naac_output, stop_naac_btn, btn_download_naac] ) # ── Tab 3: URISE UP ITI ───────────────────────────────────────── with gr.Tab("🛠️ URISE UP ITI"): with gr.Row(): with gr.Column(scale=1): gr.Markdown(""" ### URISE UP (ITI Colleges) Downloads the complete state directory of Industrial Training Institutes (ITIs) from the Uttar Pradesh URISE portal. """) btn_download_urise = gr.Button("▶ Download URISE ITI Data Now", variant="primary") with gr.Column(scale=1): urise_output = gr.Textbox( label="Download Progress", lines=12, interactive=False, placeholder="Press the download button to begin..." ) urise_output_file = gr.File( label="📥 Download URISE ITI Excel Workbook Here", visible=False, interactive=False, ) # ── Tab 4: NCVT MIS ITI ───────────────────────────────────────── with gr.Tab("🛠️ NCVT MIS ITI"): with gr.Row(): with gr.Column(scale=1): gr.Markdown(""" ### NCVT MIS (National ITI Registry) Downloads the national registry of ITI colleges from the Ministry of Skill Development (NCVT MIS). ⚠️ *Note: The government NCVT MIS server is frequently overloaded and may take a few minutes to respond, or require multiple automated retries.* """) btn_download_ncvt = gr.Button("▶ Download NCVT ITI Data Now", variant="primary") with gr.Column(scale=1): ncvt_output = gr.Textbox( label="Download Progress", lines=12, interactive=False, placeholder="Press the download button to begin..." ) ncvt_output_file = gr.File( label="📥 Download NCVT ITI Excel Workbook Here", visible=False, interactive=False, ) with gr.Tab("☁️ Dataset Masters"): gr.Markdown(""" ### 1️⃣ Download Previous Scrapes from Cloud Repository Click **'Refresh File List'** to view and redownload any previously generated spreadsheet from your secure cloud storage. """) with gr.Row(): hf_files_dropdown = gr.Dropdown(label="Available Cloud Files", choices=["(Click Refresh to fetch)"], value="(Click Refresh to fetch)") refresh_hf_btn = gr.Button("🔄 Refresh File List") with gr.Row(): fetch_hf_btn = gr.Button("☁️ Fetch Selected File", variant="primary") download_hf_btn = gr.File(label="📥 File Ready for Download", visible=False) hf_status = gr.Textbox(label="Cloud Status", interactive=False, lines=1) gr.Markdown("---") gr.Markdown(""" ### 2️⃣ Generate All-India Master Database Click below to automatically combine your latest scraped educational directories into a single unified Master Workbook, organized into separate **Colleges** and **Courses** worksheets. """) with gr.Row(): merge_master_btn = gr.Button("🔗 Generate Master Database Workbook", variant="primary") with gr.Row(): master_log = gr.Textbox(label="Generation Progress", interactive=False, lines=8) master_file_out = gr.File(label="📥 Master Database Ready", visible=False) # ── Events ──────────────────────────────────────────────────────────── dummy_btn.click(fn=dummy_gpu_func, inputs=[], outputs=[]) # AISHE Events download_btn.click( fn=_run_download_and_build, outputs=[log_output, download_btn, stop_btn, output_file], ) stop_btn.click( fn=stop_scrape_process, inputs=[], outputs=[log_output, stop_btn, download_btn], ) btn_download_urise.click( fn=_run_urise_iti_download, outputs=[urise_output, urise_output_file] ) btn_download_ncvt.click( fn=_run_ncvt_iti_download, outputs=[ncvt_output, ncvt_output_file] ) refresh_hf_btn.click( fn=fetch_hf_files_list, inputs=[], outputs=[hf_files_dropdown], ) def handle_hf_fetch(filename): path = download_file_from_hf(filename) if path: return gr.Textbox(value="Fetched successfully!"), gr.File(value=path, visible=True) return gr.Textbox(value="Failed to fetch."), gr.File(visible=False, value=None) fetch_hf_btn.click( fn=handle_hf_fetch, inputs=[hf_files_dropdown], outputs=[hf_status, download_hf_btn], ) def _run_master_merger(): log_output = [] yield "", gr.update(visible=False) try: from scraper.master_merger import merge_master_database out_dir = Path.cwd() / "output" final_path = None for msg in merge_master_database(out_dir): if isinstance(msg, Path): final_path = msg else: log_output.append(msg) yield "\n".join(log_output), gr.update(visible=False) if final_path: yield "\n".join(log_output), gr.update(value=str(final_path), visible=True) else: log_output.append("❌ Failed to generate master database.") yield "\n".join(log_output), gr.update(visible=False) except Exception as e: log_output.append(f"❌ Critical Error: {e}") yield "\n".join(log_output), gr.update(visible=False) merge_master_btn.click( fn=_run_master_merger, inputs=[], outputs=[master_log, master_file_out] ) return demo if __name__ == "__main__": import os app = build_app() app.launch( server_name="0.0.0.0" if "SPACE_ID" in os.environ else "127.0.0.1", server_port=int(os.environ.get("PORT", 7860)), inbrowser=True, theme=gr.themes.Default(primary_hue="orange", secondary_hue="gray"), css=CSS )