ChangelogUtils¶

@staticmethod
def get_changelogs_since_last_version(
    all_changelogs: List[Dict[str, Any]], versions: List[Any]
) -> List[Dict[str, Any]]:
    """Retrieves changelogs created after the most recent app version."""
    if not versions:
        print("No versions found. Returning all changelogs.")
        return all_changelogs
    try:
        valid_versions = [
            v for v in versions if isinstance(v, dict) and "createTime" in v
        ]
        if not valid_versions:
            print(
                "No valid versions with createTime found. Returning all "
                "changelogs."
            )
            return all_changelogs

        latest_version = max(
            valid_versions, key=lambda version: version["createTime"]
        )
        latest_version_timestamp_str = latest_version["createTime"]
        print(
            f"Latest version was created at: {latest_version_timestamp_str}"
        )
        latest_version_dt = datetime.datetime.fromisoformat(
            latest_version_timestamp_str.replace("Z", "+00:00")
        )
        recent_changelogs = [
            cl
            for cl in all_changelogs
            if isinstance(cl, dict)
            and "createTime" in cl
            and datetime.datetime.fromisoformat(
                cl["createTime"].replace("Z", "+00:00")
            )
            > latest_version_dt
        ]
        return recent_changelogs
    except (ValueError, KeyError, TypeError) as e:
        print(
            f"Error processing version or changelog timestamps: {e}. "
            f"Returning all changelogs."
        )
        return all_changelogs

@staticmethod
def summarize_changelogs(
    vertex_client_or_project: Any,
    changelogs: List[Dict[str, Any]],
    project_id: str = None,
) -> str:
    """Summarizes each non-evaluation changelog into a simple, specific
    one-liner."""
    resource_types_to_exclude = [
        "Version",
        "AppVersion",
        "Evaluation",
        "EvaluationRun",
    ]
    filtered_changelogs = [
        cl
        for cl in changelogs
        if cl.get("resourceType") not in resource_types_to_exclude
    ]

    if not filtered_changelogs:
        return "No user-facing changes to summarize."

    # Format each changelog entry, potentially producing multi-line
    # strings for updates
    formatted_log_entries = [
        ChangelogUtils._format_changelog_for_prompt(cl)
        for cl in filtered_changelogs
    ]

    # Combine and number the entries for the prompt
    changelog_context = ""
    entry_number = 1
    for entry in formatted_log_entries:
        # Add numbering only to the start of each logical entry
        lines = entry.strip().split("\n")
        changelog_context += f"{entry_number}. {lines[0]}\n"
        if len(lines) > 1:
            changelog_context += (
                "\n".join([f"   {line}" for line in lines[1:]]) + "\n"
            )  # Indent Original/New
        entry_number += 1

    if (
        not changelog_context.strip()
    ):  # Check if context became empty after formatting
        return "No user-facing changes to summarize."

    prompt = f"""
    You are an AI assistant that analyzes technical log entries,
    specifically focusing on 'Update' actions by comparing 'Original' and
    'New' configuration snippets. Your goal is to generate a concise,
    single-line summary describing the *exact* change that occurred for
    each entry.

    Rules:
    - **CRITICAL**: Provide one summary line for each numbered entry in
      the input. Maintain a 1-to-1 correspondence.
    - For 'Update' entries with 'Original' and 'New' snippets: Compare
      them carefully to identify the precise difference. Describe only
      that specific change (e.g., "Disabled barge-in", "Added variable
      'X'", "Updated agent instructions", "Added tool 'Y'").
    - For 'Create' or 'Delete' entries (or 'Update' entries without
      Original/New comparison data): Generate a summary based on the
      Action, ResourceType, Name, and Description provided (e.g.,
      "Created tool 'get_weather'", "Deleted agent 'old_agent'").
    - Be specific. Avoid generic phrases like "Updated settings" or
      "Changed configuration". State *what* was updated.
    - The final output must be a bulleted list, starting each line with
      '-'.

    Here is the raw changelog data to summarize:
    ---
    {changelog_context}
    ---

    Provide the specific one-line summary for each numbered entry:
    """

    try:
        # Handle if the user passes the vertex framework client or strings

        if isinstance(vertex_client_or_project, GeminiGenerate):
            response_text = vertex_client_or_project.generate(
                prompt=prompt, model_name="gemini-2.5-flash"
            )
        elif hasattr(vertex_client_or_project, "models"):
            response = vertex_client_or_project.models.generate_content(
                model="gemini-2.5-flash", contents=prompt
            )
            response_text = response.text
        else:
            cl = GeminiGenerate(
                project_id=project_id,
                location="us-central1",
                model_name="gemini-2.5-flash",
            )
            response_text = cl.generate(prompt=prompt)

        # Basic post-processing to clean up potential numbering/extra
        # whitespace
        lines = response_text.strip().split("\n") if response_text else []
        cleaned_lines = [
            line.strip() for line in lines if line.strip().startswith("-")
        ]
        return "\n".join(cleaned_lines)
    except Exception as e:
        return f"An error occurred while generating the summary: {e}"