Skip to content

Instantly share code, notes, and snippets.

@gabanox

gabanox/cinetime_s3_demo.py

Last active March 24, 2026 17:26
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save gabanox/62ba3c3792d8714f1b45e74e6f94227c to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save gabanox/62ba3c3792d8714f1b45e74e6f94227c to your computer and use it in GitHub Desktop.
Download ZIP
CineTime S3 Storage Classes Demo — AWS re/Start | Práctica manual de clases de almacenamiento sin Lifecycle Policies
Raw
cinetime_s3_demo.py
#!/usr/bin/env python3
"""
╔══════════════════════════════════════════════════════════════════════╗
║ CineTime — Demo de Clases de Almacenamiento S3 ║
║ AWS re/Start | Práctica Manual de Storage Classes ║
╚══════════════════════════════════════════════════════════════════════╝
IMPORTANTE: Este script simula mover archivos entre clases de S3
SIN usar Lifecycle Policies automáticas.
Cada operación usa copy_object() para que veas exactamente
qué llamada se hace a la API de AWS.
USO:
python3 cinetime_s3_demo.py setup → Crea bucket y archivos demo
python3 cinetime_s3_demo.py listar → Estado actual de todos los objetos
python3 cinetime_s3_demo.py escenario 1 → Estrenos (>90 días) → Standard-IA
python3 cinetime_s3_demo.py escenario 2 → Archivos RAW → Glacier Flexible
python3 cinetime_s3_demo.py escenario 3 → Thumbnails → Intelligent-Tiering
python3 cinetime_s3_demo.py escenario 4 → Catálogo general → Intelligent-Tiering
python3 cinetime_s3_demo.py escenario 5 → Contenido archivado → Glacier Deep Archive
python3 cinetime_s3_demo.py reset → Regresa todo a S3 Standard (para repetir demo)
python3 cinetime_s3_demo.py demo → Ejecuta los 5 escenarios secuencialmente
"""
import boto3
import sys
import time
import json
from botocore.exceptions import ClientError
# ─────────────────────────────────────────────────────────────
# CONFIGURACIÓN — pega aquí las credenciales de tu Lab
# ─────────────────────────────────────────────────────────────
AWS_ACCESS_KEY_ID = ""
AWS_SECRET_ACCESS_KEY = ""
AWS_SESSION_TOKEN = ""
AWS_REGION = "us-east-1"
BUCKET = "MI_BUCKET"
# ─────────────────────────────────────────────────────────────
# COLORES ANSI — para output bonito en la terminal
# ─────────────────────────────────────────────────────────────
class C:
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
# Colores de texto
BLACK = "\033[30m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
BLUE = "\033[34m"
MAGENTA = "\033[35m"
CYAN = "\033[36m"
WHITE = "\033[37m"
# Colores brillantes
BRED = "\033[91m"
BGREEN = "\033[92m"
BYELLOW = "\033[93m"
BBLUE = "\033[94m"
BMAGENTA= "\033[95m"
BCYAN = "\033[96m"
BWHITE = "\033[97m"
# Fondo
BG_BLUE = "\033[44m"
BG_CYAN = "\033[46m"
# ─────────────────────────────────────────────────────────────
# INFORMACIÓN DE CLASES S3
# ─────────────────────────────────────────────────────────────
STORAGE_CLASSES = {
"STANDARD": {
"nombre": "S3 Standard",
"color": C.BBLUE,
"costo_gb": 0.023,
"acceso": "Milisegundos",
"minimo": "Sin mínimo",
"uso_ideal": "Datos de producción activos, alto acceso",
"emoji": "⚡"
},
"STANDARD_IA": {
"nombre": "S3 Standard-IA",
"color": C.BCYAN,
"costo_gb": 0.0125,
"acceso": "Milisegundos",
"minimo": "30 días / 128 KB",
"uso_ideal": "Datos poco frecuentes, alta durabilidad",
"emoji": "📦"
},
"INTELLIGENT_TIERING": {
"nombre": "S3 Intelligent-Tiering",
"color": C.BMAGENTA,
"costo_gb": 0.023,
"acceso": "Milisegundos",
"minimo": "30 días",
"uso_ideal": "Acceso impredecible, optimización automática",
"emoji": "🧠"
},
"ONEZONE_IA": {
"nombre": "S3 One Zone-IA",
"color": C.BGREEN,
"costo_gb": 0.01,
"acceso": "Milisegundos",
"minimo": "30 días / 128 KB",
"uso_ideal": "Datos reproducibles, una sola zona AZ",
"emoji": "🗂️"
},
"GLACIER_IR": {
"nombre": "Glacier Instant Retrieval",
"color": C.BYELLOW,
"costo_gb": 0.004,
"acceso": "Milisegundos",
"minimo": "90 días / 128 KB",
"uso_ideal": "Archivos médicos, acceso trimestral",
"emoji": "❄️"
},
"GLACIER": {
"nombre": "Glacier Flexible Retrieval",
"color": C.YELLOW,
"costo_gb": 0.0036,
"acceso": "1 min – 12 horas",
"minimo": "90 días / 40 KB",
"uso_ideal": "Backups, archivos RAW, recuperación planificada",
"emoji": "🧊"
},
"DEEP_ARCHIVE": {
"nombre": "Glacier Deep Archive",
"color": C.BRED,
"costo_gb": 0.00099,
"acceso": "12 – 48 horas",
"minimo": "180 días / 40 KB",
"uso_ideal": "Retención legal, archivo permanente",
"emoji": "🏔️"
},
}
# ─────────────────────────────────────────────────────────────
# ARCHIVOS DEMO — caso CineTime Streaming
# ─────────────────────────────────────────────────────────────
DEMO_FILES = {
# ── Estrenos (< 90 días) — alta demanda, deben estar en Standard
"estrenos/serie-sombra-del-dragon-ep01.mp4.demo": {
"clase_inicial": "STANDARD",
"escenario": 1,
"clase_destino": "STANDARD_IA",
"tamano_sim": "4.2 GB",
"descripcion": "Serie original CineTime - Episodio 1 (ESTRENO)",
"contenido": """[DEMO VIDEO] CineTime Original - La Sombra del Dragón EP01
Resolución: 4K HDR | Duración: 52 min | Tamaño real: ~4.2 GB
Fecha estreno: 2026-01-05 | Días en catálogo: 78 días
Estado: EN TEMPORADA DE ESTRENO — acceso masivo simultáneo
Clase S3: STANDARD — miles de usuarios acceden concurrentemente
Métrica: 2.3M reproducciones en los primeros 30 días"""
},
"estrenos/serie-sombra-del-dragon-ep02.mp4.demo": {
"clase_inicial": "STANDARD",
"escenario": 1,
"clase_destino": "STANDARD_IA",
"tamano_sim": "4.5 GB",
"descripcion": "Serie original CineTime - Episodio 2 (ESTRENO)",
"contenido": """[DEMO VIDEO] CineTime Original - La Sombra del Dragón EP02
Resolución: 4K HDR | Duración: 58 min | Tamaño real: ~4.5 GB
Fecha estreno: 2026-01-12 | Días en catálogo: 71 días
Estado: EN TEMPORADA DE ESTRENO — acceso masivo simultáneo
Clase S3: STANDARD — latencia mínima requerida para streaming"""
},
"estrenos/cosmos-la-pelicula-4k.mp4.demo": {
"clase_inicial": "STANDARD",
"escenario": 1,
"clase_destino": "STANDARD_IA",
"tamano_sim": "18.7 GB",
"descripcion": "Película original CineTime 4K (ESTRENO)",
"contenido": """[DEMO VIDEO] CineTime Película - COSMOS: El Último Viaje
Resolución: 4K Dolby Vision | Duración: 2h 18min | Tamaño real: ~18.7 GB
Fecha estreno: 2026-02-14 | Días en catálogo: 38 días
Estado: EN TEMPORADA DE ESTRENO — pico de demanda (San Valentín)
Clase S3: STANDARD — rendimiento máximo, alta disponibilidad"""
},
# ── Archivos RAW — masivos, acceso muy raro
"raw/filming-sombra-dragon-ep01-raw.r3d.demo": {
"clase_inicial": "STANDARD",
"escenario": 2,
"clase_destino": "GLACIER",
"tamano_sim": "380 GB",
"descripcion": "Footage RAW sin editar - EP01",
"contenido": """[DEMO RAW] CineTime RAW Footage - La Sombra del Dragón EP01
Cámara: RED Komodo 6K | Formato: R3D RAW | Tamaño real: ~380 GB
Fecha filmación: 2025-10-15 | Última vez accedido: 2025-12-01
Estado: ARCHIVO RAW — solo para re-ediciones o versiones especiales
Clase S3: se moverá a GLACIER FLEXIBLE
Advertencia: recuperación tomará 3-5 horas cuando se necesite"""
},
"raw/filming-cosmos-broll-master.r3d.demo": {
"clase_inicial": "STANDARD",
"escenario": 2,
"clase_destino": "GLACIER",
"tamano_sim": "820 GB",
"descripcion": "Footage RAW B-roll película Cosmos",
"contenido": """[DEMO RAW] CineTime RAW B-Roll - COSMOS Película
Cámara: ARRI Alexa 35 | Formato: ARRIRAW | Tamaño real: ~820 GB
Fecha filmación: 2025-08-20 | Última vez accedido: 2025-11-15
Estado: ARCHIVO RAW — material descartado del corte final
Nunca se accede excepto para: auditorías del director, versiones extendidas
Guardar en Glacier Flexible: ahorro del 84% vs Standard"""
},
"raw/filming-sombra-dragon-ep02-raw.r3d.demo": {
"clase_inicial": "STANDARD",
"escenario": 2,
"clase_destino": "GLACIER",
"tamano_sim": "420 GB",
"descripcion": "Footage RAW sin editar - EP02",
"contenido": """[DEMO RAW] CineTime RAW Footage - La Sombra del Dragón EP02
Cámara: RED Komodo 6K | Formato: R3D RAW | Tamaño real: ~420 GB
Fecha filmación: 2025-10-22 | Última vez accedido: 2025-12-08
Caso de uso Glacier Flexible: si hay demanda de 'Director's Cut' en 2027,
el equipo de post-producción solicita la restauración con 24h de anticipación."""
},
# ── Thumbnails y Material Promocional — acceso moderado/impredecible
"thumbnails/sombra-dragon-poster-principal.jpg.demo": {
"clase_inicial": "STANDARD",
"escenario": 3,
"clase_destino": "INTELLIGENT_TIERING",
"tamano_sim": "2.4 MB",
"descripcion": "Poster principal serie - material promocional",
"contenido": """[DEMO THUMBNAIL] CineTime Asset - Sombra del Dragón Poster
Dimensiones: 2000x3000px | Formato: JPEG optimizado | Tamaño: ~2.4 MB
Uso: Página principal, apps móviles, smart TVs, redes sociales
Patrón de acceso: VARIABLE — alto cuando la serie está en tendencia,
bajo cuando pasa la temporada. Intelligent-Tiering lo maneja automáticamente.
IT monitorea accesos y mueve entre tiers de forma transparente."""
},
"thumbnails/cosmos-thumbnail-16x9.jpg.demo": {
"clase_inicial": "STANDARD",
"escenario": 3,
"clase_destino": "INTELLIGENT_TIERING",
"tamano_sim": "1.8 MB",
"descripcion": "Thumbnail película Cosmos (16:9)",
"contenido": """[DEMO THUMBNAIL] CineTime Asset - COSMOS Thumbnail
Dimensiones: 1920x1080px | Formato: JPEG | Tamaño: ~1.8 MB
Uso: Grid de películas, reproductor embebido, push notifications
Patrón de acceso: IMPREDECIBLE — depende de algoritmo de recomendaciones.
Si la película aparece en 'Trending' → millones de requests.
Si no → casi ninguno. Intelligent-Tiering es la clase ideal."""
},
"thumbnails/sombra-dragon-ep01-still.jpg.demo": {
"clase_inicial": "STANDARD",
"escenario": 3,
"clase_destino": "INTELLIGENT_TIERING",
"tamano_sim": "3.1 MB",
"descripcion": "Still frame EP01 para redes sociales",
"contenido": """[DEMO THUMBNAIL] CineTime Asset - Still EP01
Uso: Instagram, Twitter, previews en apps
Patrón: Viral en estreno, luego acceso esporádico cuando hay menciones."""
},
# ── Catálogo General > 1 año — acceso moderado, patrón variable
"catalogo/serie-neon-city-temporada1-completa.mp4.demo": {
"clase_inicial": "STANDARD",
"escenario": 4,
"clase_destino": "INTELLIGENT_TIERING",
"tamano_sim": "28 GB",
"descripcion": "Serie establecida - 1 año en catálogo",
"contenido": """[DEMO VIDEO] CineTime Catálogo - Neon City Temporada 1
Resolución: 1080p | Episodios: 10 | Tamaño total: ~28 GB
Fecha estreno: 2025-01-15 | Tiempo en catálogo: 14 meses
Acceso: MODERADO E IMPREDECIBLE — la serie puede volverse viral
si aparece en recomendaciones, redes sociales o binge-watching.
Intelligent-Tiering: mueve a Frequent Access tier automáticamente
si el acceso aumenta. Sin intervención manual."""
},
"catalogo/pelicula-la-ultima-frontera-2024.mp4.demo": {
"clase_inicial": "STANDARD",
"escenario": 4,
"clase_destino": "INTELLIGENT_TIERING",
"tamano_sim": "12 GB",
"descripcion": "Película de catálogo - 18 meses",
"contenido": """[DEMO VIDEO] CineTime Catálogo - La Última Frontera (2024)
Resolución: 4K | Duración: 1h 54min | Tamaño: ~12 GB
Tiempo en catálogo: 18 meses | Patrón: esporádico con picos
Intelligent-Tiering tiers:
• Frequent Access: $0.023/GB (cuando hay acceso constante)
• Infrequent Access: $0.0125/GB (después de 30 días sin acceso)
• Archive Instant: $0.004/GB (después de 90 días sin acceso)
Ideal para contenido cuyo patrón no puedes predecir."""
},
# ── Contenido fuera del catálogo activo — archivo permanente
"archivado/serie-cancelada-pixel-wars-2022.mp4.demo": {
"clase_inicial": "STANDARD",
"escenario": 5,
"clase_destino": "DEEP_ARCHIVE",
"tamano_sim": "45 GB",
"descripcion": "Serie cancelada - fuera del catálogo activo",
"contenido": """[DEMO VIDEO] CineTime Archivado - Pixel Wars Temporada 1
Resolución: 1080p | Episodios: 8 | Tamaño: ~45 GB
Fecha cancelación: 2023-06-01 | Estado: FUERA DEL CATÁLOGO
Razón del archivo: serie cancelada por bajo rating
¿Cuándo se necesitará de nuevo?
→ Si se vende la licencia a otro servicio (improbable pero posible)
→ Si hay una producción de 'revival' en el futuro
→ Para auditorías de derechos de autor
Glacier Deep Archive: $0.00099/GB — el más económico de AWS
Recuperación: 12-48 horas (aceptable para estos casos de uso)"""
},
"archivado/documental-marea-roja-2021.mp4.demo": {
"clase_inicial": "STANDARD",
"escenario": 5,
"clase_destino": "DEEP_ARCHIVE",
"tamano_sim": "8.5 GB",
"descripcion": "Documental - licencia expirada",
"contenido": """[DEMO VIDEO] CineTime Archivado - Marea Roja (Documental 2021)
Resolución: 4K | Duración: 1h 28min | Tamaño: ~8.5 GB
Razón del archivo: licencia de distribución expirada en 2024
Conservar por: posible renovación de licencia, derechos residuales
Glacier Deep Archive ahorra 95.7% vs S3 Standard en este archivo."""
},
# ── Metadatos — SIEMPRE en Standard, nunca se mueven
"metadatos/catalogo-index-v2.json.demo": {
"clase_inicial": "STANDARD",
"escenario": None,
"clase_destino": None,
"tamano_sim": "12 MB",
"descripcion": "Índice del catálogo - hot path del sistema",
"contenido": """[DEMO METADATA] CineTime - Índice de Catálogo
{"total_titulos": 4827, "series": 312, "peliculas": 891, "documentales": 156}
CLASE: S3 STANDARD — NUNCA MOVER A OTRA CLASE
Por qué: este archivo se lee en CADA búsqueda del usuario.
El reproductor lo consulta antes de mostrar cualquier contenido.
Moverlo a IA o Glacier rompería la experiencia de usuario."""
},
"metadatos/subtitulos-es-mx-pack.srt.demo": {
"clase_inicial": "STANDARD",
"escenario": None,
"clase_destino": None,
"tamano_sim": "450 MB",
"descripcion": "Pack de subtítulos ES-MX - se sirve en cada reproducción",
"contenido": """[DEMO METADATA] CineTime - Subtítulos Español México
Cobertura: 89% del catálogo | Idiomas: ES-MX, ES-ES, EN-US
Se leen en CADA reproducción de cualquier título con subtítulos.
S3 Standard es obligatorio — latencia de milisegundos requerida."""
},
}
# ─────────────────────────────────────────────────────────────
# CLIENTE S3
# ─────────────────────────────────────────────────────────────
def get_s3_client():
return boto3.client(
"s3",
region_name=AWS_REGION,
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
aws_session_token=AWS_SESSION_TOKEN,
)
# ─────────────────────────────────────────────────────────────
# HELPERS DE IMPRESIÓN
# ─────────────────────────────────────────────────────────────
def sep(char="─", width=70, color=C.DIM):
print(f"{color}{char * width}{C.RESET}")
def print_header(titulo, subtitulo=""):
print()
sep("═", 70, C.BBLUE)
print(f"{C.BOLD}{C.BBLUE} {titulo}{C.RESET}")
if subtitulo:
print(f"{C.DIM} {subtitulo}{C.RESET}")
sep("═", 70, C.BBLUE)
print()
def print_step(num, texto):
print(f"\n{C.BOLD}{C.BYELLOW} [{num}] {texto}{C.RESET}")
def print_api(llamada, params=""):
print(f"\n {C.DIM}📡 LLAMADA A LA API DE AWS:{C.RESET}")
print(f" {C.BGREEN}{llamada}{C.RESET}")
if params:
for linea in params.strip().split("\n"):
print(f" {C.DIM} {linea}{C.RESET}")
def print_ok(mensaje):
print(f" {C.BGREEN}✓ {mensaje}{C.RESET}")
def print_info(mensaje):
print(f" {C.CYAN}ℹ {mensaje}{C.RESET}")
def print_warn(mensaje):
print(f" {C.BYELLOW}⚠ {mensaje}{C.RESET}")
def print_err(mensaje):
print(f" {C.BRED}✗ {mensaje}{C.RESET}")
def print_concepto(titulo, texto):
sep("─", 70, C.DIM)
print(f" {C.BOLD}{C.BMAGENTA}💡 CONCEPTO: {titulo}{C.RESET}")
for linea in texto.strip().split("\n"):
print(f" {C.DIM}{linea}{C.RESET}")
sep("─", 70, C.DIM)
def badge_clase(clase_id):
info = STORAGE_CLASSES.get(clase_id, {"nombre": clase_id, "color": C.WHITE, "emoji": "?"})
return f"{info['color']}{C.BOLD}{info['emoji']} {info['nombre']}{C.RESET}"
def print_comparacion_costo(clase_origen, clase_destino, tamano_gb=100):
o = STORAGE_CLASSES.get(clase_origen, {})
d = STORAGE_CLASSES.get(clase_destino, {})
if not o or not d:
return
costo_o = o["costo_gb"] * tamano_gb
costo_d = d["costo_gb"] * tamano_gb
ahorro = costo_o - costo_d
pct = ((costo_o - costo_d) / costo_o * 100) if costo_o > 0 else 0
sep("─", 70, C.DIM)
print(f" {C.BOLD}💰 COMPARACIÓN DE COSTO (simulando {tamano_gb} GB/mes):{C.RESET}")
print(f" {badge_clase(clase_origen):<50} ${costo_o:>7.2f}/mes")
print(f" {badge_clase(clase_destino):<50} ${costo_d:>7.2f}/mes")
print(f" {C.BGREEN}{C.BOLD} Ahorro mensual: ${ahorro:.2f} ({pct:.0f}% menos){C.RESET}")
sep("─", 70, C.DIM)
# ─────────────────────────────────────────────────────────────
# OBTENER CLASE ACTUAL DE UN OBJETO
# ─────────────────────────────────────────────────────────────
def get_storage_class(s3, key):
try:
resp = s3.head_object(Bucket=BUCKET, Key=key)
clase = resp.get("StorageClass", "STANDARD")
return clase
except ClientError as e:
if e.response["Error"]["Code"] == "404":
return None
raise
# ─────────────────────────────────────────────────────────────
# MOVER OBJETO A NUEVA CLASE (copy_object en el mismo key)
# ─────────────────────────────────────────────────────────────
def mover_a_clase(s3, key, clase_destino, verbose=True):
clase_actual = get_storage_class(s3, key)
if clase_actual is None:
print_err(f"Objeto no encontrado: {key}")
return False
nombre_archivo = key.split("/")[-1]
info_dest = STORAGE_CLASSES.get(clase_destino, {})
if verbose:
print()
sep("─", 70, C.DIM)
print(f" {C.BOLD}{C.BWHITE}📄 Archivo: {C.CYAN}{nombre_archivo}{C.RESET}")
print(f" {C.DIM} Ruta completa: s3://{BUCKET}/{key}{C.RESET}")
print()
print(f" {'Clase actual:':20} {badge_clase(clase_actual)}")
print(f" {'Clase destino:':20} {badge_clase(clase_destino)}")
print(f" {'Tiempo de acceso:':20} {C.DIM}{info_dest.get('acceso','?')}{C.RESET}")
print(f" {'Costo por GB:':20} {C.BGREEN}${info_dest.get('costo_gb', 0):.5f}/GB/mes{C.RESET}")
print(f" {'Almacenamiento mínimo:':20} {C.DIM}{info_dest.get('minimo','?')}{C.RESET}")
if clase_actual == clase_destino:
if verbose:
print_warn(f"Ya está en {clase_destino}, no se necesita mover.")
return True
if verbose:
print_api(
"s3.copy_object(",
f'Bucket="{BUCKET}",\n'
f'CopySource={{"Bucket": "{BUCKET}", "Key": "{key}"}},\n'
f'Key="{key}",\n'
f'StorageClass="{clase_destino}",\n'
f'MetadataDirective="COPY"'
f"\n)"
)
print_info("En S3 no existe un comando 'mover' o 'cambiar clase'.")
print_info("Se COPIA el objeto sobre sí mismo con la nueva clase.")
print_info("El objeto original se reemplaza. No hay costo adicional de almacenamiento.")
print()
try:
s3.copy_object(
Bucket=BUCKET,
CopySource={"Bucket": BUCKET, "Key": key},
Key=key,
StorageClass=clase_destino,
MetadataDirective="COPY",
)
if verbose:
print_ok(f"Objeto movido exitosamente → {badge_clase(clase_destino)}")
# Verificar
clase_nueva = get_storage_class(s3, key)
if verbose:
print_api("s3.head_object( ← verificando el cambio", f'Bucket="{BUCKET}", Key="{key}")')
print_ok(f"Verificado: StorageClass = {C.BOLD}{clase_nueva}{C.RESET}")
return True
except ClientError as e:
if verbose:
print_err(f"Error AWS: {e.response['Error']['Code']} - {e.response['Error']['Message']}")
return False
# ─────────────────────────────────────────────────────────────
# COMANDO: setup
# ─────────────────────────────────────────────────────────────
def cmd_setup():
print_header(
"SETUP — Creando bucket y archivos demo de CineTime",
"Empresa de Medios / Streaming"
)
s3 = get_s3_client()
# Crear bucket
print_step("1", f"Crear bucket: {C.BCYAN}{BUCKET}{C.RESET}")
print_api("s3.create_bucket(", f'Bucket="{BUCKET}", Region="{AWS_REGION}")')
try:
s3.create_bucket(Bucket=BUCKET)
print_ok(f"Bucket creado: {BUCKET}")
except ClientError as e:
code = e.response["Error"]["Code"]
if code in ("BucketAlreadyOwnedByYou", "BucketAlreadyExists"):
print_warn("El bucket ya existe — continuando con la carga de archivos.")
else:
print_err(f"Error creando bucket: {code}")
return
# Bloquear acceso público
print_step("2", "Bloquear acceso público al bucket")
print_api("s3.put_public_access_block(...)", "BlockPublicAcls=True, RestrictPublicBuckets=True")
s3.put_public_access_block(
Bucket=BUCKET,
PublicAccessBlockConfiguration={
"BlockPublicAcls": True,
"IgnorePublicAcls": True,
"BlockPublicPolicy": True,
"RestrictPublicBuckets": True,
}
)
print_ok("Acceso público bloqueado")
# Subir archivos demo
print_step("3", f"Subiendo {len(DEMO_FILES)} archivos demo a S3 Standard")
print_concepto(
"Por qué todos comienzan en S3 Standard",
"Cuando un archivo se CREA en S3, normalmente va a Standard.\n"
"Las transiciones a otras clases se hacen después (manualmente\n"
"o con Lifecycle Policies) según cómo evoluciona el patrón de acceso."
)
subidos = 0
for key, info in DEMO_FILES.items():
prefijo = key.split("/")[0].upper()
color_prefijo = {"ESTRENOS": C.BBLUE, "RAW": C.YELLOW, "THUMBNAILS": C.BMAGENTA,
"CATALOGO": C.BCYAN, "ARCHIVADO": C.BRED, "METADATOS": C.BGREEN}.get(prefijo, C.WHITE)
print(f"\n {color_prefijo}[{prefijo}]{C.RESET} {C.DIM}{key}{C.RESET}")
print(f" {C.DIM} → Tamaño simulado: {info['tamano_sim']} | {info['descripcion']}{C.RESET}")
print_api("s3.put_object(", f'Bucket="{BUCKET}", Key="{key}", StorageClass="STANDARD")')
try:
s3.put_object(
Bucket=BUCKET,
Key=key,
Body=info["contenido"].encode("utf-8"),
ContentType="text/plain",
StorageClass="STANDARD",
Metadata={
"tamano-simulado": info["tamano_sim"],
"escenario": str(info["escenario"] or "N/A"),
"clase-destino": info["clase_destino"] or "permanente",
}
)
print_ok(f"{badge_clase('STANDARD')} — subido correctamente")
subidos += 1
except ClientError as e:
print_err(f"Error: {e.response['Error']['Message']}")
print()
sep("═", 70, C.BGREEN)
print(f" {C.BOLD}{C.BGREEN}✓ Setup completo: {subidos}/{len(DEMO_FILES)} archivos en s3://{BUCKET}/{C.RESET}")
print(f" {C.DIM} Todos en S3 Standard. Ejecuta los escenarios para 'moverlos'.{C.RESET}")
sep("═", 70, C.BGREEN)
print()
# ─────────────────────────────────────────────────────────────
# COMANDO: listar
# ─────────────────────────────────────────────────────────────
def cmd_listar():
print_header(
"ESTADO ACTUAL — Objetos en S3",
f"Bucket: {BUCKET}"
)
s3 = get_s3_client()
print_api("s3.list_objects_v2(", f'Bucket="{BUCKET}")')
print_info("head_object() se llama por cada objeto para obtener su StorageClass.")
print()
try:
resp = s3.list_objects_v2(Bucket=BUCKET)
except ClientError as e:
print_err(f"No se puede listar el bucket: {e.response['Error']['Message']}")
print_warn("Ejecuta primero: python3 cinetime_s3_demo.py setup")
return
objetos = resp.get("Contents", [])
if not objetos:
print_warn("El bucket está vacío. Ejecuta: python3 cinetime_s3_demo.py setup")
return
prefijo_actual = ""
total_costo = 0
for obj in sorted(objetos, key=lambda x: x["Key"]):
key = obj["Key"]
prefijo = key.split("/")[0]
if prefijo != prefijo_actual:
prefijo_actual = prefijo
print()
print(f" {C.BOLD}{C.BWHITE}📁 {prefijo.upper()}/{C.RESET}")
sep("─", 70, C.DIM)
clase = get_storage_class(s3, key)
info_demo = DEMO_FILES.get(key, {})
nombre = key.split("/")[-1]
tamano = info_demo.get("tamano_sim", "?")
info_clase = STORAGE_CLASSES.get(clase, {"costo_gb": 0, "acceso": "?", "emoji": "?"})
esc = info_demo.get("escenario")
esc_label = f"{C.DIM}[Esc.{esc}]{C.RESET}" if esc else f"{C.BGREEN}[FIJO]{C.RESET}"
print(f" {esc_label} {C.CYAN}{nombre:<45}{C.RESET} {badge_clase(clase)}")
print(f" {C.DIM}Tamaño: {tamano:>10} | Acceso: {info_clase['acceso']:<20} | "
f"${info_clase['costo_gb']:.5f}/GB{C.RESET}")
print()
sep("═", 70, C.BBLUE)
print(f" {C.BOLD}{len(objetos)} objetos en total.{C.RESET}")
print(f" {C.DIM}[Esc.N] = se moverá en el Escenario N | [FIJO] = siempre en Standard{C.RESET}")
sep("═", 70, C.BBLUE)
print()
# ─────────────────────────────────────────────────────────────
# ESCENARIO 1 — Estrenos → Standard-IA
# ─────────────────────────────────────────────────────────────
def cmd_escenario_1():
print_header(
"ESCENARIO 1 — Videos en Estreno → S3 Standard-IA",
"Después de 90 días, el volumen de acceso cae drásticamente"
)
s3 = get_s3_client()
print_concepto(
"¿Por qué mover los estrenos a Standard-IA después de 90 días?",
"Día 0-90: La serie es NUEVA. Miles de usuarios la ven simultáneamente.\n"
" Necesitamos S3 Standard: baja latencia, alto throughput.\n"
"\n"
"Día 90+: El 'pico de estreno' terminó. El acceso es esporádico.\n"
" Los usuarios que NO vieron la serie en el estreno la buscan\n"
" ocasionalmente. Standard-IA cuesta 46% menos que Standard.\n"
"\n"
"Clave de Standard-IA: pagas por GB almacenado (más barato) + por GET\n"
" Si el acceso es frecuente, Standard es más económico.\n"
" Si es infrecuente, Standard-IA gana."
)
archivos = {k: v for k, v in DEMO_FILES.items() if v["escenario"] == 1}
print_step("1", f"Moviendo {len(archivos)} archivos de estreno a Standard-IA...")
print_info(f"Simulando: 'Han pasado 90 días desde el estreno de La Sombra del Dragón'")
for key, info in archivos.items():
mover_a_clase(s3, key, "STANDARD_IA")
time.sleep(0.3)
print_comparacion_costo("STANDARD", "STANDARD_IA", tamano_gb=50)
print_concepto(
"¿Qué pasa si el usuario accede a un objeto en Standard-IA?",
"El acceso sigue siendo en MILISEGUNDOS — igual que Standard.\n"
"La diferencia: se cobra un fee adicional por GET (~$0.01 por 1000 requests).\n"
"Por eso Standard-IA sólo conviene cuando los accesos son POCO FRECUENTES.\n"
"Si hay muchos accesos, el fee de GET puede superar el ahorro en storage."
)
print()
# ─────────────────────────────────────────────────────────────
# ESCENARIO 2 — RAW → Glacier Flexible Retrieval
# ─────────────────────────────────────────────────────────────
def cmd_escenario_2():
print_header(
"ESCENARIO 2 — Archivos RAW → Glacier Flexible Retrieval",
"~500 TB de footage sin editar que raramente se necesita"
)
s3 = get_s3_client()
print_concepto(
"¿Por qué Glacier Flexible para los RAW y no Glacier Deep Archive?",
"Los archivos RAW son los más grandes del negocio (~500 TB).\n"
"Acceso: MUY RARO pero no imposible.\n"
"\n"
"Casos en que se necesitan:\n"
" • El director quiere hacer un 'Director's Cut' (puede ocurrir)\n"
" • Se vende el contenido a otro mercado y necesitan re-editar\n"
" • Auditoría de derechos de autor\n"
"\n"
"Glacier Flexible: recuperación en 1-5 min (Expedited) o 3-5h (Standard).\n"
"Glacier Deep Archive: 12-48 horas.\n"
"\n"
"Para producción de contenido, 3-5h es aceptable (se planifica con anticipación).\n"
"12-48h podría retrasar una producción urgente."
)
archivos = {k: v for k, v in DEMO_FILES.items() if v["escenario"] == 2}
print_step("1", f"Moviendo {len(archivos)} archivos RAW a Glacier Flexible...")
print_warn("⚠ IMPORTANTE: Una vez en Glacier, NO puedes leer el archivo directamente.")
print_warn(" Debes hacer una 'restauración temporal' antes de descargarlo.")
print_warn(" La restauración crea una copia temporal en Standard por N días.")
print()
for key, info in archivos.items():
mover_a_clase(s3, key, "GLACIER")
time.sleep(0.3)
print_comparacion_costo("STANDARD", "GLACIER", tamano_gb=500)
print_concepto(
"¿Cómo se recupera un archivo de Glacier Flexible?",
"1. s3.restore_object(Bucket=..., Key=..., RestoreRequest={\n"
" 'Days': 7, # días que estará disponible la copia\n"
" 'GlacierJobParameters': {'Tier': 'Standard'} # 3-5h\n"
" })\n"
"2. Esperar 3-5 horas (Standard) o 1-5 min (Expedited, más caro)\n"
"3. Descargar el archivo normalmente con s3.get_object()\n"
"4. Después de 'Days' días, la copia temporal se elimina automáticamente."
)
print()
# ─────────────────────────────────────────────────────────────
# ESCENARIO 3 — Thumbnails → Intelligent-Tiering
# ─────────────────────────────────────────────────────────────
def cmd_escenario_3():
print_header(
"ESCENARIO 3 — Thumbnails y Material Promocional → Intelligent-Tiering",
"Acceso impredecible: alto cuando hay tendencia, bajo el resto del tiempo"
)
s3 = get_s3_client()
print_concepto(
"¿Por qué Intelligent-Tiering para thumbnails?",
"Los thumbnails tienen un patrón de acceso que NO se puede predecir:\n"
"\n"
" Semana normal: 500K requests/día (moderado)\n"
" Si sale en Trending: 15M requests/día (explosivo)\n"
" Después de tendencia: 200K requests/día (cae)\n"
"\n"
"Con una regla fija de Lifecycle: pagarías por accesos infrecuentes\n"
" cuando en realidad el acceso es impredecible.\n"
"\n"
"Intelligent-Tiering MONITOREA el patrón automáticamente:\n"
" • 0-30 días sin acceso → mueve a Infrequent Access ($0.0125/GB)\n"
" • Nuevo acceso → vuelve a Frequent Access ($0.023/GB)\n"
" Sin latencia adicional, sin necesidad de restaurar."
)
archivos = {k: v for k, v in DEMO_FILES.items() if v["escenario"] == 3}
print_step("1", f"Moviendo {len(archivos)} assets de thumbnails a Intelligent-Tiering...")
for key, info in archivos.items():
mover_a_clase(s3, key, "INTELLIGENT_TIERING")
time.sleep(0.3)
print_comparacion_costo("STANDARD", "INTELLIGENT_TIERING", tamano_gb=2)
print_concepto(
"Diferencia entre IT y Standard-IA",
"Standard-IA: Tú decides cuándo mover. Pagas GET fee siempre.\n"
"Intelligent-Tiering: AWS decide cuándo mover. Sin GET fee adicional.\n"
" Cobra $0.0025/1000 objetos por monitoreo.\n"
"\n"
"Para objetos pequeños (<128 KB) IT no conviene (fee de monitoreo\n"
"puede superar el ahorro). Para archivos grandes como thumbnails HD: sí."
)
print()
# ─────────────────────────────────────────────────────────────
# ESCENARIO 4 — Catálogo General → Intelligent-Tiering
# ─────────────────────────────────────────────────────────────
def cmd_escenario_4():
print_header(
"ESCENARIO 4 — Catálogo General (>1 año) → Intelligent-Tiering",
"Series y películas establecidas con patrón de acceso variable"
)
s3 = get_s3_client()
print_concepto(
"¿Por qué no Standard-IA para el catálogo antiguo?",
"Standard-IA cobra por cada GET (lectura del archivo).\n"
"Un usuario que ve una película de 2h genera MUCHOS requests GET\n"
" (el reproductor pide chunks del archivo en streaming).\n"
"\n"
"Si hay 1,000 usuarios simultáneos viendo la misma película:\n"
" → Millones de GET requests → el fee de IA puede ser enorme.\n"
"\n"
"Intelligent-Tiering NO cobra fee por GET.\n"
"Si la película se vuelve viral (trending en redes), IT la mueve\n"
" automáticamente a Frequent Access tier.\n"
"Cuando el acceso baja, vuelve a Infrequent Access.\n"
"Todo automático, sin intervención manual."
)
archivos = {k: v for k, v in DEMO_FILES.items() if v["escenario"] == 4}
print_step("1", f"Moviendo {len(archivos)} títulos del catálogo a Intelligent-Tiering...")
for key, info in archivos.items():
mover_a_clase(s3, key, "INTELLIGENT_TIERING")
time.sleep(0.3)
print_comparacion_costo("STANDARD", "INTELLIGENT_TIERING", tamano_gb=200)
print_concepto(
"Los 3 tiers de Intelligent-Tiering",
"Tier 1: Frequent Access → $0.023/GB (como Standard)\n"
"Tier 2: Infrequent Access → $0.0125/GB (tras 30 días sin acceso)\n"
"Tier 3: Archive Instant Access → $0.004/GB (tras 90 días sin acceso)\n"
"\n"
"Si el usuario accede al Tier 3, se recupera en MILISEGUNDOS\n"
"y el objeto sube automáticamente al Tier 1. Transparente para la app."
)
print()
# ─────────────────────────────────────────────────────────────
# ESCENARIO 5 — Archivado → Glacier Deep Archive
# ─────────────────────────────────────────────────────────────
def cmd_escenario_5():
print_header(
"ESCENARIO 5 — Contenido Fuera del Catálogo → Glacier Deep Archive",
"Series canceladas y licencias expiradas — archivo permanente"
)
s3 = get_s3_client()
print_concepto(
"¿Cuándo usar Glacier Deep Archive?",
"Casos de uso:\n"
" ✓ Contenido que NO estará disponible para usuarios en el futuro cercano\n"
" ✓ Conservación obligatoria por contrato o ley (derechos de autor)\n"
" ✓ Backups de DR que solo se usan en catástrofes\n"
" ✓ Datos históricos para análisis eventual\n"
"\n"
"Casos donde NO usar Deep Archive:\n"
" ✗ Si hay chance de que el usuario lo pida en las próximas horas\n"
" ✗ Urgencias médicas, legales o de producción\n"
" ✗ Datos que se analizan regularmente\n"
"\n"
"Costo: $0.00099/GB/mes = $0.99 por TB/mes\n"
" S3 Standard cuesta $23 por TB/mes\n"
" Ahorro: 95.7% — dramático para petabytes."
)
archivos = {k: v for k, v in DEMO_FILES.items() if v["escenario"] == 5}
print_step("1", f"Moviendo {len(archivos)} títulos a Glacier Deep Archive...")
print_warn("⚠ Glacier Deep Archive: recuperación toma 12-48 horas.")
print_warn(" Solo para casos donde el tiempo de espera es aceptable.")
print()
for key, info in archivos.items():
mover_a_clase(s3, key, "DEEP_ARCHIVE")
time.sleep(0.3)
print_comparacion_costo("STANDARD", "DEEP_ARCHIVE", tamano_gb=1000)
print_concepto(
"¿Cómo recuperar de Deep Archive? (cuando se necesite relicenciar)",
"s3.restore_object(\n"
" Bucket=BUCKET,\n"
" Key='archivado/serie-cancelada-pixel-wars-2022.mp4.demo',\n"
" RestoreRequest={\n"
" 'Days': 30, # disponible 30 días\n"
" 'GlacierJobParameters': {'Tier': 'Standard'} # 12h\n"
" }\n"
")\n"
"\n"
"Opción 'Bulk' (más barata): hasta 48h, $0.0025/GB\n"
"Opción 'Standard': 12h, $0.02/GB"
)
print()
# ─────────────────────────────────────────────────────────────
# COMANDO: reset
# ─────────────────────────────────────────────────────────────
def cmd_reset():
print_header(
"RESET — Regresando todos los objetos a S3 Standard",
"Para poder repetir la demo desde el inicio"
)
s3 = get_s3_client()
print_warn("Todos los objetos (excepto metadatos) volverán a S3 Standard.")
print_info("Útil para repetir la demo en clase.\n")
movidos = 0
for key in DEMO_FILES.keys():
clase_actual = get_storage_class(s3, key)
if clase_actual and clase_actual != "STANDARD":
nombre = key.split("/")[-1]
print(f" {C.DIM}{nombre:<50}{C.RESET} {badge_clase(clase_actual)} → ", end="")
try:
s3.copy_object(
Bucket=BUCKET,
CopySource={"Bucket": BUCKET, "Key": key},
Key=key,
StorageClass="STANDARD",
MetadataDirective="COPY",
)
print(f"{badge_clase('STANDARD')} {C.BGREEN}✓{C.RESET}")
movidos += 1
except ClientError as e:
print(f"{C.BRED}✗ {e.response['Error']['Code']}{C.RESET}")
else:
pass
print()
print_ok(f"Reset completo: {movidos} objetos regresaron a S3 Standard.")
print()
# ─────────────────────────────────────────────────────────────
# COMANDO: demo completo
# ─────────────────────────────────────────────────────────────
def cmd_demo_completo():
print_header(
"DEMO COMPLETO — Los 5 escenarios de CineTime",
"Ejecutando toda la estrategia de almacenamiento secuencialmente"
)
print_info("Este demo simula lo que ocurriría automáticamente con Lifecycle Policies.")
print_info("Aquí lo hacemos MANUAL para entender cada operación.")
print()
escenarios = [
(1, "Estrenos (>90 días) → Standard-IA", cmd_escenario_1),
(2, "Archivos RAW → Glacier Flexible", cmd_escenario_2),
(3, "Thumbnails → Intelligent-Tiering", cmd_escenario_3),
(4, "Catálogo General → Intelligent-Tiering", cmd_escenario_4),
(5, "Contenido Archivado → Deep Archive", cmd_escenario_5),
]
for num, titulo, fn in escenarios:
print(f"\n {C.BOLD}{C.BYELLOW}━━━ Escenario {num}: {titulo} ━━━{C.RESET}")
fn()
time.sleep(1)
print()
sep("═", 70, C.BGREEN)
print(f" {C.BOLD}{C.BGREEN}✓ DEMO COMPLETO{C.RESET}")
print(f" {C.DIM} Ejecuta 'listar' para ver el estado final de todos los objetos.{C.RESET}")
print(f" {C.DIM} Ejecuta 'reset' para volver al estado inicial y repetir la demo.{C.RESET}")
sep("═", 70, C.BGREEN)
print()
# ─────────────────────────────────────────────────────────────
# MAIN
# ─────────────────────────────────────────────────────────────
def print_uso():
print(f"""
{C.BOLD}{C.BBLUE}╔══════════════════════════════════════════════════════════════════════╗
║ CineTime — Demo de Clases de Almacenamiento S3 ║
║ AWS re/Start | Práctica Manual de Storage Classes ║
╚══════════════════════════════════════════════════════════════════════╝{C.RESET}
{C.BOLD}Uso:{C.RESET}
{C.CYAN}python3 cinetime_s3_demo.py{C.RESET} {C.BYELLOW}<comando>{C.RESET} {C.DIM}[argumento]{C.RESET}
{C.BOLD}Comandos:{C.RESET}
{C.BYELLOW}setup{C.RESET} Crea el bucket y sube todos los archivos demo en S3 Standard
{C.BYELLOW}listar{C.RESET} Muestra el estado actual (clase S3) de cada objeto
{C.BYELLOW}escenario 1{C.RESET} Estrenos (>90 días) → {badge_clase('STANDARD_IA')}
{C.BYELLOW}escenario 2{C.RESET} Archivos RAW → {badge_clase('GLACIER')}
{C.BYELLOW}escenario 3{C.RESET} Thumbnails → {badge_clase('INTELLIGENT_TIERING')}
{C.BYELLOW}escenario 4{C.RESET} Catálogo general → {badge_clase('INTELLIGENT_TIERING')}
{C.BYELLOW}escenario 5{C.RESET} Contenido archivado → {badge_clase('DEEP_ARCHIVE')}
{C.BYELLOW}reset{C.RESET} Regresa todos los objetos a S3 Standard (para repetir)
{C.BYELLOW}demo{C.RESET} Ejecuta los 5 escenarios secuencialmente
{C.BOLD}Orden recomendado en clase:{C.RESET}
1. {C.DIM}python3 cinetime_s3_demo.py setup{C.RESET}
2. {C.DIM}python3 cinetime_s3_demo.py listar{C.RESET}
3. {C.DIM}python3 cinetime_s3_demo.py escenario 1{C.RESET} ← discutir con los alumnos
4. {C.DIM}python3 cinetime_s3_demo.py escenario 2{C.RESET} ← etc.
5. {C.DIM}python3 cinetime_s3_demo.py listar{C.RESET} ← ver el resultado final
""")
def main():
args = sys.argv[1:]
if not args:
print_uso()
return
cmd = args[0].lower()
if cmd == "setup":
cmd_setup()
elif cmd == "listar":
cmd_listar()
elif cmd == "escenario" and len(args) >= 2:
n = args[1]
if n == "1": cmd_escenario_1()
elif n == "2": cmd_escenario_2()
elif n == "3": cmd_escenario_3()
elif n == "4": cmd_escenario_4()
elif n == "5": cmd_escenario_5()
else:
print_err(f"Escenario '{n}' no existe. Usa del 1 al 5.")
elif cmd == "reset":
cmd_reset()
elif cmd == "demo":
cmd_demo_completo()
else:
print_err(f"Comando '{cmd}' no reconocido.")
print_uso()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment