Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+# Install uv
+RUN pip install uv
+
+# Copy requirements
+COPY requirements.txt .
+
+# Install dependencies
+RUN uv pip install --system -r requirements.txt
+
+# Copy code
+COPY . .
+
+# Expose port
+EXPOSE 7860
+
+# Run the server
+# HF Spaces expect the app to run on port 7860
+CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "7860"]

auth.py

@@ -0,0 +1,11 @@
+import requests
+from huggingface_hub import whoami
+
+def validate_hf_token(token):
+    """Validates the HF token and returns user info."""
+    try:
+        user_info = whoami(token=token)
+        return user_info
+    except Exception as e:
+        print(f"Token validation failed: {e}")
+        return None

datasets.json

@@ -0,0 +1,16 @@
+[
+    {
+        "dataset_id": "org/dataset_A",
+        "slug": "daily-signals",
+        "display_name": "Daily Signals Dataset",
+        "description": "Daily updated financial signals.",
+        "plans": [
+            {
+                "plan_id": "pro_monthly",
+                "stripe_price_id": "price_123",
+                "access_duration_days": 30
+            }
+        ],
+        "is_active": true
+    }
+]

datasets_registry.py

@@ -0,0 +1,243 @@
+import json
+import os
+import tempfile
+from typing import List, Dict, Any, Optional
+from huggingface_hub import HfApi, hf_hub_download
+from huggingface_hub.utils import EntryNotFoundError
+
+# Configuration - uses same ledger repo as subscriptions
+LEDGER_REPO = os.getenv("LEDGER_DATASET_ID", "")
+REGISTRY_FILE = "datasets.json"
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+# Fallback to local file if LEDGER_DATASET_ID not set (for local dev)
+LOCAL_REGISTRY_FILE = "datasets.json"
+
+# Initialize HF API
+api = HfApi(token=HF_TOKEN) if HF_TOKEN else None
+
+
+def _use_hf_storage() -> bool:
+    """Check if we should use HF Dataset storage."""
+    return bool(LEDGER_REPO and HF_TOKEN and api)
+
+
+def _download_registry() -> Optional[str]:
+    """Download current registry from HF Dataset."""
+    if not _use_hf_storage():
+        return None
+
+    try:
+        path = hf_hub_download(
+            repo_id=LEDGER_REPO,
+            filename=REGISTRY_FILE,
+            repo_type="dataset",
+            token=HF_TOKEN
+        )
+        return path
+    except EntryNotFoundError:
+        # File doesn't exist yet in the dataset
+        return None
+    except Exception as e:
+        print(f"Error downloading registry: {e}")
+        return None
+
+
+def _upload_registry(local_path: str) -> bool:
+    """Upload registry to HF Dataset."""
+    if not _use_hf_storage():
+        return False
+
+    try:
+        api.upload_file(
+            path_or_fileobj=local_path,
+            path_in_repo=REGISTRY_FILE,
+            repo_id=LEDGER_REPO,
+            repo_type="dataset",
+            token=HF_TOKEN,
+            commit_message=f"Update dataset registry"
+        )
+        return True
+    except Exception as e:
+        print(f"Error uploading registry: {e}")
+        return False
+
+
+def load_registry() -> List[Dict[str, Any]]:
+    """Loads the dataset registry from HF Dataset or local file."""
+    if _use_hf_storage():
+        hf_path = _download_registry()
+        if hf_path:
+            try:
+                with open(hf_path, "r") as f:
+                    return json.load(f)
+            except json.JSONDecodeError:
+                print(f"Error decoding {hf_path}")
+                return []
+
+    # Fallback to local file
+    if not os.path.exists(LOCAL_REGISTRY_FILE):
+        return []
+    try:
+        with open(LOCAL_REGISTRY_FILE, "r") as f:
+            registry = json.load(f)
+        return registry
+    except json.JSONDecodeError:
+        print(f"Error decoding {LOCAL_REGISTRY_FILE}")
+        return []
+
+
+def save_registry(registry: List[Dict[str, Any]]) -> bool:
+    """Saves the dataset registry to HF Dataset or local file."""
+    if _use_hf_storage():
+        # Create temp file with registry content
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
+            tmp_path = tmp.name
+            json.dump(registry, tmp, indent=2)
+
+        # Upload to HF
+        success = _upload_registry(tmp_path)
+
+        # Clean up temp file
+        try:
+            os.unlink(tmp_path)
+        except:
+            pass
+
+        return success
+    else:
+        # Local file storage
+        with open(LOCAL_REGISTRY_FILE, "w") as f:
+            json.dump(registry, f, indent=2)
+        return True
+
+def get_dataset_by_id(dataset_id: str) -> Optional[Dict[str, Any]]:
+    """Finds a dataset by its ID."""
+    registry = load_registry()
+    for dataset in registry:
+        if dataset.get("dataset_id") == dataset_id:
+            return dataset
+    return None
+
+def get_dataset_by_slug(slug: str) -> Optional[Dict[str, Any]]:
+    """Finds a dataset by its slug."""
+    registry = load_registry()
+    for dataset in registry:
+        if dataset.get("slug") == slug:
+            return dataset
+    return None
+
+def get_plan_by_price_id(price_id: str) -> Optional[Dict[str, Any]]:
+    """Finds a plan and its dataset by Stripe price ID."""
+    registry = load_registry()
+    for dataset in registry:
+        for plan in dataset.get("plans", []):
+            if plan.get("stripe_price_id") == price_id:
+                return {"dataset": dataset, "plan": plan}
+    return None
+
+def get_free_plan(dataset_id: str) -> Optional[Dict[str, Any]]:
+    """
+    Securely finds a free plan for a dataset.
+    Returns the plan dict if found, None otherwise.
+    """
+    dataset = get_dataset_by_id(dataset_id)
+    if not dataset:
+        return None
+
+    # Explicitly check for free markers
+    for plan in dataset.get("plans", []):
+        if plan.get("stripe_price_id") in ["free", "0", 0]:
+            return plan
+
+    return None
+
+
+def detect_dataset_format(dataset_id: str) -> Dict[str, Any]:
+    """
+    Detects the format and parquet path for a HuggingFace dataset.
+    Returns info about the dataset including the correct parquet URL pattern.
+    """
+    if not api:
+        return {
+            "dataset_id": dataset_id,
+            "error": "HF API not initialized (HF_TOKEN not set)",
+            "parquet_url_pattern": None
+        }
+
+    try:
+        # Get dataset info from main branch
+        info = api.dataset_info(dataset_id, token=HF_TOKEN)
+
+        # Check for native parquet files in main branch
+        parquet_paths = []
+        has_native_parquet = False
+
+        for sibling in info.siblings or []:
+            filename = sibling.rfilename
+            if filename.endswith('.parquet'):
+                parquet_paths.append(filename)
+                has_native_parquet = True
+
+        # Check for auto-converted parquet in refs/convert/parquet
+        has_converted_parquet = False
+        converted_parquet_paths = []
+        try:
+            convert_info = api.dataset_info(dataset_id, token=HF_TOKEN, revision='refs/convert/parquet')
+            for sibling in convert_info.siblings or []:
+                filename = sibling.rfilename
+                if filename.endswith('.parquet'):
+                    converted_parquet_paths.append(filename)
+                    has_converted_parquet = True
+        except Exception:
+            # refs/convert/parquet doesn't exist for this dataset
+            pass
+
+        # Determine the best parquet URL pattern
+        if has_native_parquet:
+            # Dataset has native parquet files in main branch
+            parquet_url_pattern = f"hf://datasets/{dataset_id}/**/*.parquet"
+            parquet_count = len(parquet_paths)
+        elif has_converted_parquet:
+            # Dataset was auto-converted, use refs/convert/parquet
+            # Note: The revision path must be URL-encoded for DuckDB
+            parquet_url_pattern = f"hf://datasets/{dataset_id}@refs%2Fconvert%2Fparquet/**/*.parquet"
+            parquet_count = len(converted_parquet_paths)
+        else:
+            # No parquet files found
+            parquet_url_pattern = None
+            parquet_count = 0
+
+        return {
+            "dataset_id": dataset_id,
+            "has_native_parquet": has_native_parquet,
+            "has_converted_parquet": has_converted_parquet,
+            "parquet_url_pattern": parquet_url_pattern,
+            "parquet_files_count": parquet_count,
+            "card_data": info.card_data.__dict__ if info.card_data else None,
+        }
+    except Exception as e:
+        return {
+            "dataset_id": dataset_id,
+            "error": str(e),
+            "parquet_url_pattern": None
+        }
+
+
+def get_parquet_url(dataset_id: str) -> str:
+    """
+    Gets the best parquet URL pattern for a dataset.
+    Checks registry first, then tries to detect automatically.
+    """
+    # Check if dataset has a stored parquet_url_pattern in registry
+    dataset = get_dataset_by_id(dataset_id)
+    if dataset and dataset.get("parquet_url_pattern"):
+        return dataset["parquet_url_pattern"]
+
+    # Try to detect the format
+    format_info = detect_dataset_format(dataset_id)
+    if format_info.get("parquet_url_pattern"):
+        return format_info["parquet_url_pattern"]
+
+    # Fallback to standard pattern
+    return f"hf://datasets/{dataset_id}/**/*.parquet"

