fix(ci): harden weekly ETL schedule

.github/workflows/etl_semanal.yml

@@ -2,7 +2,7 @@ name: ETL Weekly Data Refresh
 
 on:
   schedule:
-    - cron: "0 8 * * 1"
+    - cron: "17 8 * * 1"
   workflow_dispatch:
 
 permissions:
@@ -65,16 +65,19 @@ jobs:
             datos/github_lenguajes.csv
             datos/github_ai_repos_insights.csv
             datos/github_commits_frameworks.csv
+            datos/github_commits_frameworks_monthly.csv
             datos/github_correlacion.csv
             datos/latest/github_repos_2025.csv
             datos/latest/github_lenguajes.csv
             datos/latest/github_ai_repos_insights.csv
             datos/latest/github_commits_frameworks.csv
+            datos/latest/github_commits_frameworks_monthly.csv
             datos/latest/github_correlacion.csv
             datos/history/**/github_repos_2025.csv
             datos/history/**/github_lenguajes.csv
             datos/history/**/github_ai_repos_insights.csv
             datos/history/**/github_commits_frameworks.csv
+            datos/history/**/github_commits_frameworks_monthly.csv
             datos/history/**/github_correlacion.csv
 
   job_stackoverflow:
@@ -392,7 +395,9 @@ jobs:
           for required in \
             github_repos_2025.csv \
             github_lenguajes.csv \
+            github_ai_repos_insights.csv \
             github_commits_frameworks.csv \
+            github_commits_frameworks_monthly.csv \
             github_correlacion.csv \
             so_volumen_preguntas.csv \
             so_tasa_aceptacion.csv \
@@ -502,6 +507,9 @@ jobs:
             --project-root . \
             --expect-previous-history "${{ steps.previous_history.outputs.expect_previous_history }}"
 
+      - name: Enforce canonical source freshness guard
+        run: python scripts/check_source_freshness.py --project-root . --max-source-age-hours 192
+
       - name: Upload aggregate artifacts
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:

README.md

@@ -84,14 +84,18 @@ flutter run -d chrome
 ## Automation (GitHub Actions)
 
 1. **ETL Weekly Refresh** (`etl_semanal.yml`)
-   - Schedule: Monday `08:00 UTC`
+   - Schedule: Monday `08:17 UTC` (approximately 03:17 America/Guayaquil).
    - Runs all ETLs, Trend Score, syncs assets, validates contracts, and publishes data.
 2. **Dependency Security** (`dependency_security.yml`)
    - Schedule: Monday `09:00 UTC`
    - Runs `pip-audit` against `backend/requirements.txt`.
 3. **Deploy Frontend** (`deploy_frontend.yml`)
    - Publishes Flutter Web to GitHub Pages.
 
+### Delayed weekly run recovery
+
+GitHub Actions schedules can be delayed. First inspect **ETL Weekly Data Refresh** in the Actions tab and confirm that no run is in progress for the `etl-pipeline` concurrency group. If the scheduled run is absent or has failed, use **Run workflow** on `main` only after that check. Review the run summary and the source freshness guard before considering the published data healthy.
+
 ---
 
 ## Project Structure

scripts/check_source_freshness.py

