198 lines
7.2 KiB
Python
198 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Iterable
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.enums.upload_resource import (
|
|
UploadResourceBindStatusEnum,
|
|
UploadResourceCreatedByEnum,
|
|
UploadResourceDeletePolicyEnum,
|
|
UploadResourceEventEnum,
|
|
UploadResourceFileDeleteStatusEnum,
|
|
UploadResourceModuleEnum,
|
|
UploadResourceSourceModelEnum,
|
|
UploadResourceTypeEnum,
|
|
)
|
|
from app.models.upload_resource import UploadResource
|
|
from app.models.shot_replicate_segment import ShotReplicateSegment
|
|
from app.services.upload_resource.accounting_service import release_upload_resource_capacity
|
|
from app.services.upload_resource.core_service import record_external_upload_resource
|
|
from app.services.upload_resource.log_service import log_upload_resource_event
|
|
from app.services.upload_resource.path_resolver import normalize_storage_path, upload_url_to_storage_path
|
|
|
|
|
|
def _clean_ids(values: Iterable[str | None] | None) -> list[str]:
|
|
if not values:
|
|
return []
|
|
return [str(v).strip() for v in dict.fromkeys(values) if v and str(v).strip()]
|
|
|
|
|
|
def _clean_urls(values: Iterable[str | None] | None) -> list[str]:
|
|
if not values:
|
|
return []
|
|
return [str(v).strip() for v in dict.fromkeys(values) if v and str(v).strip()]
|
|
|
|
|
|
async def bind_upload_resources(
|
|
db: AsyncSession,
|
|
*,
|
|
user_id: str,
|
|
module: str,
|
|
source_model: str,
|
|
source_id: str,
|
|
resource_ids: Iterable[str | None] | None = None,
|
|
urls: Iterable[str | None] | None = None,
|
|
allow_common_migrate: bool = True,
|
|
) -> dict[str, int]:
|
|
ids = _clean_ids(resource_ids)
|
|
url_values = _clean_urls(urls)
|
|
storage_paths = [p for p in (upload_url_to_storage_path(url) for url in url_values) if p]
|
|
|
|
if not ids and not storage_paths:
|
|
return {"matched": 0, "bound": 0, "skipped": 0, "conflict": 0}
|
|
|
|
conditions = []
|
|
if ids:
|
|
conditions.append(UploadResource.id.in_(ids))
|
|
if storage_paths:
|
|
conditions.append(UploadResource.storage_path.in_(storage_paths))
|
|
|
|
stmt = select(UploadResource).where(UploadResource.user_id == user_id, UploadResource.deleted_at.is_(None))
|
|
if len(conditions) == 1:
|
|
stmt = stmt.where(conditions[0])
|
|
else:
|
|
from sqlalchemy import or_
|
|
stmt = stmt.where(or_(*conditions))
|
|
stmt = stmt.with_for_update()
|
|
|
|
result = await db.execute(stmt)
|
|
resources = result.scalars().all()
|
|
stats = {"matched": len(resources), "bound": 0, "skipped": 0, "conflict": 0}
|
|
|
|
for resource in resources:
|
|
if resource.source_id == source_id and resource.source_model == source_model:
|
|
stats["skipped"] += 1
|
|
continue
|
|
if resource.source_id and resource.source_id != source_id:
|
|
stats["conflict"] += 1
|
|
log_upload_resource_event(
|
|
event_type=UploadResourceEventEnum.BIND_CONFLICT.value,
|
|
module=module,
|
|
user_id=user_id,
|
|
resource_id=resource.id,
|
|
source_model=source_model,
|
|
source_id=source_id,
|
|
event_status="warning",
|
|
detail={
|
|
"current_source_model": resource.source_model,
|
|
"current_source_id": resource.source_id,
|
|
"storage_path": resource.storage_path,
|
|
},
|
|
)
|
|
continue
|
|
if resource.module != module:
|
|
if not (allow_common_migrate and resource.module == UploadResourceModuleEnum.COMMON.value):
|
|
stats["conflict"] += 1
|
|
continue
|
|
resource.module = module
|
|
resource.source_module = module
|
|
resource.source_model = source_model
|
|
resource.source_id = source_id
|
|
resource.source_module = module
|
|
resource.bind_status = UploadResourceBindStatusEnum.BOUND.value
|
|
resource.delete_policy = UploadResourceDeletePolicyEnum.MODULE_ONLY.value
|
|
stats["bound"] += 1
|
|
|
|
if stats["bound"]:
|
|
await db.flush()
|
|
log_upload_resource_event(
|
|
event_type=UploadResourceEventEnum.BIND_SUCCESS.value,
|
|
module=module,
|
|
user_id=user_id,
|
|
source_model=source_model,
|
|
source_id=source_id,
|
|
detail=stats,
|
|
)
|
|
return stats
|
|
|
|
|
|
async def release_upload_resources_by_source(
|
|
db: AsyncSession,
|
|
*,
|
|
source_model: str,
|
|
source_ids: Iterable[str],
|
|
module: str | None = None,
|
|
) -> dict[str, Any]:
|
|
ids = _clean_ids(source_ids)
|
|
if not ids:
|
|
return {"matched": 0, "released": 0, "released_size_bytes": 0, "released_resource_ids": []}
|
|
stmt = select(UploadResource).where(
|
|
UploadResource.source_model == source_model,
|
|
UploadResource.source_id.in_(ids),
|
|
UploadResource.deleted_at.is_(None),
|
|
).with_for_update()
|
|
if module:
|
|
stmt = stmt.where(UploadResource.module == module)
|
|
result = await db.execute(stmt)
|
|
resources = result.scalars().all()
|
|
now = datetime.now(timezone.utc)
|
|
stats: dict[str, Any] = {
|
|
"matched": len(resources),
|
|
"released": 0,
|
|
"released_size_bytes": 0,
|
|
"released_resource_ids": [],
|
|
}
|
|
for resource in resources:
|
|
resource_id = resource.id
|
|
size = int(resource.file_size_bytes or 0)
|
|
resource.deleted_at = now
|
|
resource.file_delete_status = UploadResourceFileDeleteStatusEnum.PENDING_DELETE.value
|
|
resource.file_delete_error = None
|
|
if await release_upload_resource_capacity(db, resource, released_at=now):
|
|
stats["released"] += 1
|
|
stats["released_size_bytes"] += size
|
|
stats["released_resource_ids"].append(resource_id)
|
|
if resources:
|
|
await db.flush()
|
|
log_upload_resource_event(
|
|
event_type=UploadResourceEventEnum.DELETE_MARKED_PENDING.value,
|
|
module=module,
|
|
source_model=source_model,
|
|
source_id=",".join(ids[:20]),
|
|
detail={k: v for k, v in stats.items() if k != "released_resource_ids"} | {"released_resource_count": len(stats["released_resource_ids"])},
|
|
)
|
|
return stats
|
|
|
|
|
|
async def record_shot_segment_upload_resource(
|
|
db: AsyncSession,
|
|
*,
|
|
segment: ShotReplicateSegment,
|
|
storage_path: str,
|
|
resource_url: str,
|
|
file_size_bytes: int,
|
|
) -> UploadResource:
|
|
return await record_external_upload_resource(
|
|
db,
|
|
user_id=segment.user_id,
|
|
module=UploadResourceModuleEnum.SHOT_REPLICATE.value,
|
|
resource_type=UploadResourceTypeEnum.SHOT_SEGMENT.value,
|
|
resource_url=resource_url,
|
|
storage_path=normalize_storage_path(storage_path),
|
|
file_size_bytes=file_size_bytes,
|
|
file_name=Path(storage_path).name,
|
|
mime_type="video/mp4",
|
|
duration_seconds=float(segment.duration_seconds or 0) if segment.duration_seconds else None,
|
|
duration_source="business",
|
|
source_model=UploadResourceSourceModelEnum.SHOT_REPLICATE_SEGMENT.value,
|
|
source_id=segment.id,
|
|
bind_status=UploadResourceBindStatusEnum.BOUND.value,
|
|
delete_policy=UploadResourceDeletePolicyEnum.MODULE_ONLY.value,
|
|
created_by=UploadResourceCreatedByEnum.SPLIT_TASK.value,
|
|
metadata={"task_set_id": segment.task_set_id, "segment_index": segment.segment_index},
|
|
created_at=segment.created_at,
|
|
)
|