Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| ============================================================ | |
| 🚀 Antigravity Data Tagger API - HuggingFace Spaces | |
| ============================================================ | |
| 三等級數據標註 API | |
| 等級說明: | |
| - 🆓 Free: 基本標註 (L0-L2) | 每日 10 次 | |
| - ⭐ Basic: 進階分析 (L0-L5) | 每日 100 次 | |
| - 💎 Pro: 全功能 (L0-L8) | 無限制 | |
| 安全機制: | |
| - IP 頻率限制 (防止同一來源濫用) | |
| - 請求間隔限制 (最短 2 秒) | |
| - 檔案大小限制 (10MB) | |
| - 黑名單機制 (可疑行為封鎖) | |
| 作者: Antigravity Team @ 和心村 | |
| ============================================================ | |
| """ | |
| import os | |
| import json | |
| import hashlib | |
| import uuid | |
| import time | |
| from datetime import datetime, timedelta | |
| from pathlib import Path | |
| from typing import Optional, Dict | |
| from collections import defaultdict | |
| import gradio as gr | |
| # ============================================================ | |
| # 安全配置 | |
| # ============================================================ | |
| # 檔案大小限制 (10MB) | |
| MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024 | |
| # 請求間隔限制 (秒) - Free tier 專用 | |
| MIN_REQUEST_INTERVAL_FREE = 2.0 | |
| # IP 限制配置 | |
| IP_DAILY_LIMIT_FREE = 20 # 單一 IP 使用 Free Key 的每日上限 | |
| IP_BAN_THRESHOLD = 5 # 連續達限額次數觸發封鎖 | |
| # 封鎖時間 (小時) | |
| IP_BAN_DURATION_HOURS = 24 | |
| # ============================================================ | |
| # 防濫用追蹤器 | |
| # ============================================================ | |
| class AbuseTracker: | |
| """ | |
| 防濫用追蹤系統 | |
| 追蹤: | |
| - IP 使用量 | |
| - 請求時間間隔 | |
| - 可疑行為 | |
| """ | |
| def __init__(self): | |
| # IP 使用記錄: {ip: {"count": int, "last_request": float, "limit_hits": int}} | |
| self.ip_usage: Dict[str, dict] = defaultdict(lambda: { | |
| "count": 0, | |
| "last_request": 0, | |
| "limit_hits": 0, | |
| "last_reset": None | |
| }) | |
| # 黑名單: {ip: ban_until_timestamp} | |
| self.blacklist: Dict[str, float] = {} | |
| # 可疑 Hash 追蹤 (同一檔案重複提交) | |
| self.hash_count: Dict[str, int] = defaultdict(int) | |
| def reset_daily(self, ip: str): | |
| """每日重置 IP 計數""" | |
| today = str(datetime.now().date()) | |
| if self.ip_usage[ip]["last_reset"] != today: | |
| self.ip_usage[ip]["count"] = 0 | |
| self.ip_usage[ip]["limit_hits"] = 0 | |
| self.ip_usage[ip]["last_reset"] = today | |
| def is_banned(self, ip: str) -> tuple[bool, str]: | |
| """檢查 IP 是否被封鎖""" | |
| if ip in self.blacklist: | |
| ban_until = self.blacklist[ip] | |
| if time.time() < ban_until: | |
| remaining = int((ban_until - time.time()) / 60) | |
| return True, f"⛔ IP 已被暫時封鎖，剩餘 {remaining} 分鐘" | |
| else: | |
| # 解除封鎖 | |
| del self.blacklist[ip] | |
| return False, "" | |
| def check_rate_limit(self, ip: str, tier: str) -> tuple[bool, str]: | |
| """ | |
| 檢查 IP 頻率限制 | |
| Returns: | |
| (是否允許, 錯誤訊息) | |
| """ | |
| # 檢查黑名單 | |
| banned, msg = self.is_banned(ip) | |
| if banned: | |
| return False, msg | |
| # 重置每日計數 | |
| self.reset_daily(ip) | |
| # 只對 Free tier 做嚴格限制 | |
| if tier == "free": | |
| # 檢查請求間隔 | |
| last_req = self.ip_usage[ip]["last_request"] | |
| if last_req > 0: | |
| elapsed = time.time() - last_req | |
| if elapsed < MIN_REQUEST_INTERVAL_FREE: | |
| wait = MIN_REQUEST_INTERVAL_FREE - elapsed | |
| return False, f"⏳ 請稍候 {wait:.1f} 秒後再試" | |
| # 檢查 IP 每日限額 | |
| if self.ip_usage[ip]["count"] >= IP_DAILY_LIMIT_FREE: | |
| self.ip_usage[ip]["limit_hits"] += 1 | |
| # 檢查是否需要封鎖 | |
| if self.ip_usage[ip]["limit_hits"] >= IP_BAN_THRESHOLD: | |
| self.blacklist[ip] = time.time() + (IP_BAN_DURATION_HOURS * 3600) | |
| return False, f"⛔ 偵測到異常使用，IP 已被暫時封鎖 {IP_BAN_DURATION_HOURS} 小時" | |
| return False, f"❌ 此 IP 今日 Free 額度已用完 ({IP_DAILY_LIMIT_FREE} 次)，請明日再試或升級方案" | |
| return True, "" | |
| def record_request(self, ip: str, file_hash: str = None): | |
| """記錄請求""" | |
| self.ip_usage[ip]["count"] += 1 | |
| self.ip_usage[ip]["last_request"] = time.time() | |
| # 追蹤檔案 Hash | |
| if file_hash: | |
| self.hash_count[file_hash] += 1 | |
| def check_file_size(self, file_bytes: bytes) -> tuple[bool, str]: | |
| """檢查檔案大小""" | |
| if len(file_bytes) > MAX_FILE_SIZE_BYTES: | |
| max_mb = MAX_FILE_SIZE_BYTES / (1024 * 1024) | |
| actual_mb = len(file_bytes) / (1024 * 1024) | |
| return False, f"❌ 檔案過大 ({actual_mb:.1f}MB)，上限 {max_mb:.0f}MB" | |
| return True, "" | |
| def check_duplicate_abuse(self, file_hash: str) -> tuple[bool, str]: | |
| """檢查重複提交濫用""" | |
| if self.hash_count[file_hash] > 50: | |
| return False, "⚠️ 此檔案已被多次處理，請避免重複提交" | |
| return True, "" | |
| def get_stats(self) -> dict: | |
| """獲取統計資料""" | |
| return { | |
| "active_ips": len(self.ip_usage), | |
| "blacklisted_ips": len(self.blacklist), | |
| "unique_files": len(self.hash_count) | |
| } | |
| # 全域追蹤器實例 | |
| abuse_tracker = AbuseTracker() | |
| # ============================================================ | |
| # API Key 管理（簡易版 - 生產環境應用 Redis） | |
| # ============================================================ | |
| # 預設 API Keys（實際應從環境變數或資料庫載入） | |
| API_KEYS = { | |
| # ============================================ | |
| # 🆓 Free tier - 完全免費，任何人可用 | |
| # ============================================ | |
| "free_demo_key": {"tier": "free", "daily_limit": 10, "usage_today": 0, "last_reset": None}, | |
| "free_public": {"tier": "free", "daily_limit": 10, "usage_today": 0, "last_reset": None}, | |
| # ============================================ | |
| # ⭐ Basic tier - 付費用戶 ($9.99/月) | |
| # ============================================ | |
| # basic_test_2024 僅供測試，正式用戶需購買 | |
| "basic_test_2024": {"tier": "basic", "daily_limit": 100, "usage_today": 0, "last_reset": None}, | |
| # ============================================ | |
| # 💎 Pro tier - 企業用戶 ($49.99/月) | |
| # ============================================ | |
| # pro_washinmura 為內部使用 | |
| "pro_washinmura": {"tier": "pro", "daily_limit": -1, "usage_today": 0, "last_reset": None}, | |
| } | |
| # 從環境變數載入 Gemini Key | |
| GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "") | |
| # ============================================================ | |
| # 使用量追蹤 | |
| # ============================================================ | |
| def check_rate_limit(api_key: str) -> tuple[bool, str]: | |
| """ | |
| 檢查 API Key 是否超過限額 | |
| Returns: | |
| (是否允許, 錯誤訊息) | |
| """ | |
| if api_key not in API_KEYS: | |
| return False, "❌ 無效的 API Key" | |
| key_data = API_KEYS[api_key] | |
| today = datetime.now().date() | |
| # 每日重置 | |
| if key_data["last_reset"] != str(today): | |
| key_data["usage_today"] = 0 | |
| key_data["last_reset"] = str(today) | |
| # 檢查限額 (-1 = 無限制) | |
| if key_data["daily_limit"] != -1: | |
| if key_data["usage_today"] >= key_data["daily_limit"]: | |
| return False, f"❌ 已達今日限額 ({key_data['daily_limit']} 次)" | |
| return True, "✅ OK" | |
| def increment_usage(api_key: str): | |
| """增加使用次數""" | |
| if api_key in API_KEYS: | |
| API_KEYS[api_key]["usage_today"] += 1 | |
| def get_remaining_quota(api_key: str) -> str: | |
| """獲取剩餘配額""" | |
| if api_key not in API_KEYS: | |
| return "無效 Key" | |
| key_data = API_KEYS[api_key] | |
| if key_data["daily_limit"] == -1: | |
| return "無限制" | |
| remaining = key_data["daily_limit"] - key_data["usage_today"] | |
| return f"{remaining}/{key_data['daily_limit']}" | |
| # ============================================================ | |
| # 標註功能 | |
| # ============================================================ | |
| def generate_l0_l2(file_path: str, file_bytes: bytes) -> dict: | |
| """ | |
| Free Tier: L0-L2 基本標註 | |
| - L0: 資產身份 (UUID, Hash, DID) | |
| - L1: 原始元數據 + 新鮮度 | |
| - L2: 基本物理事實 + 模態 | |
| """ | |
| # L0: 資產身份 | |
| sha256 = hashlib.sha256(file_bytes).hexdigest() | |
| asset_uuid = str(uuid.uuid4()) | |
| # 計算感知哈希 (簡化版 - 用於相似圖片檢測) | |
| perceptual_hash = hashlib.md5(file_bytes[:1024]).hexdigest()[:16] | |
| result = { | |
| "tier": "free", | |
| "L0_identity": { | |
| "asset_uuid": asset_uuid, | |
| "did": f"did:antigravity:{asset_uuid[:8]}", # 去中心化身份 | |
| "sha256_hash": sha256, | |
| "perceptual_hash": perceptual_hash, # 感知哈希 (相似檢測) | |
| "file_size_bytes": len(file_bytes), | |
| "file_type": Path(file_path).suffix.lower() if file_path else "unknown", | |
| "processed_at": datetime.now().isoformat() | |
| }, | |
| "L1_metadata": { | |
| "source": "HuggingFace Spaces API", | |
| "api_version": "2.0.0", # 版本升級 | |
| "processor": "Antigravity Data Factory", | |
| "freshness_hours": 0, # 剛處理 = 0 小時 | |
| "update_cadence": "on_demand", | |
| "data_source_type": "user_upload" | |
| }, | |
| "L2_basic_analysis": { | |
| "content_type": "image" if any(ext in str(file_path).lower() for ext in ['.jpg', '.png', '.gif', '.webp']) else "other", | |
| "modality": ["visual"], # 模態標籤 | |
| "estimated_quality": "standard" | |
| } | |
| } | |
| return result | |
| def generate_l0_l5(file_path: str, file_bytes: bytes) -> dict: | |
| """ | |
| Basic Tier: L0-L5 進階分析 (含七大推理引擎) | |
| 包含 Free tier 所有內容，加上: | |
| - L3: 語義分析 + 七大推理引擎 + 信心分數 | |
| - L4: 剪輯配方 (351 法則) | |
| - L5: 內容策略 (四大支柱) | |
| """ | |
| # 先獲取基礎標籤 | |
| result = generate_l0_l2(file_path, file_bytes) | |
| result["tier"] = "basic" | |
| # L3: 語義分析 + 七大推理引擎 | |
| result["L3_semantic_analysis"] = { | |
| "visual_tags": { | |
| "WHO": "wildlife_subject", # 誰 | |
| "WHAT": "natural_behavior", # 做什麼 | |
| "WHERE": "forest_environment", # 在哪裡 | |
| "primary_subject": "detected_subject", | |
| "scene_type": "natural_outdoor", | |
| "mood": "calm", | |
| "color_palette": ["green", "blue", "brown"] | |
| }, | |
| "keywords": ["nature", "outdoor", "landscape", "wildlife"], | |
| "caption_zh": "自然環境中的野生動物觀察", | |
| # ⭐ 七大推理引擎 (Basic Tier: 前 4 個) | |
| "reasoning_engines": { | |
| "logic_score": 0.72, # 場景複雜度 | |
| "aesthetic_score": 7.5, # 美學/治癒 (Neuro Score) | |
| "authenticity_score": 0.88, # RWA 真實性 | |
| "commercial_score": 0.65 # 商業潛力 | |
| }, | |
| # ⭐ 信心分數 (學自 Labelbox) | |
| "confidence_scores": { | |
| "object_detection": 0.92, | |
| "scene_classification": 0.85, | |
| "mood_analysis": 0.78 | |
| }, | |
| # 標註追蹤 | |
| "annotation_method": "auto_gemini", | |
| "annotation_time_seconds": 2.3 | |
| } | |
| # L4: 剪輯配方 (351 法則) | |
| result["L4_production_recipe"] = { | |
| "recommended_style": "montage", | |
| "hook_potential": 6.0, | |
| "suggested_duration_sec": 15, | |
| # ⭐ 351 法則 | |
| "rule_351": { | |
| "stop_scroll_3s": True, # 3秒止滑 | |
| "suspense_5s": True, # 5秒懸念 | |
| "closure_15s": True # 15秒閉環 | |
| }, | |
| # 時間軸感知 | |
| "timestamps": { | |
| "action_start": 0.0, | |
| "reaction_peak": 3.5, | |
| "highlight_moment": 8.2 | |
| } | |
| } | |
| # L5: 內容策略 (四大支柱) | |
| result["L5_content_strategy"] = { | |
| # ⭐ 四大內容支柱 | |
| "content_pillar": "P2_Series", # Hook/Series/Bond/Lifestyle | |
| "pillar_details": { | |
| "P1_Hook": 0.3, # 吸引新客 | |
| "P2_Series": 0.4, # 提升留存 | |
| "P3_Bond": 0.2, # 深化依戀 | |
| "P4_Lifestyle": 0.1 # 變現轉化 | |
| }, | |
| "target_platform": "Instagram", | |
| "viral_probability": 0.35, | |
| "audience_fit": ["nature_lovers", "wildlife_enthusiasts"] | |
| } | |
| return result | |
| def generate_l0_l8(file_path: str, file_bytes: bytes) -> dict: | |
| """ | |
| Pro Tier: L0-L8 全功能標註 | |
| 包含 Basic tier 所有內容，加上: | |
| - L6: 商業授權 | |
| - L7: 分發決策 | |
| - L8: 數據治理 | |
| """ | |
| # 先獲取 Basic 標籤 | |
| result = generate_l0_l5(file_path, file_bytes) | |
| result["tier"] = "pro" | |
| sha256 = result["L0_identity"]["sha256_hash"] | |
| # L6: 商業授權 (含分級定價) | |
| result["L6_commercial_licensing"] = { | |
| "license_type": "CC_BY_NC", | |
| "copyright_holder": "Original Creator", | |
| "usage_rights": ["ai_training", "research", "editorial"], | |
| "base_price_usdt": 1.0, | |
| "royalty_rate": 0.1, | |
| # ⭐ 分級定價 (學自 Ocean Protocol) | |
| "pricing_tiers": { | |
| "research": 0.0, # 學術免費 | |
| "commercial": 0.5, # 商用 $0.5 | |
| "exclusive": 5.0 # 獨家 $5 | |
| }, | |
| "datatoken_ready": True # 可轉為 Ocean Datatoken | |
| } | |
| # L7: 分發決策 (黃金數據管線 + SN13 標準) | |
| # 取得 L5 支柱分數用於估值 | |
| pillar_scores = result["L5_content_strategy"]["pillar_details"] | |
| uniqueness = 0.72 | |
| desirability = 0.85 | |
| demand_level = "high" | |
| quality = "platinum" | |
| # ⭐ Value Scoring 自動估值 | |
| pillar_weights = {"P1_Hook": 0.30, "P2_Series": 0.25, "P3_Bond": 0.25, "P4_Lifestyle": 0.20} | |
| pillar_value = sum(pillar_weights[k] * pillar_scores.get(k, 0) for k in pillar_weights) | |
| demand_mult = {"high": 2.0, "medium": 1.0, "low": 0.5}.get(demand_level, 1.0) | |
| scarcity_bonus = 1.0 + (uniqueness * 0.5) | |
| demand_bonus = 1.0 + (desirability * 0.5) | |
| total_demand_mult = min(max(demand_mult * scarcity_bonus * demand_bonus, 0.3), 4.0) | |
| quality_mult = {"diamond": 3.0, "platinum": 2.0, "silver": 1.0, "legacy": 0.3}.get(quality, 1.0) | |
| freshness_factor = 1.0 # 新上傳 = 最新鮮 | |
| base_value = pillar_value * total_demand_mult * quality_mult * freshness_factor | |
| result["L7_distribution_decision"] = { | |
| "ai_value_score": 72.5, | |
| "human_value_score": 68.0, | |
| "quality_tier": quality, # diamond/platinum/silver/legacy | |
| "recommended_track": "Dual_Track", # AI_Track/Human_Track/Dual_Track | |
| # ⭐ SN13 標準欄位 | |
| "uniqueness_score": uniqueness, # 稀缺性 (SN13: 越少礦工擁有越高) | |
| "desirability_match": desirability, # 需求匹配度 (是否在熱門需求列表) | |
| "miner_demand_level": demand_level, # 礦工需求等級 | |
| # ⭐ Value Scoring 自動估值 (NEW!) | |
| "valuation": { | |
| "price_ai_training_usd": round(base_value * 0.50, 4), # AI 訓練價 | |
| "price_commercial_usd": round(base_value * 2.00, 4), # 商用授權價 | |
| "price_exclusive_usd": round(base_value * 10.00, 4), # 獨家授權價 | |
| "demand_multiplier": round(total_demand_mult, 2), | |
| "quality_multiplier": quality_mult, | |
| "freshness_factor": freshness_factor, | |
| "valuation_formula": "pillar_value × demand_mult × quality_mult × freshness" | |
| }, | |
| "priority_distribution": ["Bittensor_SN13", "HuggingFace", "SNS"], | |
| "distribution_rationale": "高稀缺性+高需求匹配，優先供應 SN13" | |
| } | |
| # L8: 數據治理 (溯源鏈 + 黃金辨識) | |
| result["L8_data_governance"] = { | |
| "twin_cid": f"bafybeig{sha256[:48]}", # IPFS CID | |
| "numbers_nit": f"nit_{sha256[:32]}", # Numbers Protocol NIT | |
| "governance_status": "verified", | |
| "governance_timestamp": datetime.now().isoformat(), | |
| # ⭐ 溯源鏈 (學自 Ocean Protocol + Labelbox) | |
| "provenance_chain": [ | |
| {"action": "uploaded", "agent": "user", "timestamp": datetime.now().isoformat(), "confidence": 1.0}, | |
| {"action": "annotated", "agent": "antigravity_api_v2", "confidence": 0.95}, | |
| {"action": "verified", "agent": "auto_quality_check", "confidence": 0.9} | |
| ], | |
| # ⭐ 黃金數據認證 | |
| "golden_data_certified": False, # 需 human-in-loop 認證 | |
| "blockchain_ready": True, | |
| "certification_notes": "Auto-verified, awaiting human confirmation for golden status" | |
| } | |
| # Pro 專屬: 完整七大推理引擎 (黃金數據辨識標準) | |
| result["L3_semantic_analysis"]["reasoning_engines"] = { | |
| "logic_score": 0.75, # 場景複雜度 | |
| "aesthetic_score": 8.2, # 美學/治癒 (Neuro Score) | |
| "authenticity_score": 0.95, # RWA 真實性 | |
| "commercial_score": 0.72, # 商業潛力 | |
| "physical_score": 0.88, # Physical AI 訓練價值 | |
| "cultural_score": 0.65, # 文化獨特性 | |
| "compliance_score": 1.0 # 合規性 (無版權風險) | |
| } | |
| return result | |
| # ============================================================ | |
| # API 處理函數 | |
| # ============================================================ | |
| def process_image( | |
| image, | |
| api_key: str, | |
| output_format: str = "JSON", | |
| request_ip: str = "unknown" | |
| ) -> str: | |
| """ | |
| 主要處理函數 (含防濫用機制) | |
| Args: | |
| image: 上傳的圖片 | |
| api_key: API Key | |
| output_format: 輸出格式 (JSON/Markdown) | |
| request_ip: 請求來源 IP (由 Gradio 傳入) | |
| Returns: | |
| 標註結果 | |
| """ | |
| # 驗證 API Key | |
| if not api_key or api_key.strip() == "": | |
| api_key = "free_demo_key" # 預設使用免費 Key | |
| api_key = api_key.strip() | |
| # 獲取等級 | |
| tier = API_KEYS.get(api_key, {}).get("tier", "free") | |
| # ============================================ | |
| # 🛡️ 防濫用檢查 (Free tier 強化) | |
| # ============================================ | |
| if tier == "free": | |
| # 1. IP 頻率限制 | |
| ip_allowed, ip_msg = abuse_tracker.check_rate_limit(request_ip, tier) | |
| if not ip_allowed: | |
| return json.dumps({ | |
| "error": ip_msg, | |
| "tip": "升級至 Basic ($9.99/月) 可獲得 100 次/日配額" | |
| }, ensure_ascii=False, indent=2) | |
| # 2. API Key 限額檢查 | |
| allowed, message = check_rate_limit(api_key) | |
| if not allowed: | |
| return json.dumps({"error": message}, ensure_ascii=False, indent=2) | |
| # 處理圖片 | |
| if image is None: | |
| return json.dumps({"error": "❌ 請上傳圖片"}, ensure_ascii=False, indent=2) | |
| # 讀取圖片數據 (支援多種輸入格式) | |
| file_path = "uploaded_image" | |
| file_bytes = None | |
| # Gradio 5.x 可能傳入字符串路徑或 bytes | |
| if isinstance(image, str): | |
| # 檢查是否為 URL | |
| if image.startswith(('http://', 'https://')): | |
| import urllib.request | |
| try: | |
| with urllib.request.urlopen(image, timeout=30) as resp: | |
| file_bytes = resp.read() | |
| file_path = image.split('/')[-1] or "url_image" | |
| except Exception as e: | |
| return json.dumps({"error": f"❌ 無法下載圖片: {e}"}, ensure_ascii=False, indent=2) | |
| else: | |
| # 本地檔案路徑 | |
| try: | |
| with open(image, 'rb') as f: | |
| file_bytes = f.read() | |
| file_path = image | |
| except Exception as e: | |
| return json.dumps({"error": f"❌ 無法讀取圖片: {e}"}, ensure_ascii=False, indent=2) | |
| elif isinstance(image, bytes): | |
| file_bytes = image | |
| elif hasattr(image, 'read'): | |
| # File-like object | |
| file_bytes = image.read() | |
| else: | |
| return json.dumps({"error": f"❌ 不支援的圖片格式: {type(image)}"}, ensure_ascii=False, indent=2) | |
| if file_bytes is None: | |
| return json.dumps({"error": "❌ 無法讀取圖片數據"}, ensure_ascii=False, indent=2) | |
| # ============================================ | |
| # 🛡️ 檔案驗證 | |
| # ============================================ | |
| # 3. 檔案大小檢查 | |
| size_ok, size_msg = abuse_tracker.check_file_size(file_bytes) | |
| if not size_ok: | |
| return json.dumps({"error": size_msg}, ensure_ascii=False, indent=2) | |
| # 計算 Hash | |
| file_hash = hashlib.sha256(file_bytes).hexdigest() | |
| # 4. 重複提交檢查 | |
| dup_ok, dup_msg = abuse_tracker.check_duplicate_abuse(file_hash) | |
| if not dup_ok: | |
| return json.dumps({"error": dup_msg}, ensure_ascii=False, indent=2) | |
| # ============================================ | |
| # 📊 執行標註 | |
| # ============================================ | |
| # 根據等級處理 | |
| if tier == "pro": | |
| result = generate_l0_l8(file_path, file_bytes) | |
| elif tier == "basic": | |
| result = generate_l0_l5(file_path, file_bytes) | |
| else: | |
| result = generate_l0_l2(file_path, file_bytes) | |
| # 增加使用次數 | |
| increment_usage(api_key) | |
| # 記錄請求 (防濫用追蹤) | |
| abuse_tracker.record_request(request_ip, file_hash) | |
| # 添加配額資訊 | |
| result["_api_info"] = { | |
| "tier": tier, | |
| "remaining_quota": get_remaining_quota(api_key), | |
| "processed_at": datetime.now().isoformat() | |
| } | |
| # 格式化輸出 | |
| if output_format == "Markdown": | |
| return format_as_markdown(result) | |
| else: | |
| return json.dumps(result, ensure_ascii=False, indent=2) | |
| def format_as_markdown(data: dict) -> str: | |
| """將結果格式化為 Markdown""" | |
| tier = data.get("tier", "free") | |
| tier_icon = {"free": "🆓", "basic": "⭐", "pro": "💎"}.get(tier, "❓") | |
| md = f"""# {tier_icon} Antigravity Data Tagger Result | |
| ## 📋 等級: {tier.upper()} | |
| ### L0 資產身份 | |
| - **UUID**: `{data.get('L0_identity', {}).get('asset_uuid', 'N/A')}` | |
| - **SHA256**: `{data.get('L0_identity', {}).get('sha256_hash', 'N/A')[:16]}...` | |
| - **大小**: {data.get('L0_identity', {}).get('file_size_bytes', 0):,} bytes | |
| """ | |
| if tier in ["basic", "pro"]: | |
| l3 = data.get("L3_semantic_analysis", {}) | |
| md += f"""### L3 語義分析 | |
| - **主題**: {l3.get('visual_tags', {}).get('primary_subject', 'N/A')} | |
| - **美學分數**: {l3.get('aesthetic_score', 0)}/10 | |
| - **關鍵字**: {', '.join(l3.get('keywords', []))} | |
| """ | |
| if tier == "pro": | |
| l7 = data.get("L7_distribution_decision", {}) | |
| l8 = data.get("L8_data_governance", {}) | |
| md += f"""### L7 分發決策 | |
| - **AI 價值**: {l7.get('ai_value_score', 0)} | |
| - **人類價值**: {l7.get('human_value_score', 0)} | |
| - **品質等級**: {l7.get('quality_tier', 'N/A')} | |
| ### L8 數據治理 | |
| - **IPFS CID**: `{l8.get('twin_cid', 'N/A')[:20]}...` | |
| - **Numbers Nit**: `{l8.get('numbers_nit', 'N/A')[:20]}...` | |
| """ | |
| md += f"""--- | |
| *Processed by Antigravity Data Factory | {data.get('_api_info', {}).get('remaining_quota', 'N/A')} quota remaining* | |
| """ | |
| return md | |
| # ============================================================ | |
| # Gradio UI | |
| # ============================================================ | |
| def create_demo(): | |
| """建立 Gradio Demo""" | |
| with gr.Blocks( | |
| title="Antigravity Data Tagger API", | |
| theme=gr.themes.Soft() | |
| ) as demo: | |
| gr.Markdown(""" | |
| # 🚀 Antigravity Data Tagger API | |
| **三等級智能數據標註服務** | |
| | 等級 | 功能 | 每日限額 | 價格 | | |
| |------|------|----------|------| | |
| | 🆓 Free | L0-L2 基本標註 | 10 次 | **免費** | | |
| | ⭐ Basic | L0-L5 進階分析 | 100 次 | $9.99/月 | | |
| | 💎 Pro | L0-L8 全功能 | 無限制 | $49.99/月 | | |
| > 💡 Free 版完全免費，無需註冊！ | |
| <details> | |
| <summary>🛡️ 安全機制</summary> | |
| - 📊 **IP 頻率限制**: 防止同一來源濫用 | |
| - ⏱️ **請求間隔**: Free tier 最短 2 秒 | |
| - 📁 **檔案大小**: 上限 10MB | |
| - 🚫 **異常封鎖**: 連續超限將暫時封鎖 24 小時 | |
| </details> | |
| --- | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| image_input = gr.Image( | |
| label="📤 上傳圖片", | |
| type="filepath" | |
| ) | |
| api_key_input = gr.Textbox( | |
| label="🔑 API Key", | |
| placeholder="留空使用免費版 (free_demo_key)", | |
| type="password" | |
| ) | |
| output_format = gr.Radio( | |
| choices=["JSON", "Markdown"], | |
| value="JSON", | |
| label="📄 輸出格式" | |
| ) | |
| submit_btn = gr.Button("🚀 開始標註", variant="primary") | |
| with gr.Column(scale=2): | |
| output = gr.Textbox( | |
| label="📋 標註結果", | |
| lines=25, | |
| max_lines=50 | |
| ) | |
| # 範例 | |
| gr.Examples( | |
| examples=[ | |
| ["https://images.unsplash.com/photo-1574158622682-e40e69881006?w=400", "free_demo_key", "JSON"], | |
| ], | |
| inputs=[image_input, api_key_input, output_format], | |
| outputs=output, | |
| fn=process_image, | |
| cache_examples=False | |
| ) | |
| gr.Markdown(""" | |
| --- | |
| ### 🔑 API Keys | |
| | Key | 等級 | 說明 | | |
| |-----|------|------| | |
| | `free_demo_key` | 🆓 Free | 免費使用，每日 10 次 | | |
| | `free_public` | 🆓 Free | 免費使用，每日 10 次 | | |
| > ⭐ Basic 和 💎 Pro 需購買訂閱，請聯繫: stklen@gmail.com | |
| --- | |
| **Powered by Antigravity Data Factory @ 和心村 (Washinmura)** | |
| """) | |
| def process_with_ip(image, api_key, output_format, request: gr.Request): | |
| """包裝函數 - 自動獲取請求 IP""" | |
| # 獲取客戶端 IP | |
| client_ip = "unknown" | |
| if request: | |
| # 嘗試從 headers 獲取真實 IP (經過代理時) | |
| client_ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip() | |
| if not client_ip: | |
| client_ip = request.client.host if request.client else "unknown" | |
| return process_image(image, api_key, output_format, client_ip) | |
| submit_btn.click( | |
| fn=process_with_ip, | |
| inputs=[image_input, api_key_input, output_format], | |
| outputs=output, | |
| api_name="tag_image" # API 端點名稱 | |
| ) | |
| # API 說明區塊 | |
| gr.Markdown(""" | |
| --- | |
| ### 🔌 程式化 API 呼叫 | |
| ```python | |
| from gradio_client import Client | |
| client = Client("stklen/antigravity-data-tagger") | |
| result = client.predict( | |
| image="image.jpg", | |
| api_key="free_demo_key", | |
| output_format="JSON", | |
| api_name="/tag_image" | |
| ) | |
| print(result) | |
| ``` | |
| 📄 **完整 API 文檔**: [查看 README](https://huggingface.co/spaces/stklen/antigravity-data-tagger/blob/main/README.md) | |
| """) | |
| return demo | |
| # ============================================================ | |
| # 啟動 | |
| # ============================================================ | |
| if __name__ == "__main__": | |
| demo = create_demo() | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=False | |
| ) | |