PlanOpticon
| a6b6869… | leo | 1 | """Base interface for cloud source integrations.""" |
| a6b6869… | leo | 2 | |
| a6b6869… | leo | 3 | import logging |
| a6b6869… | leo | 4 | from abc import ABC, abstractmethod |
| a6b6869… | leo | 5 | from pathlib import Path |
| a6b6869… | leo | 6 | from typing import List, Optional |
| a6b6869… | leo | 7 | |
| a6b6869… | leo | 8 | from pydantic import BaseModel, Field |
| a6b6869… | leo | 9 | |
| a6b6869… | leo | 10 | logger = logging.getLogger(__name__) |
| a6b6869… | leo | 11 | |
| a6b6869… | leo | 12 | |
| a6b6869… | leo | 13 | class SourceFile(BaseModel): |
| a6b6869… | leo | 14 | """A file available in a cloud source.""" |
| 829e24a… | leo | 15 | |
| a6b6869… | leo | 16 | name: str = Field(description="File name") |
| a6b6869… | leo | 17 | id: str = Field(description="Provider-specific file identifier") |
| a6b6869… | leo | 18 | size_bytes: Optional[int] = Field(default=None, description="File size in bytes") |
| a6b6869… | leo | 19 | mime_type: Optional[str] = Field(default=None, description="MIME type") |
| a6b6869… | leo | 20 | modified_at: Optional[str] = Field(default=None, description="Last modified timestamp") |
| a6b6869… | leo | 21 | path: Optional[str] = Field(default=None, description="Path within the source folder") |
| a6b6869… | leo | 22 | |
| a6b6869… | leo | 23 | |
| a6b6869… | leo | 24 | class BaseSource(ABC): |
| a6b6869… | leo | 25 | """Abstract base class for cloud source integrations.""" |
| a6b6869… | leo | 26 | |
| a6b6869… | leo | 27 | @abstractmethod |
| a6b6869… | leo | 28 | def authenticate(self) -> bool: |
| a6b6869… | leo | 29 | """Authenticate with the cloud provider. Returns True on success.""" |
| a6b6869… | leo | 30 | ... |
| a6b6869… | leo | 31 | |
| a6b6869… | leo | 32 | @abstractmethod |
| a6b6869… | leo | 33 | def list_videos( |
| a6b6869… | leo | 34 | self, |
| a6b6869… | leo | 35 | folder_id: Optional[str] = None, |
| a6b6869… | leo | 36 | folder_path: Optional[str] = None, |
| a6b6869… | leo | 37 | patterns: Optional[List[str]] = None, |
| a6b6869… | leo | 38 | ) -> List[SourceFile]: |
| a6b6869… | leo | 39 | """List video files in a folder.""" |
| a6b6869… | leo | 40 | ... |
| a6b6869… | leo | 41 | |
| a6b6869… | leo | 42 | @abstractmethod |
| a6b6869… | leo | 43 | def download( |
| a6b6869… | leo | 44 | self, |
| a6b6869… | leo | 45 | file: SourceFile, |
| a6b6869… | leo | 46 | destination: Path, |
| a6b6869… | leo | 47 | ) -> Path: |
| a6b6869… | leo | 48 | """Download a file to a local path. Returns the local path.""" |
| a6b6869… | leo | 49 | ... |
| a6b6869… | leo | 50 | |
| a6b6869… | leo | 51 | def download_all( |
| a6b6869… | leo | 52 | self, |
| a6b6869… | leo | 53 | files: List[SourceFile], |
| a6b6869… | leo | 54 | destination_dir: Path, |
| a6b6869… | leo | 55 | ) -> List[Path]: |
| 287a3bb… | leo | 56 | """Download multiple files to a directory, preserving subfolder structure.""" |
| a6b6869… | leo | 57 | destination_dir.mkdir(parents=True, exist_ok=True) |
| a6b6869… | leo | 58 | paths = [] |
| a6b6869… | leo | 59 | for f in files: |
| 287a3bb… | leo | 60 | # Use path (with subfolder) if available, otherwise just name |
| 287a3bb… | leo | 61 | relative = f.path if f.path else f.name |
| 287a3bb… | leo | 62 | dest = destination_dir / relative |
| a6b6869… | leo | 63 | try: |
| a6b6869… | leo | 64 | local_path = self.download(f, dest) |
| a6b6869… | leo | 65 | paths.append(local_path) |
| 287a3bb… | leo | 66 | logger.info(f"Downloaded: {relative}") |
| a6b6869… | leo | 67 | except Exception as e: |
| 287a3bb… | leo | 68 | logger.error(f"Failed to download {relative}: {e}") |
| a6b6869… | leo | 69 | return paths |