""" DataFlow SaaS — Campaign Processor

Módulo de procesamiento de campañas de marketing para DataFlow SaaS. Aplica todos los patrones de gestión de recursos Python de la skill gestion-recursos-python:

• Pattern 1 — Class-Based Context Manager (sync DB connection) • Pattern 2 — Async Context Manager (async DB pool) • Pattern 3 — @contextmanager / @asynccontextmanager • Pattern 4 — Unconditional Resource Release (FileProcessor) • Pattern 6 — Streaming con estado acumulado (StreamingResult) • Pattern 7 — Acumulación eficiente O(n) • Pattern 8 — Métricas de streaming (TTFB, total_time_ms) • Pattern 9 — ExitStack para múltiples ficheros

Genera una salida de consola que simula el flujo completo sin deps externas. """

from future import annotations

import csv import io import time import random from contextlib import contextmanager, asynccontextmanager, ExitStack from dataclasses import dataclass, field from pathlib import Path from types import TracebackType from typing import Generator, AsyncGenerator, Iterator

─────────────────────────────────────────────────────────────────────────────

Stubs de infraestructura (simulan psycopg / asyncpg sin deps reales)

─────────────────────────────────────────────────────────────────────────────

class _FakeConnection: """Simula una conexión psycopg síncrona.""" def init(self, dsn: str) -> None: self.dsn = dsn self.closed = False print(f" [DB] Conexión abierta → {dsn}")

def execute(self, query: str, params: tuple = ()) -> list[dict]:
    print(f"  [DB] Ejecutando: {query[:60]}...")
    return [{"status": "ok", "rows_affected": random.randint(1, 20)}]

def close(self) -> None:
    if not self.closed:
        self.closed = True
        print(f"  [DB] Conexión cerrada ← {self.dsn}")

class _FakePool: """Simula un pool asyncpg.""" def init(self, dsn: str, min_size: int, max_size: int) -> None: self.dsn = dsn self.min_size = min_size self.max_size = max_size self.closed = False print(f" [POOL] Pool creado ({min_size}..{max_size}) → {dsn}")

def acquire(self) -> "_FakePoolConn":
    return _FakePoolConn(self)

async def close(self) -> None:
    self.closed = True
    print(f"  [POOL] Pool cerrado ← {self.dsn}")

class _FakePoolConn: """Simula una conexión extraída del pool.""" def init(self, pool: _FakePool) -> None: self._pool = pool

async def __aenter__(self) -> "_FakePoolConn":
    print("  [POOL] Conexión adquirida del pool")
    return self

async def __aexit__(self, *_: object) -> None:
    print("  [POOL] Conexión devuelta al pool")

async def execute(self, query: str, *args: object) -> list[dict]:
    print(f"  [POOL] Async query: {query[:60]}...")
    return [{"campaign_id": random.randint(1000, 9999), "status": "inserted"}]

async def fetch(self, query: str, *args: object) -> list[dict]:
    return await self.execute(query, *args)

─────────────────────────────────────────────────────────────────────────────

Pattern 1 — Class-Based Context Manager (sync)

─────────────────────────────────────────────────────────────────────────────

class DatabaseConnection: """ Conexión PostgreSQL síncrona con cleanup garantizado. Patrón 1: enter / exit de clase. """

def __init__(self, dsn: str) -> None:
    self._dsn = dsn
    self._conn: _FakeConnection | None = None

def connect(self) -> None:
    self._conn = _FakeConnection(self._dsn)

def close(self) -> None:
    if self._conn is not None:
        self._conn.close()
        self._conn = None

def execute(self, query: str, params: tuple = ()) -> list[dict]:
    if self._conn is None:
        raise RuntimeError("No hay conexión activa")
    return self._conn.execute(query, params)

def __enter__(self) -> "DatabaseConnection":
    self.connect()
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    self.close()
    # No suprimimos excepciones → retornamos None (falsy)

─────────────────────────────────────────────────────────────────────────────

Pattern 2 — Async Context Manager

─────────────────────────────────────────────────────────────────────────────

class AsyncDatabasePool: """ Pool de conexiones async PostgreSQL. Patrón 2: aenter / aexit. """

