fix: fix readthedocs build

pyproject.toml

@@ -37,6 +37,7 @@ aioftp = "^0.21.4"
 dbfread = "2.0.7"
 bigtree = "^0.12.2"
 pyreaddbc = ">=2.0.4"
+PyYAML = "^6.0.1"
 dotenv = "^0.9.9"
 boto3 = "^1.42.89"
 typer = "^0.24.1"

pysus/cli/management.py

@@ -114,18 +114,16 @@ def check(
         help="Path to a file with ACCESS_KEY/SECRET_KEY/DADOSGOV_TOKEN",
     ),
 ):
-    """Check every source against the S3 databases.
-
-    By default this is a dry run: it only reports which files would need
-    to be updated/uploaded (``needs_update``) and which are already at
-    the most updated version (``skipped``), without touching S3. Pass
-    ``--apply`` to actually download, convert, upload and catalog the
-    outdated files. Use ``--json`` to stream machine-readable results.
-
-    A run is resumable: each completed file is appended to a journal
-    (``--resume``, or derived from ``--reupload-before``), and a paused
-    run can be resumed with the same command to skip already-processed
-    files.
+    """Check a database against its FTP origin to see if it needs updating.
+
+    By default this is a dry run: it classifies every mirrored file as
+    ``missing`` (not on S3), ``outdated`` (the FTP file is more recent or
+    has a different size) or ``current``, and prints a per-database table
+    with a "needs update" / "up to date" verdict — without touching S3.
+
+    Pass ``--apply`` to actually download, convert, upload and catalog the
+    outstanding files (the full sync pipeline). Pass ``--json`` to stream
+    machine-readable results.
     """
     from pysus.api.client import _run_sync
     from pysus.management.sync import SyncEngine
@@ -137,9 +135,67 @@ def check(
         dadosgov_token=env.get("DADOSGOV_TOKEN"),
     )
 
+    datasets = [d.upper() for d in name] if name else None
+
+    def _flush() -> None:
+        import sys
+
+        sys.stdout.flush()
+        sys.stderr.flush()
+
+    def _print_check_report(checks) -> None:
+        total_missing = sum(len(c.missing) for c in checks.values())
+        total_outdated = sum(len(c.outdated) for c in checks.values())
+        total_current = sum(len(c.current) for c in checks.values())
+
+        typer.echo(
+            f"{'DATABASE':<30}{'MISSING':>9}{'OUTDATED':>10}"
+            f"{'CURRENT':>9}  STATUS"
+        )
+        typer.echo("-" * 78)
+        for ds in sorted(checks):
+            c = checks[ds]
+            verdict = "needs update" if c.needs_update else "up to date"
+            typer.echo(
+                f"{ds:<30}{len(c.missing):>9}{len(c.outdated):>10}"
+                f"{len(c.current):>9}  {verdict}"
+            )
+        typer.echo("-" * 78)
+        typer.echo(
+            f"{'TOTAL':<30}{total_missing:>9}{total_outdated:>10}"
+            f"{total_current:>9}"
+        )
+        if total_missing or total_outdated:
+            typer.echo(
+                f"\n{total_missing + total_outdated} of "
+                f"{total_missing + total_outdated + total_current} file(s) "
+                "need updating — re-run with --apply to mirror them."
+            )
+
+    def _print_check_json(checks) -> None:
+        for ds in sorted(checks):
+            typer.echo(json.dumps({"dataset": ds, **checks[ds].summary()}))
+
+    async def _run_check() -> None:
+        await engine.__aenter__(lock=False)
+        try:
+            checks = await engine.check(datasets=datasets)
+        finally:
+            await engine.__aexit__(None, None, None)
+        _flush()
+        if json_out:
+            _print_check_json(checks)
+        else:
+            _print_check_report(checks)
+        _flush()
+
+    if not apply:
+        _run_sync(_run_check())
+        return
+
     journal = _journal_path(resume, reupload_before)
     resume_keys = set()
-    if apply and journal is not None and journal.exists():
+    if journal is not None and journal.exists():
         resume_keys = load_journal_keys(journal)
 
     counts: dict[str, int] = {}
@@ -172,25 +228,19 @@ def on_outcome(outcome) -> None:
         if total % 500 == 0:
             typer.echo(f"progress: {counts}", err=True)
 
-    def _flush() -> None:
-        import sys
-
-        sys.stdout.flush()
-        sys.stderr.flush()
-
-    async def _run():
+    async def _run() -> dict[str, int]:
         async with engine:
             report = await engine.run(
-                datasets=[d.upper() for d in name] if name else None,
+                datasets=datasets,
                 force=force,
                 reupload_before=_parse_date(reupload_before),
-                dry_run=not apply,
+                dry_run=False,
                 workers=workers,
                 ftp_connections=ftp_connections,
-                checkpoint_every=checkpoint_every if apply else None,
+                checkpoint_every=checkpoint_every,
                 on_outcome=on_outcome,
                 resume=resume_keys or None,
-                journal=journal if apply else None,
+                journal=journal,
             )
         summary = report.summary()
         _flush()
@@ -211,5 +261,5 @@ async def _run():
         return summary
 
     summary = _run_sync(_run())
-    if summary["failed"]:
+    if summary and summary["failed"]:
         raise typer.Exit(code=1)

pysus/data/dbf_reader.py

@@ -256,31 +256,29 @@ def stream_dbf_fast(
     dtype = schema.build_dtype()
 
     with open(path, "rb") as fh:
-        fh.seek(schema.header_len)
-        raw = fh.read(n * rl)
-
-    for start in range(0, n, chunk_size):
-        end = min(start + chunk_size, n)
-        chunk_n = end - start
-        chunk_raw = raw[start * rl : end * rl]
-
-        records: np.ndarray = np.frombuffer(
-            chunk_raw, dtype=dtype, count=chunk_n
-        )
-        records = records[records["_deleted"] != b"*"]  # skip deleted rows
-        chunk_n = len(records)
-
-        data = {}
-        for fld in schema.fields:
-            col: np.ndarray = records[fld.name]
-            decoded = np.empty(chunk_n, dtype=object)
-            for i in range(chunk_n):
-                val = col[i]
-                b = val if isinstance(val, bytes) else val.tobytes()
-                decoded[i] = _decode(b)
-            data[fld.name] = decoded
-
-        yield pd.DataFrame(data)
+        for start in range(0, n, chunk_size):
+            end = min(start + chunk_size, n)
+            chunk_n = end - start
+            fh.seek(schema.header_len + start * rl)
+            chunk_raw = fh.read(chunk_n * rl)
+
+            records: np.ndarray = np.frombuffer(
+                chunk_raw, dtype=dtype, count=chunk_n
+            )
+            records = records[records["_deleted"] != b"*"]  # skip deleted rows
+            chunk_n = len(records)
+
+            data = {}
+            for fld in schema.fields:
+                col: np.ndarray = records[fld.name]
+                decoded = np.empty(chunk_n, dtype=object)
+                for i in range(chunk_n):
+                    val = col[i]
+                    b = val if isinstance(val, bytes) else val.tobytes()
+                    decoded[i] = _decode(b)
+                data[fld.name] = decoded
+
+            yield pd.DataFrame(data)
 
 
 def _find_field(schema: DBFSchema, name: str) -> DBFField:

pysus/management/records.py

@@ -335,6 +335,87 @@ def to_dict(self) -> dict[str, Any]:
         }
 
 
+#: Freshness classification for a mirrored file vs its origin.
+FRESH_MISSING = "missing"  # absent from the S3 mirror
+FRESH_OUTDATED = "outdated"  # present but the origin file is more recent
+FRESH_CURRENT = "current"  # present and up to date
+
+
+def freshness_status(comparison: FileComparison) -> tuple[str, str]:
+    """Classify a logical file against its mirrored S3 (ducklake) copy.
+
+    A file that exists on an origin (FTP/DadosGov/Saude) but has no S3
+    artifact is ``missing``. When an S3 artifact exists, the mirror is
+    ``outdated`` if any origin record is *newer* (its ``modified`` origin
+    date is later than the recorded ``source_modified``) **or** has a
+    *different* ``size`` than the recorded ``source_size``. Otherwise it
+    is ``current``.
+
+    Returns a ``(status, reason)`` tuple where *status* is one of
+    :data:`FRESH_MISSING`, :data:`FRESH_OUTDATED` or :data:`FRESH_CURRENT`.
+    """
+    s3 = comparison._pick("ducklake")
+    if s3 is None:
+        return FRESH_MISSING, "no mirror artifact in the S3 catalog"
+
+    reasons: list[str] = []
+    for record in comparison.records:
+        if record.origin == "ducklake":
+            continue
+        # 1) origin modification date ("origin date") newer than mirrored.
+        if (
+            record.modified is not None
+            and s3.source_modified is not None
+            and record.modified > s3.source_modified
+        ):
+            reasons.append(
+                f"{record.origin} modified {record.modified:%Y-%m-%d} is "
+                f"newer than the mirrored {s3.source_modified:%Y-%m-%d}"
+            )
+        # 2) origin size differs from the mirrored source size.
+        if record.size and s3.source_size and record.size != s3.source_size:
+            reasons.append(
+                f"{record.origin} size {record.size} differs from the "
+                f"mirrored {s3.source_size}"
+            )
+
+    if reasons:
+        return FRESH_OUTDATED, "; ".join(reasons)
+    return FRESH_CURRENT, "mirror is up to date"
+
+
+@dataclass
+class DatabaseCheck:
+    """Aggregated freshness check for one database."""
+
+    dataset: str
+    missing: list[str] = field(default_factory=list)
+    outdated: list[str] = field(default_factory=list)
+    current: list[str] = field(default_factory=list)
+
+    @property
+    def needs_update(self) -> bool:
+        """True when any file is missing or outdated in this database."""
+        return bool(self.missing or self.outdated)
+
+    def add(self, status: str, label: str, reason: str = "") -> None:
+        bucket = {
+            FRESH_MISSING: self.missing,
+            FRESH_OUTDATED: self.outdated,
+            FRESH_CURRENT: self.current,
+        }.get(status)
+        if bucket is not None:
+            bucket.append(label if not reason else f"{label} — {reason}")
+
+    def summary(self) -> dict[str, int]:
+        return {
+            "missing": len(self.missing),
+            "outdated": len(self.outdated),
+            "current": len(self.current),
+            "needs_update": self.needs_update,
+        }
+
+
 @dataclass
 class SnapshotDiff:
     """Difference between two snapshots of the same origin."""
@@ -432,10 +513,13 @@ def write_journal_line(path: Path, outcome: SyncOutcome) -> None:
 
 
 def load_journal_keys(path: Path) -> set[IdentityKey]:
-    """Return identity keys already processed in a prior run.
+    """Return identity keys already transferred in a prior run.
 
-    Both ``uploaded`` and ``failed`` entries are included so a resumed run
-    skips files that were already transferred or could not be downloaded.
+    Only ``uploaded`` entries are included so a resumed run skips files
+    that were transferred and never re-downloads them. ``failed`` entries
+    are deliberately excluded: transient failures (e.g. a throttled FTP
+    server or a dropped connection) must be retried on the next run, not
+    dropped permanently.
     """
     keys: set[IdentityKey] = set()
     if not path.exists():
@@ -448,7 +532,7 @@ def load_journal_keys(path: Path) -> set[IdentityKey]:
             data = json.loads(line)
         except json.JSONDecodeError:
             continue
-        if data.get("status") not in ("uploaded", "failed"):
+        if data.get("status") != "uploaded":
             continue
         try:
             keys.add(