chore: run vulture and remove dead code from the repo

pysus/api/ducklake/functional.py

@@ -98,84 +98,6 @@ async def download_http(
                 raise e
 
 
-async def download_s3(
-    remote_path: str,
-    local_path: Path,
-    access_key: str | None = None,
-    secret_key: str | None = None,
-    callback: Callable[[int, int], None] | None = None,
-) -> None:
-    """
-    Download *remote_path* to *local_path* using
-    boto3 with optional credentials.
-
-    Parameters
-    ----------
-    remote_path : str
-        Object key within the bucket.
-    local_path : Path
-        Local destination path.
-    access_key : str, optional
-        S3 access key ID.
-    secret_key : str, optional
-        S3 secret access key.
-    callback : Callable[[int, int], None], optional
-        Progress callback receiving ``(downloaded, total)`` bytes.
-    """
-    max_retries = 5
-
-    def _get_client_args():
-        args: dict = {
-            "service_name": "s3",
-            "endpoint_url": f"https://{types.S3_ENDPOINT}",
-            "region_name": types.S3_REGION,
-        }
-        if access_key and secret_key:
-            args["aws_access_key_id"] = access_key
-            args["aws_secret_access_key"] = secret_key
-            args["config"] = Config(signature_version="s3v4")
-        else:
-            args["config"] = Config(signature_version=UNSIGNED)
-        return args
-
-    def _get_total_size(client_args) -> int:
-        try:
-            client = boto3.client(**client_args)
-            meta = client.head_object(Bucket=types.S3_BUCKET, Key=remote_path)
-            return int(meta.get("ContentLength", 0))
-        except Exception:  # noqa
-            return 0
-
-    def _download(client_args, total_size: int):
-        client = boto3.client(**client_args)
-        downloaded = 0
-
-        def boto_callback(bytes_amount):
-            nonlocal downloaded
-            downloaded += bytes_amount
-            if callback:
-                callback(downloaded, total_size)
-
-        client.download_file(
-            Bucket=types.S3_BUCKET,
-            Key=remote_path,
-            Filename=str(local_path),
-            Callback=boto_callback if callback else None,
-        )
-
-    for attempt in range(max_retries):
-        try:
-            client_args = _get_client_args()
-            total_size = await to_thread.run_sync(_get_total_size, client_args)
-            await to_thread.run_sync(_download, client_args, total_size)
-            return
-        except Exception as e:  # noqa
-            if attempt < max_retries - 1:
-                await sleep(1)
-            else:
-                raise e
-
-
 async def upload_s3(
     local_path: Path,
     remote_path: str,

pysus/api/transform/streaming.py

@@ -15,7 +15,6 @@
 
 from collections.abc import Generator
 from pathlib import Path
-from typing import Any
 
 import pandas as pd
 
@@ -55,46 +54,3 @@ def stream_parquet(
         df = pd.read_parquet(path, columns=columns)
         for i in range(0, len(df), chunk_size):
             yield df.iloc[i : i + chunk_size]
-
-
-def stream_with_progress(
-    path: str | Path,
-    chunk_size: int = 10000,
-    callback: Any = None,
-) -> Generator[pd.DataFrame, None, None]:
-    """Stream with progress callback.
-
-    Parameters
-    ----------
-    path : str or Path
-        Path to Parquet file.
-    chunk_size : int
-        Rows per chunk.
-    callback : callable, optional
-        Progress callback function ``(current, total) -> None``.
-
-    Yields
-    ------
-    pd.DataFrame
-        DataFrame chunks.
-    """
-    try:
-        import pyarrow.parquet as pq
-
-        parquet_file = pq.ParquetFile(path)
-        total_rows = parquet_file.metadata.num_rows
-
-        for i, batch in enumerate(
-            parquet_file.iter_batches(batch_size=chunk_size)
-        ):
-            if callback:
-                callback(min((i + 1) * chunk_size, total_rows), total_rows)
-            yield batch.to_pandas()
-
-    except ImportError:
-        df = pd.read_parquet(path)
-        total = len(df)
-        for i in range(0, total, chunk_size):
-            if callback:
-                callback(min(i + chunk_size, total), total)
-            yield df.iloc[i : i + chunk_size]

pysus/management/compare.py

@@ -203,14 +203,3 @@ def content_fingerprint(
     for row in sampled:
         digest.update(repr(row).encode())
     return digest.hexdigest()
-
-
-def byte_hash(
-    path, algorithm: str = "sha256", chunk_size: int = 1024 * 1024
-) -> str:
-    """Compute the byte-level hash of a local file (exact identity only)."""
-    hash_obj = hashlib.new(algorithm)
-    with open(path, "rb") as f:
-        while chunk := f.read(chunk_size):
-            hash_obj.update(chunk)
-    return hash_obj.hexdigest()

pysus/management/normalize.py

@@ -20,29 +20,12 @@
 import boto3
 from botocore.config import Config
 
-from .records import compose_s3_key, parquet_key
+from .records import compose_s3_key
 
 _BUCKET = "pysus"
 _ENDPOINT = "nbg1.your-objectstorage.com"
 _REGION = "nbg1"
 
-_SCAN_PREFIXES = (
-    "public/data/dadosgov/",
-    "public/data/ftp/ciha/",
-    "data/ftp/",
-)
-
-_FORMAT_RANK = {"csv": 0, "json": 1, "xml": 2, "xlsx": 3}
-
-
-def _format_of(name: str) -> str | None:
-    """Return the format token of a name, if present (csv/json/xml)."""
-    lower = name.lower()
-    for fmt in _FORMAT_RANK:
-        if f".{fmt}." in lower or lower.endswith(f"_{fmt}"):
-            return fmt
-    return None
-
 
 @dataclass
 class ObjectRename:
@@ -152,128 +135,6 @@ def _list_objects(self, prefix: str) -> list[tuple[str, int]]:
                 objects.append((obj["Key"], obj["Size"]))
         return objects
 
-    def _object_exists(self, key: str) -> bool:
-        try:
-            self.client.head_object(Bucket=_BUCKET, Key=key)
-            return True
-        except Exception:  # noqa
-            return False
-
-    def survey_objects(self) -> tuple[list[ObjectRename], list[str]]:
-        """Find non-canonical *parquet* object keys and resolve collisions.
-
-        Only objects whose key ends in ``.parquet`` are candidates — raw
-        source objects (``.dbc``, ``.dbf``, ``.zip``...) are never renamed
-        (that would mislabel content; converting them is ETL work, not a
-        rename).
-
-        When csv/json/xml parquet variants normalize to the same key, the
-        format highest in ``_FORMAT_RANK`` (csv first) is kept and the
-        others are scheduled for deletion.
-        """
-        renames: list[ObjectRename] = []
-        deletes: list[str] = []
-        by_canonical: dict[str, list[tuple[str, int]]] = defaultdict(list)
-
-        for prefix in _SCAN_PREFIXES:
-            for key, size in self._list_objects(prefix):
-                if not key.endswith(".parquet"):
-                    self.raw_objects.append(key)
-                    continue
-                base = key.rsplit("/", 1)[-1]
-                canonical = key.rsplit("/", 1)[0] + "/" + parquet_key(base)
-                by_canonical[canonical].append((key, size))
-
-        for canonical, items in by_canonical.items():
-            if len(items) == 1:
-                key, size = items[0]
-                if key != canonical:
-                    renames.append(
-                        ObjectRename(old=key, new=canonical, size=size)
-                    )
-                continue
-
-            ranked = sorted(
-                items,
-                key=lambda kv: (
-                    _FORMAT_RANK.get(_format_of(kv[0]) or "", 99),
-                    kv[1],
-                ),
-            )
-            winner_key, winner_size = ranked[0]
-            if winner_key != canonical:
-                renames.append(
-                    ObjectRename(
-                        old=winner_key, new=canonical, size=winner_size
-                    )
-                )
-            for loser_key, _ in ranked[1:]:
-                deletes.append(loser_key)
-
-        return renames, deletes
-
-    def survey_catalog(
-        self, catalog_dir: Path
-    ) -> tuple[list[CatalogPathFix], list[CatalogRowDelete]]:
-        """Build catalog path fixes and duplicate-row deletions.
-
-        * Rows whose path ends ``.dbc`` and whose object exists on S3 are
-          raw artifacts: kept as-is (converting is ETL work).
-        * Rows whose path ends ``.dbc`` and whose object is missing while a
-          sibling ``.parquet`` row exists are duplicate stale rows: deleted.
-        * Other non-canonical rows are fixed when the canonical object
-          exists (e.g. ``*.csv.parquet`` -> ``*.parquet`` after the object
-          rename).
-        """
-        import duckdb
-
-        fixes: list[CatalogPathFix] = []
-        row_deletes: list[CatalogRowDelete] = []
-        catalog = catalog_dir.name.removesuffix(".duckdb").removeprefix(
-            "catalog_"
-        )
-        con = duckdb.connect(str(catalog_dir), read_only=True)
-        try:
-            rows = con.execute("SELECT path FROM pysus.files").fetchall()
-        finally:
-            con.close()
-
-        parquet_paths = {p for (p,) in rows if p.endswith(".parquet")}
-
-        for (path,) in rows:
-            base = path.rsplit("/", 1)[-1]
-            canonical = path.rsplit("/", 1)[0] + "/" + parquet_key(base)
-            if canonical == path:
-                continue
-
-            if path.endswith(".dbc"):
-                if self._object_exists(path):
-                    self.raw_objects.append(path)
-                    continue
-                sibling = path.rsplit("/", 1)[0] + "/" + parquet_key(base)
-                if sibling in parquet_paths:
-                    row_deletes.append(
-                        CatalogRowDelete(
-                            catalog=catalog,
-                            path=path,
-                            reason=f"stale duplicate of {sibling}",
-                        )
-                    )
-                else:
-                    self.broken_rows.append((catalog, path))
-                continue
-
-            if self._object_exists(canonical):
-                fixes.append(
-                    CatalogPathFix(
-                        catalog=catalog, old_path=path, new_path=canonical
-                    )
-                )
-            else:
-                self.broken_rows.append((catalog, path))
-
-        return fixes, row_deletes
-
     # ------------------------------------------------------------------
     # apply
     # ------------------------------------------------------------------
@@ -552,7 +413,7 @@ def survey_relayout(
                     )
                 continue
 
-            winner, winner_src = existing[0]
+            winner, _ = existing[0]
             plan.object_renames.append(ObjectRename(old=winner, new=new_key))
             plan.catalog_fixes.append(
                 CatalogPathFix(

pysus/management/sync.py

@@ -1687,18 +1687,6 @@ async def _fix_misparsed_metadata(
             except Exception as exc:  # noqa
                 error(f"repair failed for {record.path}: {exc}")
 
-    @staticmethod
-    def _pick_source(comparison: FileComparison) -> FileRecord | None:
-        """Return the artifact to ingest (ftp > dadosgov > saude)."""
-        record = comparison._pick("ftp")
-        if record is None or record.file is None:
-            record = comparison._pick("dadosgov")
-        if record is None or record.file is None:
-            record = comparison._pick("saude")
-        if record is None or record.file is None:
-            return None
-        return record
-
     @staticmethod
     def _label(comparison: FileComparison) -> str:
         key = comparison.key