def __init__(self, dsn: str, min_size: int = 2, max_size: int = 10) -> None:
    self._dsn = dsn
    self._min_size = min_size
    self._max_size = max_size
    self._pool: _FakePool | None = None

async def __aenter__(self) -> "AsyncDatabasePool":
    self._pool = _FakePool(self._dsn, self._min_size, self._max_size)
    return self

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    if self._pool is not None:
        await self._pool.close()

async def insert_campaign_row(self, row: dict) -> dict:
    if self._pool is None:
        raise RuntimeError("Pool no inicializado")
    async with self._pool.acquire() as conn:
        return await conn.fetch(
            "INSERT INTO campaign_results (campaign_id, clicks, conversions) "
            "VALUES ($1, $2, $3) RETURNING id",
            row["campaign_id"], row["clicks"], row["conversions"],
        )

─────────────────────────────────────────────────────────────────────────────

Pattern 3 — @contextmanager / @asynccontextmanager

─────────────────────────────────────────────────────────────────────────────

@contextmanager def timed_block(name: str) -> Generator[None, None, None]: """Cronometra un bloque de código y registra la duración.""" start = time.perf_counter() try: yield finally: elapsed = time.perf_counter() - start print(f" [TIMER] '{name}' completado en {elapsed * 1000:.1f} ms")

@asynccontextmanager async def database_transaction(conn: _FakePoolConn) -> AsyncGenerator[_FakePoolConn, None]: """ Gestiona una transacción async con COMMIT / ROLLBACK automáticos. Patrón 3 async: @asynccontextmanager. """ await conn.execute("BEGIN") try: yield conn await conn.execute("COMMIT") except Exception: await conn.execute("ROLLBACK") raise

─────────────────────────────────────────────────────────────────────────────

Pattern 4 — Unconditional Resource Release

─────────────────────────────────────────────────────────────────────────────

class CampaignCSVProcessor: """ Lee y valida un CSV de campaña con cleanup garantizado de file handles y ficheros temporales. Patrón 4: exit siempre limpia, aunque haya excepción. """

REQUIRED_COLUMNS = {"campaign_id", "clicks", "impressions", "conversions", "spend_eur"}

def __init__(self, path: str) -> None:
    self._path = path
    self._file: io.TextIOWrapper | None = None
    self._temp_files: list[Path] = []

def __enter__(self) -> "CampaignCSVProcessor":
    self._file = open(self._path, newline="", encoding="utf-8")
    print(f"  [CSV] Abierto: {self._path}")
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    # Cierra el fichero principal
    if self._file is not None:
        self._file.close()
        print(f"  [CSV] Cerrado: {self._path}")

    # Limpia ficheros temporales (best-effort)
    for tmp in self._temp_files:
        try:
            tmp.unlink()
            print(f"  [CSV] Temporal eliminado: {tmp}")
        except OSError:
            pass

    # Devuelve False (None) → propaga excepciones

def read_rows(self) -> list[dict]:
    if self._file is None:
        raise RuntimeError("Fichero no abierto")
    reader = csv.DictReader(self._file)
    missing = self.REQUIRED_COLUMNS - set(reader.fieldnames or [])
    if missing:
        raise ValueError(f"Columnas faltantes en CSV: {missing}")
    rows = []
    for row in reader:
        rows.append({
            "campaign_id": row["campaign_id"],
            "clicks": int(row["clicks"]),
            "impressions": int(row["impressions"]),
            "conversions": int(row["conversions"]),
            "spend_eur": float(row["spend_eur"]),
        })
    return rows

─────────────────────────────────────────────────────────────────────────────

Pattern 6 + 7 + 8 — Streaming con estado acumulado + métricas

─────────────────────────────────────────────────────────────────────────────

@dataclass class StreamingResult: """ Acumula chunks de un informe de campaña de forma eficiente (O(n)). Patrón 7: list + join en lugar de concatenación de strings. """ chunks: list[str] = field(default_factory=list) _finalized: bool = False

@property
def content(self) -> str:
    return "".join(self.chunks)

