Upload folder using huggingface_hub

.dockerignore

@@ -0,0 +1,7 @@
+__pycache__/
+*.py[co]
+*.log
+.env
+.venv/
+.git
+.gitignore

.env.example

@@ -0,0 +1,21 @@
+# Telegram bot token obtained from BotFather
+BOT_TOKEN=123456789:ABCDEF-your-token
+
+# Custom Telegram API URL (required for Hugging Face deployment with Cloudflare Worker)
+# Example: TELEGRAM_API_URL=https://your-worker.your-subdomain.workers.dev/telegram-api
+TELEGRAM_API_URL=
+
+# Backend endpoint that will receive the normalized proxy payload (optional)
+BACKEND_URL=https://your-backend.example.com/api/proxies
+
+# Optional bearer token sent as Authorization header to the backend
+BACKEND_TOKEN=change-me
+
+# Optional overrides for testing behaviour
+# PROXY_TEST_URL=https://api.ipify.org?format=json
+# PROXY_TEST_TIMEOUT=10
+# MAX_CONCURRENT_TESTS=5
+
+# Comma or semicolon separated Telegram user IDs allowed to use the bot
+# Example: ALLOWED_USER_IDS=123456789,987654321
+ALLOWED_USER_IDS=

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+COPY requirements.txt ./
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . ./
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,119 @@
 ---
-title: Gencard
-emoji: 😻
-colorFrom: gray
-colorTo: gray
+title: Telegram SOCKS5 Proxy Bot
 sdk: docker
-pinned: false
+app_port: 7860
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Telegram SOCKS5 Proxy Bot
+
+This bot normalizes SOCKS5 proxy strings sent via Telegram, forwards them to an optional backend, and tests whether the proxies are working.
+
+## Telegram setup
+
+1. 在 Telegram 中寻找 @BotFather，发送 `/newbot` 创建机器人并获取 `BOT_TOKEN`。
+2. 使用 @userinfobot 或内置方法记录你的 Telegram 用户 ID。
+3. 在 `.env` 中填写 `BOT_TOKEN`，并将允许访问的用户 ID 列在 `ALLOWED_USER_IDS`（逗号或分号分隔）。若留空则任何人都可以使用机器人。
+
+## Prerequisites
+
+- Python 3.10+
+- A Telegram bot token from @BotFather
+
+## Installation
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+```
+
+Copy the environment template and fill in the values you need:
+
+```bash
+cp .env.example .env
+```
+
+Edit `.env` and add at least your `BOT_TOKEN`. Optionally set:
+
+- `BACKEND_URL` and `BACKEND_TOKEN` if you want to push normalized proxies to an HTTP endpoint
+- `PROXY_TEST_URL`, `PROXY_TEST_TIMEOUT`, and `MAX_CONCURRENT_TESTS` to tune validation behaviour
+- `ALLOWED_USER_IDS` to restrict the bot to specific Telegram accounts
+
+## Running the bot
+
+```bash
+python bot.py
+```
+
+The bot uses long polling by default. You can containerize or run the script under a process supervisor if you need persistent availability.
+
+## Docker
+
+Build and run directly:
+
+```bash
+docker build -t socks5-bot .
+docker run --rm --env-file .env socks5-bot
+```
+
+Or keep it running with `docker-compose` (uses `docker-compose.yml` in this repo):
+
+```bash
+docker-compose up --build -d
+```
+
+Stop the compose stack when needed:
+
+```bash
+docker-compose down
+```
+
+## Deploying to Hugging Face Spaces
+
+Hugging Face 仅开放 7860 端口，因此空间需要以 Docker 方式运行：
+
+1. 在 Hugging Face 新建 Space，选择 **Docker** 模式，并将此仓库推送过去。
+2. 在 Space 的 *Settings → Secrets* 中配置 `.env` 中用到的环境变量（例如 `BOT_TOKEN`、`BACKEND_URL` 等）。
+3. 该仓库的 `Dockerfile` 会启动 `uvicorn app:app --port 7860`，并暴露一个 `/` 和 `/healthz` 健康检查端点以满足 Hugging Face 的存活检测，Telegram 机器人在后台轮询。
+4. 如需更新配置，修改 Secrets 后在 *Actions → Restart* 重启 Space 即可。
+
+## Using the bot
+
+- Send one proxy per line in any of these formats:
+  - `user:password@host:port`
+  - `host:port:user:password`
+  - Lines with spaces, commas, pipes, or tabs between elements are also normalized
+- The bot replies with the proxies it accepted, the test result for each proxy, any rejected lines, and optional backend feedback.
+
+## Backend payload
+
+When `BACKEND_URL` is set, the bot `POST`s JSON like the following:
+
+```json
+{
+  "proxies": [
+    "user:password@host:1080"
+  ]
+}
+```
+
+If `BACKEND_TOKEN` is set it is sent as a Bearer token in the `Authorization` header.
+
+## Notes
+
+- The proxy test makes a GET request to the configured test URL using the SOCKS5 proxy. Pick an endpoint that can be reached through the proxies you receive.
+- Be mindful of the rate limits imposed by Telegram and the remote test endpoint.
+
+## Solving Network Restrictions (e.g., on Hugging Face)
+
+If you're deploying on platforms like Hugging Face Spaces that block access to `api.telegram.org`, you can use the Cloudflare Worker proxy solution:
+
+1. Create a Cloudflare account at https://www.cloudflare.com/
+2. Create a new Worker in your Cloudflare dashboard
+3. Copy the content from `telegram-api-proxy-worker.js` and paste it into your Worker
+4. Deploy the Worker and note your Worker URL (e.g., `https://your-worker.your-subdomain.workers.dev`)
+5. Update your environment variables to include `TELEGRAM_API_URL` pointing to your worker (append `/telegram-api` to the URL):
+   - `TELEGRAM_API_URL=https://your-worker.your-subdomain.workers.dev/telegram-api`
+6. The Docker setup already uses the patched version that supports custom API URLs
+
+This solution allows your bot to work even when direct access to the Telegram API is blocked.

