Task: gitlab.list-info
Goal: Tell me the email address of the contributor who has the most commits to branch main in repo CellularPrivacy/Android-IMSI-Catcher-Detector
Task configuration:
attribute: email address
namespace: CellularPrivacy
project_name: Android-IMSI-Catcher-Detector
branch_name: main

Validation code:
from typing import Any, Dict, List
from urllib.parse import quote

import requests
import json
import os


def _get_headers():
    return {
        "Content-Type": "application/json",
        "Accept": "application/json",
        **json.loads(os.getenv("GITLAB_EXTRA_HTTP_HEADERS", "{}")),
    }


def _get_project_id(config):
    return quote(
        f"{config['namespace']}/{config['project_name']}", safe=""
    ).replace(".", "%2E")


def _get_contributors(project_id, ref=None, per_page=100):
    base_url = os.getenv("GITLAB_BASE_URL", "").rstrip("/")
    params = {"order_by": "commits", "sort": "desc", "per_page": per_page}
    if ref:
        params["ref"] = ref
    url = base_url + "/api/v4/projects/" + project_id + "/repository/contributors"
    response = requests.get(url=url, headers=_get_headers(), params=params)
    if response.status_code != 200:
        return []
    return response.json()


def _search_user(query):
    """Search for a GitLab user by name or email."""
    base_url = os.getenv("GITLAB_BASE_URL", "").rstrip("/")
    response = requests.get(
        base_url + "/api/v4/users",
        headers=_get_headers(),
        params={"search": query},
    )
    if response.status_code != 200 or not response.json():
        return None
    return response.json()[0]


def _get_user_profile(user_id):
    base_url = os.getenv("GITLAB_BASE_URL", "").rstrip("/")
    response = requests.get(
        base_url + f"/api/v4/users/{user_id}",
        headers=_get_headers(),
    )
    if response.status_code != 200:
        return None
    return response.json()


def _check_contributor_info(content_lower, contributor, fields):
    """Check if the agent's output contains the requested fields for a contributor."""
    name = contributor.get("name", "")
    email = contributor.get("email", "")
    commits = contributor.get("commits", 0)

    for field in fields:
        field_lower = field.lower()
        if "email" in field_lower:
            if email.lower() not in content_lower:
                return False
        elif "last name" in field_lower:
            parts = name.strip().split()
            if len(parts) > 1:
                last_name = parts[-1].lower()
                if last_name not in content_lower:
                    return False
        elif "commit" in field_lower:
            if str(commits) not in content_lower:
                return False
        elif "name" in field_lower:
            if name.lower() not in content_lower:
                return False

    return True


def _check_user_profile_info(content_lower, user_profile, fields):
    """Check profile-level attributes (followers, location, username)."""
    for field in fields:
        field_lower = field.lower()
        if "follower" in field_lower:
            followers = str(user_profile.get("followers", ""))
            if followers not in content_lower:
                return False
        if "location" in field_lower:
            location = (user_profile.get("location") or "").lower()
            if location and location not in content_lower:
                return False
        if "account name" in field_lower or "username" in field_lower:
            username = user_profile.get("username", "").lower()
            if username not in content_lower:
                return False

    return True


def _validate_top_contributors(content, config):
    """314: List attributes of top 3 contributors to a repo."""
    project_id = _get_project_id(config)
    contributors = _get_contributors(project_id, per_page=3)
    if not contributors:
        return 0

    top3 = contributors[:3]
    fields_str = config.get("fields", config.get("attribute", "name"))
    fields = [f.strip() for f in fields_str.replace(" and ", ", ").split(",")]

    content_lower = content.lower()
    for contributor in top3:
        if not _check_contributor_info(content_lower, contributor, fields):
            return 0

    return 1


def _validate_project_members(content, config):
    """349: List members who have access to my repo."""
    project_id = _get_project_id(config)
    base_url = os.getenv("GITLAB_BASE_URL", "").rstrip("/")

    # Get all members (including inherited)
    members = []
    page = 1
    while True:
        response = requests.get(
            base_url + "/api/v4/projects/" + project_id + "/members/all",
            headers=_get_headers(),
            params={"page": page, "per_page": 100},
        )
        if response.status_code != 200:
            return 0
        result = response.json()
        members.extend(result)
        if len(result) < 100:
            break
        page += 1

    if not members:
        return 0

    # "Who else" — exclude the current user
    current_user_url = base_url + "/api/v4/user"
    current_resp = requests.get(current_user_url, headers=_get_headers())
    current_username = ""
    if current_resp.status_code == 200:
        current_username = current_resp.json().get("username", "").lower()

    other_members = [
        m for m in members if m["username"].lower() != current_username
    ]

    if not other_members:
        return 0

    content_lower = content.lower()
    for member in other_members:
        if member["username"].lower() not in content_lower:
            return 0

    return 1


def _validate_top_branch_contributor(content, config):
    """784: Tell me attributes of the top contributor to a specific branch."""
    project_id = _get_project_id(config)
    branch = config["branch_name"]
    contributors = _get_contributors(project_id, ref=branch, per_page=1)
    if not contributors:
        return 0

    top = contributors[0]
    attr_str = config["attribute"]
    fields = [f.strip() for f in attr_str.replace(" and ", ", ").split(",")]

    content_lower = content.lower()

    # Check contributor-level info (name, email, commits)
    if not _check_contributor_info(content_lower, top, fields):
        return 0

    # Check profile-level info (followers, location, username) if needed
    needs_profile = any(
        kw in attr_str.lower()
        for kw in ["follower", "location", "account name", "username"]
    )
    if needs_profile:
        # Try to find the user by name or email
        user = _search_user(top["name"]) or _search_user(top["email"])
        if not user:
            return 0
        profile = _get_user_profile(user["id"])
        if not profile:
            return 0
        if not _check_user_profile_info(content_lower, profile, fields):
            return 0

    return 1


def validate(messages: List[Dict[str, str]], config: Dict[str, Any]) -> int:

    if not messages:
        raise ValueError("Messages cannot be empty")

    if "namespace" not in config or "project_name" not in config:
        raise ValueError("Target must contain 'namespace' and 'project_name'")

    content = messages[-1].get("content", "")

    if "fields" in config:
        return _validate_top_contributors(content, config)
    elif "attribute" in config and "branch_name" in config:
        return _validate_top_branch_contributor(content, config)
    else:
        return _validate_project_members(content, config)