Spaces:
Sleeping
Sleeping
| """ | |
| KSTools Version Manager - 版本系統資料庫連接 | |
| 專門處理版本系統 (Project 2) 的資料庫操作 | |
| """ | |
| import os | |
| from supabase import create_client, Client | |
| from typing import Optional | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| class VersionDatabase: | |
| """版本系統 Supabase 資料庫客戶端""" | |
| def __init__(self): | |
| # 明確使用版本系統的環境變數 | |
| self.url: str = os.environ.get("SUPABASE_VERSION_URL", "") | |
| self.anon_key: str = os.environ.get("SUPABASE_VERSION_ANON_KEY", "") | |
| self.service_key: str = os.environ.get("SUPABASE_VERSION_SERVICE_KEY", "") | |
| self.client: Optional[Client] = None | |
| if self.url and self.service_key and not self.url.startswith('your-') and not self.service_key.startswith('your-'): | |
| try: | |
| self.client = create_client(self.url, self.service_key) | |
| logger.info("✅ Version System Supabase connection initialized") | |
| except Exception as e: | |
| logger.error(f"❌ Version System connection failed: {e}") | |
| self.client = None | |
| else: | |
| logger.warning("⚠️ Version System Supabase not configured") | |
| def get_client(self) -> Optional[Client]: | |
| """取得版本系統 Supabase 客戶端""" | |
| if not self.client: | |
| logger.error("❌ Version System Supabase client 未初始化") | |
| return self.client | |
| def is_connected(self) -> bool: | |
| """檢查版本系統是否連接成功""" | |
| connected = self.client is not None | |
| logger.info(f"🔍 Version System Supabase connection status: {'Connected' if connected else 'Not connected'}") | |
| return connected | |
| async def upload_file(self, file_path: str, file_data: bytes, content_type: str = 'application/zip') -> str: | |
| """上傳檔案到 Supabase Storage""" | |
| if not self.client: | |
| raise Exception("Version System Supabase client not initialized") | |
| try: | |
| # 上傳到 plugin-releases bucket | |
| response = self.client.storage.from_('plugin-releases').upload( | |
| file_path, | |
| file_data, | |
| file_options={"content-type": content_type} | |
| ) | |
| # 檢查上傳是否成功 | |
| if hasattr(response, 'error') and response.error: | |
| raise Exception(f"Upload failed: {response.error}") | |
| # 取得公開 URL | |
| url_response = self.client.storage.from_('plugin-releases').get_public_url(file_path) | |
| # 確保返回正確的 URL 字串 | |
| if hasattr(url_response, 'public_url'): | |
| public_url = url_response.public_url | |
| elif isinstance(url_response, str): | |
| public_url = url_response | |
| else: | |
| public_url = str(url_response) | |
| logger.info(f"✅ File uploaded successfully: {file_path} -> {public_url}") | |
| return public_url | |
| except Exception as e: | |
| logger.error(f"❌ File upload failed: {e}") | |
| raise | |
| def get_public_url(self, file_path: str) -> str: | |
| """取得檔案的公開 URL""" | |
| if not self.client: | |
| raise Exception("Version System Supabase client not initialized") | |
| return self.client.storage.from_('plugin-releases').get_public_url(file_path) | |
| # 全域版本系統資料庫實例 | |
| version_db = VersionDatabase() |