app.py

@@ -0,0 +1,44 @@
+from contextlib import asynccontextmanager, suppress
+
+from fastapi import FastAPI, Response
+
+from bot import build_application
+
+
+telegram_application = None
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Startup
+    global telegram_application
+    telegram_application = build_application()
+    await telegram_application.initialize()
+    await telegram_application.start()
+    if telegram_application.updater:
+        await telegram_application.updater.start_polling()
+
+    yield
+
+    # Shutdown
+    if telegram_application:
+        if telegram_application.updater:
+            with suppress(Exception):
+                await telegram_application.updater.stop()
+        with suppress(Exception):
+            await telegram_application.stop()
+        with suppress(Exception):
+            await telegram_application.shutdown()
+
+
+app = FastAPI(title="Telegram Credit Card Generator Bot", lifespan=lifespan)
+
+
+@app.get("/")
+async def root() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.get("/healthz")
+async def health() -> Response:
+    return Response(content="ok", media_type="text/plain")

bot.py

@@ -0,0 +1,299 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import re
+import random
+from typing import Optional
+
+from dotenv import load_dotenv
+from telegram import Update
+from telegram.ext import (
+    ApplicationBuilder,
+    CommandHandler,
+    ContextTypes,
+    MessageHandler,
+    filters,
+)
+
+logging.basicConfig(
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    level=logging.INFO,
+)
+logger = logging.getLogger(__name__)
+
+
+# 信用卡生成相关函数
+def luhn_checksum(card_number: str) -> int:
+    """计算Luhn校验和"""
+    def digits_of(n):
+        return [int(d) for d in str(n)]
+    
+    digits = digits_of(card_number)
+    odd_digits = digits[-1::-2]  # 从右到左，奇数位
+    even_digits = digits[-2::-2]  # 从右到左，偶数位
+    
+    checksum = 0
+    checksum += sum(odd_digits)
+    
+    for d in even_digits:
+        checksum += sum(digits_of(d * 2))
+    
+    return checksum % 10
+
+
+def is_luhn_valid(card_number: str) -> bool:
+    """验证卡号是否符合Luhn算法"""
+    return luhn_checksum(card_number) == 0
+
+
+def generate_luhn_valid_number(prefix: str, length: int = 16) -> str:
+    """生成符合Luhn算法的卡号"""
+    # 确保前缀是数字字符串
+    prefix = str(prefix)
+    
+    # 如果前缀已经达到指定长度，直接验证并返回
+    if len(prefix) >= length:
+        if is_luhn_valid(prefix[:length]):
+            return prefix[:length]
+        else:
+            raise ValueError("前缀长度已达到但不符合Luhn算法")
+    
+    # 生成随机数字填充到长度-1（最后一位是校验位）
+    remaining_length = length - len(prefix) - 1
+    random_digits = ''.join([str(random.randint(0, 9)) for _ in range(remaining_length)])
+    
+    # 组合前缀、随机数字和临时校验位0
+    card_number = prefix + random_digits + '0'
+    
+    # 计算正确的校验位
+    checksum = luhn_checksum(card_number)
+    correct_check_digit = (10 - checksum) % 10
+    
+    # 替换最后一位为正确的校验位
+    valid_card_number = card_number[:-1] + str(correct_check_digit)
+    
+    return valid_card_number
+
+
+def parse_and_generate_card(input_string: str) -> dict:
+    """解析输入字符串并生成完整的信用卡信息"""
+    # 更灵活的正则表达式匹配：前缀|月份|年份|CVV
+    # 前缀可以包含x，CVV部分可以包含x或省略
+    pattern = r'^(\d+x*)\|(\d{1,2})\|(\d{2,4})(?:\|(x*))?$'
+    match = re.match(pattern, input_string)
+    
+    if not match:
+        raise ValueError("输入格式不正确，应为：前缀|月份|年份|CVV (CVV可选)")
+    
+    prefix_part = match.group(1)
+    month = match.group(2)
+    year = match.group(3)
+    cvv_part = match.group(4) if match.group(4) else 'xxx'  # 默认CVV长度
+    
+    # 分离前缀中的数字和x
+    prefix_match = re.match(r'^(\d+)(x*)$', prefix_part)
+    if not prefix_match:
+        raise ValueError("前缀格式不正确，应为数字后跟x")
+    
+    prefix = prefix_match.group(1)
+    
+    # 确定卡号总长度（通常是16位，但Amex是15位）
+    if prefix.startswith('34') or prefix.startswith('37'):
+        total_length = 15  # American Express
+    else:
+        total_length = 16  # Visa, Mastercard, Discover等
+    
+    # 生成完整的卡号
+    full_card_number = generate_luhn_valid_number(prefix, total_length)
+    
+    # 格式化月份和年份
+    month = month.zfill(2)
+    if len(year) == 2:
+        # 假设是20xx年
+        year = '20' + year
+    
+    # 生成随机CVV（通常是3位，Amex是4位）
+    if prefix.startswith('34') or prefix.startswith('37'):
+        cvv_length = 4
+    else:
+        cvv_length = 3
+    
+    cvv = ''.join([str(random.randint(0, 9)) for _ in range(cvv_length)])
+    
+    return {
+        'card_number': full_card_number,
+        'expiry_month': month,
+        'expiry_year': year,
+        'cvv': cvv,
+        'formatted': f"{full_card_number}|{month}|{year}|{cvv}"
+    }
+
+
+def is_credit_card_format(text: str) -> bool:
+    """检测文本是否为信用卡格式"""
+    # 检测格式：数字+x|数字|数字|可选的x
+    pattern = r'^\d+x*\|\d{1,2}\|\d{2,4}(?:\|x*)?$'
+    return bool(re.match(pattern, text.strip()))
+
+
+def generate_multiple_cards(input_string: str, count: int = 10) -> list:
+    """根据输入格式生成多张信用卡"""
+    cards = []
+    for _ in range(count):
+        try:
+            card_info = parse_and_generate_card(input_string)
+            cards.append(card_info)
+        except ValueError as e:
+            logger.warning(f"生成信用卡失败: {e}")
+            continue
+    return cards
+
+
+def _parse_allowed_users(raw_value: str) -> set[int]:
+    """解析允许的用户ID列表"""
+    if not raw_value.strip():
+        return set()
+    allowed = set()
+    for token in raw_value.replace(";", ",").split(","):
+        token = token.strip()
+        if not token:
+            continue
+        try:
+            allowed.add(int(token))
+        except ValueError:
+            logger.warning("Ignoring invalid user id token: %s", token)
+    return allowed
+
+
+async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """处理/start命令"""
+    if not await _ensure_authorized(update, context):
+        return
+
+    await update.message.reply_text(
+        "欢迎使用信用卡生成机器人！\n\n"
+        "请发送信用卡格式，我将为您生成10张随机信用卡。\n"
+        "格式示例：\n"
+        "• 440393xxxxx|3|27|xxx\n"
+        "• 424242|11|28\n"
+        "• 34xxxxx|6|25|xxxx (American Express)\n\n"
+        "格式说明：前缀|月份|年份|CVV(CVV可选)"
+    )
+
+
+async def handle_text(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """处理文本消息"""
+    assert update.message is not None
+    if not await _ensure_authorized(update, context):
+        return
+
+    text = update.message.text or ""
+    
+    # 检查是否为信用卡格式
+    if is_credit_card_format(text):
+        await handle_credit_card_request(update, text)
+    else:
+        await update.message.reply_text(
+            "格式不正确！请使用以下格式：\n"
+            "• 440393xxxxx|3|27|xxx\n"
+            "• 424242|11|28\n\n"
+            "格式说明：前缀|月份|年份|CVV(CVV可选)"
+        )
+
+
+async def handle_credit_card_request(update: Update, input_format: str) -> None:
+    """处理信用卡生成请求"""
+    try:
+        # 生成10张信用卡
+        cards = generate_multiple_cards(input_format, 10)
+        
+        if not cards:
+            await update.message.reply_text("生成信用卡失败，请检查输入格式。")
+            return
+        
+        # 格式化输出
+        response_lines = [f"🎉 成功生成 {len(cards)} 张信用卡：\n"]
+        
+        for i, card in enumerate(cards, 1):
+            response_lines.append(
+                f"{i}. {card['formatted']}"
+            )
+        
+        # 分批发送消息，避免消息过长
+        batch_size = 5
+        for i in range(0, len(response_lines), batch_size * 2):  # 每批包含标题+5张卡
+            batch = response_lines[i:i + batch_size * 2]
+            await update.message.reply_text("\n".join(batch))
+            
+    except Exception as e:
+        logger.exception("处理信用卡请求时出错")
+        await update.message.reply_text(f"生成信用卡时出错：{str(e)}")
+
+
+def build_application():
+    """构建Telegram应用"""
+    load_dotenv()
+
+    bot_token = os.getenv("BOT_TOKEN")
+    if not bot_token:
+        raise RuntimeError("BOT_TOKEN is not set")
+
+    # Create the application builder
+    application_builder = ApplicationBuilder().token(bot_token)
+
+    # Check for custom API URL (for proxy)
+    custom_api_url = os.getenv("TELEGRAM_API_URL", "").strip()
+    if custom_api_url:
+        logger.info(f"Using custom Telegram API URL: {custom_api_url}")
+        api_base = f"{custom_api_url}/bot"
+        file_base = f"{custom_api_url}/file/bot"
+        logger.info(f"Setting API base to: {api_base}")
+        logger.info(f"Setting file base to: {file_base}")
+        application_builder = application_builder.base_url(api_base).base_file_url(file_base)
+    else:
+        logger.info("Using default Telegram API URL")
+
+    allowed_ids = _parse_allowed_users(os.getenv("ALLOWED_USER_IDS", ""))
+
+    application = application_builder.build()
+
+    # Store the custom API URL in the application's bot_data
+    if custom_api_url:
+        application.bot_data["custom_api_url"] = custom_api_url
+
+    application.bot_data["allowed_ids"] = allowed_ids
+
+    application.add_handler(CommandHandler("start", start))
+    application.add_handler(MessageHandler(filters.TEXT & (~filters.COMMAND), handle_text))
+
+    return application
+
+
+def main() -> None:
+    """主函数"""
+    application = build_application()
+    application.run_polling()
+
+
+if __name__ == "__main__":
+    main()
+
+
+async def _ensure_authorized(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
+    """确保用户有权限使用机器人"""
+    allowed_ids: set[int] = context.application.bot_data.get("allowed_ids", set())
+    if not allowed_ids:
+        return True
+
+    user = update.effective_user
+    if user and user.id in allowed_ids:
+        return True
+
+    message = update.effective_message
+    if message:
+        await message.reply_text("您没有权限使用此机器人。")
+
+    logger.info("Unauthorized access attempt by user id %s", user.id if user else "unknown")
+    return False

docker-compose.yml

@@ -0,0 +1,7 @@
+version: "3.9"
+services:
+  bot:
+    build: .
+    restart: unless-stopped
+    env_file:
+      - .env

requirements.txt

@@ -0,0 +1,4 @@
+python-telegram-bot>=20.8,<20.9
+python-dotenv>=1.0.0
+fastapi>=0.110
+uvicorn[standard]>=0.27

telegram-api-proxy-worker.js

@@ -0,0 +1,87 @@
+// Cloudflare Worker script to proxy Telegram API requests
+// This script intercepts requests to the Telegram Bot API and forwards them
+// allowing your bot to work even when direct access to api.telegram.org is blocked
+
+export default {
+  async fetch(request, env, ctx) {
+    // Get the original request URL
+    const originalUrl = new URL(request.url);
+    
+    // Handle CORS preflight requests
+    if (request.method === 'OPTIONS') {
+      return new Response(null, {
+        headers: {
+          'Access-Control-Allow-Origin': '*',
+          'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
+          'Access-Control-Allow-Headers': 'Content-Type, Authorization',
+        },
+      });
+    }
+
+    // Check if this is a request to our worker
+    // If the path starts with /telegram-api, forward to Telegram API
+    if (originalUrl.pathname.startsWith('/telegram-api')) {
+      // Extract the path after /telegram-api
+      // The full path should be like /telegram-api/bot{TOKEN}/{METHOD}
+      const telegramApiPath = originalUrl.pathname.replace('/telegram-api', '');
+      
+      // Construct the target URL for Telegram API
+      const telegramApiUrl = `https://api.telegram.org${telegramApiPath}`;
+      
+      // Clone the original request to modify it
+      // Remove the Cloudflare-specific headers that might cause issues
+      const newHeaders = new Headers(request.headers);
+      newHeaders.delete('cf-ray');
+      newHeaders.delete('cf-connecting-ip');
+      newHeaders.delete('x-forwarded-for');
+      
+      const modifiedRequest = new Request(telegramApiUrl, {
+        method: request.method,
+        headers: newHeaders,
+        body: request.body,
+        redirect: 'follow'
+      });
+      
+      try {
+        // Send request to Telegram API
+        const response = await fetch(modifiedRequest);
+        
+        // Return the response from Telegram API with CORS headers
+        const modifiedResponse = new Response(response.body, {
+          status: response.status,
+          statusText: response.statusText,
+          headers: {
+            ...response.headers,
+            'Access-Control-Allow-Origin': '*',
+            'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
+            'Access-Control-Allow-Headers': 'Content-Type, Authorization',
+          }
+        });
+        
+        return modifiedResponse;
+      } catch (error) {
+        console.error('Proxy error:', error);
+        return new Response(JSON.stringify({ 
+          error: 'Proxy error', 
+          details: error.message,
+          message: 'Failed to connect to Telegram API'
+        }), {
+          status: 502, // Bad Gateway
+          headers: { 
+            'Content-Type': 'application/json',
+            'Access-Control-Allow-Origin': '*',
+          }
+        });
+      }
+    }
+    
+    // If not a Telegram API request, return info about the proxy
+    return new Response('Telegram API Proxy Worker - Ready to forward requests to Telegram API', {
+      status: 200,
+      headers: { 
+        'Content-Type': 'text/plain',
+        'Access-Control-Allow-Origin': '*',
+      }
+    });
+  }
+};