scripts/run_learning_cycle.py

"""Hourly learning cycle: test solvers, record results, analyze, optimize.

Run locally:  python scripts/run_learning_cycle.py
Run via GH Action: triggered by .github/workflows/learn.yml

This simulates captcha solves across all solver types to gather
accuracy data over time. The learning DB is stored at data/learning.db
and exported as a GitHub Actions artifact.
"""
from __future__ import annotations

import json
import os
import random
import sys
import time
from pathlib import Path

# Add parent to path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from captcha_solver.learning.db import LearningDB
from captcha_solver.learning.trainer import Trainer


def generate_math_samples(count: int = 20) -> list[dict]:
    """Generate math captcha test cases: one per method (regex, llm, tesseract)."""
    samples = []
    ops = [
        (lambda a, b: a + b, "+"),
        (lambda a, b: a - b, "-"),
        (lambda a, b: a * b, "*"),
    ]
    for _ in range(count):
        a = random.randint(1, 99)
        b = random.randint(1, min(99, a if False else 99))
        op_fn, op_str = random.choice(ops)
        if op_str == "-" and b > a:
            a, b = b, a
        result = op_fn(a, b)
        samples.append({
            "type": "math",
            "hint": f"{a} {op_str} {b} = ?",
            "expected": str(result),
        })
    return samples


def simulate_solve(sample: dict, solver: str) -> dict:
    """Simulate a solver attempt (no real models needed for CI)."""
    start = time.time()
    # Simulate latency
    latency = random.uniform(0.05, 0.5)
    time.sleep(latency * 0.01)  # 1% of real time for speed

    # Simulated accuracy per solver
    accuracy_map = {
        "regex": 0.92,
        "llm": 0.88,
        "tesseract": 0.65,
        "florence2": 0.75,
        "moondream2": 0.50,
        "qwen": 0.70,
        "whisper": 0.80,
    }
    accuracy = accuracy_map.get(solver, 0.5)

    is_correct = random.random() < accuracy
    answer = sample["expected"] if is_correct else str(random.randint(0, 999))
    elapsed = int((time.time() - start) * 1000)

    return {
        "answer": answer,
        "expected": sample["expected"],
        "correct": is_correct,
        "latency_ms": elapsed,
        "confidence": accuracy if is_correct else accuracy * 0.3,
    }


def run_cycle(output_dir: str | Path | None = None) -> dict:
    """Run one full learning cycle.

    Args:
        output_dir: if set, saves results as JSON here (for GH artifact).
    """
    db = LearningDB()
    trainer = Trainer(db)

    # Start cycle
    cycle_id = db.start_cycle()
    print(f"[cycle {cycle_id}] starting...")

    solvers = ["regex", "llm", "tesseract", "florence2", "moondream2", "qwen", "whisper"]
    samples = generate_math_samples(30)

    total = 0
    correct = 0
    latencies = []

    for sample in samples:
        solver = random.choice(solvers)
        result = simulate_solve(sample, solver)
        db.record_attempt(
            captcha_type=sample["type"],
            solver_used=solver,
            answer=result["answer"],
            expected=result["expected"],
            correct=result["correct"],
            confidence=result.get("confidence"),
            latency_ms=result["latency_ms"],
            hint=sample.get("hint"),
            run_source="learning_cycle",
        )
        total += 1
        if result["correct"]:
            correct += 1
        latencies.append(result["latency_ms"])
        status = "OK" if result["correct"] else "FAIL"
        print(f"  [{solver:12}] {sample['hint']:20} -> {result['answer']:6} ({status})")

    avg_lat = sum(latencies) / max(len(latencies), 1)
    db.finish_cycle(cycle_id, total, correct, avg_lat)

    # Analyze
    analysis = trainer.optimize(cycle_id)

    print(f"\n[cycle {cycle_id}] complete: {correct}/{total} correct ({correct/max(total,1)*100:.1f}%), avg {avg_lat:.0f}ms")

    result = {
        "cycle_id": cycle_id,
        "total": total,
        "correct": correct,
        "accuracy_pct": round(correct / max(total, 1) * 100, 1),
        "avg_latency_ms": round(avg_lat),
        "changes": analysis,
        "stats": db.summary(),
        "recommendations": analysis,
    }

    if output_dir:
        out = Path(output_dir)
        out.mkdir(parents=True, exist_ok=True)
        (out / "cycle_result.json").write_text(json.dumps(result, indent=2, default=str), encoding="utf-8")
        print(f"\nresult saved to {out / 'cycle_result.json'}")

    return result


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", "-o", default=None, help="Output directory for results JSON")
    args = parser.parse_args()
    result = run_cycle(args.output)
    print(f"\naccuracy: {result['accuracy_pct']}% over {result['total']} solves")