Spaces:
Sleeping
Sleeping
File size: 3,489 Bytes
5486429 79f9175 5486429 79f9175 5486429 79f9175 5486429 8843db1 5486429 8843db1 5486429 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """
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() |