requirements.txt

@@ -0,0 +1,9 @@
+mcp
+huggingface_hub
+stripe
+fastapi
+uvicorn
+python-dotenv
+duckdb
+pandas
+numpy

server.py

@@ -0,0 +1,629 @@
+from mcp.server.fastmcp import FastMCP, Context
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from dotenv import load_dotenv
+import os
+import json
+import datasets_registry
+import subscriptions_ledger
+import stripe_webhook
+import auth
+
+load_dotenv()
+
+# Initialize MCP Server
+mcp = FastMCP("datapass")
+
+# Admin API Secret for secure admin endpoints
+ADMIN_API_SECRET = os.getenv("ADMIN_API_SECRET")
+
+
+def verify_admin_secret(provided_secret: str) -> bool:
+    """Verify the admin secret matches."""
+    if not ADMIN_API_SECRET:
+        return False  # Fail closed if not configured
+    return provided_secret == ADMIN_API_SECRET
+
+
+def get_admin_secret_from_request(request: Request, body_data: dict = None) -> str:
+    """Extract admin secret from header (preferred) or body (fallback)."""
+    header_secret = request.headers.get("X-Admin-Secret", "")
+    if header_secret:
+        return header_secret
+    if body_data:
+        return body_data.get("admin_secret", "")
+    return ""
+
+
+def _get_access_token_from_context(ctx: Context) -> str | None:
+    """Extract access token from the Authorization header in the request context."""
+    try:
+        if ctx.request_context and ctx.request_context.request:
+            auth_header = ctx.request_context.request.headers.get("Authorization", "")
+            if auth_header.startswith("Bearer "):
+                return auth_header[7:]  # Remove "Bearer " prefix
+    except Exception as e:
+        print(f"Error extracting access token: {e}")
+    return None
+
+
+def _validate_token_for_dataset(access_token: str, dataset_id: str = None):
+    """Validate access token and optionally check dataset access."""
+    if not access_token:
+        return None, "No access token provided. Please include Authorization header with Bearer token."
+
+    token_info = subscriptions_ledger.validate_access_token(access_token)
+    if not token_info:
+        return None, "Invalid or expired access token."
+
+    if dataset_id and token_info["dataset_id"] != dataset_id:
+        return None, f"Access token not valid for dataset '{dataset_id}'. Your token is for '{token_info['dataset_id']}'."
+
+    return token_info, None
+
+
+# =============================================================================
+# Public MCP Tools (no auth required)
+# =============================================================================
+
+@mcp.tool()
+def get_dataset_catalog():
+    """
+    Returns the catalog of all available datasets in the marketplace.
+    Use this to discover datasets you can subscribe to.
+    """
+    registry = datasets_registry.load_registry()
+    active_datasets = [d for d in registry if d.get("is_active")]
+    return active_datasets
+
+
+# =============================================================================
+# Authenticated MCP Tools (require access token via Authorization header)
+# =============================================================================
+
+@mcp.tool()
+def get_my_datasets(ctx: Context):
+    """
+    Returns the list of datasets you have access to based on your subscription.
+    The access token is automatically read from your Authorization header.
+    """
+    access_token = _get_access_token_from_context(ctx)
+    if not access_token:
+        return {"error": "No access token provided. Please include Authorization header with Bearer token."}
+
+    token_info = subscriptions_ledger.validate_access_token(access_token)
+    if not token_info:
+        return {"error": "Invalid or expired access token."}
+
+    # Return info about the dataset this token grants access to
+    dataset_id = token_info["dataset_id"]
+    subscription = token_info.get("subscription", {})
+
+    # Get dataset details from registry
+    registry = datasets_registry.load_registry()
+    dataset = next((d for d in registry if d["dataset_id"] == dataset_id), None)
+
+    return {
+        "datasets": [{
+            "dataset_id": dataset_id,
+            "display_name": dataset.get("display_name", dataset_id) if dataset else dataset_id,
+            "description": dataset.get("description", "") if dataset else "",
+            "subscription_end": subscription.get("subscription_end"),
+            "plan_id": subscription.get("plan_id")
+        }],
+        "user": token_info["hf_user"]
+    }
+
+
+def _get_duckdb_connection(server_hf_token: str):
+    """Create a DuckDB connection with HF token configured."""
+    import duckdb
+    con = duckdb.connect(database=':memory:')
+    con.execute("INSTALL httpfs;")
+    con.execute("LOAD httpfs;")
+    con.execute(f"CREATE SECRET hf_token (TYPE huggingface, TOKEN '{server_hf_token}');")
+    return con
+
+
+@mcp.tool()
+def get_dataset_schema(dataset_id: str, ctx: Context):
+    """
+    Returns the schema (column names and types) of a dataset you have access to.
+    Use this to understand the data structure before querying.
+
+    Args:
+        dataset_id: The ID of the dataset (e.g., 'waroca/prompts')
+    """
+    access_token = _get_access_token_from_context(ctx)
+    token_info, error = _validate_token_for_dataset(access_token, dataset_id)
+    if error:
+        return {"error": error}
+
+    # Use server's HF token to access the dataset
+    server_hf_token = os.getenv("HF_TOKEN")
+    if not server_hf_token:
+        return {"error": "Server HF token not configured"}
+
+    # Get the correct parquet URL pattern for this dataset
+    dataset_url = datasets_registry.get_parquet_url(dataset_id)
+
+    try:
+        con = _get_duckdb_connection(server_hf_token)
+
+        # Get schema
+        result = con.execute(f"DESCRIBE SELECT * FROM read_parquet('{dataset_url}')").fetchall()
+        schema = [{"column": row[0], "type": row[1]} for row in result]
+
+        # Get row count
+        count_result = con.execute(f"SELECT COUNT(*) FROM read_parquet('{dataset_url}')").fetchone()
+        row_count = count_result[0] if count_result else 0
+
+        return {
+            "dataset_id": dataset_id,
+            "schema": schema,
+            "row_count": row_count
+        }
+    except Exception as e:
+        return {"error": f"Failed to get schema: {str(e)}"}
+
+
+@mcp.tool()
+def get_dataset_sample(dataset_id: str, ctx: Context, num_rows: int = 5):
+    """
+    Returns a sample of rows from a dataset you have access to.
+    Use this to preview the data before running queries.
+
+    Args:
+        dataset_id: The ID of the dataset (e.g., 'waroca/prompts')
+        num_rows: Number of sample rows to return (default: 5, max: 20)
+    """
+    access_token = _get_access_token_from_context(ctx)
+    token_info, error = _validate_token_for_dataset(access_token, dataset_id)
+    if error:
+        return {"error": error}
+
+    # Limit rows
+    num_rows = min(max(1, num_rows), 20)
+
+    server_hf_token = os.getenv("HF_TOKEN")
+    if not server_hf_token:
+        return {"error": "Server HF token not configured"}
+
+    # Get the correct parquet URL pattern for this dataset
+    dataset_url = datasets_registry.get_parquet_url(dataset_id)
+
+    try:
+        con = _get_duckdb_connection(server_hf_token)
+        result = con.execute(f"SELECT * FROM read_parquet('{dataset_url}') LIMIT {num_rows}").fetchdf()
+        return {
+            "dataset_id": dataset_id,
+            "sample": result.to_dict(orient='records'),
+            "num_rows": len(result)
+        }
+    except Exception as e:
+        return {"error": f"Failed to get sample: {str(e)}"}
+
+
+@mcp.tool()
+def query_dataset(dataset_id: str, query: str, ctx: Context):
+    """
+    Execute a SQL query against a dataset you have access to.
+    The dataset is available as a table named 'data'.
+
+    Args:
+        dataset_id: The ID of the dataset (e.g., 'waroca/prompts')
+        query: SQL query to execute. Use 'data' as the table name.
+               Example: "SELECT * FROM data WHERE column > 10 LIMIT 100"
+    """
+    access_token = _get_access_token_from_context(ctx)
+    token_info, error = _validate_token_for_dataset(access_token, dataset_id)
+    if error:
+        return {"error": error}
+
+    server_hf_token = os.getenv("HF_TOKEN")
+    if not server_hf_token:
+        return {"error": "Server HF token not configured"}
+
+    # Get the correct parquet URL pattern for this dataset
+    dataset_url = datasets_registry.get_parquet_url(dataset_id)
+
+    try:
+        con = _get_duckdb_connection(server_hf_token)
+        con.execute(f"CREATE OR REPLACE VIEW data AS SELECT * FROM read_parquet('{dataset_url}');")
+        result = con.execute(query).fetchdf()
+        # Wrap in dict to ensure proper JSON serialization
+        return {"results": result.to_dict(orient='records'), "row_count": len(result)}
+    except Exception as e:
+        return {"error": f"Query execution failed: {str(e)}"}
+
+
+@mcp.tool()
+def query_dataset_natural_language(dataset_id: str, question: str, ctx: Context):
+    """
+    Query a dataset using natural language. The question will be converted to SQL automatically.
+
+    Args:
+        dataset_id: The ID of the dataset (e.g., 'waroca/prompts')
+        question: Your question in plain English.
+                  Example: "What are the top 10 most common values in the category column?"
+    """
+    access_token = _get_access_token_from_context(ctx)
+    token_info, error = _validate_token_for_dataset(access_token, dataset_id)
+    if error:
+        return {"error": error}
+
+    server_hf_token = os.getenv("HF_TOKEN")
+    if not server_hf_token:
+        return {"error": "Server HF token not configured"}
+
+    # Get the correct parquet URL pattern for this dataset
+    dataset_url = datasets_registry.get_parquet_url(dataset_id)
+
+    try:
+        con = _get_duckdb_connection(server_hf_token)
+
+        # Get schema for context
+        schema_result = con.execute(f"DESCRIBE SELECT * FROM read_parquet('{dataset_url}')").fetchall()
+        schema_str = "\n".join([f"  - {row[0]}: {row[1]}" for row in schema_result])
+
+        # Generate SQL using LLM
+        from huggingface_hub import InferenceClient
+        client = InferenceClient(token=server_hf_token)
+
+        system_prompt = "You are a SQL expert. Convert natural language questions into DuckDB SQL queries. Return ONLY the SQL query, nothing else. Do not explain."
+
+        user_prompt = f"""Table name: data
+Schema:
+{schema_str}
+
+Question: {question}
+
+Return ONLY the SQL query, nothing else."""
+
+        response = client.chat_completion(
+            model="Qwen/Qwen2.5-Coder-32B-Instruct",
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            max_tokens=300
+        )
+        generated_sql = response.choices[0].message.content.strip()
+
+        # Clean up if model adds markdown
+        if "```sql" in generated_sql:
+            generated_sql = generated_sql.split("```sql")[1].split("```")[0].strip()
+        elif "```" in generated_sql:
+            generated_sql = generated_sql.split("```")[1].split("```")[0].strip()
+
+        # Remove trailing semicolon if present (we'll add it if needed)
+        generated_sql = generated_sql.rstrip(';').strip()
+
+        # Execute the query
+        con.execute(f"CREATE OR REPLACE VIEW data AS SELECT * FROM read_parquet('{dataset_url}');")
+        result = con.execute(generated_sql).fetchdf()
+
+        return {
+            "question": question,
+            "generated_sql": generated_sql,
+            "result": result.to_dict(orient='records')
+        }
+    except Exception as e:
+        return {"error": f"Natural language query failed: {str(e)}"}
+
+
+# =============================================================================
+# Admin Helper Functions (used by HTTP API endpoints, not exposed as MCP tools)
+# =============================================================================
+
+def _admin_get_subscriber_stats():
+    """Returns subscriber statistics."""
+    all_subs = subscriptions_ledger.get_all_subscriptions()
+    stats = {}
+    active_count = 0
+
+    from datetime import datetime
+
+    for sub in all_subs.values():
+        d_id = sub["dataset_id"]
+        if d_id not in stats:
+            stats[d_id] = {"total": 0, "active": 0}
+
+        stats[d_id]["total"] += 1
+
+        end_str = sub.get("subscription_end", "")
+        if end_str:
+            try:
+                if end_str.endswith("Z"):
+                    end_str = end_str[:-1]
+                end_date = datetime.fromisoformat(end_str)
+                if end_date > datetime.utcnow():
+                    stats[d_id]["active"] += 1
+                    active_count += 1
+            except:
+                pass
+
+    return {
+        "by_dataset": stats,
+        "total_active": active_count,
+        "total_subscriptions": len(all_subs)
+    }
+
+
+def _admin_get_subscribers_for_dataset(dataset_id: str):
+    """Returns list of subscribers for a dataset."""
+    all_subs = subscriptions_ledger.get_all_subscriptions()
+    subscribers = []
+
+    from datetime import datetime
+
+    for sub in all_subs.values():
+        if sub["dataset_id"] == dataset_id:
+            end_str = sub.get("subscription_end", "")
+            is_active = False
+            if end_str:
+                try:
+                    if end_str.endswith("Z"):
+                        end_str = end_str[:-1]
+                    end_date = datetime.fromisoformat(end_str)
+                    is_active = end_date > datetime.utcnow()
+                except:
+                    pass
+
+            subscribers.append({
+                **sub,
+                "is_active": is_active
+            })
+
+    return subscribers
+
+
+def _admin_update_dataset_registry(dataset_id: str, action: str, data: str):
+    """Updates the dataset registry."""
+    registry = datasets_registry.load_registry()
+
+    if action == "add":
+        new_dataset = json.loads(data)
+        registry.append(new_dataset)
+    elif action == "toggle_active":
+        for d in registry:
+            if d["dataset_id"] == dataset_id:
+                d["is_active"] = not d.get("is_active", False)
+    elif action == "delete":
+        registry = [d for d in registry if d["dataset_id"] != dataset_id]
+    elif action == "update":
+        updated_dataset = json.loads(data)
+        for i, d in enumerate(registry):
+            if d["dataset_id"] == dataset_id:
+                registry[i] = updated_dataset
+                break
+
+    success = datasets_registry.save_registry(registry)
+    if success:
+        return {"status": "success"}
+    else:
+        return {"error": "Failed to save registry"}
+
+
+# =============================================================================
+# HTTP API Routes (for frontend, not MCP)
+# =============================================================================
+
+@mcp.custom_route("/webhook", methods=["POST"])
+async def webhook_handler(request: Request):
+    return await stripe_webhook.handle_stripe_webhook(request)
+
+
+async def _subscribe_free_logic(dataset_id: str, hf_token: str):
+    user_info = auth.validate_hf_token(hf_token)
+    if not user_info:
+        return {"error": "Invalid HF token"}
+
+    hf_user = user_info["name"]
+    registry = datasets_registry.load_registry()
+    dataset = next((d for d in registry if d["dataset_id"] == dataset_id), None)
+
+    if not dataset or not dataset.get("is_active"):
+        return {"error": "Dataset not found or inactive"}
+
+    free_plan = datasets_registry.get_free_plan(dataset_id)
+
+    if not free_plan:
+        return {"error": "No free plan available for this dataset."}
+
+    from datetime import datetime, timedelta
+
+    duration_days = free_plan.get("access_duration_days", 1)  # Default to 1-day trial
+    end_date = datetime.utcnow() + timedelta(days=duration_days)
+    access_token = subscriptions_ledger.generate_access_token()
+
+    ledger_entry = {
+        "event_id": f"free_{hf_user}_{datetime.utcnow().timestamp()}",
+        "hf_user": hf_user,
+        "dataset_id": dataset_id,
+        "plan_id": free_plan["plan_id"],
+        "subscription_start": datetime.utcnow().isoformat() + "Z",
+        "subscription_end": end_date.isoformat() + "Z",
+        "source": "free_tier",
+        "access_token": access_token,
+        "created_at": datetime.utcnow().isoformat() + "Z"
+    }
+
+    subscriptions_ledger.append_subscription_event(ledger_entry)
+    return {"status": "success", "message": "Subscribed successfully", "access_token": access_token}
+
+
+async def _create_checkout_session_logic(dataset_id: str, hf_token: str):
+    user_info = auth.validate_hf_token(hf_token)
+    if not user_info:
+        return {"error": "Invalid HF token"}
+
+    hf_user = user_info["name"]
+    registry = datasets_registry.load_registry()
+    dataset = next((d for d in registry if d["dataset_id"] == dataset_id), None)
+
+    if not dataset or not dataset.get("is_active"):
+        return {"error": "Dataset not found or inactive"}
+
+    if not dataset.get("plans"):
+        return {"error": "No plans available for this dataset"}
+
+    plan = dataset["plans"][0]
+    price_id = plan.get("stripe_price_id")
+
+    if price_id in ["free", "0", 0]:
+        return {"error": "This is a free dataset, use subscribe_free instead"}
+
+    if not price_id:
+        return {"error": "Price ID not configured"}
+
+    try:
+        import stripe
+        stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
+
+        checkout_session = stripe.checkout.Session.create(
+            payment_method_types=['card'],
+            line_items=[{'price': price_id, 'quantity': 1}],
+            mode='subscription',
+            metadata={'hf_user': hf_user, 'dataset_id': dataset_id},
+            success_url='https://huggingface.co/spaces/waroca/monetization-frontend?success=true',
+            cancel_url='https://huggingface.co/spaces/waroca/monetization-frontend?canceled=true',
+        )
+        return {"checkout_url": checkout_session.url}
+    except Exception as e:
+        return {"error": str(e)}
+
+
+@mcp.custom_route("/api/subscribe_free", methods=["POST"])
+async def api_subscribe_free(request: Request):
+    data = await request.json()
+    result = await _subscribe_free_logic(data.get("dataset_id"), data.get("hf_token"))
+    return JSONResponse(result)
+
+
+@mcp.custom_route("/api/create_checkout_session", methods=["POST"])
+async def api_create_checkout_session(request: Request):
+    data = await request.json()
+    result = await _create_checkout_session_logic(data.get("dataset_id"), data.get("hf_token"))
+    return JSONResponse(result)
+
+
+@mcp.custom_route("/api/user_subscriptions", methods=["POST"])
+async def api_user_subscriptions(request: Request):
+    """Get subscriptions for the current user."""
+    data = await request.json()
+    hf_user = data.get("hf_user")
+    hf_token = data.get("hf_token")
+
+    if not hf_user:
+        return JSONResponse({"error": "hf_user required"}, status_code=400)
+
+    if not hf_token:
+        return JSONResponse({"error": "hf_token required"}, status_code=401)
+
+    user_info = auth.validate_hf_token(hf_token)
+    if not user_info:
+        return JSONResponse({"error": "Invalid HF token"}, status_code=401)
+
+    if user_info.get("name") != hf_user:
+        return JSONResponse({"error": "Token does not match user"}, status_code=403)
+
+    subs = subscriptions_ledger.get_user_subscriptions(hf_user)
+    return JSONResponse(subs)
+
+
+@mcp.custom_route("/api/catalog", methods=["GET"])
+async def api_catalog(request: Request):
+    """Public endpoint for getting the dataset catalog."""
+    catalog = get_dataset_catalog()
+    return JSONResponse(catalog)
+
+
+# =============================================================================
+# Admin API Routes
+# =============================================================================
+
+@mcp.custom_route("/api/admin/subscriber_stats", methods=["POST"])
+async def api_admin_subscriber_stats(request: Request):
+    data = await request.json()
+    admin_secret = get_admin_secret_from_request(request, data)
+
+    if not verify_admin_secret(admin_secret):
+        return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+    result = _admin_get_subscriber_stats()
+    return JSONResponse(result)
+
+
+@mcp.custom_route("/api/admin/subscribers_for_dataset", methods=["POST"])
+async def api_admin_subscribers_for_dataset(request: Request):
+    data = await request.json()
+    admin_secret = get_admin_secret_from_request(request, data)
+    dataset_id = data.get("dataset_id")
+
+    if not verify_admin_secret(admin_secret):
+        return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+    if not dataset_id:
+        return JSONResponse({"error": "dataset_id required"}, status_code=400)
+
+    result = _admin_get_subscribers_for_dataset(dataset_id)
+    return JSONResponse(result)
+
+
+@mcp.custom_route("/api/admin/update_registry", methods=["POST"])
+async def api_admin_update_registry(request: Request):
+    data = await request.json()
+    admin_secret = get_admin_secret_from_request(request, data)
+    dataset_id = data.get("dataset_id", "")
+    action = data.get("action")
+    payload = data.get("data", "")
+
+    if not verify_admin_secret(admin_secret):
+        return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+    if not action:
+        return JSONResponse({"error": "action required"}, status_code=400)
+
+    result = _admin_update_dataset_registry(dataset_id, action, payload)
+    return JSONResponse(result)
+
+
+@mcp.custom_route("/api/admin/catalog", methods=["POST"])
+async def api_admin_catalog(request: Request):
+    data = await request.json()
+    admin_secret = get_admin_secret_from_request(request, data)
+
+    if not verify_admin_secret(admin_secret):
+        return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+    registry = datasets_registry.load_registry()
+    return JSONResponse(registry)
+
+
+@mcp.custom_route("/api/admin/detect_format", methods=["POST"])
+async def api_admin_detect_format(request: Request):
+    """Detect the format and parquet path for a HuggingFace dataset."""
+    data = await request.json()
+    admin_secret = get_admin_secret_from_request(request, data)
+    dataset_id = data.get("dataset_id")
+
+    if not verify_admin_secret(admin_secret):
+        return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+    if not dataset_id:
+        return JSONResponse({"error": "dataset_id required"}, status_code=400)
+
+    result = datasets_registry.detect_dataset_format(dataset_id)
+    return JSONResponse(result)
+
+
+# =============================================================================
+# App Initialization
+# =============================================================================
+
+app = mcp.sse_app()
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

