"""Nightly learning-activity ETL: observe a large frame without publishing it whole.

A separate workload from the fixed term review. Synthetic event extracts contain
500,000 rows by default. Only a small course aggregate is written to disk.
"""
import argparse
import os
from pathlib import Path
import signal
import time

import numpy as np
import pandas as pd
import plotsrv as ps

PORT = int(os.environ.get("PLOTSRV_PORT", "8000"))
OUTPUT = Path(__file__).resolve().parent / "output"


@ps.view(observe=True, label="Activity extract", section="Activity pipeline", host="127.0.0.1", port=PORT)
def extract_activity(rows, batch):
    """Large synthetic event extract; observed distributions describe bounded samples."""
    rng = np.random.default_rng(100 + batch)
    minutes = np.round(np.clip(rng.lognormal(2.8, .65, rows), 1, 180), 1)
    scores = np.round(np.clip(rng.normal(72, 15, rows), 0, 100), 1)
    # Some sessions have no assessment. Every fifth batch models an import fault.
    scores[rng.random(rows) < (.18 if batch % 5 == 4 else .08)] = np.nan
    minutes[rng.random(rows) < .004] = -1  # invalid upstream duration
    return pd.DataFrame({
        "duration_minutes": minutes,
        "assessment_score": scores,
        "resources_opened": rng.poisson(4, rows),
        "course_id": rng.integers(1, 7, rows),
        "campus_id": rng.integers(1, 5, rows),
        "learner_id": rng.integers(10000, 35000, rows),
        "event_id": np.arange(rows) + batch * rows,
    })


@ps.view(observe=True, label="Validated activity", section="Activity pipeline", host="127.0.0.1", port=PORT)
def validate_activity(frame):
    """Reject invalid durations; preserve legitimate missing assessment scores."""
    return frame.loc[frame.duration_minutes.between(0, 180)].copy()


def run_batch(rows, batch):
    started = time.perf_counter()
    raw = extract_activity(rows, batch)
    clean = validate_activity(raw)
    # These are full-data business computations, independent of observation samples.
    aggregate = clean.groupby("course_id").agg(
        activity_events=("event_id", "size"),
        learning_minutes=("duration_minutes", "sum"),
        assessed_events=("assessment_score", "count"),
        mean_assessment_score=("assessment_score", "mean"),
    ).round(1).reset_index()
    OUTPUT.mkdir(exist_ok=True)
    aggregate.to_csv(OUTPUT / "activity-by-course.csv", index=False)
    metrics = {
        "batch_number": batch + 1,
        "source_rows": len(raw),
        "accepted_rows": len(clean),
        "rejected_rows": len(raw) - len(clean),
        "unassessed_pct": round(float(100 * clean.assessment_score.isna().mean()), 2),
        "source_memory_mb": round(float(raw.memory_usage(deep=True).sum()) / 1_000_000, 2),
        "elapsed_seconds": round(time.perf_counter() - started, 3),
    }
    ps.publish_view(metrics, observe=True, label="Batch metrics", section="Activity pipeline", host="127.0.0.1", port=PORT)
    ps.flush_views(timeout=5)
    print(f"Activity batch: {metrics}; observation delivery: {ps.get_observation_stats()}", flush=True)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--rows", type=int, default=500_000)
    parser.add_argument("--repeat", action="store_true", help="Run a new batch every 30 seconds")
    args = parser.parse_args()
    if args.rows < 1:
        parser.error("--rows must be positive")
    def stop(signum, frame):
        raise KeyboardInterrupt
    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    # Reuse only the launch readiness check; the business pipelines are separate.
    from pipeline import wait_for_server
    wait_for_server(f"http://127.0.0.1:{PORT}/status")
    try:
        batch = 0
        while True:
            run_batch(args.rows, batch)
            if not args.repeat:
                break
            batch += 1
            time.sleep(30)
    except KeyboardInterrupt:
        pass
    finally:
        ps.flush_views(timeout=2)


if __name__ == "__main__":
    main()