@@ -0,0 +1,164 @@
+"""Reject canonical source snapshots that are too old to publish."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+DEFAULT_MAX_SOURCE_AGE_HOURS = 192
+REQUIRED_SOURCE_DATASETS = {
+    "github": (
+        "github_ai_repos_insights",
+        "github_commits_frameworks",
+        "github_commits_frameworks_monthly",
+        "github_correlacion",
+        "github_lenguajes",
+        "github_repos_2025",
+    ),
+    "stackoverflow": (
+        "so_volumen_preguntas",
+        "so_tasa_aceptacion",
+        "so_tendencias_mensuales",
+    ),
+    "reddit": (
+        "reddit_sentimiento_frameworks",
+        "reddit_temas_emergentes",
+    ),
+}
+
+
+def _parse_utc_timestamp(value: object) -> datetime | None:
+    if not isinstance(value, str) or not value.endswith("Z"):
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
+    except ValueError:
+        return None
+
+
+def _oldest_required_timestamp(
+    dataset_summaries: object,
+    required_datasets: tuple[str, ...],
+) -> tuple[str, datetime] | None:
+    if not isinstance(dataset_summaries, list):
+        return None
+
+    summaries = {
+        str(summary.get("dataset")): summary
+        for summary in dataset_summaries
+        if isinstance(summary, dict)
+    }
+    timestamps: list[tuple[str, datetime]] = []
+    for dataset in required_datasets:
+        summary = summaries.get(dataset)
+        timestamp = _parse_utc_timestamp(summary.get("updated_at_utc")) if summary else None
+        if timestamp is None:
+            return None
+        timestamps.append((str(summary["updated_at_utc"]), timestamp))
+
+    return min(timestamps, key=lambda item: item[1])
+
+
+def _source_error(
+    dataset_summaries: object,
+    source: str,
+    required_datasets: tuple[str, ...],
+    reference_at: datetime,
+) -> str | None:
+    if not isinstance(dataset_summaries, list):
+        return f"Source freshness unavailable: {source} has no canonical dataset summaries"
+
+    seen_datasets: set[str] = set()
+    required_names = set(required_datasets)
+    for summary in dataset_summaries:
+        if not isinstance(summary, dict):
+            continue
+        dataset = str(summary.get("dataset"))
+        if dataset not in required_names:
+            continue
+        if dataset in seen_datasets:
+            return f"Source freshness invalid: {source} has duplicate required dataset {dataset}"
+        seen_datasets.add(dataset)
+
+    summaries = {
+        str(summary.get("dataset")): summary
+        for summary in dataset_summaries
+        if isinstance(summary, dict)
+    }
+    for dataset in required_datasets:
+        summary = summaries.get(dataset)
+        if summary is None:
+            return f"Source freshness unavailable: {source} is missing required dataset {dataset}"
+        timestamp = _parse_utc_timestamp(summary.get("updated_at_utc"))
+        if timestamp is None:
+            return f"Source freshness unavailable: {source} has invalid updated_at_utc for {dataset}"
+        if timestamp > reference_at:
+            return f"Source freshness invalid: {source} has future updated_at_utc for {dataset}"
+    return None
+
+
+def check_source_freshness(
+    project_root: Path | str,
+    *,
+    max_source_age_hours: int = DEFAULT_MAX_SOURCE_AGE_HOURS,
+) -> dict[str, object]:
+    """Validate source timestamps against the manifest generation time."""
+    manifest_path = Path(project_root) / "frontend" / "assets" / "data" / "run_manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    if not isinstance(manifest, dict):
+        raise ValueError("Source freshness unavailable: run manifest must be a JSON object")
+
+    reference_at = _parse_utc_timestamp(manifest.get("generated_at_utc"))
+    if reference_at is None:
+        raise ValueError("Source freshness unavailable: manifest generated_at_utc is invalid")
+
+    source_updated_at_utc: dict[str, str] = {}
+    errors: list[str] = []
+    dataset_summaries = manifest.get("dataset_summaries")
+    for source, required_datasets in REQUIRED_SOURCE_DATASETS.items():
+        source_error = _source_error(dataset_summaries, source, required_datasets, reference_at)
+        if source_error is not None:
+            errors.append(source_error)
+            continue
+
+        oldest = _oldest_required_timestamp(dataset_summaries, required_datasets)
+        if oldest is None:
+            errors.append(f"Source freshness unavailable: {source} has no valid canonical updated_at_utc")
+            continue
+        updated_at_raw, updated_at = oldest
+        source_updated_at_utc[source] = updated_at_raw
+        age_hours = (reference_at - updated_at).total_seconds() / 3600
+        if age_hours > max_source_age_hours:
+            errors.append(
+                f"Source freshness stale: {source} is {age_hours:.1f}h old "
+                f"(maximum {max_source_age_hours}h)"
+            )
+
+    if errors:
+        raise ValueError("; ".join(errors))
+
+    return {
+        "max_source_age_hours": max_source_age_hours,
+        "source_updated_at_utc": source_updated_at_utc,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--project-root", type=Path, default=Path("."))
+    parser.add_argument("--max-source-age-hours", type=int, default=DEFAULT_MAX_SOURCE_AGE_HOURS)
+    args = parser.parse_args()
+
+    try:
+        check_source_freshness(args.project_root, max_source_age_hours=args.max_source_age_hours)
+    except ValueError as error:
+        print(f"Source freshness guard failed: {error}")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_check_source_freshness.py

@@ -0,0 +1,141 @@
+import json
+
+import pytest
+
+from scripts.check_source_freshness import check_source_freshness
+
+
+CANONICAL_DATASETS = {
+    "github": (
+        "github_ai_repos_insights",
+        "github_commits_frameworks",
+        "github_commits_frameworks_monthly",
+        "github_correlacion",
+        "github_lenguajes",
+        "github_repos_2025",
+    ),
+    "stackoverflow": (
+        "so_volumen_preguntas",
+        "so_tasa_aceptacion",
+        "so_tendencias_mensuales",
+    ),
+    "reddit": (
+        "reddit_sentimiento_frameworks",
+        "reddit_temas_emergentes",
+    ),
+}
+
+
+def _write_manifest(
+    tmp_path,
+    *,
+    updated_at_by_dataset=None,
+    missing_datasets=(),
+    prepended_summaries=(),
+):
+    assets_root = tmp_path / "frontend" / "assets" / "data"
+    assets_root.mkdir(parents=True)
+    timestamps = {
+        dataset: "2026-08-31T08:17:00Z"
+        for datasets in CANONICAL_DATASETS.values()
+        for dataset in datasets
+    }
+    timestamps.update(updated_at_by_dataset or {})
+    payload = {
+        "generated_at_utc": "2026-08-31T08:17:00Z",
+        "dataset_summaries": list(prepended_summaries)
+        + [
+            {"dataset": dataset, "updated_at_utc": updated_at}
+            for dataset, updated_at in timestamps.items()
+            if dataset not in missing_datasets
+        ]
+        + [
+            {"dataset": "trend_score", "updated_at_utc": "2026-08-31T08:17:00Z"},
+        ],
+    }
+    (assets_root / "run_manifest.json").write_text(json.dumps(payload), encoding="utf-8")
+
+
+def test_allows_mixed_source_freshness_within_weekly_grace(tmp_path):
+    _write_manifest(
+        tmp_path,
+        updated_at_by_dataset={
+            **{
+                dataset: "2026-08-24T08:17:00Z"
+                for dataset in CANONICAL_DATASETS["github"]
+                + CANONICAL_DATASETS["stackoverflow"]
+            },
+            "reddit_sentimiento_frameworks": "2026-08-31T08:16:00Z",
+            "reddit_temas_emergentes": "2026-08-31T08:17:00Z",
+        },
+    )
+
+    result = check_source_freshness(tmp_path)
+
+    assert result["source_updated_at_utc"] == {
+        "github": "2026-08-24T08:17:00Z",
+        "stackoverflow": "2026-08-24T08:17:00Z",
+        "reddit": "2026-08-31T08:16:00Z",
+    }
+
+
+def test_rejects_source_older_than_weekly_grace(tmp_path):
+    _write_manifest(
+        tmp_path,
+        updated_at_by_dataset={"github_lenguajes": "2026-08-23T08:16:59Z"},
+    )
+
+    with pytest.raises(ValueError, match="Source freshness stale: github"):
+        check_source_freshness(tmp_path)
+
+
+def test_rejects_missing_source_timestamp_without_manifest_fallback(tmp_path):
+    _write_manifest(
+        tmp_path,
+        updated_at_by_dataset={"github_lenguajes": "not-a-timestamp"},
+    )
+
+    with pytest.raises(ValueError, match="Source freshness unavailable: github"):
+        check_source_freshness(tmp_path)
+
+
+def test_rejects_required_timestamp_later_than_manifest_generation(tmp_path):
+    _write_manifest(
+        tmp_path,
+        updated_at_by_dataset={"github_lenguajes": "2026-08-31T09:17:00Z"},
+    )
+
+    with pytest.raises(ValueError, match="Source freshness invalid: github"):
+        check_source_freshness(tmp_path)
+
+
+def test_rejects_duplicate_required_dataset_even_when_later_entry_is_fresh(tmp_path):
+    _write_manifest(
+        tmp_path,
+        prepended_summaries=(
+            {"dataset": "github_lenguajes", "updated_at_utc": "2026-08-20T08:17:00Z"},
+        ),
+    )
+
+    with pytest.raises(ValueError, match="Source freshness invalid: github"):
+        check_source_freshness(tmp_path)
+
+
+def test_rejects_stale_required_sibling_even_when_another_github_dataset_is_fresh(tmp_path):
+    _write_manifest(
+        tmp_path,
+        updated_at_by_dataset={
+            "github_lenguajes": "2026-08-20T08:17:00Z",
+            "github_repos_2025": "2026-08-31T08:17:00Z",
+        },
+    )
+
+    with pytest.raises(ValueError, match="Source freshness stale: github"):
+        check_source_freshness(tmp_path)
+
+
+def test_rejects_missing_required_canonical_dataset(tmp_path):
+    _write_manifest(tmp_path, missing_datasets=("github_lenguajes",))
+
+    with pytest.raises(ValueError, match="Source freshness unavailable: github"):
+        check_source_freshness(tmp_path)