changed method of fetching crontab schedule

api/scrape_schedule.py

@@ -1,15 +1,22 @@
 """Reports when a scrape last ran and when cron will next run it.
 
 "Last ran" is read from the mtime of the scraper's output file - true
-regardless of how/when the scrape was actually invoked. "Next run" requires
-telling this module the cron schedule via an env var (there's no reliable,
-portable way to introspect another process's crontab from here), given as
-a standard 5-field cron expression, e.g. "0 * * * *".
+regardless of how/when the scrape was actually invoked. "Next run" needs the
+cron schedule, sourced in order of preference:
+
+1. An explicit env var (SCRAPE_CRON / ARCHIVE_SCRAPE_CRON) - always wins,
+   works anywhere.
+2. The current OS user's crontab (``discover_cron_expr``) - only works when
+   this process runs as the same user whose personal crontab (``crontab -e``)
+   contains the scrape job, and the ``crontab`` binary is available. Not
+   portable beyond that (different user, no cron, system crontab/cron.d
+   instead of a personal one, containers split across hosts, ...).
 """
 
+import subprocess
 from datetime import datetime
 from pathlib import Path
-from typing import Optional
+from typing import List, Optional, Tuple
 
 from croniter import croniter
 
@@ -18,6 +25,58 @@ def _now_local() -> datetime:
     return datetime.now().astimezone()
 
 
+def _read_crontab_lines() -> List[str]:
+    """Return the invoking OS user's crontab lines, or [] if unavailable."""
+    try:
+        result = subprocess.run(
+            ["crontab", "-l"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return []
+    if result.returncode != 0:
+        return []
+    return result.stdout.splitlines()
+
+
+def _parse_crontab_line(line: str) -> Optional[Tuple[str, str]]:
+    """Split one crontab line into (schedule, command); None if not a job."""
+    line = line.strip()
+    if not line or line.startswith("#"):
+        return None
+    if line.startswith("@"):
+        parts = line.split(maxsplit=1)
+        if len(parts) != 2:
+            return None
+        return parts[0], parts[1]
+    parts = line.split(maxsplit=5)
+    if len(parts) != 6:
+        return None
+    return " ".join(parts[:5]), parts[5]
+
+
+def discover_cron_expr(command_hint: str, *, archive: bool) -> Optional[str]:
+    """Find a matching scrape job in the current user's crontab.
+
+    ``command_hint`` is a substring identifying the scrape command (e.g.
+    "scraper.py"); ``archive`` picks the "--archive" line vs. the plain one.
+    Returns None if the crontab can't be read or no line matches.
+    """
+    for line in _read_crontab_lines():
+        parsed = _parse_crontab_line(line)
+        if parsed is None:
+            continue
+        schedule, command = parsed
+        if command_hint not in command:
+            continue
+        if ("--archive" in command) != archive:
+            continue
+        return schedule
+    return None
+
+
 def _file_mtime_local(path: Path) -> Optional[datetime]:
     try:
         return datetime.fromtimestamp(path.stat().st_mtime).astimezone()