stripe_webhook.py

@@ -0,0 +1,142 @@
+import stripe
+import os
+import json
+from fastapi import Request, HTTPException
+from datetime import datetime, timedelta
+import datasets_registry
+import subscriptions_ledger
+
+from fastapi.responses import JSONResponse
+
+async def handle_stripe_webhook(request: Request):
+    payload = await request.body()
+    sig_header = request.headers.get("stripe-signature")
+    webhook_secret = os.getenv("STRIPE_WEBHOOK_SECRET")
+    
+    try:
+        event = stripe.Webhook.construct_event(
+            payload, sig_header, webhook_secret
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail="Invalid payload")
+    except stripe.error.SignatureVerificationError as e:
+        raise HTTPException(status_code=400, detail="Invalid signature")
+        
+    event_type = event["type"]
+    data_object = event["data"]["object"]
+    
+    if event_type == "checkout.session.completed":
+        await handle_checkout_session(data_object, event["id"])
+    elif event_type == "invoice.paid":
+        await handle_invoice_paid(data_object, event["id"])
+        
+    return JSONResponse({"status": "success"})
+
+async def handle_checkout_session(session, event_id):
+    # Metadata should contain hf_user and dataset_id
+    metadata = session.get("metadata", {})
+    hf_user = metadata.get("hf_user")
+    dataset_id = metadata.get("dataset_id")
+    
+    if not hf_user or not dataset_id:
+        print(f"Missing metadata in session {session['id']}")
+        return
+
+    # Determine plan and duration
+    # In a real scenario, we might look up the line items to find the price ID
+    # For simplicity, we assume the first line item's price ID matches our registry
+    # Or we can pass the plan_id in metadata too.
+    
+    # Let's try to get price_id from the session if possible, or rely on metadata
+    # If subscription mode, we might need to fetch the subscription details
+    subscription_id = session.get("subscription")
+    if subscription_id:
+        # It's a subscription
+        sub = stripe.Subscription.retrieve(subscription_id)
+        price_id = sub["items"]["data"][0]["price"]["id"]
+        current_period_end = sub["current_period_end"]
+        end_date = datetime.fromtimestamp(current_period_end)
+    else:
+        # One-time payment?
+        # For this hackathon, let's assume subscriptions.
+        print("Non-subscription checkout not fully supported yet.")
+        return
+
+    plan_info = datasets_registry.get_plan_by_price_id(price_id)
+    if not plan_info:
+        print(f"Unknown price_id {price_id}")
+        return
+        
+    plan_id = plan_info["plan"]["plan_id"]
+
+    # Generate unique access token for this subscription
+    access_token = subscriptions_ledger.generate_access_token()
+
+    ledger_entry = {
+        "event_id": event_id,
+        "hf_user": hf_user,
+        "dataset_id": dataset_id,
+        "plan_id": plan_id,
+        "subscription_start": datetime.utcnow().isoformat() + "Z",
+        "subscription_end": end_date.isoformat() + "Z",
+        "source": "stripe",
+        "access_token": access_token,
+        "created_at": datetime.utcnow().isoformat() + "Z",
+        "stripe_customer_id": session.get("customer"),
+        "stripe_subscription_id": subscription_id
+    }
+
+    subscriptions_ledger.append_subscription_event(ledger_entry)
+
+async def handle_invoice_paid(invoice, event_id):
+    subscription_id = invoice.get("subscription")
+    if not subscription_id:
+        return
+        
+    # We need to find the user and dataset associated with this subscription
+    # We can query Stripe or look up in our ledger if we stored subscription_id
+    # For now, let's assume we can get it from the subscription metadata in Stripe
+    
+    sub = stripe.Subscription.retrieve(subscription_id)
+    metadata = sub.get("metadata", {})
+    hf_user = metadata.get("hf_user")
+    dataset_id = metadata.get("dataset_id")
+    
+    if not hf_user or not dataset_id:
+        # Try to find from previous ledger entries? 
+        # For simplicity, we assume metadata is preserved on the subscription object in Stripe
+        print(f"Missing metadata in subscription {subscription_id}")
+        return
+
+    price_id = sub["items"]["data"][0]["price"]["id"]
+    current_period_end = sub["current_period_end"]
+    end_date = datetime.fromtimestamp(current_period_end)
+    
+    plan_info = datasets_registry.get_plan_by_price_id(price_id)
+    if not plan_info:
+        return
+
+    plan_id = plan_info["plan"]["plan_id"]
+
+    # For renewals, try to preserve the existing access token
+    existing_sub = subscriptions_ledger.get_active_subscription(hf_user, dataset_id)
+    if existing_sub and existing_sub.get("access_token"):
+        access_token = existing_sub["access_token"]
+    else:
+        access_token = subscriptions_ledger.generate_access_token()
+
+    ledger_entry = {
+        "event_id": event_id,
+        "hf_user": hf_user,
+        "dataset_id": dataset_id,
+        "plan_id": plan_id,
+        "subscription_start": datetime.utcnow().isoformat() + "Z",  # Or period start
+        "subscription_end": end_date.isoformat() + "Z",
+        "source": "stripe",
+        "access_token": access_token,
+        "created_at": datetime.utcnow().isoformat() + "Z",
+        "stripe_customer_id": invoice.get("customer"),
+        "stripe_subscription_id": subscription_id
+    }
+
+    subscriptions_ledger.append_subscription_event(ledger_entry)

