Files
xiaoxia-saas/packages/domain/asset.py
T
Xiaoxia AI 80a57ff0fe Phase 6: Complete admin pages (Analytics, Monitor, Logs) and Phase 7: Asset management foundation
Phase 6 completion:
- Add Analytics page with charts (user growth, revenue, retention)
- Add SystemMonitor page (CPU, memory, services health)
- Add LogViewer page (log search, filtering, details)
- Update sidebar navigation with Admin submenu
- Add recharts to package.json for data visualization

Phase 7 Day 1-2:
- Create Asset and AssetLibrary domain entities
- Define AssetRepository and AssetLibraryRepository interfaces
- Implement InMemory repositories for testing
- Implement PostgreSQL repository adapter
- Update SQLAlchemy models for assets and asset_libraries
- Add database migration script 004_asset_management.sql

Architecture: Strict Clean Architecture compliance
Testing: InMemory adapters ready for unit tests
Database: Migration script with indexes and foreign keys
2026-06-17 17:41:48 +08:00

106 lines
3.3 KiB
Python

"""
Asset 实体 - 素材
"""
from datetime import datetime
from typing import Optional
from enum import Enum
class AssetType(str, Enum):
"""素材类型"""
VIDEO = "video"
IMAGE = "image"
AUDIO = "audio"
class AssetStatus(str, Enum):
"""素材状态"""
UPLOADING = "uploading"
READY = "ready"
PROCESSING = "processing"
ERROR = "error"
class ClassificationStatus(str, Enum):
"""分类状态"""
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
class Asset:
"""素材实体"""
def __init__(
self,
id: str,
workspace_id: str,
project_id: str,
asset_library_id: str,
name: str,
file_type: AssetType,
file_size: int,
file_url: str,
uploaded_by_user_id: str,
status: AssetStatus = AssetStatus.UPLOADING,
thumbnail_url: Optional[str] = None,
duration: Optional[float] = None,
width: Optional[int] = None,
height: Optional[int] = None,
fps: Optional[float] = None,
codec: Optional[str] = None,
classification_status: ClassificationStatus = ClassificationStatus.PENDING,
classification_result: Optional[dict] = None,
quality_score: Optional[float] = None,
created_at: Optional[datetime] = None,
updated_at: Optional[datetime] = None,
):
self.id = id
self.workspace_id = workspace_id
self.project_id = project_id
self.asset_library_id = asset_library_id
self.name = name
self.file_type = file_type
self.file_size = file_size
self.file_url = file_url
self.uploaded_by_user_id = uploaded_by_user_id
self.status = status
self.thumbnail_url = thumbnail_url
self.duration = duration
self.width = width
self.height = height
self.fps = fps
self.codec = codec
self.classification_status = classification_status
self.classification_result = classification_result
self.quality_score = quality_score
self.created_at = created_at or datetime.utcnow()
self.updated_at = updated_at or datetime.utcnow()
def to_dict(self) -> dict:
"""转换为字典"""
return {
"id": self.id,
"workspace_id": self.workspace_id,
"project_id": self.project_id,
"asset_library_id": self.asset_library_id,
"name": self.name,
"file_type": self.file_type.value,
"file_size": self.file_size,
"file_url": self.file_url,
"thumbnail_url": self.thumbnail_url,
"duration": self.duration,
"width": self.width,
"height": self.height,
"fps": self.fps,
"codec": self.codec,
"status": self.status.value,
"classification_status": self.classification_status.value,
"classification_result": self.classification_result,
"quality_score": self.quality_score,
"uploaded_by_user_id": self.uploaded_by_user_id,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}