Upload 6 files

.gitignore

@@ -0,0 +1,78 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+# Usually these files are written by a python script from a template
+# before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+.pytest_cache/
+.hypothesis/
+
+# Environments
+.env
+.venv
+venv/
+ENV/
+env/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# VS Code settings
+.vscode/
+
+# Temp files
+*.tmp 

app.py

@@ -0,0 +1,521 @@
+import logging
+import os
+
+import gradio as gr
+from dotenv import load_dotenv
+
+from llm_interface import ERROR_503_DICT  # Import error dict
+from llm_interface import parse_qwen_response, query_qwen_endpoint
+
+# Updated prompt imports for new order
+from prompts import format_privacy_prompt, format_summary_highlights_prompt
+
+# Import helper functions from other modules
+from utils import list_cached_spaces  # Added import
+from utils import (
+    check_report_exists,
+    download_cached_reports,
+    get_space_code_files,
+    upload_reports_to_dataset,
+)
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+# Load environment variables from .env file
+# This is important to ensure API keys and endpoints are loaded before use
+load_dotenv()
+
+# --- Constants ---
+HF_TOKEN = os.getenv("HF_TOKEN")
+DATASET_ID = "yjernite/spaces-privacy-reports"
+CACHE_INFO_MSG = "\n\n*(Report retrieved from cache)*"
+DEFAULT_SELECTION = "HuggingFaceTB/SmolVLM2"
+
+TRUNCATION_WARNING = """**⚠️ Warning:** The input data (code and/or prior analysis) was too long for the AI model's context limit and had to be truncated. The analysis below may be incomplete or based on partial information.\n\n---\n\n"""
+
+ERROR_503_USER_MESSAGE = """**503 Service Unavailable**: It appears that the analysis model endpoint is currently down or starting up. 
+
+You have a few options:
+
+*   **Wait & Retry:** Try clicking "Get Space Report" again in ~3-5 minutes. Endpoints often scale down to save resources and take a short time to wake up.
+*   **Select Cached Report:** Use the dropdown above to view a report for a Space that has already been analyzed.
+*   **Request Analysis:** If the error persists, please open an issue or discussion in the Space's Community tab requesting analysis for your target Space ID. We can run the job manually when the endpoint is available.
+"""
+
+
+def get_space_report_wrapper(
+    selected_cached_space: str | None,
+    new_space_id: str | None,
+    progress=gr.Progress(track_tqdm=True),
+):
+    """
+    Wrapper function to decide whether to fetch cache or run live analysis.
+    Handles the logic based on Dropdown and Textbox inputs.
+    Yields tuples of Gradio updates.
+    """
+    target_space_id = None
+    source = "new"  # Assume new input unless dropdown is chosen
+
+    # Prioritize new_space_id if provided
+    if new_space_id and new_space_id.strip():
+        target_space_id = new_space_id.strip()
+        if target_space_id == selected_cached_space:
+            source = "dropdown_match"  # User typed ID that exists in dropdown
+        else:
+            source = "new"
+    elif selected_cached_space:
+        target_space_id = selected_cached_space
+        source = "dropdown"
+
+    if not target_space_id:
+        # No input provided
+        return (
+            gr.update(
+                value="Please select an existing report or enter a new Space ID.",
+                visible=True,
+            ),
+            gr.update(value="", visible=False),
+            gr.update(visible=True, open=True),
+            gr.update(visible=False),
+        )
+
+    # Validate format
+    if "/" not in target_space_id:
+        return (
+            gr.update(
+                value=f"Invalid Space ID format: '{target_space_id}'. Use 'owner/name'.",
+                visible=True,
+            ),
+            gr.update(value="", visible=False),
+            gr.update(visible=True, open=True),
+            gr.update(visible=False),
+        )
+
+    logging.info(f"Request received for: '{target_space_id}' (Source: {source})")
+
+    # --- Cache Handling ---
+    # If the user explicitly selected from the dropdown, try to fetch it directly.
+    if source == "dropdown":
+        progress(
+            0.1, desc="Fetching cached report..."
+        )  # Simple progress for cache fetch
+        yield (
+            gr.update(value="Fetching selected cached report...", visible=True),
+            gr.update(value="", visible=True),
+            gr.update(visible=True, open=True),
+            gr.update(visible=True, open=False),
+        )
+        try:
+            cached_reports = download_cached_reports(
+                target_space_id, DATASET_ID, HF_TOKEN
+            )
+            summary_report = (
+                cached_reports.get("summary", "Error: Cached summary not found.")
+                + CACHE_INFO_MSG
+            )
+            privacy_report = (
+                cached_reports.get("privacy", "Error: Cached privacy report not found.")
+                + CACHE_INFO_MSG
+            )
+            logging.info(
+                f"Successfully displayed cached reports for selected '{target_space_id}'."
+            )
+            progress(1.0, desc="Complete (from cache)")
+            yield (
+                gr.update(value=summary_report, visible=True),
+                gr.update(value=privacy_report, visible=True),
+                gr.update(visible=True, open=True),
+                gr.update(visible=True, open=True),
+            )
+        except Exception as e:
+            error_msg = f"Failed to download cached report for selected '{target_space_id}': {e}"
+            logging.error(error_msg)
+            progress(1.0, desc="Error")
+            yield (
+                gr.update(value=error_msg, visible=True),
+                gr.update(value="", visible=False),
+                gr.update(visible=True, open=True),
+                gr.update(visible=False),
+            )
+
+    # --- Live Analysis or Check Cache for New Input ---
+    # If it came from the textbox OR was a dropdown match, we first check cache, then run live.
+    else:  # source == "new" or source == "dropdown_match"
+        # This generator now performs the full analysis if needed
+        # Yield intermediate updates from the generator
+        # Important: Need to use a loop to consume the generator
+        final_update = None
+        for update_tuple in _run_live_analysis(target_space_id, progress):
+            yield update_tuple
+            final_update = update_tuple  # Keep track of the last update
+        yield final_update  # Return the very last state
+
+
+def _run_live_analysis(space_id: str, progress=gr.Progress(track_tqdm=True)):
+    """
+    Performs the full analysis pipeline: cache check, code fetch, LLM calls, upload.
+    Yields tuples of Gradio updates.
+    (This contains the logic previously in analyze_space_privacy, minus initial input handling)
+    """
+    steps = 8  # Steps for the full pipeline
+    privacy_truncated = False
+    summary_truncated = False
+
+    # --- Step 1: Check Cache --- (Check again for new/matched input)
+    progress(1 / steps, desc="Step 1/8: Checking cache...")
+    logging.info(f"Step 1/8: Checking cache for '{space_id}'...")
+    yield (
+        gr.update(value="Checking cache for existing reports...", visible=True),
+        gr.update(value="", visible=True),
+        gr.update(visible=True, open=True),
+        gr.update(visible=True, open=False),
+    )
+    found_in_cache = False
+    if HF_TOKEN:
+        try:
+            found_in_cache = check_report_exists(space_id, DATASET_ID, HF_TOKEN)
+        except Exception as e:
+            logging.warning(f"Cache check failed: {e}. Proceeding.")
+            yield (
+                gr.update(
+                    value="Cache check failed, proceeding with live analysis...",
+                    visible=True,
+                ),
+                gr.update(value="", visible=True),
+                gr.update(visible=True, open=True),
+                gr.update(visible=True, open=False),
+            )
+
+    if found_in_cache:
+        logging.info(f"Cache hit for {space_id}. Downloading.")
+        progress(2 / steps, desc="Step 2/8: Cache hit! Downloading reports...")
+        yield (
+            gr.update(value="Cache hit! Downloading reports...", visible=True),
+            gr.update(value="", visible=True),
+            gr.update(visible=True, open=True),
+            gr.update(visible=True, open=False),
+        )
+        try:
+            cached_reports = download_cached_reports(space_id, DATASET_ID, HF_TOKEN)
+            summary_report = (
+                cached_reports.get("summary", "Error: Cached summary not found.")
+                + CACHE_INFO_MSG
+            )
+            privacy_report = (
+                cached_reports.get("privacy", "Error: Cached privacy report not found.")
+                + CACHE_INFO_MSG
+            )
+            logging.info(f"Successfully displayed cached reports for {space_id}.")
+            progress(8 / steps, desc="Complete (from cache)")
+            yield (
+                gr.update(value=summary_report, visible=True),
+                gr.update(value=privacy_report, visible=True),
+                gr.update(visible=True, open=True),
+                gr.update(visible=True, open=True),
+            )
+            return  # End generation here if cache successful
+        except Exception as e:
+            logging.warning(f"Cache download failed for {space_id}: {e}. Proceeding.")
+            yield (
+                gr.update(
+                    value="Cache download failed, proceeding with live analysis...",
+                    visible=True,
+                ),
+                gr.update(value="", visible=True),
+                gr.update(visible=True, open=True),
+                gr.update(visible=True, open=False),
+            )
+    else:
+        logging.info(f"Cache miss for {space_id}. Performing live analysis.")
+        yield (
+            gr.update(value="Cache miss. Fetching code...", visible=True),
+            gr.update(value="", visible=True),
+            gr.update(visible=True, open=True),
+            gr.update(visible=True, open=False),
+        )
+
+    # --- Step 2: Fetch Code Files (if not cached) ---
+    progress(2 / steps, desc="Step 2/8: Fetching code files...")
+    logging.info("Step 2/8: Fetching code files...")
+    code_files = get_space_code_files(space_id)
+    if not code_files:
+        error_msg = f"Could not retrieve code files for '{space_id}'. Check ID and ensure it's a public Space."
+        logging.warning(error_msg)
+        yield (
+            gr.update(value=f"**Error:**\n{error_msg}", visible=True),
+            gr.update(value="Analysis Canceled", visible=True),
+            gr.update(visible=True, open=True),
+            gr.update(visible=True, open=False),
+        )
+        return  # End generation on error
+
+    # --- Step 3: Generate DETAILED Privacy Report (LLM Call 1) ---
+    progress(
+        3 / steps, desc="Step 3/8: Generating detailed privacy report (AI Call 1)..."
+    )
+    logging.info("Step 3/8: Generating detailed privacy analysis report...")
+    yield (
+        gr.update(value="Generating detailed privacy report...", visible=True),
+        gr.update(value="Generating detailed privacy report via AI...", visible=True),
+        gr.update(visible=True, open=True),
+        gr.update(visible=True, open=True),
+    )
+    privacy_prompt_messages, privacy_truncated = format_privacy_prompt(
+        space_id, code_files
+    )
+
+    # --- Check for 503 after query ---
+    privacy_api_response = query_qwen_endpoint(privacy_prompt_messages, max_tokens=3072)
+    if privacy_api_response == ERROR_503_DICT:
+        logging.warning("LLM Call 1 failed with 503.")
+        yield (
+            gr.update(
+                value=ERROR_503_USER_MESSAGE, visible=True
+            ),  # Show 503 message in summary area
+            gr.update(value="", visible=False),  # Clear privacy area
+            gr.update(visible=True, open=True),  # Keep summary open
+            gr.update(visible=False),  # Hide privacy accordion
+        )
+        return  # Stop analysis
+
+    detailed_privacy_report = parse_qwen_response(privacy_api_response)
+
+    if "Error:" in detailed_privacy_report:
+        logging.error(
+            f"Failed to generate detailed privacy report: {detailed_privacy_report}"
+        )
+        yield (
+            gr.update(value="Analysis Halted due to Error", visible=True),
+            gr.update(
+                value=f"**Error Generating Detailed Privacy Report:**\n{detailed_privacy_report}",
+                visible=True,
+            ),
+            gr.update(visible=True, open=True),
+            gr.update(visible=True, open=True),
+        )
+        return  # End generation on error
+    if privacy_truncated:
+        detailed_privacy_report = TRUNCATION_WARNING + detailed_privacy_report
+
+    yield (
+        gr.update(value="Extracting model info...", visible=True),
+        gr.update(value=detailed_privacy_report, visible=True),
+        gr.update(visible=True, open=True),
+        gr.update(visible=True, open=True),
+    )
+
+    # --- Step 4: Extract Model IDs ---
+    progress(4 / steps, desc="Step 4/8: Extracting model IDs...")
+    logging.info("Step 4/8: Extracting potential model IDs...")
+
+    # --- Step 5: Fetch Model Descriptions ---
+    progress(5 / steps, desc="Step 5/8: Fetching model descriptions...")
+    logging.info("Step 5/8: Fetching model descriptions...")
+    yield (
+        gr.update(value="Fetching model descriptions...", visible=True),
+        gr.update(),
+        gr.update(),
+        gr.update(),
+    )
+    # --- Step 6: Generate Summary + Highlights Report (LLM Call 2) ---
+    progress(6 / steps, desc="Step 6/8: Generating summary & highlights (AI Call 2)...")
+    logging.info("Step 6/8: Generating summary and highlights report...")
+    yield (
+        gr.update(value="Generating summary & highlights via AI...", visible=True),
+        gr.update(),
+        gr.update(),
+        gr.update(),
+    )
+    summary_highlights_prompt_messages, summary_truncated = (
+        format_summary_highlights_prompt(space_id, code_files, detailed_privacy_report)
+    )
+
+    # --- Check for 503 after query ---
+    summary_highlights_api_response = query_qwen_endpoint(
+        summary_highlights_prompt_messages, max_tokens=2048
+    )
+    if summary_highlights_api_response == ERROR_503_DICT:
+        logging.warning("LLM Call 2 failed with 503.")
+        yield (
+            gr.update(
+                value=ERROR_503_USER_MESSAGE, visible=True
+            ),  # Show 503 message in summary area
+            gr.update(
+                value=detailed_privacy_report, visible=True
+            ),  # Keep previous report visible
+            gr.update(visible=True, open=True),  # Keep summary open
+            gr.update(visible=True, open=True),  # Keep privacy open
+        )
+        return  # Stop analysis
+
+    summary_highlights_report = parse_qwen_response(summary_highlights_api_response)
+
+    if "Error:" in summary_highlights_report:
+        logging.error(
+            f"Failed to generate summary/highlights report: {summary_highlights_report}"
+        )
+        yield (
+            gr.update(
+                value=f"**Error Generating Summary/Highlights:**\n{summary_highlights_report}",
+                visible=True,
+            ),
+            gr.update(value=detailed_privacy_report, visible=True),
+            gr.update(visible=True, open=True),
+            gr.update(visible=True, open=True),
+        )
+        return  # End generation on error
+    if summary_truncated:
+        summary_highlights_report = TRUNCATION_WARNING + summary_highlights_report
+
+    # Yield summary report before attempting upload
+    yield (
+        gr.update(value=summary_highlights_report, visible=True),
+        gr.update(value=detailed_privacy_report, visible=True),
+        gr.update(visible=True, open=True),
+        gr.update(visible=True, open=True),
+    )
+
+    # --- Step 7: Upload to Cache ---
+    progress(7 / steps, desc="Step 7/8: Uploading results to cache...")
+    logging.info("Step 7/8: Attempting to upload results to dataset cache...")
+    try:
+        if (
+            HF_TOKEN
+            and not found_in_cache
+            and "Error:" not in detailed_privacy_report
+            and "Error:" not in summary_highlights_report
+        ):
+            summary_to_save = summary_highlights_report.replace(
+                TRUNCATION_WARNING, ""
+            ).replace(CACHE_INFO_MSG, "")
+            privacy_to_save = detailed_privacy_report.replace(
+                TRUNCATION_WARNING, ""
+            ).replace(CACHE_INFO_MSG, "")
+            upload_reports_to_dataset(
+                space_id=space_id,
+                summary_report=summary_to_save,
+                detailed_report=privacy_to_save,
+                dataset_id=DATASET_ID,
+                hf_token=HF_TOKEN,
+            )
+        elif not HF_TOKEN:
+            logging.warning("Skipping cache upload as HF_TOKEN is not set.")
+        elif found_in_cache:
+            logging.info("Skipping cache upload as results were loaded from cache.")
+    except Exception as e:
+        logging.error(f"Non-critical error during report upload: {e}")
+
+    logging.info("Step 8/8: Analysis complete.")
+    progress(8 / steps, desc="Step 8/8: Analysis Complete!")
+
+    # --- Step 8: Yield Final Results --- (Ensure final state is correct)
+    yield (
+        gr.update(value=summary_highlights_report, visible=True),
+        gr.update(value=detailed_privacy_report, visible=True),
+        gr.update(visible=True, open=True),
+        gr.update(visible=True, open=True),
+    )
+
+
+# --- Load Initial Data Function (for demo.load) ---
+def load_cached_list():
+    """Fetches the list of cached spaces and determines the default selection."""
+    print("Running demo.load: Fetching list of cached spaces...")
+    # Use os.getenv here directly as HF_TOKEN might be loaded after initial import
+    token = os.getenv("HF_TOKEN")
+    cached_list = list_cached_spaces(DATASET_ID, token)
+    default_value = DEFAULT_SELECTION if DEFAULT_SELECTION in cached_list else None
+    if not cached_list:
+        print(
+            "WARNING: No cached spaces found or failed to fetch list during demo.load."
+        )
+    # Return an update object for the dropdown using gr.update()
+    return gr.update(choices=cached_list, value=default_value)
+
+
+# --- Gradio Interface Definition ---
+# Use HTML/CSS for centering the title
+TITLE = "<div style='text-align: center;'><h1>🤗 Space Privacy Analyzer 🕵️</h1></div>\n<div style='text-align: center;'><h4>Automatic code Data transfer review powered by <a href='https://huggingface.co/Qwen/Qwen2.5-Coder-32B-Instruct' target='_blank'>Qwen2.5-Coder-32B-Instruct</a></h4></div>"
+
+DESCRIPTION = """
+### What Privacy Questions do 🤗 Spaces Raise?
+
+[Hugging Face Spaces](https://huggingface.co/spaces) offer a convenient way to build and share demos leveraging AI models.
+In most cases, the code for these demos is open source &mdash; which provides a unique opportunity to check **how they manage the privacy** of the data in use.
+
+This demo leverages a code analysis model ([Qwen2.5-Coder-32B-Instruct](https://huggingface.co/Qwen/Qwen2.5-Coder-32B-Instruct)) to help explore these questions in two steps:
+it first obtains and **parses the code** to identify data inputs, AI model use, API calls, and data transfer, then generates a summary of the app's function and **key privacy points**.
+
+Use the dropdown menu below to explore the [reports generated for some popular Spaces](https://huggingface.co/datasets/yjernite/spaces-privacy-reports/tree/main), or enter a new Space ID to query your own 👇
+
+*Please note the following limitations:*
+- *The model may easily miss important details in the code, especially when it leverages docker files or external libraries.*
+- *This app uses the base Qwen Coder model without specific adaptation to the task. We'd love to discuss how to improve this, if you want to participate feel free to open a discussion!*
+"""
+
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    gr.Markdown(TITLE)  # This will now render the centered HTML
+
+    with gr.Row():
+        with gr.Column(scale=1):  # Left column for inputs
+            gr.Markdown(DESCRIPTION)
+
+            cached_spaces_dropdown = gr.Dropdown(
+                label="Select Existing Report",
+                info="Select a Space whose report has been previously generated.",
+                choices=[],  # Initialize empty, will be populated by demo.load
+                value=None,  # Initialize empty
+            )
+
+            space_id_input = gr.Textbox(
+                label="Or Enter New Space ID",
+                placeholder="owner/space-name",
+                info="Enter a new Space ID to analyze (takes precedence over selection).",
+            )
+
+            analyze_button = gr.Button("Get Space Report", variant="primary", scale=1)
+
+        with gr.Column(scale=1):  # Right column for outputs
+            # Define Accordions first, open by default, hidden initially
+            summary_accordion = gr.Accordion(
+                "Summary & Privacy Highlights", open=True, visible=True
+            )
+            privacy_accordion = gr.Accordion(
+                "Detailed Privacy Analysis Report", open=False, visible=True
+            )
+            with summary_accordion:
+                summary_markdown = gr.Markdown(
+                    "Enter or select a Space ID and click Get Report.",
+                    show_copy_button=True,
+                )
+            with privacy_accordion:
+                privacy_markdown = gr.Markdown(
+                    "Detailed report will appear here.", show_copy_button=True
+                )
+
+    # --- Event Listeners ---
+
+    # Load event to populate the dropdown when the UI loads for a user session
+    demo.load(fn=load_cached_list, inputs=None, outputs=cached_spaces_dropdown)
+
+    # Button click event
+    analyze_button.click(
+        fn=get_space_report_wrapper,
+        inputs=[cached_spaces_dropdown, space_id_input],
+        outputs=[
+            summary_markdown,
+            privacy_markdown,
+            summary_accordion,
+            privacy_accordion,
+        ],
+        show_progress="full",
+    )
+
+# --- Application Entry Point ---
+
+if __name__ == "__main__":
+    logging.info("Starting Gradio application...")
+    demo.launch()