def add_chunk(self, chunk: str) -> None:
    if self._finalized:
        raise RuntimeError("No se puede añadir a un resultado finalizado")
    self.chunks.append(chunk)

def finalize(self) -> str:
    self._finalized = True
    return self.content

def stream_campaign_report(rows: list[dict]) -> Generator[tuple[str, dict], None, None]: """ Genera el informe de campaña chunk a chunk, acumulando contenido y midiendo TTFB y tiempo total (Patrones 6, 7, 8).

Yields: (accumulated_content_so_far, metrics_snapshot)
"""
result = StreamingResult()
start = time.perf_counter()
first_chunk_time: float | None = None
chunk_count = 0
total_bytes = 0

# Cabecera
header = "# Informe de Campaña — DataFlow SaaS\n\n"
header += f"Procesadas {len(rows)} filas\n\n"
header += f"{'campaign_id':<15} {'clicks':>8} {'conv':>6} {'CTR%':>7} {'CPA €':>8}\n"
header += "-" * 50 + "\n"

for chunk in [header]:
    if first_chunk_time is None:
        first_chunk_time = time.perf_counter() - start
    result.add_chunk(chunk)
    chunk_count += 1
    total_bytes += len(chunk.encode())
    time.sleep(0.002)  # Simula latencia de red/BD
    yield result.content, {
        "ttfb_ms": round((first_chunk_time or 0) * 1000, 2),
        "elapsed_ms": round((time.perf_counter() - start) * 1000, 2),
        "chunks": chunk_count,
        "bytes": total_bytes,
    }

# Filas de datos
for row in rows:
    ctr = row["clicks"] / row["impressions"] * 100 if row["impressions"] else 0
    cpa = row["spend_eur"] / row["conversions"] if row["conversions"] else 0
    line = (
        f"{row['campaign_id']:<15} {row['clicks']:>8,} "
        f"{row['conversions']:>6,} {ctr:>7.2f} {cpa:>8.2f}\n"
    )
    result.add_chunk(line)
    chunk_count += 1
    total_bytes += len(line.encode())
    time.sleep(0.001)
    yield result.content, {
        "ttfb_ms": round((first_chunk_time or 0) * 1000, 2),
        "elapsed_ms": round((time.perf_counter() - start) * 1000, 2),
        "chunks": chunk_count,
        "bytes": total_bytes,
    }

# Totales
total_clicks = sum(r["clicks"] for r in rows)
total_conv = sum(r["conversions"] for r in rows)
total_spend = sum(r["spend_eur"] for r in rows)
total_imp = sum(r["impressions"] for r in rows)
avg_ctr = total_clicks / total_imp * 100 if total_imp else 0
avg_cpa = total_spend / total_conv if total_conv else 0

footer = (
    "-" * 50 + "\n"
    f"{'TOTALES':<15} {total_clicks:>8,} {total_conv:>6,} "
    f"{avg_ctr:>7.2f} {avg_cpa:>8.2f}\n"
)
result.add_chunk(footer)
chunk_count += 1
total_bytes += len(footer.encode())
total_time = time.perf_counter() - start

yield result.content, {
    "ttfb_ms": round((first_chunk_time or 0) * 1000, 2),
    "total_time_ms": round(total_time * 1000, 2),
    "chunk_count": chunk_count,
    "total_bytes": total_bytes,
    "status": "complete",
}

─────────────────────────────────────────────────────────────────────────────

Pattern 9 — ExitStack para múltiples ficheros CSV

─────────────────────────────────────────────────────────────────────────────

def process_multiple_csv_files(paths: list[str]) -> list[dict]: """ Abre y procesa múltiples CSVs de campañas a la vez. Patrón 9: ExitStack garantiza el cierre de todos aunque uno falle. """ all_rows: list[dict] = [] with ExitStack() as stack: processors = [ stack.enter_context(CampaignCSVProcessor(p)) for p in paths ] for proc in processors: all_rows.extend(proc.read_rows()) return all_rows

─────────────────────────────────────────────────────────────────────────────

Demo — simula el flujo completo

─────────────────────────────────────────────────────────────────────────────

