API Reference

class PolisKMeans(KMeans):
    """
    A modified version of scikit-learn's KMeans that allows partial initialization
    with user-supplied cluster centers and custom fallback strategies.

    This subclass extends `sklearn.cluster.KMeans` with additional features
    around centroid initialization. Outside the behavior documented, it retains
    all other parameters and behavior from the base KMeans implementation.

    Parameters
    ----------

    init : {'k-means++', 'random', 'polis'}, default='k-means++'
        Strategy to initialize any missing cluster centers if `init_centers` is
        not fully specified. The strategies are:

        - 'k-means++': Smart centroid initialization (same as scikit-learn default)
        - 'random': Random selection of initial centers from the data (same as scikit-learn)
        - 'polis': Selects the first unique data points in `X` as initial centers.
            - This strategy is deterministic for any stable set of `X`, while
            determinism in the other strategies depends on `random_state`.

        !!! note
            Unlike `KMeans` parent class, we prevent passing `ndarray` args
            here, and expect `init_centers` to handle that use-case.

    init_centers : ndarray of shape (n_clusters, n_features), optional
        Initial cluster centers to use. May contain fewer (or more) than `n_clusters`:

        - If more, the extras will be trimmed
        - If fewer, the remaining will be filled using the `init` strategy

    Attributes
    ----------

    init_centers_used_ : ndarray of shape (n_clusters, n_features)
        The full array of initial cluster centers actually used to initialize the algorithm,
        including both `init_centers` and any centers generated from the `init` strategy.

    See Also
    --------

    `sklearn.cluster.KMeans` : Original implementation with full parameter list.
    """
    def __init__(
        self,
        n_clusters=8,
        init="k-means++",  # or 'random', 'polis'
        init_centers: Optional[ArrayLike] = None,  # array-like, optional
        n_init="auto",
        max_iter=300,
        tol=1e-4,
        verbose=0,
        random_state=None,
        copy_x=True,
        algorithm="lloyd",
    ):
        super().__init__(
            n_clusters=n_clusters,
            init=init,  # will override via set_params, with our center selection logic below
            n_init=n_init,
            max_iter=max_iter,
            tol=tol,
            verbose=verbose,
            random_state=random_state,
            copy_x=copy_x,
            algorithm=algorithm,
        )
        self._init_strategy = init
        self.init_centers = init_centers
        self.init_centers_used_ = None

    def _generate_centers(self, X, x_squared_norms, n_to_generate, random_state):
        if not isinstance(self._init_strategy, str):
            raise ValueError("Internal error: _strategy must be a string.")

        if self._init_strategy == "k-means++":
            centers, _ = kmeans_plusplus(
                X, n_clusters=n_to_generate,
                random_state=random_state,
                x_squared_norms=x_squared_norms
            )
        elif self._init_strategy == "random":
            indices = random_state.choice(X.shape[0], n_to_generate, replace=False)
            centers = X[indices]
        elif self._init_strategy == "polis":
            unique_X = np.unique(X, axis=0)
            if len(unique_X) < n_to_generate:
                raise ValueError("Not enough unique rows in X for 'polis' strategy.")
            centers = unique_X[:n_to_generate]
        else:
            raise ValueError(f"Unsupported init strategy: {self._init_strategy}")
        return centers

    def fit(self, X, y=None, sample_weight=None):
        X = check_array(X, accept_sparse="csr", dtype=[np.float64, np.float32]) # type:ignore
        random_state = check_random_state(self.random_state)
        x_squared_norms = np.sum(X ** 2, axis=1)

        # Determine init_centers_used_
        if self.init_centers is not None:
            init_array = np.array(self.init_centers)
            if init_array.ndim != 2 or init_array.shape[1] != X.shape[1]:
                raise ValueError("init_centers must be of shape (n, n_features)")

            n_given = init_array.shape[0]
            if n_given > self.n_clusters:
                init_array = init_array[:self.n_clusters]
            elif n_given < self.n_clusters:
                needed = self.n_clusters - n_given
                extra = self._generate_centers(X, x_squared_norms, needed, random_state)
                init_array = np.vstack([init_array, extra])
            self.init_centers_used_ = init_array.copy()
        else:
            self.init_centers_used_ = self._generate_centers(
                X, x_squared_norms, self.n_clusters, random_state
            )

        # Override the init param passed to sklearn with actual centers.
        # We take control of the initialization strategy (`k-means++`, `random`,
        # `polis`, etc) in our own code.
        super().set_params(init=self.init_centers_used_)

        return super().fit(X, y=y, sample_weight=sample_weight)

