app.py

# app.py - Simplified version that definitely works on HF Spaces
import os
import re
import json
from datetime import datetime
from typing import List, Dict, Any

import gradio as gr
import requests
import asyncio
from threading import Thread

# Configuration
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
HF_TOKEN = os.getenv("HF_TOKEN")

# Storage for operations
tag_operations_store: List[Dict[str, Any]] = []

# Recognized tags
RECOGNIZED_TAGS = {
    "pytorch", "tensorflow", "jax", "transformers", "diffusers",
    "text-generation", "text-classification", "question-answering",
    "text-to-image", "image-classification", "object-detection",
    "fill-mask", "token-classification", "translation", "summarization",
    "feature-extraction", "sentence-similarity", "zero-shot-classification",
    "image-to-text", "automatic-speech-recognition", "audio-classification",
    "voice-activity-detection", "depth-estimation", "image-segmentation",
    "video-classification", "reinforcement-learning", "tabular-classification",
    "tabular-regression", "time-series-forecasting", "graph-ml", "robotics",
    "computer-vision", "nlp", "cv", "multimodal", "gguf", "safetensors",
    "llamacpp", "onnx", "mlx", "machine-learning", "deep-learning"
}

def extract_tags_from_text(text: str) -> List[str]:
    """Extract potential tags from discussion text"""
    text_lower = text.lower()
    explicit_tags = []

    # Pattern 1: "tag: something" or "tags: something"
    tag_pattern = r"tags?:\s*([a-zA-Z0-9-_,\s]+)"
    matches = re.findall(tag_pattern, text_lower)
    for match in matches:
        tags = [tag.strip() for tag in match.split(",")]
        explicit_tags.extend(tags)

    # Pattern 2: "#hashtag" style
    hashtag_pattern = r"#([a-zA-Z0-9-_]+)"
    hashtag_matches = re.findall(hashtag_pattern, text_lower)
    explicit_tags.extend(hashtag_matches)

    # Pattern 3: Look for recognized tags mentioned in natural text
    mentioned_tags = []
    for tag in RECOGNIZED_TAGS:
        if tag in text_lower:
            mentioned_tags.append(tag)

    # Combine and deduplicate
    all_tags = list(set(explicit_tags + mentioned_tags))

    # Filter to only include recognized tags or explicitly mentioned ones
    valid_tags = []
    for tag in all_tags:
        if tag in RECOGNIZED_TAGS or tag in explicit_tags:
            valid_tags.append(tag)

    return valid_tags

def process_tags_sync(all_tags: List[str], repo_name: str) -> List[str]:
    """Process tags using direct HuggingFace Hub API calls (synchronous)"""
    print(f"🔧 Processing tags: {all_tags} for repo: {repo_name}")
    result_messages = []
    
    if not HF_TOKEN:
        error_msg = "No HF_TOKEN configured"
        print(f"❌ {error_msg}")
        return [error_msg]
    
    try:
        from huggingface_hub import HfApi, model_info, dataset_info, space_info, ModelCard, ModelCardData
        from huggingface_hub.utils import HfHubHTTPError
        from huggingface_hub import CommitOperationAdd
        
        hf_api = HfApi(token=HF_TOKEN)
        
        # Determine repository type
        repo_type = None
        repo_info = None
        
        for repo_type_to_try in ["model", "dataset", "space"]:
            try:
                print(f"🔍 Trying to access {repo_name} as {repo_type_to_try}...")
                if repo_type_to_try == "model":
                    repo_info = model_info(repo_id=repo_name, token=HF_TOKEN)
                elif repo_type_to_try == "dataset":
                    repo_info = dataset_info(repo_id=repo_name, token=HF_TOKEN)
                elif repo_type_to_try == "space":
                    repo_info = space_info(repo_id=repo_name, token=HF_TOKEN)
                
                repo_type = repo_type_to_try
                print(f"✅ Found repository as {repo_type}")
                break
                
            except HfHubHTTPError as e:
                if "404" in str(e):
                    continue
                else:
                    print(f"❌ Error accessing as {repo_type_to_try}: {e}")
                    continue
            except Exception as e:
                print(f"❌ Unexpected error for {repo_type_to_try}: {e}")
                continue
        
        if not repo_type or not repo_info:
            error_msg = f"Repository '{repo_name}' not found"
            return [f"Error: {error_msg}"]
        
        current_tags = repo_info.tags if repo_info.tags else []
        print(f"🏷️ Current tags: {current_tags}")
        
        # Process each tag
        for tag in all_tags:
            try:
                if tag in current_tags:
                    msg = f"Tag '{tag}': Already exists"
                    result_messages.append(msg)
                    continue
                
                # Add the new tag
                updated_tags = current_tags + [tag]
                
                # Create/update model card
                try:
                    card = ModelCard.load(repo_name, token=HF_TOKEN, repo_type=repo_type)
                    if not hasattr(card, "data") or card.data is None:
                        card.data = ModelCardData()
                except HfHubHTTPError:
                    card = ModelCard("")
                    card.data = ModelCardData()
                
                # Update tags
                card_dict = card.data.to_dict()
                card_dict["tags"] = updated_tags
                card.data = ModelCardData(**card_dict)
                
                pr_title = f"Add '{tag}' tag"
                pr_description = f"""
## Add tag: {tag}

This PR adds the `{tag}` tag to the {repo_type} repository.

**Changes:**
- Added `{tag}` to {repo_type} tags
- Updated from {len(current_tags)} to {len(updated_tags)} tags

**Current tags:** {", ".join(current_tags) if current_tags else "None"}
**New tags:** {", ".join(updated_tags)}

*This PR was created automatically by the HF Tagging Bot.*
"""
                
                commit_info = hf_api.create_commit(
                    repo_id=repo_name,
                    repo_type=repo_type,
                    operations=[
                        CommitOperationAdd(
                            path_in_repo="README.md", 
                            path_or_fileobj=str(card).encode("utf-8")
                        )
                    ],
                    commit_message=pr_title,
                    commit_description=pr_description,
                    token=HF_TOKEN,
                    create_pr=True,
                )
                
                pr_url = getattr(commit_info, 'pr_url', str(commit_info))
                msg = f"Tag '{tag}': [PR created]({pr_url})"
                result_messages.append(msg)
                
            except Exception as tag_error:
                error_msg = f"Tag '{tag}': Error - {str(tag_error)}"
                result_messages.append(error_msg)
        
        return result_messages
        
    except Exception as e:
        error_msg = f"Processing failed: {str(e)}"
        return [error_msg]