def _create_sample_csv(path: str) -> None: """Crea un CSV de ejemplo en disco para la demo.""" rows = [ {"campaign_id": "CMP-2024-001", "clicks": 4_230, "impressions": 85_000, "conversions": 142, "spend_eur": 1_840.50}, {"campaign_id": "CMP-2024-002", "clicks": 7_810, "impressions": 210_000, "conversions": 389, "spend_eur": 5_220.00}, {"campaign_id": "CMP-2024-003", "clicks": 1_990, "impressions": 42_000, "conversions": 67, "spend_eur": 780.25}, {"campaign_id": "CMP-2024-004", "clicks": 12_450, "impressions": 380_000, "conversions": 712, "spend_eur": 9_100.00}, {"campaign_id": "CMP-2024-005", "clicks": 3_100, "impressions": 65_000, "conversions": 98, "spend_eur": 1_250.75}, ] with open(path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=rows[0].keys()) writer.writeheader() writer.writerows(rows)

def run_demo() -> None: import tempfile, os

DSN = "postgresql://dataflow:secret@db.dataflow.io:5432/campaigns"
sep = "─" * 60

print(f"\n{'═' * 60}")
print("  DataFlow SaaS — Campaign Processor (Demo)")
print(f"  Skill: gestion-recursos-python")
print(f"{'═' * 60}\n")

# ── 1. Crear CSV de muestra ──────────────────────────────────
with tempfile.NamedTemporaryFile(
    mode="w", suffix=".csv", delete=False, encoding="utf-8"
) as tmp:
    csv_path = tmp.name
_create_sample_csv(csv_path)
print(f"[DEMO] CSV de muestra creado en: {csv_path}\n")

# ── 2. Pattern 1: sync DB connection ────────────────────────
print(f"{sep}")
print("PATRÓN 1 — Class-Based Context Manager (sync DB)")
print(f"{sep}")
with timed_block("sync DB upsert"):
    with DatabaseConnection(DSN) as db:
        result = db.execute(
            "INSERT INTO campaign_sync_log (ts) VALUES (NOW()) RETURNING id"
        )
        print(f"  [DB] Resultado: {result}")
print()

# ── 3. Pattern 4: CSV con cleanup garantizado ───────────────
print(f"{sep}")
print("PATRÓN 4 — CampaignCSVProcessor (unconditional cleanup)")
print(f"{sep}")
with timed_block("lectura CSV"):
    with CampaignCSVProcessor(csv_path) as proc:
        rows = proc.read_rows()
        print(f"  [CSV] {len(rows)} filas leídas correctamente")
print()

# ── 4. Patterns 6+7+8: streaming del informe ────────────────
print(f"{sep}")
print("PATRONES 6/7/8 — Streaming con acumulación O(n) + métricas")
print(f"{sep}")
final_content = ""
final_metrics: dict = {}
for content, metrics in stream_campaign_report(rows):
    final_content = content
    final_metrics = metrics
    if metrics.get("status") == "complete":
        break

print(final_content)
print(f"  Métricas finales de streaming:")
for k, v in final_metrics.items():
    print(f"    {k}: {v}")
print()

# ── 5. Pattern 9: ExitStack con múltiples CSVs ──────────────
print(f"{sep}")
print("PATRÓN 9 — ExitStack para múltiples ficheros CSV")
print(f"{sep}")
# Creamos un segundo CSV
with tempfile.NamedTemporaryFile(
    mode="w", suffix=".csv", delete=False, encoding="utf-8"
) as tmp2:
    csv_path2 = tmp2.name
_create_sample_csv(csv_path2)

with timed_block("multi-CSV ExitStack"):
    all_rows = process_multiple_csv_files([csv_path, csv_path2])
    print(f"  [ExitStack] Total filas combinadas: {len(all_rows)}")
print()

# ── Limpieza ─────────────────────────────────────────────────
os.unlink(csv_path)
os.unlink(csv_path2)

print(f"{'═' * 60}")
print("  Demo completada — todos los recursos liberados correctamente")
print(f"{'═' * 60}\n")

if name == "main": run_demo()