class PolisKMeansDownsampler(BaseEstimator, TransformerMixin):
    """
    A transformer that fits `PolisKMeans` and returns the cluster centers as the
    downsampled dataset.

    This will support mimicking "base clusters" from the Polis platform.

    This enables use in sklearn pipelines, where intermediate steps
    are expected to implement both `fit` and `transform`.
    """
    def __init__(self,
        n_clusters=100,
        random_state=None,
        init="k-means++",
        init_centers=None,
    ):
        self.n_clusters = n_clusters
        self.random_state = random_state
        self.init = init
        self.init_centers = init_centers
        self.kmeans_ = None

    def fit(self, X, y=None):
        self.kmeans_ = PolisKMeans(
            n_clusters=self.n_clusters,
            random_state=self.random_state,
            init=self.init,
            init_centers=self.init_centers,
        )
        self.kmeans_.fit(X)
        return self

    def transform(self, X, y=None):
        return self.kmeans_.cluster_centers_ if self.kmeans_ else None

class GridSearchNonCV(GridSearchCV):
    """
    `sklearn.model_selection.GridSearchCV`, but modified to score against the
    full dataset (ie. not cross-validated).

    Normally, `GridSearchCV` splits up the `X` data and scores each "fold" of data.
    This is identical, but we automatically use the full dataset in each fold.
    """
    def __init__(self, estimator, param_grid, scoring=None, refit=True, **kwargs):
        # Default CV is a single fold: train = test = full dataset
        self._default_cv = None  # we'll set it in fit() when we have data size

        # User can override cv via kwargs
        self._user_provided_cv = 'cv' in kwargs
        super().__init__(
            estimator=estimator,
            param_grid=param_grid,
            scoring=scoring,
            refit=refit,
            cv=None,  # will be overwritten in fit()
            **kwargs
        )

    def fit(self, X, y=None, **fit_params):
        if not self._user_provided_cv:
            # Create full-fold cross-validation only if user didn’t specify their own.
            # This scores the full dataset for each branch of grid.
            # See: https://scikit-learn.org/stable/modules/generated/sklearn.model_selection.GridSearchCV.html#:~:text=An%20iterable%20yielding%20(train%2C%20test)%20splits%20as%20arrays%20of%20indices.
            n_samples = len(X)
            full_idx = np.arange(n_samples)
            train_idx = full_idx
            test_idx = full_idx
            # Use full dataset for training/testing
            full_fold = [(train_idx, test_idx)]
            self.cv = full_fold
        return super().fit(X, y, **fit_params)

class SparsityAwareScaler(BaseEstimator, TransformerMixin):
    """
    Scale projected points (participant or statements) based on sparsity of vote
    matrix, to account for any small number of votes by a participant and
    prevent those participants from bunching up in the center.

    Attributes:
        capture_step (str | int | None): Name or index of the capture step in the pipeline.
        X_sparse (np.ndarray | None): A sparse array with shape (n_features,)
    """
    def __init__(self, capture_step: Optional[str | int] = None, X_sparse: Optional[Array1D | Array2D] = None):
        self.capture_step = capture_step
        self.X_sparse = X_sparse

    # See: https://scikit-learn.org/stable/modules/generated/sklearn.utils.Tags.html#sklearn.utils.Tags
    def __sklearn_tags__(self):
        tags = super().__sklearn_tags__()
        # Suppresses warning caused by fit() not being required before usage in transform().
        tags.requires_fit = False
        return tags

    def fit(self, X, y=None):
        return self

    def transform(self, X):
        scaling_factors = self._calculate_scaling_factors()
        return X * scaling_factors[:, np.newaxis]

    def inverse_transform(self, X):
        scaling_factors = self._calculate_scaling_factors()
        return X / scaling_factors[:, np.newaxis]


    def _get_pipeline_step(self, step):
        """
        Fetch the parent pipeline when available via PatchedPipeline usage.
        """
        parent = getattr(self, "_parent_pipeline", None)
        if parent is None:
            raise RuntimeError(
                f"{self.__class__.__name__} cannot resolve `capture_step={step}` "
                "because it is not being used inside a `PatchedPipeline`. "
                "Either use a `PatchedPipeline` or pass `X_sparse` directly."
            )
        if isinstance(step, str):
            return parent.named_steps[step]
        elif isinstance(step, int):
            return parent.steps[step][1]
        else:
            raise ValueError("`capture_step` must be a string (name) or int (index).")

    def _resolve_X_sparse(self):
        """
        Resolve X_sparse (a sparse vote matrix) from argument or prior capture step.
        """
        if self.X_sparse is not None:
            return self.X_sparse

        capture = self._get_pipeline_step(self.capture_step)
        if not hasattr(capture, "X_captured_"):
            raise AttributeError(
                f"Step '{self.capture_step}' does not contain `.X_captured_`. "
                f"Did you run `fit/transform` on the pipeline?"
            )
        return capture.X_captured_

    def _calculate_scaling_factors(self):
        X_sparse = self._resolve_X_sparse()
        return calculate_scaling_factors(X_sparse=X_sparse)

