| name | media-expert |
| version | 1.0.0 |
| description | Expert-level media production, content management, streaming, broadcasting, and media technology systems |
| category | domains |
| tags | ["media","video","streaming","broadcast","cms","production"] |
| allowed-tools | ["Read","Write","Edit"] |
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
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]
codec: Optional[str]
bitrate_kbps: Optional[int]
frame_rate: Optional[float]
created_at: datetime
created_by:
status: AssetStatus
tags: []
metadata: [, ]
checksum:
:
package_id:
title:
assets: []
created_at: datetime
scheduled_publish: [datetime]
expiration_date: [datetime]
distribution_channels: []
:
():
.assets = {}
.packages = {}
.workflows = []
() -> MediaAsset:
checksum = ._calculate_checksum(file_path)
tech_metadata = ._extract_metadata(file_path)
asset = MediaAsset(
asset_id=._generate_asset_id(),
title=metadata[],
description=metadata.get(, ),
media_type=MediaType(metadata[]),
file_path=file_path,
file_size_bytes=tech_metadata[],
duration_seconds=tech_metadata.get(),
resolution=tech_metadata.get(),
codec=tech_metadata.get(),
bitrate_kbps=tech_metadata.get(),
frame_rate=tech_metadata.get(),
created_at=datetime.now(),
created_by=metadata[],
status=AssetStatus.DRAFT,
tags=metadata.get(, []),
metadata=metadata.get(, {}),
checksum=checksum
)
.assets[asset.asset_id] = asset
._trigger_workflows(asset)
asset
() -> :
{
: ,
: ,
: ,
: ,
: ,
:
}
() -> :
hashlib.sha256(file_path.encode()).hexdigest()
() -> :
asset = .assets.get(asset_id)
asset:
{: }
transcode_jobs = []
profile output_profiles:
job = {
: ._generate_job_id(),
: asset_id,
: profile[],
: profile[],
: profile[],
: profile[],
: ,
: ,
: datetime.now() + timedelta(hours=)
}
transcode_jobs.append(job)
{
: asset_id,
: transcode_jobs,
: (transcode_jobs)
}
() -> [MediaAsset]:
results = []
asset .assets.values():
=
query:
keywords = query[].lower()
keywords asset.title.lower() keywords asset.description.lower():
=
query asset.media_type.value != query[]:
=
query asset.status.value != query[]:
=
query:
required_tags = (query[])
asset_tags = (asset.tags)
required_tags.issubset(asset_tags):
=
query asset.created_at < query[]:
=
:
results.append(asset)
results
() -> ContentPackage:
package = ContentPackage(
package_id=._generate_package_id(),
title=package_data[],
assets=package_data[],
created_at=datetime.now(),
scheduled_publish=package_data.get(),
expiration_date=package_data.get(),
distribution_channels=package_data[]
)
.packages[package.package_id] = package
package
() -> :
asset = .assets.get(asset_id)
asset:
{: }
analysis = {
: asset_id,
: [, , ],
: [, , ],
: ,
: ,
: ,
: [, ],
: [, , ],
:
}
analysis
():
() -> :
uuid
() -> :
uuid
() -> :
uuid
Video Streaming Platform
class VideoStreamingPlatform:
"""Video streaming and delivery system"""
def __init__(self):
self.streams = {}
self.viewers = {}
self.cdn_nodes = {}
def start_live_stream(self, stream_data: dict) -> dict:
"""Start live video stream"""
stream_id = self._generate_stream_id()
stream = {
'stream_id': stream_id,
'title': stream_data['title'],
'description': stream_data.get('description', ''),
'streamer_id': stream_data['streamer_id'],
'status': 'live',
'started_at': datetime.now(),
'viewer_count': 0,
'peak_viewers': 0,
'ingest_url': f'rtmp://ingest.example.com/live/{stream_id}',
'playback_urls': {
'hls': f'https://cdn.example.com/live/{stream_id}/playlist.m3u8',
'dash': f'https://cdn.example.com/live/{stream_id}/manifest.mpd'
},
'quality_profiles': ['1080p', '720p', '480p', ]
}
.streams[stream_id] = stream
stream
() -> :
hls_variants = [
{
: ,
: ,
: ,
:
},
{
: ,
: ,
: ,
:
},
{
: ,
: ,
: ,
:
},
{
: ,
: ,
: ,
:
}
]
{
: asset_id,
: ,
: ,
: hls_variants
}
() -> :
metrics = {
: stream_id,
: viewer_id,
: ,
: ,
: ,
: ,
: ,
: ,
: ,
:
}
qoe_score = ._calculate_qoe(metrics)
metrics[] = qoe_score
metrics
() -> :
score =
score -= metrics[] *
metrics[] > :
score -=
score -= metrics[] *
(, score)
() -> :
drm = {
: asset_id,
: {
: {
: ,
: [, ]
},
: {
: ,
:
},
: {
:
}
},
: ,
:
}
drm
() -> :
nearest_node = ._find_nearest_cdn_node(viewer_location)
{
: asset_id,
: nearest_node[],
: nearest_node[],
: nearest_node[],
: nearest_node[],
:
}
() -> :
{
: ,
: ,
: ,
:
}
() -> :
uuid
Broadcast Automation
class BroadcastAutomationSystem:
"""Broadcast scheduling and automation"""
def __init__(self):
self.schedule = []
self.playlists = {}
def create_broadcast_schedule(self, channel: str, date: datetime, programming: List[dict]) -> dict:
"""Create daily broadcast schedule"""
schedule_items = []
current_time = date.replace(hour=0, minute=0, second=0)
for program in programming:
item = {
'channel': channel,
'start_time': current_time,
'end_time': current_time + timedelta(seconds=program['duration']),
'program_title': program['title'],
'asset_id': program['asset_id'],
'type': program['type'],
'metadata': program.get('metadata', {})
}
schedule_items.append(item)
current_time = item['end_time']
self.schedule.extend(schedule_items)
return {
'channel': channel,
'date': date.date().isoformat(),
'total_items': len(schedule_items),
: (schedule_items[-][] - schedule_items[][]).seconds,
: schedule_items[:]
}
() -> :
playlist = {
: ._generate_playlist_id(),
: ,
: []
}
playlist
() -> :
status = {
: channel,
: ,
: ,
: ,
: ,
: ,
: {
: ,
: ,
:
},
: []
}
status
() -> :
uuid
Best Practices
Media Production
- Use standardized workflows
- Implement version control
- Maintain proper backups
- Use collaborative tools
- Implement quality control
- Document production processes
- Use industry-standard formats
Content Management
- Implement robust metadata schema
- Use consistent naming conventions
- Enable full-text search
- Implement access controls
- Maintain audit trails
- Use automated workflows
- Implement archival policies
Streaming Delivery
- Use adaptive bitrate streaming
- Implement CDN for global delivery
- Monitor QoE metrics
- Optimize for mobile devices
- Implement DRM when required
- Use low-latency protocols for live
- Monitor buffer ratios
Broadcasting
- Implement redundant systems
- Automate scheduling
- Monitor signal quality
- Maintain emergency protocols
- Use backup playout systems
- Implement proper logging
- Conduct regular testing
Anti-Patterns
❌ No backup systems
❌ Poor metadata management
❌ Single bitrate streaming
❌ No CDN implementation
❌ Ignoring QoE metrics
❌ Manual scheduling processes
❌ No DRM for premium content
❌ Poor asset organization
❌ No disaster recovery plan
Resources