llm_interface.py

@@ -0,0 +1,154 @@
+import logging
+import os
+
+from dotenv import load_dotenv
+from huggingface_hub import InferenceClient
+from huggingface_hub.inference._generated.types import ChatCompletionOutput
+from huggingface_hub.utils import HfHubHTTPError
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+# Load environment variables from .env file
+# load_dotenv() # Removed: This should be loaded only at the main entry point (app.py)
+load_dotenv()  # Restored: Ensure env vars are loaded when this module is imported/used
+
+HF_TOKEN = os.getenv("HF_TOKEN")
+HF_INFERENCE_ENDPOINT_URL = os.getenv("HF_INFERENCE_ENDPOINT_URL")
+
+# Default parameters for the LLM call
+DEFAULT_MAX_TOKENS = 2048
+DEFAULT_TEMPERATURE = 0.1  # Lower temperature for more deterministic analysis
+
+# Special dictionary to indicate a 503 error
+ERROR_503_DICT = {"error_type": "503", "message": "Service Unavailable"}
+
+
+def query_qwen_endpoint(
+    formatted_prompt: list[dict[str, str]], max_tokens: int = DEFAULT_MAX_TOKENS
+) -> ChatCompletionOutput | dict | None:
+    """
+    Queries the specified Qwen Inference Endpoint with the formatted prompt.
+
+    Args:
+        formatted_prompt: A list of message dictionaries for the chat completion API.
+        max_tokens: The maximum number of tokens to generate.
+
+    Returns:
+        The ChatCompletionOutput object from the inference client,
+        a specific dictionary (ERROR_503_DICT) if a 503 error occurs,
+        or None if another error occurs.
+    """
+    if not HF_INFERENCE_ENDPOINT_URL:
+        logging.error("HF_INFERENCE_ENDPOINT_URL environment variable not set.")
+        return None
+    if not HF_TOKEN:
+        logging.warning(
+            "HF_TOKEN environment variable not set. Requests might fail if the endpoint requires authentication."
+        )
+        # Depending on endpoint config, it might still work without token
+
+    logging.info(f"Querying Inference Endpoint: {HF_INFERENCE_ENDPOINT_URL}")
+    client = InferenceClient(model=HF_INFERENCE_ENDPOINT_URL, token=HF_TOKEN)
+
+    try:
+        response = client.chat_completion(
+            messages=formatted_prompt,
+            max_tokens=max_tokens,
+            temperature=DEFAULT_TEMPERATURE,
+            # Qwen models often benefit from setting stop sequences if known,
+            # but we'll rely on max_tokens and model's natural stopping for now.
+            # stop=["<|im_end|>"] # Example stop token if needed for specific Qwen finetunes
+        )
+        logging.info("Successfully received response from Inference Endpoint.")
+        return response
+    except HfHubHTTPError as e:
+        # Check specifically for 503 Service Unavailable
+        if e.response is not None and e.response.status_code == 503:
+            logging.warning(
+                f"Encountered 503 Service Unavailable from endpoint: {HF_INFERENCE_ENDPOINT_URL}"
+            )
+            return ERROR_503_DICT  # Return special dict for 503
+        else:
+            # Handle other HTTP errors
+            logging.error(f"HTTP error querying Inference Endpoint: {e}")
+            if e.response is not None:
+                logging.error(f"Response details: {e.response.text}")
+            return None  # Return None for other HTTP errors
+    except Exception as e:
+        logging.error(f"An unexpected error occurred querying Inference Endpoint: {e}")
+        return None
+
+
+def parse_qwen_response(response: ChatCompletionOutput | dict | None) -> str:
+    """
+    Parses the response from the Qwen model to extract the generated text.
+    Handles potential None or error dict inputs.
+
+    Args:
+        response: The ChatCompletionOutput object, ERROR_503_DICT, or None.
+
+    Returns:
+        The extracted response text as a string, or an error message string.
+    """
+    if response is None:
+        return "Error: Failed to get response from the language model."
+
+    # Check if it's our specific 503 error signal before trying to parse as ChatCompletionOutput
+    if isinstance(response, dict) and response.get("error_type") == "503":
+        return f"Error: {response['error_type']} {response['message']}"
+
+    # Check if it's likely the expected ChatCompletionOutput structure
+    if not hasattr(response, "choices"):
+        logging.error(
+            f"Unexpected response type received by parse_qwen_response: {type(response)}. Content: {response}"
+        )
+        return "Error: Received an unexpected response format from the language model endpoint."
+
+    try:
+        # Access the generated content according to the ChatCompletionOutput structure
+        if response.choices and len(response.choices) > 0:
+            content = response.choices[0].message.content
+            if content:
+                logging.info("Successfully parsed response content.")
+                return content.strip()
+            else:
+                logging.warning("Response received, but content is empty.")
+                return "Error: Received an empty response from the language model."
+        else:
+            logging.warning("Response received, but no choices found.")
+            return "Error: No response choices found in the language model output."
+    except AttributeError as e:
+        # This might catch cases where response looks like the object but lacks expected attributes
+        logging.error(
+            f"Attribute error parsing response: {e}. Response structure might be unexpected."
+        )
+        logging.error(f"Raw response object: {response}")
+        return "Error: Could not parse the structure of the language model response."
+    except Exception as e:
+        logging.error(f"An unexpected error occurred parsing the response: {e}")
+        return "Error: An unexpected error occurred while parsing the language model response."
+
+
+# Example Usage (for testing - requires .env setup and potentially prompts.py)
+# if __name__ == '__main__':
+#     # This example assumes you have a prompts.py that can generate a test prompt
+#     try:
+#         from prompts import format_code_for_analysis
+#         # Create a dummy prompt for testing
+#         test_files = {"app.py": "print('hello')"}
+#         test_prompt = format_code_for_analysis("test/minimal", test_files)
+#         print("--- Sending Test Prompt ---")
+#         print(test_prompt)
+#         api_response = query_qwen_endpoint(test_prompt)
+#         print("\n--- Raw API Response ---")
+#         print(api_response)
+#         print("\n--- Parsed Response ---")
+#         parsed_text = parse_qwen_response(api_response)
+#         print(parsed_text)
+#     except ImportError:
+#         print("Could not import prompts.py for testing. Run this test from the project root.")
+#     except Exception as e:
+#         print(f"An error occurred during testing: {e}")