def process_webhook_comment(webhook_data: Dict[str, Any]):
    """Process webhook to detect and add tags"""
    try:
        comment_content = webhook_data["comment"]["content"]
        discussion_title = webhook_data["discussion"]["title"]
        repo_name = webhook_data["repo"]["name"]
        discussion_num = webhook_data["discussion"]["num"]
        comment_author = webhook_data["comment"]["author"].get("id", "unknown")

        print(f"📝 Processing comment from {comment_author} on {repo_name}")
        print(f"📝 Comment: {comment_content}")

        # Extract tags
        comment_tags = extract_tags_from_text(comment_content)
        title_tags = extract_tags_from_text(discussion_title)
        all_tags = list(set(comment_tags + title_tags))

        if not all_tags:
            print("❌ No tags found")
            return "No recognizable tags found"

        print(f"🏷️ Found tags: {all_tags}")
        result_messages = process_tags_sync(all_tags, repo_name)

        # Store interaction
        interaction = {
            "timestamp": datetime.now().isoformat(),
            "repo": repo_name,
            "discussion_title": discussion_title,
            "discussion_num": discussion_num,
            "comment_author": comment_author,
            "detected_tags": all_tags,
            "results": result_messages,
        }
        tag_operations_store.append(interaction)

        # Keep only last 100 operations
        if len(tag_operations_store) > 100:
            tag_operations_store.pop(0)

        return " | ".join(result_messages)

    except Exception as e:
        error_msg = f"Error processing webhook: {str(e)}"
        print(f"❌ {error_msg}")
        return error_msg

def handle_webhook_request(request: gr.Request):
    """Handle webhook via Gradio's request handling"""
    try:
        print(f"📥 Received request: {request.method}")
        
        if request.method != "POST":
            return {"error": "Method not allowed"}
        
        # Get request data
        if hasattr(request, 'json') and callable(request.json):
            payload = request.json()
        else:
            # Fallback for different request formats
            return {"error": "Could not parse JSON"}
        
        print(f"📥 Webhook payload: {payload.get('event', {})}")
        
        # Verify webhook secret if configured
        if WEBHOOK_SECRET:
            webhook_secret = request.headers.get("X-Webhook-Secret")
            if webhook_secret != WEBHOOK_SECRET:
                print("❌ Invalid webhook secret")
                return {"error": "Invalid webhook secret"}

        event = payload.get("event", {})
        scope = event.get("scope")
        action = event.get("action")

        # Only process discussion comment creation (not PRs)
        if (scope in ["discussion", "discussion.comment"] and 
            action == "create" and 
            not payload.get("discussion", {}).get("isPullRequest", False)):
            
            # Process webhook in a separate thread
            def process_in_background():
                process_webhook_comment(payload)
            
            thread = Thread(target=process_in_background)
            thread.start()
            
            return {"status": "processing"}

        return {"status": "ignored"}
        
    except Exception as e:
        print(f"❌ Webhook error: {e}")
        return {"error": str(e)}

