Create app.py

app.py

@@ -0,0 +1,299 @@
+import gradio as gr
+import requests
+import re
+import time
+import logging
+from functools import lru_cache
+from urllib.parse import urlparse, urljoin
+from typing import Dict, Any, Optional, List
+from dataclasses import dataclass
+from datetime import datetime
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s'
+)
+
+@dataclass
+class ProxyResponse:
+    """代理响应数据类"""
+    status: int
+    content: str
+    headers: Dict[str, str]
+    redirect_url: Optional[str] = None
+    error: Optional[str] = None
+
+class Config:
+    """配置类"""
+    ASSET_URL = "https://1pages.nbid.bid/"
+    PREFIX = "/"
+    JSDELIVR = 0
+    CACHE_TTL = 3600
+    MAX_RETRIES = 3
+    TIMEOUT = 10
+    RATE_LIMIT = {
+        "window_ms": 15 * 60 * 1000,  # 15分钟
+        "max": 100  # 限制每个IP最多100个请求
+    }
+    WHITE_LIST: List[str] = []  # 白名单
+    
+    # 请求头
+    DEFAULT_HEADERS = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+    }
+    
+    # CORS设置
+    CORS = {
+        "allow_origins": ["*"],
+        "allow_methods": ["GET", "POST", "OPTIONS"],
+        "allow_headers": ["*"],
+        "max_age": 1728000
+    }
+    
+    # URL模式
+    PATTERNS = {
+        "releases": r"^(?:https?:\/\/)?github\.com\/.+?\/.+?\/(?:releases|archive)\/.*$",
+        "blob": r"^(?:https?:\/\/)?github\.com\/.+?\/.+?\/(?:blob|raw)\/.*$",
+        "git": r"^(?:https?:\/\/)?github\.com\/.+?\/.+?\/(?:info|git-).*$",
+        "raw": r"^(?:https?:\/\/)?raw\.(?:githubusercontent|github)\.com\/.+?\/.+?\/.+?\/.+$",
+        "gist": r"^(?:https?:\/\/)?gist\.(?:githubusercontent|github)\.com\/.+?\/.+?\/.+$",
+        "tags": r"^(?:https?:\/\/)?github\.com\/.+?\/.+?\/tags.*$"
+    }
+
+class RateLimiter:
+    """请求频率限制器"""
+    def __init__(self):
+        self.request_records: Dict[str, List[float]] = {}
+    
+    def is_allowed(self, ip: str) -> bool:
+        now = time.time() * 1000
+        window_start = now - Config.RATE_LIMIT["window_ms"]
+        
+        if ip not in self.request_records:
+            self.request_records[ip] = []
+        
+        # 清理过期记录
+        self.request_records[ip] = [t for t in self.request_records[ip] if t > window_start]
+        
+        if len(self.request_records[ip]) >= Config.RATE_LIMIT["max"]:
+            return False
+        
+        self.request_records[ip].append(now)
+        return True
+    
+    def get_remaining(self, ip: str) -> int:
+        """获取剩余请求次数"""
+        if ip not in self.request_records:
+            return Config.RATE_LIMIT["max"]
+        return Config.RATE_LIMIT["max"] - len(self.request_records[ip])
+
+class GitHubProxy:
+    """GitHub代理核心类"""
+    def __init__(self):
+        self.rate_limiter = RateLimiter()
+        self.session = requests.Session()
+        self.session.headers.update(Config.DEFAULT_HEADERS)
+    
+    def check_url(self, url: str) -> bool:
+        """检查URL是否匹配GitHub模式"""
+        return any(re.search(pattern, url, re.I) for pattern in Config.PATTERNS.values())
+    
+    def check_white_list(self, url: str) -> bool:
+        """检查白名单"""
+        if not Config.WHITE_LIST:
+            return True
+        return any(white_item in url for white_item in Config.WHITE_LIST)
+    
+    @lru_cache(maxsize=1000)
+    def fetch_github_content(self, url: str, method: str = "GET") -> ProxyResponse:
+        """获取GitHub内容（带缓存）"""
+        try:
+            response = self.session.request(
+                method=method,
+                url=url,
+                timeout=Config.TIMEOUT,
+                allow_redirects=False
+            )
+            
+            headers = dict(response.headers)
+            
+            # 处理重定向
+            if response.is_redirect:
+                redirect_url = response.headers["Location"]
+                if self.check_url(redirect_url):
+                    redirect_url = Config.PREFIX + redirect_url
+                return ProxyResponse(
+                    status=response.status_code,
+                    content="",
+                    headers=headers,
+                    redirect_url=redirect_url
+                )
+            
+            return ProxyResponse(
+                status=response.status_code,
+                content=response.text,
+                headers=headers
+            )
+            
+        except requests.Timeout:
+            return ProxyResponse(
+                status=504,
+                content="Request Timeout",
+                headers={},
+                error="请求超时"
+            )
+        except Exception as e:
+            logging.error(f"Fetch error: {str(e)}")
+            return ProxyResponse(
+                status=500,
+                content=str(e),
+                headers={},
+                error="服务器内部错误"
+            )
+    
+    def proxy_request(self, url: str, request: gr.Request) -> Dict[str, Any]:
+        """处理代理请求"""
+        # 记录请求
+        logging.info(f"Proxy request from {request.client.host} to {url}")
+        
+        # 检查频率限制
+        if not self.rate_limiter.is_allowed(request.client.host):
+            return {
+                "status": 429,
+                "content": "Too Many Requests",
+                "headers": {},
+                "error": "请求过于频繁，请稍后再试",
+                "rate_limit": {
+                    "remaining": self.rate_limiter.get_remaining(request.client.host),
+                    "reset": int((time.time() * 1000 + Config.RATE_LIMIT["window_ms"]) / 1000)
+                }
+            }
+        
+        # 检查白名单
+        if not self.check_white_list(url):
+            return {
+                "status": 403,
+                "content": "Access Denied",
+                "headers": {},
+                "error": "访问被拒绝"
+            }
+        
+        # 处理URL
+        if not url.startswith(("http://", "https://")):
+            url = "https://" + url
+        
+        # 检查URL是否为GitHub链接
+        if not self.check_url(url):
+            return {
+                "status": 400,
+                "content": "Invalid GitHub URL",
+                "headers": {},
+                "error": "无效的GitHub URL"
+            }
+        
+        # 处理jsDelivr重定向
+        if Config.JSDELIVR and re.search(Config.PATTERNS["blob"], url):
+            url = url.replace("/blob/", "@").replace("github.com", "cdn.jsdelivr.net/gh")
+            return {
+                "status": 302,
+                "content": "",
+                "headers": {"Location": url},
+                "redirect_url": url
+            }
+        
+        # 获取内容
+        response = self.fetch_github_content(url)
+        result = {
+            "status": response.status,
+            "content": response.content,
+            "headers": response.headers,
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        if response.redirect_url:
+            result["redirect_url"] = response.redirect_url
+        if response.error:
+            result["error"] = response.error
+            
+        return result
+
+def create_interface():
+    """创建Gradio界面"""
+    proxy = GitHubProxy()
+    
+    with gr.Blocks(title="GitHub Proxy", theme=gr.themes.Soft()) as app:
+        gr.Markdown("""
+        # 🚀 GitHub Proxy
+        
+        ### 功能特点
+        - ✨ 支持多种GitHub URL格式
+        - 🔄 自动处理重定向
+        - 💾 响应缓存
+        - ⚡ CDN加速支持
+        - 🛡️ 请求频率限制
+        
+        ### 支持的URL类型
+        - GitHub Release/Archive
+        - GitHub Raw/Blob
+        - GitHub Gist
+        - Raw GitHub Content
+        """)
+        
+        with gr.Row():
+            url_input = gr.Textbox(
+                label="GitHub URL",
+                placeholder="输入GitHub URL，例如：github.com/user/repo/blob/master/file.txt",
+                scale=4
+            )
+            submit_btn = gr.Button("获取内容", scale=1)
+        
+        with gr.Row():
+            with gr.Column():
+                status = gr.Textbox(label="状态")
+                headers = gr.JSON(label="响应头")
+            with gr.Column():
+                content = gr.Textbox(label="内容", max_lines=20)
+                error = gr.Textbox(label="错误信息", visible=False)
+        
+        def handle_request(url: str, request: gr.Request):
+            result = proxy.proxy_request(url, request)
+            
+            # 更新UI
+            error_visible = "error" in result
+            error_msg = result.get("error", "")
+            
+            return {
+                status: f"状态码: {result['status']}",
+                headers: result["headers"],
+                content: result["content"],
+                error: error_msg,
+                error: gr.update(visible=error_visible, value=error_msg)
+            }
+        
+        submit_btn.click(
+            fn=handle_request,
+            inputs=[url_input],
+            outputs=[status, headers, content, error]
+        )
+        
+        # 添加示例
+        gr.Examples(
+            examples=[
+                ["github.com/microsoft/vscode/blob/main/README.md"],
+                ["raw.githubusercontent.com/microsoft/vscode/main/README.md"],
+                ["gist.github.com/username/gist_id/raw/file.txt"]
+            ],
+            inputs=url_input
+        )
+    
+    return app
+
+if __name__ == "__main__":
+    app = create_interface()
+    app.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        show_error=True,
+        quiet=False
+    )