prompts.py

@@ -0,0 +1,267 @@
+import logging
+from collections import defaultdict
+
+MAX_FILE_CONTENT_LENGTH = 10000  # Limit the length of individual file contents
+MAX_MODEL_DESC_LENGTH = 1500  # Limit the length of fetched model descriptions
+MAX_PROMPT_CHARS = 110000  # Approx < 30k tokens (using ~4 chars/token heuristic)
+
+# Prompt for the first LLM call: Detailed Privacy Analysis
+PRIVACY_SYSTEM_PROMPT = (
+    "You are a helpful AI assistant specialized in analyzing Hugging Face Spaces code for privacy concerns. "
+    "Your goal is to identify data flows and potential privacy risks based *only* on the provided code files. "
+    "Analyze the following aspects and provide relevant code snippets (formatted as Markdown code blocks) as evidence for each point. "
+    "**Crucially, include the filename for each code snippet.** Example: `(filename.py)`\n\n"
+    "**Note:** If the app uses externally defined or unaccessible code to upload or process data, say so.\n\n"
+    "1.  **Data Inputs:**\n"
+    "    - What types of user data does the application accept as input (e.g., text, images, audio, files)?\n"
+    "    - Where in the code are these inputs defined (e.g., Gradio input widgets, file uploads)? Provide the filename and code snippet.\n\n"
+    "2.  **Processing Services & Data Transmission:**\n"
+    "    - What specific internal or external APIs, models, or services are used to process the input data?\n"
+    "    - What specific AI models or services are used to process the input data? Are any of these Hugging Face-hosted models?\n"
+    "    - Where in the code are these services called (e.g., `requests.post`, `InferenceClient`, specific API endpoint URLs) or defined (e.g., `transformers` library)? Provide the filename and code snippet.\n"
+    "    - Is it likely that user data is transmitted to these external services, and what kind of data is transmitted by each service or API? Mention if the services are known (like Hugging Face Inference API/Endpoints) or potentially unknown third parties.\n\n"
+    "3.  **Execution Environment & Potential Local Processing:**\n"
+    "    - Does the code indicate that models or significant processing might run *locally* within the Space container? Provide the filename and code snippet.\n"
+    "    - Does the code explicitly use external *inference services* to query AI models? If so, reiterate the relevant code snippet from point 2 with filename.\n\n"
+    "    - Does the code mention interactions with remote databases (e.g., `sqlite`, `postgres`, `mysql`, `redis`, `mongodb`, etc.), storage (e.g., `s3`, `gcs`, `azure`, etc.), or Cloud-based data services? If so, provide the filename and code snippet.\n\n"
+    "4.  **Explicit Data Storage/Logging:**\n"
+    "    - Is there any code that explicitly stores user input or results to files, databases, or external logging services? Provide the filename and code snippet.\n\n"
+    "5.  **Overall Privacy Risk Summary:**\n"
+    "    - Based ONLY on the evidence from the code snippets above, provide a concise summary paragraph highlighting the main potential privacy considerations or risks.\n\n"
+    "Format your entire response clearly using Markdown. Ensure all code snippets include filename and are properly formatted."
+)
+
+# Prompt for the second LLM call: Space Summary + Privacy Highlights
+SUMMARY_HIGHLIGHTS_SYSTEM_PROMPT = (
+    "You are an AI assistant reviewing a Hugging Face Space. You have been provided with: "
+    "(1) the application code, and "
+    "(2) a detailed preliminary privacy analysis report."
+    "Your task is to generate a summary report containing two parts:\n\n"
+    "**Part 1: Space Summary**\n"
+    "- Based on the code and privacy analysis report, provide a concise summary (4-6 sentences max) of what the application does from a user's perspective.\n\n"
+    "**Part 2: Privacy Highlights**\n"
+    "- Using information from the preliminary privacy report (cross-referencing code/descriptions as needed), list the following key privacy aspects:\n"
+    "  1.  **Data Inputs:** List the main types of data provided to the application with a brief description for each. List where the data is used or stored by the application.\n"
+    "  2.  **AI Models/Services:** List the core AI models or services used. For each, specify: Is it run locally or remotely? What library or service is used, or is the code defined within the app?\n"
+    "  3.  **Other Remote Data or Dataset Calls:** List any other identified remote data calls that might upload or transmit data outside of the app (e.g., to databases, external APIs not covered above, cloud storage).\n"
+    "  4.  **Libraries Suggesting Data Transmission:** List libraries used (e.g., `requests`, `gradio[sharing]`) that might implicitly or explicitly transmit data, suggesting where users might look for more details (e.g., library documentation, specific code sections).\n\n"
+    "Format the entire response clearly using Markdown. Do not include the preliminary privacy report itself in your output."
+)
+
+
+def _generate_file_structure(code_files: dict[str, str]) -> str:
+    """Generates a tree-like textual representation of the file structure."""
+    tree = defaultdict(dict)
+    files = sorted(code_files.keys())
+
+    for fpath in files:
+        parts = fpath.split("/")
+        node = tree
+        for i, part in enumerate(parts):
+            if i == len(parts) - 1:  # It's a file
+                node[part] = None  # Mark as file
+            else:  # It's a directory
+                if part not in node:
+                    node[part] = defaultdict(dict)  # Create dir node if not exists
+                # Check if we previously marked this as a file (edge case where dir name = file name at higher level)
+                elif node[part] is None:
+                    node[part] = defaultdict(dict)  # Convert file marker to dir
+                node = node[part]  # Move deeper
+
+    output_lines = ["Project File Structure:"]
+
+    def build_tree_lines(node, prefix=""):
+        # Sort items: directories first (defaultdict instances), then files (keys with None value)
+        items = sorted(
+            node.items(),
+            key=lambda item: isinstance(item[1], defaultdict),
+            reverse=True,
+        )
+
+        pointers = ["├── " for _ in range(len(items) - 1)] + ["└── "]
+        for pointer, (name, sub_node) in zip(pointers, items):
+            output_lines.append(prefix + pointer + name)
+            if isinstance(sub_node, defaultdict):  # It's a directory
+                extension = "│   " if pointer == "├── " else "    "
+                build_tree_lines(sub_node, prefix + extension)
+
+    build_tree_lines(tree)
+    print("\n".join(output_lines))
+    return "\n".join(output_lines)
+
+
+def _format_code_files_for_prompt(code_files: dict[str, str]) -> str:
+    """Formats the code files into a single string for the prompt, sorted by depth and path."""
+
+    def sort_key(filepath):
+        parts = filepath.split("/")
+        depth = len(parts) - 1
+        dir_path = "/".join(parts[:-1]) if depth > 0 else ""
+        filename = parts[-1]
+        return (depth, dir_path, filename)
+
+    sorted_filenames = sorted(code_files.keys(), key=sort_key)
+
+    output_parts = []
+    for filename in sorted_filenames:
+        content = code_files[filename]
+        print(f"--- File: {filename} ---")
+        print(content[:128])
+        output_parts.append(
+            f"--- File: {filename} ---\n```\n{content[:MAX_FILE_CONTENT_LENGTH]}{'\n... [truncated]' if len(content) > MAX_FILE_CONTENT_LENGTH else ''}\n```"
+        )
+
+    return "\n".join(output_parts)
+
+
+def format_privacy_prompt(
+    space_id: str, code_files: dict[str, str]
+) -> tuple[list[dict[str, str]], bool]:
+    """
+    Formats the prompt for the initial detailed privacy analysis task.
+    Returns messages list and a boolean indicating if truncation occurred.
+    """
+    was_truncated = False
+    file_structure = _generate_file_structure(code_files)
+    formatted_code = _format_code_files_for_prompt(code_files)
+
+    # Define components for length calculation
+    prompt_header = f"Please perform a detailed privacy analysis for the Hugging Face Space '{space_id}'.\n\n{file_structure}\n\nCode Files Content:\n"
+    base_length = len(prompt_header) + len(PRIVACY_SYSTEM_PROMPT)
+
+    # Check if formatted code needs truncation for the overall prompt
+    available_chars_for_code = MAX_PROMPT_CHARS - base_length
+    if available_chars_for_code < 0:  # Header itself is too long (unlikely)
+        available_chars_for_code = 0
+        was_truncated = True
+
+    if len(formatted_code) > available_chars_for_code:
+        formatted_code = (
+            formatted_code[:available_chars_for_code]
+            + "\n... [Code Section Truncated Due to Overall Prompt Length] ..."
+        )
+        was_truncated = True
+        logging.warning(
+            f"Privacy prompt code section truncated for Space ID {space_id} due to overall length."
+        )
+
+    user_content = prompt_header + formatted_code
+
+    messages = [
+        {"role": "system", "content": PRIVACY_SYSTEM_PROMPT},
+        {"role": "user", "content": user_content},
+    ]
+    return messages, was_truncated
+
+
+def format_summary_highlights_prompt(
+    space_id: str, code_files: dict[str, str], detailed_privacy_report: str
+) -> tuple[list[dict[str, str]], bool]:
+    """
+    Formats the prompt for the final summary + highlights report.
+    Returns messages list and a boolean indicating if truncation occurred.
+    """
+    was_truncated = False
+    file_structure = _generate_file_structure(code_files)
+    formatted_code = _format_code_files_for_prompt(code_files)
+
+    # Define components for length calculation
+    prompt_header = f"Please generate a final summary and privacy highlights report for the Hugging Face Space '{space_id}'.\n\n"
+    report_header = "**Preliminary Detailed Privacy Report:**\n---\n"
+    report_footer = "\n---\n\n"
+    support_header = f"**Supporting Information:**\n{file_structure}\n\n"
+    code_header = "**Original Code Files Content:**\n"
+
+    base_length = (
+        len(prompt_header)
+        + len(report_header)
+        + len(report_footer)
+        + len(support_header)
+        + len(code_header)
+        + len(SUMMARY_HIGHLIGHTS_SYSTEM_PROMPT)
+    )
+    available_chars_total = MAX_PROMPT_CHARS - base_length
+
+    if available_chars_total < 0:  # Base structure is too long
+        logging.error(
+            f"Base prompt structure for summary highlights exceeds limit for Space ID {space_id}. Cannot proceed effectively."
+        )
+        # Return minimal user content to avoid errors, but flag truncation heavily
+        user_content = (
+            prompt_header
+            + report_header
+            + "[TRUNCATED DUE TO LENGTH]"
+            + report_footer
+            + support_header
+            + code_header
+            + "[TRUNCATED DUE TO LENGTH]"
+        )
+        was_truncated = True
+    else:
+        # Prioritize truncating the detailed report first
+        available_chars_for_report = available_chars_total - len(
+            formatted_code
+        )  # Reserve space for code
+        if available_chars_for_report < 0:
+            available_chars_for_report = 0  # Cannot fit report
+
+        if len(detailed_privacy_report) > available_chars_for_report:
+            detailed_privacy_report = (
+                detailed_privacy_report[:available_chars_for_report]
+                + "\n... [Detailed Privacy Report Truncated Due to Overall Prompt Length] ..."
+            )
+            was_truncated = True
+            logging.warning(
+                f"Summary prompt detailed report section truncated for Space ID {space_id}."
+            )
+
+        # Now check code length again with (potentially truncated) report length
+        available_chars_for_code = available_chars_total - len(detailed_privacy_report)
+        if available_chars_for_code < 0:
+            available_chars_for_code = 0  # Cannot fit code
+
+        if len(formatted_code) > available_chars_for_code:
+            formatted_code = (
+                formatted_code[:available_chars_for_code]
+                + "\n... [Code Section Truncated Due to Overall Prompt Length] ..."
+            )
+            was_truncated = True
+            logging.warning(
+                f"Summary prompt code section truncated for Space ID {space_id}."
+            )
+
+        # Assemble the final user content
+        user_content = (
+            prompt_header
+            + report_header
+            + detailed_privacy_report
+            + report_footer
+            + support_header
+            + code_header
+            + formatted_code
+        )
+
+    messages = [
+        {"role": "system", "content": SUMMARY_HIGHLIGHTS_SYSTEM_PROMPT},
+        {"role": "user", "content": user_content},
+    ]
+    return messages, was_truncated
+
+
+# Example usage (for testing)
+# if __name__ == '__main__':
+#     test_files = {
+#         "app.py": "import gradio as gr\n\ndef greet(name):\n    # Potentially send data to external service?\n    # requests.post('http://example.com/log', json={'user': name})\n    return f'Hello {name}!'",
+#         "requirements.txt": "gradio\nrequests",
+#         "nested/utils.py": "def helper():\n    pass",
+#         "README.md": "This should be ignored.", # Example of a file that *should* be filtered out before reaching here
+#         "very_long_file.py": "print('hello' * 5000)" # Test truncation
+#     }
+#     # Typically, files like README.md would be filtered by get_space_code_files in utils.py
+#     # We include it here just for demo purposes if you were to test prompts.py directly.
+#     filtered_test_files = {k: v for k, v in test_files.items() if not k.endswith('.md')}
+#     prompt_messages = format_code_for_analysis("test/space", filtered_test_files)
+#     print("--- System Prompt ---")
+#     print(prompt_messages[0]['content'])
+#     print("\n--- User Prompt ---")
+#     print(prompt_messages[1]['content'])

