media-expert

What this skill does

# Media Expert

Expert guidance for media production, content management systems, video streaming, broadcasting systems, and modern media technology solutions.

## Core Concepts

### Media Production
- Video production workflows
- Audio production and mixing
- Post-production and editing
- Visual effects (VFX)
- Color grading and correction
- Animation and motion graphics
- Live production

### Streaming and Broadcasting
- Video streaming platforms
- Content Delivery Networks (CDN)
- Adaptive bitrate streaming
- Live broadcasting
- OTT (Over-the-Top) platforms
- Digital rights management (DRM)
- Transcoding and encoding

### Technologies
- Media Asset Management (MAM)
- Digital Asset Management (DAM)
- Broadcast automation
- IP-based media production
- Cloud production workflows
- AI for content analysis
- Virtual production

### Standards and Protocols
- SMPTE standards
- HLS (HTTP Live Streaming)
- DASH (Dynamic Adaptive Streaming over HTTP)
- RTMP/RTSP protocols
- NDI (Network Device Interface)
- MXF (Material Exchange Format)
- Metadata standards (Dublin Core, IPTC)

## Content Management System

```python
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional, Dict
from decimal import Decimal
from enum import Enum
import hashlib

class MediaType(Enum):
    VIDEO = "video"
    AUDIO = "audio"
    IMAGE = "image"
    DOCUMENT = "document"

class AssetStatus(Enum):
    DRAFT = "draft"
    IN_REVIEW = "in_review"
    APPROVED = "approved"
    PUBLISHED = "published"
    ARCHIVED = "archived"

@dataclass
class MediaAsset:
    """Media asset information"""
    asset_id: str
    title: str
    description: str
    media_type: MediaType
    file_path: str
    file_size_bytes: int
    duration_seconds: Optional[float]
    resolution: Optional[str]  # e.g., "1920x1080"
    codec: Optional[str]
    bitrate_kbps: Optional[int]
    frame_rate: Optional[float]
    created_at: datetime
    created_by: str
    status: AssetStatus
    tags: List[str]
    metadata: Dict[str, str]
    checksum: str

@dataclass
class ContentPackage:
    """Content package for distribution"""
    package_id: str
    title: str
    assets: List[str]  # Asset IDs
    created_at: datetime
    scheduled_publish: Optional[datetime]
    expiration_date: Optional[datetime]
    distribution_channels: List[str]

class MediaAssetManagementSystem:
    """Media asset management and workflow"""

    def __init__(self):
        self.assets = {}
        self.packages = {}
        self.workflows = []

    def ingest_asset(self, file_path: str, metadata: dict) -> MediaAsset:
        """Ingest media asset into system"""
        # Calculate checksum
        checksum = self._calculate_checksum(file_path)

        # Extract technical metadata
        tech_metadata = self._extract_metadata(file_path)

        asset = MediaAsset(
            asset_id=self._generate_asset_id(),
            title=metadata['title'],
            description=metadata.get('description', ''),
            media_type=MediaType(metadata['media_type']),
            file_path=file_path,
            file_size_bytes=tech_metadata['file_size'],
            duration_seconds=tech_metadata.get('duration'),
            resolution=tech_metadata.get('resolution'),
            codec=tech_metadata.get('codec'),
            bitrate_kbps=tech_metadata.get('bitrate'),
            frame_rate=tech_metadata.get('frame_rate'),
            created_at=datetime.now(),
            created_by=metadata['created_by'],
            status=AssetStatus.DRAFT,
            tags=metadata.get('tags', []),
            metadata=metadata.get('custom_metadata', {}),
            checksum=checksum
        )

        self.assets[asset.asset_id] = asset

        # Trigger automated workflows
        self._trigger_workflows(asset)

        return asset

    def _extract_metadata(self, file_path: str) -> dict:
        """Extract technical metadata from media file"""
        # Would use ffprobe or similar tool
        # Simulated metadata
        return {
            'file_size': 1073741824,  # 1 GB
            'duration': 3600.0,  # 1 hour
            'resolution': '1920x1080',
            'codec': 'h264',
            'bitrate': 5000,
            'frame_rate': 29.97
        }

    def _calculate_checksum(self, file_path: str) -> str:
        """Calculate file checksum for integrity"""
        # In production, would read actual file
        return hashlib.sha256(file_path.encode()).hexdigest()

    def transcode_asset(self, asset_id: str, output_profiles: List[dict]) -> dict:
        """Transcode asset to multiple formats"""
        asset = self.assets.get(asset_id)
        if not asset:
            return {'error': 'Asset not found'}

        transcode_jobs = []

        for profile in output_profiles:
            job = {
                'job_id': self._generate_job_id(),
                'asset_id': asset_id,
                'profile_name': profile['name'],
                'target_resolution': profile['resolution'],
                'target_bitrate': profile['bitrate'],
                'target_codec': profile['codec'],
                'status': 'queued',
                'progress_percent': 0,
                'estimated_completion': datetime.now() + timedelta(hours=1)
            }
            transcode_jobs.append(job)

        return {
            'asset_id': asset_id,
            'transcode_jobs': transcode_jobs,
            'total_jobs': len(transcode_jobs)
        }

    def search_assets(self, query: dict) -> List[MediaAsset]:
        """Search assets by metadata"""
        results = []

        for asset in self.assets.values():
            match = True

            # Text search
            if 'keywords' in query:
                keywords = query['keywords'].lower()
                if keywords not in asset.title.lower() and keywords not in asset.description.lower():
                    match = False

            # Media type filter
            if 'media_type' in query and asset.media_type.value != query['media_type']:
                match = False

            # Status filter
            if 'status' in query and asset.status.value != query['status']:
                match = False

            # Tag filter
            if 'tags' in query:
                required_tags = set(query['tags'])
                asset_tags = set(asset.tags)
                if not required_tags.issubset(asset_tags):
                    match = False

            # Date range
            if 'created_after' in query and asset.created_at < query['created_after']:
                match = False

            if match:
                results.append(asset)

        return results

    def create_content_package(self, package_data: dict) -> ContentPackage:
        """Create content package for distribution"""
        package = ContentPackage(
            package_id=self._generate_package_id(),
            title=package_data['title'],
            assets=package_data['asset_ids'],
            created_at=datetime.now(),
            scheduled_publish=package_data.get('scheduled_publish'),
            expiration_date=package_data.get('expiration_date'),
            distribution_channels=package_data['channels']
        )

        self.packages[package.package_id] = package

        return package

    def analyze_content(self, asset_id: str) -> dict:
        """AI-powered content analysis"""
        asset = self.assets.get(asset_id)
        if not asset:
            return {'error': 'Asset not found'}

        # Simulate AI analysis
        analysis = {
            'asset_id': asset_id,
            'detected_objects': ['person', 'car', 'building'],
            'detected_scenes': ['outdoor', 'daytime', 'urban'],
            'faces_detected': 3,
            'speech_to_text': 'Transcribed content would appear here...',
            'sentiment': 'positive',
            'content_categories': ['news', 'documentary'],
            'suggested_tags': ['urban', 'int