inventory¶

def list_repos(self) -> pl.DataFrame:
    """Fetch and parse the public repositories for the specified GitHub user.

    Checks the local cache first (unless force_refresh=True). Returns a Polars DataFrame
    with the columns 'name', 'html_url', and 'description'.
    """
    self._inventory_df = self._retrieve_repos()
    return self._inventory_df

def review_version_changes(
    self,
    from_v: str = "first",
    to_v: str = "latest",
) -> pl.DataFrame:
    """Compare repository metadata across two versions (placeholder).

    Currently returns a trivial DataFrame.
    """
    return pl.DataFrame({"from_v": [from_v], "to_v": [to_v]})

def walk_file_trees(
    self,
    pattern: str = "**",
    no_recurse: bool = False,
    skip_larger_than_mb: int | None = None,
) -> pl.DataFrame:
    """Walk (recursively enumerate) files in each repository via UPath.

    Discovers (but does not read) file paths that match a given glob pattern.

    Args:
    ----
        pattern: Glob pattern for file listing. By default "**" (recursive).
        no_recurse: If True, uses "*" (non-recursive) instead of the default "**".
        skip_larger_than_mb: If set, skip listing files larger than this many MB.
                             By default, None (don't skip based on size).

    Returns:
    -------
        A Polars DataFrame with columns:
            - "repository_name": str
            - "file_path": str
            - "is_directory": bool
            - "file_size_bytes": int

    """
    if self._inventory_df is None:
        self.list_repos()
    if no_recurse:
        pattern = "*"
    records = []
    for row in self._inventory_df.to_dicts():
        repo_name = row["name"]
        default_branch = row["default_branch"]
        ghpath = UPath(
            "/",
            protocol="github",
            org=self.username,
            repo=repo_name,
            sha=default_branch,
            username=self.username,
            token=self.token,
        )
        for p in ghpath.glob(pattern):
            if is_dir := p.is_dir():
                file_size_bytes = 0
            else:
                file_size_bytes = p.stat().st_size
                if skip_larger_than_mb is not None:
                    threshold_bytes = skip_larger_than_mb * 1_048_576
                    if file_size_bytes > threshold_bytes:
                        continue
            records.append(
                {
                    "repository_name": repo_name,
                    "file_path": os.path.join(*p.parts),
                    "is_directory": is_dir,
                    "file_size_bytes": file_size_bytes,
                },
            )
    files = pl.DataFrame(
        records,
        schema={
            "repository_name": pl.String,
            "file_path": pl.String,
            "is_directory": pl.Boolean,
            "file_size_bytes": pl.Int64,
        },
    )
    files.hopper.add_filters(*self.filter_exprs)
    files.hopper.add_selects(*self.select_exprs)
    files.hopper.add_addcols(*self.addcols_exprs)
    files = files.hopper.apply_ready_exprs()
    self.filter_exprs = tuple(files.hopper.list_filters())
    self.select_exprs = tuple(files.hopper.list_selects())
    self.addcols_exprs = tuple(files.hopper.list_addcols())
    return files

def read_files(
    self,
    pattern: str = "**",
    no_recurse: bool = False,
    skip_larger_than_mb: int | None = None,
) -> pl.DataFrame:
    """Read *all* file contents in each matched repository path.

    This enumerates all files (via walk_file_trees) that match `pattern`,
    then reads their text content if they are not directories.

    Args:
    ----
        pattern: Glob pattern for file listing. Default "**" means recursive.
        no_recurse: If True, uses "*" instead of "**".
        skip_larger_than_mb: Optional size limit in MB. If set, skip any file above it.

    Returns:
    -------
        A Polars DataFrame with columns:
            - "repository_name": str
            - "file_path": str
            - "file_size_bytes": int
            - "content": str (file content, or empty if directory/failed)

    """
    # First, get a listing
    file_tree = self.walk_file_trees(
        pattern=pattern,
        no_recurse=no_recurse,
        skip_larger_than_mb=skip_larger_than_mb,
    )
    if file_tree.is_empty():
        return file_tree

    # We'll accumulate a new DataFrame with the file content
    rows = []
    for row in file_tree.to_dicts():
        if row["is_directory"]:
            # Skip directories; no content to read
            rows.append(
                {
                    "repository_name": row["repository_name"],
                    "file_path": row["file_path"],
                    "file_size_bytes": row["file_size_bytes"],
                    "content": "",
                },
            )
            continue

        repo_name = row["repository_name"]
        file_path = row["file_path"]
        # We'll look up the default branch from the known repos
        default_branch = (
            self._inventory_df.filter(pl.col("name") == repo_name)
            .select("default_branch")
            .item()
        )

        ghpath = UPath(
            "/",
            protocol="github",
            org=self.username,
            repo=repo_name,
            sha=default_branch,
            username=self.username,
            token=self.token,
        )
        p = ghpath / file_path
        try:
            content_str = p.read_text()
        except Exception:
            content_str = ""

        rows.append(
            {
                "repository_name": repo_name,
                "file_path": file_path,
                "file_size_bytes": row["file_size_bytes"],
                "content": content_str,
            },
        )
    return pl.DataFrame(rows)