subscriptions_ledger.py

@@ -0,0 +1,252 @@
+import json
+import os
+import secrets
+import tempfile
+from datetime import datetime
+from typing import Dict, Any, Optional
+from huggingface_hub import HfApi, hf_hub_download
+from huggingface_hub.utils import EntryNotFoundError
+
+# Configuration for HF Dataset-based ledger
+LEDGER_REPO = os.getenv("LEDGER_DATASET_ID", "")
+LEDGER_FILE = "subscriptions.jsonl"
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+# Fallback to local file if LEDGER_DATASET_ID not set (for local dev)
+LOCAL_LEDGER_FILE = "../subscriptions_ledger/subscriptions.jsonl"
+
+# Initialize HF API
+api = HfApi(token=HF_TOKEN) if HF_TOKEN else None
+
+
+def _use_hf_storage() -> bool:
+    """Check if we should use HF Dataset storage."""
+    return bool(LEDGER_REPO and HF_TOKEN and api)
+
+
+def _download_ledger() -> Optional[str]:
+    """Download current ledger from HF Dataset."""
+    if not _use_hf_storage():
+        return None
+
+    try:
+        path = hf_hub_download(
+            repo_id=LEDGER_REPO,
+            filename=LEDGER_FILE,
+            repo_type="dataset",
+            token=HF_TOKEN
+        )
+        return path
+    except EntryNotFoundError:
+        # File doesn't exist yet in the dataset
+        return None
+    except Exception as e:
+        print(f"Error downloading ledger: {e}")
+        return None
+
+
+def _upload_ledger(local_path: str) -> bool:
+    """Upload ledger to HF Dataset."""
+    if not _use_hf_storage():
+        return False
+
+    try:
+        api.upload_file(
+            path_or_fileobj=local_path,
+            path_in_repo=LEDGER_FILE,
+            repo_id=LEDGER_REPO,
+            repo_type="dataset",
+            token=HF_TOKEN,
+            commit_message=f"Update subscriptions ledger - {datetime.utcnow().isoformat()}"
+        )
+        return True
+    except Exception as e:
+        print(f"Error uploading ledger: {e}")
+        return False
+
+
+def _get_ledger_path() -> str:
+    """Get the path to read the ledger from."""
+    if _use_hf_storage():
+        hf_path = _download_ledger()
+        if hf_path:
+            return hf_path
+
+    # Fallback to local file
+    return LOCAL_LEDGER_FILE
+
+
+def append_subscription_event(event: Dict[str, Any]) -> bool:
+    """
+    Appends a subscription event to the ledger.
+    If using HF Dataset, downloads, appends, and re-uploads.
+    """
+    # Ensure timestamp is present
+    if "created_at" not in event:
+        event["created_at"] = datetime.utcnow().isoformat() + "Z"
+
+    if _use_hf_storage():
+        # Download current ledger
+        current_path = _download_ledger()
+
+        # Create temp file to work with
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.jsonl', delete=False) as tmp:
+            tmp_path = tmp.name
+
+            # Copy existing content if any
+            if current_path and os.path.exists(current_path):
+                with open(current_path, 'r') as f:
+                    tmp.write(f.read())
+
+            # Append new event
+            tmp.write(json.dumps(event) + "\n")
+
+        # Upload back to HF
+        success = _upload_ledger(tmp_path)
+
+        # Clean up temp file
+        try:
+            os.unlink(tmp_path)
+        except:
+            pass
+
+        return success
+    else:
+        # Local file storage
+        parent_dir = os.path.dirname(LOCAL_LEDGER_FILE)
+        if parent_dir:
+            os.makedirs(parent_dir, exist_ok=True)
+        with open(LOCAL_LEDGER_FILE, "a") as f:
+            f.write(json.dumps(event) + "\n")
+        return True
+
+
+def get_all_subscriptions() -> Dict[tuple, Dict[str, Any]]:
+    """
+    Reads the ledger and folds events to find the latest state for each (user, dataset).
+    Returns a dict: key=(hf_user, dataset_id), value=subscription_record
+    """
+    subscriptions = {}
+
+    ledger_path = _get_ledger_path()
+
+    if not ledger_path or not os.path.exists(ledger_path):
+        return subscriptions
+
+    try:
+        with open(ledger_path, "r") as f:
+            for line in f:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    event = json.loads(line)
+                    hf_user = event.get("hf_user")
+                    dataset_id = event.get("dataset_id")
+
+                    if hf_user and dataset_id:
+                        key = (hf_user, dataset_id)
+                        # Append-only ledger: latest record is current state
+                        subscriptions[key] = event
+                except json.JSONDecodeError:
+                    continue
+    except Exception as e:
+        print(f"Error reading ledger: {e}")
+
+    return subscriptions
+
+
+def get_active_subscription(hf_user: str, dataset_id: str) -> Optional[Dict[str, Any]]:
+    """Checks if a user has an active subscription to a dataset."""
+    all_subs = get_all_subscriptions()
+    sub = all_subs.get((hf_user, dataset_id))
+
+    if not sub:
+        return None
+
+    # Check expiry
+    subscription_end = sub.get("subscription_end")
+    if not subscription_end:
+        return None
+
+    try:
+        # Handle Z suffix if present
+        if subscription_end.endswith("Z"):
+            subscription_end = subscription_end[:-1]
+        end_date = datetime.fromisoformat(subscription_end)
+        if end_date > datetime.utcnow():
+            return sub
+    except ValueError:
+        print(f"Error parsing date: {subscription_end}")
+        return None
+
+    return None
+
+
+def get_user_subscriptions(hf_user: str) -> list:
+    """Get all subscriptions for a specific user."""
+    all_subs = get_all_subscriptions()
+    user_subs = []
+
+    for (user, dataset_id), sub in all_subs.items():
+        if user == hf_user:
+            # Add active status
+            end_str = sub.get("subscription_end", "")
+            is_active = False
+            if end_str:
+                try:
+                    if end_str.endswith("Z"):
+                        end_str = end_str[:-1]
+                    end_date = datetime.fromisoformat(end_str)
+                    is_active = end_date > datetime.utcnow()
+                except:
+                    pass
+
+            user_subs.append({
+                **sub,
+                "is_active": is_active
+            })
+
+    return user_subs
+
+
+def generate_access_token() -> str:
+    """Generate a secure random access token."""
+    return f"hfdata_{secrets.token_urlsafe(32)}"
+
+
+def validate_access_token(access_token: str) -> Optional[Dict[str, Any]]:
+    """
+    Validate an access token and return the subscription info if valid.
+    Returns None if token is invalid or subscription expired.
+    """
+    all_subs = get_all_subscriptions()
+
+    for (hf_user, dataset_id), sub in all_subs.items():
+        if sub.get("access_token") == access_token:
+            # Check if subscription is still active
+            end_str = sub.get("subscription_end", "")
+            if end_str:
+                try:
+                    if end_str.endswith("Z"):
+                        end_str = end_str[:-1]
+                    end_date = datetime.fromisoformat(end_str)
+                    if end_date > datetime.utcnow():
+                        return {
+                            "hf_user": hf_user,
+                            "dataset_id": dataset_id,
+                            "subscription": sub
+                        }
+                except:
+                    pass
+            return None  # Token found but subscription expired
+
+    return None  # Token not found
+
+
+def get_subscription_by_token(access_token: str) -> Optional[Dict[str, Any]]:
+    """
+    Get subscription details by access token.
+    Alias for validate_access_token for clarity.
+    """
+    return validate_access_token(access_token)