def create_gradio_interface():
    """Create Gradio interface"""
    with gr.Blocks(title="HF Tagging Bot", theme=gr.themes.Soft()) as interface:
        gr.Markdown("# 🏷️ HuggingFace Tagging Bot")
        gr.Markdown("*Automatically adds tags to repositories when mentioned in discussions*")
        
        with gr.Tab("🏠 Status"):
            status_text = gr.HTML(f"""
<h2>Bot Configuration</h2>
<ul>
<li>🔑 <strong>HF Token</strong>: {'✅ Configured' if HF_TOKEN else '❌ Missing'}</li>
<li>🔐 <strong>Webhook Secret</strong>: {'✅ Configured' if WEBHOOK_SECRET else '❌ Missing'}</li>
<li>📊 <strong>Operations Processed</strong>: {len(tag_operations_store)}</li>
</ul>

<h2>Webhook Endpoint</h2>
<p><code>https://asmaa105-hf-tagging-bot.hf.space/webhook</code></p>

<h2>Setup Instructions</h2>
<ol>
<li>Go to your repository Settings → Webhooks</li>
<li>Add webhook URL above</li>
<li>Select "Discussion comments" events</li>
<li>Add your webhook secret if configured</li>
</ol>
            """)
            
        with gr.Tab("📝 Operations Log"):
            def get_recent_operations():
                if not tag_operations_store:
                    return "No operations yet. The webhook is ready to receive requests."
                
                recent = tag_operations_store[-10:]
                output = []
                for op in reversed(recent):
                    output.append(f"""
**{op['repo']}** - {op['timestamp'][:19]}
- 👤 Author: {op['comment_author']}
- 🏷️ Tags: {', '.join(op['detected_tags']) if op['detected_tags'] else 'None'}
- 📋 Results: {' | '.join(op['results'][:2])}{'...' if len(op['results']) > 2 else ''}
---""")
                return "\n".join(output)
            
            operations_display = gr.Textbox(
                label="Recent Operations",
                value=get_recent_operations(),
                lines=15,
                interactive=False
            )
            
            refresh_btn = gr.Button("🔄 Refresh Log")
            refresh_btn.click(fn=get_recent_operations, outputs=operations_display)
            
        with gr.Tab("🧪 Test Webhook"):
            gr.Markdown("### Test Webhook Processing")
            test_repo = gr.Textbox(label="Repository", value="asmaa105/streamlitweb1")
            test_comment = gr.Textbox(label="Test Comment", value="Please add tags: pytorch, transformers", lines=3)
            test_btn = gr.Button("🔧 Test Processing")
            test_result = gr.Textbox(label="Result", lines=5, interactive=False)
            
            def test_webhook_processing(repo, comment):
                try:
                    # Create mock webhook data
                    mock_webhook = {
                        "event": {"action": "create", "scope": "discussion"},
                        "comment": {"content": comment, "author": {"id": "test-user"}},
                        "discussion": {"title": "Test", "num": 1, "isPullRequest": False},
                        "repo": {"name": repo}
                    }
                    
                    result = process_webhook_comment(mock_webhook)
                    return f"✅ Test completed!\n\nResult: {result}"
                except Exception as e:
                    return f"❌ Test failed: {str(e)}"
            
            test_btn.click(fn=test_webhook_processing, inputs=[test_repo, test_comment], outputs=test_result)
            
        with gr.Tab("🏷️ Supported Tags"):
            gr.Markdown(f"""
## Supported Tags ({len(RECOGNIZED_TAGS)} total)
{', '.join(sorted(RECOGNIZED_TAGS))}

## Tag Detection Examples
- **Explicit**: `tag: pytorch` or `tags: pytorch, transformers`
- **Hashtag**: `#pytorch #transformers`
- **Natural**: "This model uses pytorch and transformers"
            """)
            
            gr.Markdown("### Test Tag Detection")
            test_input = gr.Textbox(
                label="Test Comment",
                placeholder="Enter a comment to test tag detection...",
                lines=3,
                value="Please add tags: pytorch, transformers"
            )
            test_output = gr.Textbox(
                label="Detected Tags",
                lines=2,
                interactive=False
            )
            test_btn = gr.Button("🔍 Test Detection")
            
            def test_tag_detection(text):
                if not text:
                    return "Enter some text to test"
                tags = extract_tags_from_text(text)
                if tags:
                    return f"Found {len(tags)} tags: {', '.join(tags)}"
                else:
                    return "No tags detected in this text"
            
            test_btn.click(fn=test_tag_detection, inputs=test_input, outputs=test_output)

    return interface

# Create the Gradio interface
demo = create_gradio_interface()

# Add webhook handling via Gradio's API
if hasattr(demo, 'add_api_route'):
    demo.add_api_route("/webhook", handle_webhook_request, methods=["POST"])

if __name__ == "__main__":
    print("🚀 HF Tagging Bot - Simplified version starting")
    print(f"🔑 HF_TOKEN: {'✅ Configured' if HF_TOKEN else '❌ Missing'}")
    print(f"🔐 Webhook Secret: {'✅ Configured' if WEBHOOK_SECRET else '❌ Missing'}")
    print("🌐 Webhook endpoint: /webhook")
    demo.launch(server_name="0.0.0.0", server_port=7860)