class Loader:
    """
    A comprehensive data loader for Polis conversation data.

    The Loader class provides a unified interface for loading Polis conversation data
    from multiple sources including API endpoints, CSV exports, and local JSON files.
    It handles data validation, caching, rate limiting, and export functionality.

    The class automatically determines the appropriate loading strategy based on the
    provided parameters and can export data in both JSON and CSV formats compatible
    with the Polis platform.

    Args:
        polis_instance_url (str, optional): Base URL of the Polis instance. Defaults to "https://pol.is".
        filepaths (list[str], optional): List of local file paths to load data from. Defaults to [].
        polis_id (str, optional): Generic Polis identifier (report ID starting with 'r' or conversation ID).
        conversation_id (str, optional): Specific conversation ID for API requests.
        report_id (str, optional): Specific report ID for API requests or CSV exports.
        is_cache_enabled (bool, optional): Enable HTTP request caching. Defaults to True.
        output_dir (str, optional): Directory path for exporting loaded data. If provided, data is automatically exported.
        data_source (str, optional): Data source type ("api" or "csv_export"). Defaults to "api".
        directory_url (str, optional): Direct URL to CSV export directory. If provided, forces csv_export mode.

    Attributes:
        votes_data (list[dict]): Loaded vote data with participant_id, statement_id, vote, and modified fields.
        comments_data (list[dict]): Loaded statement/comment data with text, metadata, and statistics.
        math_data (dict): Mathematical analysis data including PCA projections and group clusters.
        conversation_data (dict): Conversation metadata including topic, description, and settings.
        report_data (dict): Report metadata when loaded via report_id.
        skipped_dup_votes (list[dict]): Duplicate votes that were filtered out during processing.

    Examples:
        Load from API using conversation ID:
        >>> loader = Loader(conversation_id="12345")

        Load from CSV export using report ID:
        >>> loader = Loader(report_id="r67890", data_source="csv_export")

        Load from local files:
        >>> loader = Loader(filepaths=["votes.json", "comments.json", "math-pca2.json"])

        Load and export to directory:
        >>> loader = Loader(conversation_id="12345", output_dir="./exported_data")
    """

    def __init__(
        self,
        polis_instance_url=None,
        filepaths=[],
        polis_id=None,
        conversation_id=None,
        report_id=None,
        is_cache_enabled=True,
        output_dir=None,
        data_source="api",
        directory_url=None,
    ):
        self.polis_instance_url = polis_instance_url or "https://pol.is"
        self.polis_id = report_id or conversation_id or polis_id
        self.conversation_id = conversation_id
        self.report_id = report_id
        self.is_cache_enabled = is_cache_enabled
        self.output_dir = output_dir
        self.data_source = data_source
        self.filepaths = filepaths
        self.directory_url = directory_url

        self.votes_data = []
        self.comments_data = []
        self.math_data = {}
        self.conversation_data = {}
        self.report_data = {}
        self.skipped_dup_votes = []

        if self.filepaths:
            self.load_file_data()
        elif (
            self.conversation_id
            or self.report_id
            or self.polis_id
            or self.directory_url
        ):
            self.populate_polis_ids()
            self.init_http_client()
            if self.directory_url:
                self.data_source = "csv_export"

            if self.data_source == "csv_export":
                self.load_remote_export_data()
            elif self.data_source == "api":
                self.load_api_data()
            else:
                raise ValueError("Unknown data_source: {}".format(self.data_source))

        if self.output_dir:
            self.dump_data(self.output_dir)

    def populate_polis_ids(self):
        """
        Normalize and populate Polis ID fields from the provided identifiers.

        This method handles the logic for determining conversation_id and report_id
        from the generic polis_id parameter. (Report IDs start with 'r', while
        conversation IDs start with a number.)
        """
        if self.polis_id:
            # If polis_id set, set report or conversation ID.
            if self.polis_id[0] == "r":
                self.report_id = self.polis_id
            elif self.polis_id[0].isdigit():
                self.conversation_id = self.polis_id
        else:
            # If not set, write it from what's provided.
            self.polis_id = self.report_id or self.conversation_id

    # Deprecated.
    def dump_data(self, output_dir):
        """
        Export loaded data to JSON files in the specified directory.

        Args:
            output_dir (str): Directory path where JSON files will be written.

        Note:
            This method is deprecated. Use export_data() instead.
        """
        self.export_data(output_dir, format="json")

    def export_data(self, output_dir, format="csv"):
        """
        Export loaded data to files in the specified format.

        Args:
            output_dir (str): Directory path where files will be written.
            format (str): Export format, either "json" or "csv". Defaults to "csv".

        The CSV format exports multiple files compatible with Polis platform:
        - votes.csv: Individual vote records
        - comments.csv: Statement/comment data with metadata
        - comment-groups.csv: Group-specific voting statistics per statement
        - participant-votes.csv: Participant voting patterns and group assignments
        - summary.csv: Conversation summary statistics
        """
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)

        if format == "json":
            self._export_data_json(output_dir)
        elif format == "csv":
            self._export_data_csv(output_dir)

    def _export_data_json(self, output_dir):
        if self.votes_data:
            with open(output_dir + "/votes.json", "w") as f:
                f.write(json.dumps(self.votes_data, indent=4))

        if self.comments_data:
            with open(output_dir + "/comments.json", "w") as f:
                f.write(json.dumps(self.comments_data, indent=4))

        if self.math_data:
            with open(output_dir + "/math-pca2.json", "w") as f:
                f.write(json.dumps(self.math_data, indent=4))

        if self.conversation_data:
            with open(output_dir + "/conversation.json", "w") as f:
                f.write(json.dumps(self.conversation_data, indent=4))

    def _export_data_csv(self, output_dir):
        self._write_polis_votes(output_dir)
        self._write_polis_comments(output_dir)
        self._write_polis_comment_groups(output_dir)
        self._write_polis_participant_votes(output_dir)
        self._write_polis_summary(output_dir)

    def _write_polis_votes(self, output_dir):
        """
        POLIS format:
            timestamp,datetime,comment-id,voter-id,vote
        """
        if not self.votes_data:
            return

        sorted_votes_data = sorted(
            self.votes_data, key=lambda x: (x["statement_id"], x["participant_id"])
        )
        with open(output_dir + "/votes.csv", "w") as f:
            writer = csv.writer(f)
            headers = ["timestamp", "datetime", "comment-id", "voter-id", "vote"]
            writer.writerow(headers)
            for entry in sorted_votes_data:
                ts, dt_str = self._format_polis_times(entry["modified"])
                row = [
                    ts,
                    dt_str,
                    entry["statement_id"],
                    entry["participant_id"],
                    entry["vote"],
                ]
                writer.writerow(row)

    def _write_polis_comments(self, output_dir):
        """
        POLIS format:
            timestamp,datetime,comment-id,author-id,agrees,disagrees,moderated,comment-body
        """
        if not self.comments_data:
            return

        with open(output_dir + "/comments.csv", "w") as f:
            headers = [
                "timestamp",
                "datetime",
                "comment-id",
                "author-id",
                "agrees",
                "disagrees",
                "moderated",
                "comment-body",
            ]
            f.write(",".join(headers) + "\n")
            # Sort comments_data by 'created' timestamp before writing
            sorted_comments = sorted(
                self.comments_data,
                key=lambda x: (x["statement_id"], x["participant_id"]),
            )
            for entry in sorted_comments:
                ts, dt_str = self._format_polis_times(entry["created"])
                single_quote = '"'
                double_quote = '""'
                row = [
                    ts,
                    dt_str,
                    entry["statement_id"],
                    entry["participant_id"],
                    entry["agree_count"],
                    entry["disagree_count"],
                    entry["moderated"],
                    f'"{str(entry["txt"]).replace(single_quote, double_quote)}"',
                ]
                f.write(",".join([str(item) for item in row]) + "\n")

    def _format_polis_times(self, time):
        """Convert timestamp or ISO string to Polis datetime format."""
        from dateutil import parser

        try:
            if isinstance(time, (int, float)):
                # Handle timestamps
                timestamp = int(str(time)[:10]) if time > 10**10 else time
                date_obj = datetime.fromtimestamp(timestamp, tz=timezone.utc)
            else:
                # Handle string inputs in ISO format
                date_obj = parser.parse(time)
                if date_obj.tzinfo is None:
                    date_obj = date_obj.replace(tzinfo=timezone.utc)

            date_obj = date_obj.astimezone(timezone.utc)
            dt_str = date_obj.strftime(
                "%a %b %d %Y %H:%M:%S GMT+0000 (Coordinated Universal Time)"
            )

            return int(date_obj.timestamp()), dt_str
        except (ValueError, OSError) as error:
            raise ValueError(f"Timestamp is not in a recognizable format: {error}")

    def _write_polis_comment_groups(self, output_dir):
        """
        POLIS format:
            comment-id,comment,total-votes,total-agrees,total-disagrees,total-passes,group-a-votes,group-a-agrees,group-a-disagrees,group-a-passes,group-[next alphabetic identifier (b)]-votes,[repeat 'votes/agrees/disagrees/passes' with alphabetic identifier...]

        Each row represents a comment with total votes & votes by group
        """
        if not self.comments_data or not self.math_data:
            return

        group_votes = self.math_data.get("group-votes", {})
        group_clusters = self.math_data.get("group-clusters", [])
        group_ids = [group["id"] for group in group_clusters]
        # Map group indices to letters: 0 -> 'a', 1 -> 'b', etc.
        group_letters = [chr(ord("a") + i) for i in range(len(group_ids))]

        with open(output_dir + "/comment-groups.csv", "w") as f:
            # Build header dynamically based on available groups
            header = [
                "comment-id",
                "comment",
                "total-votes",
                "total-agrees",
                "total-disagrees",
                "total-passes",
            ]
            for i, group in enumerate(group_clusters):
                if i < len(group_letters):
                    group_letter = group_letters[i]
                    header.extend(
                        [
                            f"group-{group_letter}-votes",
                            f"group-{group_letter}-agrees",
                            f"group-{group_letter}-disagrees",
                            f"group-{group_letter}-passes",
                        ]
                    )
            f.write(",".join(header))
            f.write("\n")
            rows = []
            sorted_comments_data = sorted(
                self.comments_data, key=lambda x: x["statement_id"]
            )
            for comment in sorted_comments_data:
                comment_id = str(comment["statement_id"])
                row = [
                    comment_id,
                    comment["txt"]
                    if comment["txt"][0] == '"'
                    else '"' + comment["txt"] + '"',
                    comment["count"],
                    comment["agree_count"],
                    comment["disagree_count"],
                    comment["pass_count"],
                ]

                # Add group-specific data
                for i, group in enumerate(group_clusters):
                    if i < len(group_letters):
                        group_id = str(group["id"])
                        if (
                            group_id in group_votes
                            and comment_id in group_votes[group_id]["votes"]
                        ):
                            vote_data = group_votes[group_id]["votes"][comment_id]
                            total_votes = (
                                vote_data["A"] + vote_data["D"] + vote_data["S"]
                            )
                            row.extend(
                                [
                                    total_votes,
                                    vote_data["A"],  # agrees
                                    vote_data["D"],  # disagrees
                                    vote_data["S"],  # passes (skips)
                                ]
                            )
                        else:
                            # No votes from this group for this comment
                            row.extend([0, 0, 0, 0])
                rows.append(row)
                f.write(",".join([str(item) for item in row]) + "\n")

    def _write_polis_participant_votes(self, output_dir):
        """
        POLIS format:
            participant,group-id,n-comments,n-votes,n-agree,n-disagree,0,1,2,3,...

        Each row represents a participant with:
        - participant: participant ID
        - group-id: which group they belong to (if any)
        - n-comments: number of comments they made
        - n-votes: total number of votes they cast
        - n-agree: number of agree votes
        - n-disagree: number of disagree votes
        - 0,1,2,3...: their vote on each comment (1=agree, -1=disagree, 0=pass, empty=no vote)
        """
        if not self.votes_data:
            return

        # Get all unique participant IDs and statement IDs
        participant_ids = set()
        statement_ids = set()
        for vote in self.votes_data:
            participant_ids.add(vote["participant_id"])
            statement_ids.add(vote["statement_id"])

        # Sort to ensure consistent order
        sorted_participant_ids = sorted(participant_ids)
        sorted_statement_ids = sorted(statement_ids)

        # Build participant vote matrix
        participant_votes = {}
        for vote in self.votes_data:
            pid = vote["participant_id"]
            sid = vote["statement_id"]
            if pid not in participant_votes:
                participant_votes[pid] = {}
            participant_votes[pid][sid] = vote["vote"]

        # Get participant group assignments from math data
        participant_groups = {}
        if self.math_data and "group-clusters" in self.math_data:
            for group in self.math_data["group-clusters"]:
                group_id = group["id"]
                for member in group["members"]:
                    participant_groups[member] = group_id

        # Count comments per participant
        participant_comment_counts = {}
        if self.comments_data:
            for comment in self.comments_data:
                pid = comment["participant_id"]
                participant_comment_counts[pid] = (
                    participant_comment_counts.get(pid, 0) + 1
                )

        with open(output_dir + "/participant-votes.csv", "w") as f:
            # Build header
            header = [
                "participant",
                "group-id",
                "n-comments",
                "n-votes",
                "n-agree",
                "n-disagree",
            ]
            header.extend([str(sid) for sid in sorted_statement_ids])
            f.write(",".join(header) + "\n")

            # Write participant data
            for pid in sorted_participant_ids:
                participant_vote_data = participant_votes.get(pid, {})

                # Count votes
                n_votes = len(participant_vote_data)
                n_agree = sum(1 for v in participant_vote_data.values() if v == 1)
                n_disagree = sum(1 for v in participant_vote_data.values() if v == -1)

                # Get group assignment
                group_id = participant_groups.get(pid, "")

                # Get comment count
                n_comments = participant_comment_counts.get(pid, 0)

                row = [pid, group_id, n_comments, n_votes, n_agree, n_disagree]

                # Add vote for each statement
                for sid in sorted_statement_ids:
                    vote = participant_vote_data.get(sid, "")
                    row.append(vote)

                f.write(",".join([str(item) for item in row]) + "\n")

    def _write_polis_summary(self, output_dir):
        """
        POLIS format:
            topic,[string]
            url,http://pol.is/[report_id]
            voters,[num]
            voters-in-conv,[num]
            commenters,[num]
            comments,[num]
            groups,[num]
            conversation-description,[string]
        """
        if not self.conversation_data:
            return

        # Calculate summary statistics
        total_voters = (
            len(set(vote["participant_id"] for vote in self.votes_data))
            if self.votes_data
            else 0
        )
        total_commenters = (
            len(set(comment["participant_id"] for comment in self.comments_data))
            if self.comments_data
            else 0
        )
        total_comments = len(self.comments_data) if self.comments_data else 0
        total_groups = (
            len(self.math_data.get("group-clusters", [])) if self.math_data else 0
        )

        # Get conversation details
        topic = self.conversation_data.get("topic", "")
        description = self.conversation_data.get("description", "")
        if description:
            description = (
                description.replace("\n", "\\n")
                .replace("\r", "\\r")
                .replace("\t", "\\t")
            )

        # Build URL
        url = (
            f"{self.polis_instance_url}/{self.conversation_id}"
            if self.conversation_id
            else self.polis_id
            if self.polis_id
            else self.report_id
        )

        with open(output_dir + "/summary.csv", "w") as f:
            f.write(f'topic,"{topic}"\n')
            f.write(f"url,{url}\n")
            f.write(f"voters,{total_voters}\n")
            f.write(f"voters-in-conv,{total_voters}\n")
            f.write(f"commenters,{total_commenters}\n")
            f.write(f"comments,{total_comments}\n")
            f.write(f"groups,{total_groups}\n")
            f.write(f'conversation-description,"{description}"\n')

    def init_http_client(self):
        """
        Initialize HTTP session with rate limiting, caching, and Cloudflare bypass.

        Sets up a requests session with:
        - Rate limiting (5 requests per second)
        - Optional SQLite-based response caching (1 hour expiration)
        - Cloudflare bypass adapter for the Polis instance
        - Random user agent headers
        """
        # Throttle requests, but disable when response is already cached.
        if self.is_cache_enabled:
            # Source: https://github.com/JWCook/requests-ratelimiter/tree/main?tab=readme-ov-file#custom-session-example-requests-cache
            self.session = CachedLimiterSession(
                per_second=5,
                expire_after=timedelta(hours=1),
                cache_name="test_cache.sqlite",
                bucket_class=SQLiteBucket,
                bucket_kwargs={
                    "path": "test_cache.sqlite",
                    "isolation_level": "EXCLUSIVE",
                    "check_same_thread": False,
                },
            )
        else:
            self.session = LimiterSession(per_second=5)
        adapter = CloudflareBypassHTTPAdapter()
        self.session.mount(self.polis_instance_url, adapter)
        self.session.headers = {
            "User-Agent": ua.random,
        }

    def get_polis_export_directory_url(self, report_id):
        """
        Generate the CSV export directory URL for a given report ID.

        Args:
            report_id (str): The report ID (typically starts with 'r').

        Returns:
            str: Full URL to the CSV export directory endpoint.
        """
        return f"{self.polis_instance_url}/api/v3/reportExport/{report_id}/"

    def _is_statement_meta_field_missing(self):
        if self.comments_data:
            return self.comments_data[0]["is_meta"] is None
        else:
            # No statements loaded, so can't say.
            return False

    def load_remote_export_data(self):
        """
        Load data from remote CSV export endpoints.

        Downloads and processes CSV files from Polis export directory, including:
        - comments.csv: Statement data
        - votes.csv: Vote records

        Handles missing is_meta field by falling back to API data when necessary.
        Automatically filters duplicate votes, keeping the most recent.

        Raises:
            ValueError: If CSV export URL cannot be determined or API fallback fails.
        """
        if self.directory_url:
            directory_url = self.directory_url
        elif self.report_id:
            directory_url = self.get_polis_export_directory_url(self.report_id)
        else:
            raise ValueError(
                "Cannot determine CSV export URL without report_id or directory_url"
            )

        self.load_remote_export_data_comments(directory_url)
        self.load_remote_export_data_votes(directory_url)

        # Supplement is_meta statement field via API if missing.
        # See: https://github.com/polis-community/red-dwarf/issues/55
        if self._is_statement_meta_field_missing():
            import warnings

            warnings.warn(
                "CSV import is missing is_meta field. Attempting to load comments data from API instead..."
            )
            try:
                if self.report_id and not self.conversation_id:
                    self.load_api_data_report()
                    self.conversation_id = self.report_data["conversation_id"]
                self.load_api_data_comments()
            except Exception:
                raise ValueError(
                    " ".join(
                        [
                            "Due to an upstream bug, we must patch CSV exports using the API,",
                            "so conversation_id or report_id is required.",
                            "See: https://github.com/polis-community/red-dwarf/issues/56",
                        ]
                    )
                )

        # When multiple votes (same tid and pid), keep only most recent (vs first).
        self.filter_duplicate_votes(keep="recent")
        # self.load_remote_export_data_summary()
        # self.load_remote_export_data_participant_votes()
        # self.load_remote_export_data_comment_groups()

    def load_remote_export_data_comments(self, directory_url):
        """
        Load statement/comment data from remote CSV export.

        Args:
            directory_url (str): Base URL of the CSV export directory.
        """
        r = self.session.get(directory_url + "comments.csv")
        comments_csv = r.text
        reader = csv.DictReader(StringIO(comments_csv))
        self.comments_data = [
            Statement(**c).model_dump(mode="json") for c in list(reader)
        ]

    def load_remote_export_data_votes(self, directory_url):
        """
        Load vote data from remote CSV export.

        Args:
            directory_url (str): Base URL of the CSV export directory.
        """
        r = self.session.get(directory_url + "votes.csv")
        votes_csv = r.text
        reader = csv.DictReader(StringIO(votes_csv))
        self.votes_data = [
            Vote(**vote).model_dump(mode="json") for vote in list(reader)
        ]

    def filter_duplicate_votes(self, keep="recent"):
        """
        Remove duplicate votes from the same participant on the same statement.

        Args:
            keep (str): Which vote to keep when duplicates found.
                       "recent" keeps the most recent vote, "first" keeps the earliest.

        The filtered duplicate votes are stored in self.skipped_dup_votes for reference.

        Raises:
            ValueError: If keep parameter is not "recent" or "first".
        """
        if keep not in {"recent", "first"}:
            raise ValueError("Invalid value for 'keep'. Use 'recent' or 'first'.")

        # Sort by modified time (descending for "recent", ascending for "first")
        if keep == "recent":
            reverse_sort = True
        else:
            reverse_sort = False
        sorted_votes = sorted(
            self.votes_data, key=lambda x: x["modified"], reverse=reverse_sort
        )

        filtered_dict = {}
        for v in sorted_votes:
            key = (v["participant_id"], v["statement_id"])
            if key not in filtered_dict:
                filtered_dict[key] = v
            else:
                # Append skipped votes
                self.skipped_dup_votes.append(v)

        self.votes_data = list(filtered_dict.values())

    def load_remote_export_data_summary(self):
        # r = self.session.get(self.polis_instance_url + "/api/v3/reportExport/{}/summary.csv".format(self.report_id))
        # summary_csv = r.text
        # print(summary_csv)
        raise NotImplementedError

    def load_remote_export_data_participant_votes(self):
        # r = self.session.get(self.polis_instance_url + "/api/v3/reportExport/{}/participant-votes.csv".format(self.report_id))
        # participant_votes_csv = r.text
        # print(participant_votes_csv)
        raise NotImplementedError

    def load_remote_export_data_comment_groups(self):
        # r = self.session.get(self.polis_instance_url + "/api/v3/reportExport/{}/comment-groups.csv".format(self.report_id))
        # comment_groups_csv = r.text
        # print(comment_groups_csv)
        raise NotImplementedError

    def load_file_data(self):
        """
        Load data from local JSON files specified in self.filepaths.

        Automatically detects file types based on filename patterns:
        - votes.json: Vote records
        - comments.json: Statement/comment data
        - conversation.json: Conversation metadata
        - math-pca2.json: Mathematical analysis results

        Raises:
            ValueError: If a file type cannot be determined from its name.
        """
        for f in self.filepaths:
            if f.endswith("votes.json"):
                self.load_file_data_votes(file=f)
            elif f.endswith("comments.json"):
                self.load_file_data_comments(file=f)
            elif f.endswith("conversation.json"):
                self.load_file_data_conversation(file=f)
            elif f.endswith("math-pca2.json"):
                self.load_file_data_math(file=f)
            else:
                raise ValueError("Unknown file type")

    def load_file_data_votes(self, file=None):
        """
        Load vote data from a local JSON file.

        Args:
            file (str): Path to the votes JSON file.
        """
        with open(file) as f:
            votes_data = json.load(f)

        votes_data = [Vote(**vote).model_dump(mode="json") for vote in votes_data]
        self.votes_data = votes_data

    def load_file_data_comments(self, file=None):
        """
        Load statement/comment data from a local JSON file.

        Args:
            file (str): Path to the comments JSON file.
        """
        with open(file) as f:
            comments_data = json.load(f)

        comments_data = [Statement(**c).model_dump(mode="json") for c in comments_data]
        self.comments_data = comments_data

    def load_file_data_conversation(self, file=None):
        """
        Load conversation metadata from a local JSON file.

        Args:
            file (str): Path to the conversation JSON file.
        """
        with open(file) as f:
            convo_data = json.load(f)

        self.conversation_data = convo_data

    def load_file_data_math(self, file=None):
        """
        Load mathematical analysis data from a local JSON file.

        Args:
            file (str): Path to the math-pca2 JSON file.
        """
        with open(file) as f:
            math_data = json.load(f)

        self.math_data = math_data

    def load_api_data(self):
        """
        Load complete dataset from Polis API endpoints.

        Loads data in the following order:
        1. Report data (if report_id provided) to get conversation_id
        2. Conversation metadata
        3. Comments/statements data
        4. Mathematical analysis data (PCA, clustering)
        5. Individual participant votes (up to participant count from math data)

        Automatically handles vote sign correction for API data and resolves
        any conflicts between report_id and conversation_id parameters.

        Raises:
            ValueError: If report_id conflicts with conversation_id.
        """
        if self.report_id:
            self.load_api_data_report()
            convo_id_from_report_id = self.report_data["conversation_id"]
            if self.conversation_id and (
                self.conversation_id != convo_id_from_report_id
            ):
                raise ValueError("report_id conflicts with conversation_id")
            self.conversation_id = convo_id_from_report_id

        self.load_api_data_conversation()
        self.load_api_data_comments()
        self.load_api_data_math()
        # TODO: Add a way to do this without math data, for example
        # by checking until 5 empty responses in a row.
        # This is the best place to check though, as `voters`
        # in summary.csv omits some participants.
        participant_count = self.math_data["n"]
        # DANGER: This is potentially an issue that throws everything off by missing some participants.
        self.load_api_data_votes(last_participant_id=participant_count)

    def load_api_data_report(self):
        """
        Load report metadata from the Polis API.

        Uses the report_id to fetch report information and extract the associated
        conversation_id for subsequent API calls.
        """
        params = {
            "report_id": self.report_id,
        }
        r = self.session.get(self.polis_instance_url + "/api/v3/reports", params=params)
        reports = json.loads(r.text)
        self.report_data = reports[0]

    def load_api_data_conversation(self):
        """
        Load conversation metadata from the Polis API.

        Fetches conversation details including topic, description, and settings
        using the conversation_id.
        """
        params = {
            "conversation_id": self.conversation_id,
        }
        r = self.session.get(
            self.polis_instance_url + "/api/v3/conversations", params=params
        )
        convo = json.loads(r.text)
        self.conversation_data = convo

    def load_api_data_math(self):
        """
        Load mathematical analysis data from the Polis API.

        Fetches PCA projections, clustering results, and group statistics
        from the math/pca2 endpoint.
        """
        params = {
            "conversation_id": self.conversation_id,
        }
        r = self.session.get(
            self.polis_instance_url + "/api/v3/math/pca2", params=params
        )
        math = json.loads(r.text)
        self.math_data = math

    def load_api_data_comments(self):
        """
        Load statement/comment data from the Polis API.

        Fetches all statements with moderation status and voting patterns
        included in the response.
        """
        params = {
            "conversation_id": self.conversation_id,
            "moderation": "true",
            "include_voting_patterns": "true",
        }
        r = self.session.get(
            self.polis_instance_url + "/api/v3/comments", params=params
        )
        comments = json.loads(r.text)
        comments = [Statement(**c).model_dump(mode="json") for c in comments]
        self.comments_data = comments

    def fix_participant_vote_sign(self):
        """
        Correct vote sign inversion in API data.

        The Polis API returns votes with inverted signs compared to the expected
        format (e.g., agree votes come as -1 instead of 1). This method fixes
        the inversion by negating all vote values.
        """
        """For data coming from the API, vote signs are inverted (e.g., agree is -1)"""
        for item in self.votes_data:
            item["vote"] = -item["vote"]

    def load_api_data_votes(self, last_participant_id=None):
        """
        Load individual participant votes from the Polis API.

        Args:
            last_participant_id (int): Maximum participant ID to fetch votes for.
                                     Typically obtained from math data participant count.

        Iterates through all participant IDs from 0 to last_participant_id and
        fetches their vote records. Automatically applies vote sign correction.
        """
        for pid in range(0, last_participant_id + 1):
            params = {
                "pid": pid,
                "conversation_id": self.conversation_id,
            }
            r = self.session.get(
                self.polis_instance_url + "/api/v3/votes", params=params
            )
            participant_votes = json.loads(r.text)
            participant_votes = [
                Vote(**vote).model_dump(mode="json") for vote in participant_votes
            ]
            self.votes_data.extend(participant_votes)

        self.fix_participant_vote_sign()

    def fetch_pid(self, xid):
        """
        Fetch internal participant ID (pid) for a given external ID (xid).

        Args:
            xid (str): External participant identifier.

        Returns:
            int: Internal participant ID used by Polis system.
        """
        params = {
            "pid": "mypid",
            "xid": xid,
            "conversation_id": self.conversation_id,
        }
        r = self.session.get(
            self.polis_instance_url + "/api/v3/participationInit", params=params
        )
        data = json.loads(r.text)

        return data["ptpt"]["pid"]

    def fetch_xid_to_pid_mappings(self, xids=[]):
        """
        Create mapping dictionary from external IDs to internal participant IDs.

        Args:
            xids (list[str]): List of external participant identifiers.

        Returns:
            dict: Mapping of external IDs to internal participant IDs.
        """
        mappings = {}
        for xid in xids:
            pid = self.fetch_pid(xid)
            mappings[xid] = pid

        return mappings