requirements.txt

@@ -0,0 +1,4 @@
+gradio
+huggingface-hub
+python-dotenv
+requests 

utils.py

@@ -0,0 +1,507 @@
+import logging
+import os
+import re
+import tempfile
+
+from huggingface_hub import HfApi, hf_hub_download
+from huggingface_hub.utils import EntryNotFoundError, RepositoryNotFoundError
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+# Files/extensions to definitely include
+INCLUDE_PATTERNS = [
+    ".py",
+    "requirements.txt",
+    "Dockerfile",
+    ".js",
+    ".jsx",
+    ".ts",
+    ".tsx",
+    ".html",
+    ".css",
+    ".svelte",
+    ".vue",
+    ".json",
+    ".yaml",
+    ".yml",
+    ".toml",
+    "Procfile",
+    ".sh",
+]
+
+# Files/extensions/folders to ignore
+IGNORE_PATTERNS = [
+    ".git",
+    ".hfignore",
+    "README.md",
+    "LICENSE",
+    "__pycache__",
+    ".ipynb_checkpoints",
+    ".png",
+    ".jpg",
+    ".jpeg",
+    ".gif",
+    ".svg",
+    ".ico",
+    ".mp3",
+    ".wav",
+    ".mp4",
+    ".mov",
+    ".avi",
+    ".onnx",
+    ".pt",
+    ".pth",
+    ".bin",
+    ".safetensors",
+    ".tflite",
+    ".pickle",
+    ".pkl",
+    ".joblib",
+    ".parquet",
+    ".csv",
+    ".tsv",
+    ".zip",
+    ".tar.gz",
+    ".gz",
+    ".ipynb",
+    ".DS_Store",
+    "node_modules",
+]
+
+# Regex to find potential Hugging Face model IDs (e.g., "org/model-name", "user/model-name")
+# This is a simple heuristic and might catch non-model strings or miss complex cases.
+HF_MODEL_ID_PATTERN = re.compile(r"([\"\'])([\w\-.]+/[\w\-\.]+)\1\'")
+
+# Max length for model descriptions to keep prompts manageable
+MAX_MODEL_DESC_LENGTH = 1500
+
+SUMMARY_FILENAME = "summary_highlights.md"
+PRIVACY_FILENAME = "privacy_report.md"
+
+
+def _is_relevant_file(filename):
+    """Check if a file should be included based on patterns."""
+    # Ignore files matching ignore patterns (case-insensitive check for some)
+    lower_filename = filename.lower()
+    if any(
+        pattern in lower_filename
+        for pattern in [".git", ".hfignore", "readme.md", "license"]
+    ):
+        return False
+    if any(
+        filename.endswith(ext) for ext in IGNORE_PATTERNS if ext.startswith(".")
+    ):  # Check extensions
+        return False
+    if any(
+        part == pattern
+        for part in filename.split("/")
+        for pattern in IGNORE_PATTERNS
+        if "." not in pattern and "/" not in pattern
+    ):  # Check directory/file names
+        return False
+    if filename in IGNORE_PATTERNS:  # Check full filenames
+        return False
+
+    # Include files matching include patterns
+    if any(filename.endswith(ext) for ext in INCLUDE_PATTERNS if ext.startswith(".")):
+        return True
+    if any(filename == pattern for pattern in INCLUDE_PATTERNS if "." not in pattern):
+        return True
+
+    # Default to False if not explicitly included (safer)
+    # logging.debug(f"File '{filename}' excluded by default.")
+    return False
+
+
+def get_space_code_files(space_id: str) -> dict[str, str]:
+    """
+    Downloads relevant code and configuration files from a Hugging Face Space.
+
+    Args:
+        space_id: The ID of the Hugging Face Space (e.g., 'gradio/hello_world').
+
+    Returns:
+        A dictionary where keys are filenames and values are file contents as strings.
+        Returns an empty dictionary if the space is not found or has no relevant files.
+    """
+    code_files = {}
+    api = HfApi()
+
+    try:
+        logging.info(f"Fetching file list for Space: {space_id}")
+        repo_files = api.list_repo_files(repo_id=space_id, repo_type="space")
+        logging.info(f"Found {len(repo_files)} total files in {space_id}.")
+
+        relevant_files = [f for f in repo_files if _is_relevant_file(f)]
+        logging.info(f"Identified {len(relevant_files)} relevant files for download.")
+
+        for filename in relevant_files:
+            try:
+                logging.debug(f"Downloading {filename} from {space_id}...")
+                file_path = hf_hub_download(
+                    repo_id=space_id,
+                    filename=filename,
+                    repo_type="space",
+                    # Consider adding use_auth_token=os.getenv("HF_TOKEN") if accessing private spaces later
+                )
+                with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
+                    content = f.read()
+                code_files[filename] = content
+                logging.debug(f"Successfully read content of {filename}")
+            except EntryNotFoundError:
+                logging.warning(
+                    f"File {filename} listed but not found in repo {space_id}."
+                )
+            except UnicodeDecodeError:
+                logging.warning(
+                    f"Could not decode file {filename} from {space_id} as UTF-8. Skipping."
+                )
+            except OSError as e:
+                logging.warning(f"OS error reading file {filename} from cache: {e}")
+            except Exception as e:
+                logging.error(
+                    f"Unexpected error downloading or reading file {filename} from {space_id}: {e}"
+                )
+
+    except RepositoryNotFoundError:
+        logging.error(f"Space repository '{space_id}' not found.")
+        return {}
+    except Exception as e:
+        logging.error(f"Failed to list or process files for space {space_id}: {e}")
+        return {}
+
+    logging.info(
+        f"Successfully retrieved content for {len(code_files)} files from {space_id}."
+    )
+    return code_files
+
+
+def extract_hf_model_ids(code_files: dict[str, str]) -> set[str]:
+    """
+    Extracts potential Hugging Face model IDs mentioned in code files.
+
+    Args:
+        code_files: Dictionary of {filename: content}.
+
+    Returns:
+        A set of unique potential model IDs found.
+    """
+    potential_ids = set()
+    for filename, content in code_files.items():
+        # Limit search to relevant file types
+        if filename.endswith((".py", ".json", ".yaml", ".yml", ".toml", ".md")):
+            try:
+                matches = HF_MODEL_ID_PATTERN.findall(content)
+                for _, model_id in matches:
+                    # Basic validation: must contain exactly one '/'
+                    if model_id.count("/") == 1:
+                        # Avoid adding common paths that look like IDs
+                        if not any(
+                            part in model_id.lower()
+                            for part in ["http", "www", "@", " ", ".", ":"]
+                        ):  # Check if '/' is only separator
+                            if len(model_id) < 100:  # Avoid overly long strings
+                                potential_ids.add(model_id)
+            except Exception as e:
+                logging.warning(f"Regex error processing file {filename}: {e}")
+
+    logging.info(f"Extracted {len(potential_ids)} potential model IDs.")
+    # Add simple filter for very common false positives if needed
+    # potential_ids = {id for id in potential_ids if id not in ['user/repo']}
+    return potential_ids
+
+
+def get_model_descriptions(model_ids: set[str]) -> dict[str, str]:
+    """
+    Fetches the README.md content (description) for a set of model IDs.
+
+    Args:
+        model_ids: A set of Hugging Face model IDs.
+
+    Returns:
+        A dictionary mapping model_id to its description string (or an error message).
+    """
+    descriptions = {}
+    if not model_ids:
+        return descriptions
+
+    logging.info(f"Fetching descriptions for {len(model_ids)} models...")
+    for model_id in model_ids:
+        try:
+            # Check if the model exists first (optional but good practice)
+            # api.model_info(model_id)
+
+            # Download README.md
+            readme_path = hf_hub_download(
+                repo_id=model_id,
+                filename="README.md",
+                repo_type="model",
+                # Add token if needing to access private/gated models - unlikely for Space analysis
+                # use_auth_token=os.getenv("HF_TOKEN"),
+                error_if_not_found=True,  # Raise error if README doesn't exist
+            )
+            with open(readme_path, "r", encoding="utf-8", errors="ignore") as f:
+                description = f.read()
+            descriptions[model_id] = description[:MAX_MODEL_DESC_LENGTH] + (
+                "... [truncated]" if len(description) > MAX_MODEL_DESC_LENGTH else ""
+            )
+            logging.debug(f"Successfully fetched description for {model_id}")
+        except RepositoryNotFoundError:
+            logging.warning(f"Model repository '{model_id}' not found.")
+            descriptions[model_id] = "[Model repository not found]"
+        except EntryNotFoundError:
+            logging.warning(f"README.md not found in model repository '{model_id}'.")
+            descriptions[model_id] = "[README.md not found in model repository]"
+        except Exception as e:
+            logging.error(f"Error fetching description for model '{model_id}': {e}")
+            descriptions[model_id] = f"[Error fetching description: {e}]"
+
+    logging.info(f"Finished fetching descriptions for {len(descriptions)} models.")
+    return descriptions
+
+
+def list_cached_spaces(dataset_id: str, hf_token: str | None) -> list[str]:
+    """Lists the space IDs (owner/name) that have cached reports in the dataset repository."""
+    if not hf_token:
+        logging.warning("HF Token not provided, cannot list cached spaces.")
+        return []
+    try:
+        api = HfApi(token=hf_token)
+        # Get all filenames in the dataset repository
+        all_files = api.list_repo_files(repo_id=dataset_id, repo_type="dataset")
+
+        # Extract unique directory paths that look like owner/space_name
+        # by checking if they contain our specific report files.
+        space_ids = set()
+        for f_path in all_files:
+            # Check if the file is one of our report files
+            if f_path.endswith(f"/{PRIVACY_FILENAME}") or f_path.endswith(
+                f"/{SUMMARY_FILENAME}"
+            ):
+                # Extract the directory path part (owner/space_name)
+                parts = f_path.split("/")
+                if len(parts) == 3:  # Expecting owner/space_name/filename.md
+                    owner_slash_space_name = "/".join(parts[:-1])
+                    # Basic validation: owner and space name shouldn't start with '.'
+                    if not parts[0].startswith(".") and not parts[1].startswith("."):
+                        space_ids.add(owner_slash_space_name)
+
+        sorted_space_ids = sorted(list(space_ids))
+        logging.info(
+            f"Found {len(sorted_space_ids)} cached space reports in {dataset_id} via HfApi."
+        )
+        return sorted_space_ids
+
+    except RepositoryNotFoundError:
+        logging.warning(
+            f"Dataset {dataset_id} not found or empty when listing cached spaces."
+        )
+        return []
+    except Exception as e:
+        logging.error(f"Error listing cached spaces in {dataset_id} via HfApi: {e}")
+        return []  # Return empty list on error
+
+
+def check_report_exists(space_id: str, dataset_id: str, hf_token: str | None) -> bool:
+    """Checks if report files already exist in the target dataset repo using HfApi."""
+    print(
+        f"[Debug Cache Check] Checking for space_id: '{space_id}' in dataset: '{dataset_id}'"
+    )  # DEBUG
+    if not hf_token:
+        logging.warning("HF Token not provided, cannot check dataset cache.")
+        print("[Debug Cache Check] No HF Token, returning False.")  # DEBUG
+        return False
+    try:
+        api = HfApi(token=hf_token)
+        # List ALL files in the repo
+        print(f"[Debug Cache Check] Listing ALL files in repo '{dataset_id}'")  # DEBUG
+        all_repo_files = api.list_repo_files(repo_id=dataset_id, repo_type="dataset")
+        # DEBUG: Optionally print a subset if the list is huge
+        # print(f"[Debug Cache Check] First 10 files returned by API: {all_repo_files[:10]}")
+
+        # Construct the exact paths we expect for the target space_id
+        expected_summary_path = f"{space_id}/{SUMMARY_FILENAME}"
+        expected_privacy_path = f"{space_id}/{PRIVACY_FILENAME}"
+        print(
+            f"[Debug Cache Check] Expecting summary file: '{expected_summary_path}'"
+        )  # DEBUG
+        print(
+            f"[Debug Cache Check] Expecting privacy file: '{expected_privacy_path}'"
+        )  # DEBUG
+
+        # Check if both expected paths exist in the full list of files
+        summary_exists = expected_summary_path in all_repo_files
+        privacy_exists = expected_privacy_path in all_repo_files
+        exists = summary_exists and privacy_exists
+        print(
+            f"[Debug Cache Check] Summary exists in full list: {summary_exists}"
+        )  # DEBUG
+        print(
+            f"[Debug Cache Check] Privacy exists in full list: {privacy_exists}"
+        )  # DEBUG
+        print(f"[Debug Cache Check] Overall exists check result: {exists}")  # DEBUG
+        return exists
+
+    except RepositoryNotFoundError:
+        logging.warning(
+            f"Dataset repository {dataset_id} not found or not accessible during check."
+        )
+        print(
+            f"[Debug Cache Check] Repository {dataset_id} not found, returning False."
+        )  # DEBUG
+    except Exception as e:
+        # ... (error handling remains the same) ...
+        print(f"[Debug Cache Check] Exception caught: {type(e).__name__}: {e}")  # DEBUG
+        # Note: 404 check based on path_in_repo is no longer applicable here
+        # We rely on RepositoryNotFoundError or general Exception
+        logging.error(
+            f"Error checking dataset {dataset_id} for {space_id} via HfApi: {e}"
+        )
+        print("[Debug Cache Check] Other exception, returning False.")  # DEBUG
+    return False  # Treat errors as cache miss
+
+
+def download_cached_reports(
+    space_id: str, dataset_id: str, hf_token: str | None
+) -> dict[str, str]:
+    """Downloads cached reports from the dataset repo. Raises error on failure."""
+    if not hf_token:
+        raise ValueError("HF Token required to download cached reports.")
+
+    logging.info(
+        f"Attempting to download cached reports for {space_id} from {dataset_id}..."
+    )
+    reports = {}
+    # Define paths relative to dataset root for hf_hub_download
+    summary_repo_path = f"{space_id}/{SUMMARY_FILENAME}"
+    privacy_repo_path = f"{space_id}/{PRIVACY_FILENAME}"
+    try:
+        # Download summary
+        summary_path_local = hf_hub_download(
+            repo_id=dataset_id,
+            filename=summary_repo_path,
+            repo_type="dataset",
+            token=hf_token,
+        )
+        with open(summary_path_local, "r", encoding="utf-8") as f:
+            reports["summary"] = f.read()
+        logging.info(f"Successfully downloaded cached summary for {space_id}.")
+
+        # Download privacy report
+        privacy_path_local = hf_hub_download(
+            repo_id=dataset_id,
+            filename=privacy_repo_path,
+            repo_type="dataset",
+            token=hf_token,
+        )
+        with open(privacy_path_local, "r", encoding="utf-8") as f:
+            reports["privacy"] = f.read()
+        logging.info(f"Successfully downloaded cached privacy report for {space_id}.")
+
+        return reports
+
+    except EntryNotFoundError as e:
+        # More specific error based on which file failed
+        missing_file = (
+            summary_repo_path if summary_repo_path in str(e) else privacy_repo_path
+        )
+        logging.error(
+            f"Cache download error: Report file {missing_file} not found for {space_id} in {dataset_id}. {e}"
+        )
+        raise FileNotFoundError(
+            f"Cached report file {missing_file} not found for {space_id}"
+        ) from e
+    except RepositoryNotFoundError as e:
+        logging.error(f"Cache download error: Dataset repo {dataset_id} not found. {e}")
+        raise FileNotFoundError(f"Dataset repo {dataset_id} not found") from e
+    except Exception as e:
+        logging.error(
+            f"Unexpected error downloading cached reports for {space_id} from {dataset_id}: {e}"
+        )
+        raise IOError(f"Failed to download cached reports for {space_id}") from e
+
+
+def upload_reports_to_dataset(
+    space_id: str,
+    summary_report: str,
+    detailed_report: str,
+    dataset_id: str,
+    hf_token: str | None,
+):
+    """Uploads the generated reports to the specified dataset repository."""
+    if not hf_token:
+        logging.warning("HF Token not provided, skipping dataset report upload.")
+        return
+
+    logging.info(
+        f"Attempting to upload reports for {space_id} to dataset {dataset_id}..."
+    )
+    api = HfApi(token=hf_token)
+
+    # Sanitize space_id for path safety (though HF Hub usually handles this)
+    safe_space_id = space_id.replace("..", "")
+
+    try:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            summary_path_local = os.path.join(tmpdir, SUMMARY_FILENAME)
+            privacy_path_local = os.path.join(tmpdir, PRIVACY_FILENAME)
+
+            with open(summary_path_local, "w", encoding="utf-8") as f:
+                f.write(summary_report)
+            with open(privacy_path_local, "w", encoding="utf-8") as f:
+                f.write(detailed_report)
+
+            commit_message = f"Add privacy analysis reports for Space: {safe_space_id}"
+            repo_url = api.create_repo(
+                repo_id=dataset_id,
+                repo_type="dataset",
+                exist_ok=True,
+            )
+            logging.info(f"Ensured dataset repo {repo_url} exists.")
+
+            api.upload_file(
+                path_or_fileobj=summary_path_local,
+                path_in_repo=f"{safe_space_id}/{SUMMARY_FILENAME}",
+                repo_id=dataset_id,
+                repo_type="dataset",
+                commit_message=commit_message,
+            )
+            logging.info(f"Successfully uploaded summary report for {safe_space_id}.")
+
+            api.upload_file(
+                path_or_fileobj=privacy_path_local,
+                path_in_repo=f"{safe_space_id}/{PRIVACY_FILENAME}",
+                repo_id=dataset_id,
+                repo_type="dataset",
+                commit_message=commit_message,
+            )
+            logging.info(
+                f"Successfully uploaded detailed privacy report for {safe_space_id}."
+            )
+
+    except Exception as e:
+        logging.error(
+            f"Failed to upload reports for {safe_space_id} to dataset {dataset_id}: {e}"
+        )
+
+
+# Example usage (for testing)
+# if __name__ == '__main__':
+#     # Make sure HF_TOKEN is set if accessing private spaces or for higher rate limits
+#     from dotenv import load_dotenv
+#     load_dotenv()
+#     # test_space = "gradio/hello_world"
+#     test_space = "huggingface-projects/diffusers-gallery" # A more complex example
+#     # test_space = "nonexistent/space" # Test not found
+#     files_content = get_space_code_files(test_space)
+#     if files_content:
+#         print(f"\n--- Files retrieved from {test_space} ---")
+#         for name in files_content.keys():
+#             print(f"- {name}")
+#         # print("\n--- Content of app.py (first 200 chars) ---")
+#         # print(files_content.get("app.py", "app.py not found")[:200])
+#     else:
+#         print(f"Could not retrieve files from {test_space}")