27 lines
1017 B
Python
27 lines
1017 B
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from review_agent.models import FileSummaryBatchAttachment
|
|
|
|
from ..paths import resolve_storage_path
|
|
from ..services.archive import ARCHIVE_EXTENSIONS, extract_archive
|
|
from .base import BaseSkill, SkillResult, WorkflowContext
|
|
|
|
|
|
class ArchiveExtractSkill(BaseSkill):
|
|
name = "archive_extract"
|
|
|
|
def run(self, context: WorkflowContext) -> SkillResult:
|
|
extracted_count = 0
|
|
target_dir = Path(context.batch.work_dir or "")
|
|
if not target_dir:
|
|
return SkillResult(success=True, data={"extracted_count": 0})
|
|
|
|
for binding in FileSummaryBatchAttachment.objects.filter(batch=context.batch):
|
|
path = resolve_storage_path(binding.attachment.storage_path)
|
|
if path.suffix.lower().lstrip(".") not in ARCHIVE_EXTENSIONS:
|
|
continue
|
|
extracted_count += len(extract_archive(path, target_dir))
|
|
return SkillResult(success=True, data={"extracted_count": extracted_count})
|