Approved RPGF3 projects on EAS

RPGF3_projects_on_EAS.py

import json
import os
import requests


PROJECT_APP_SCHEMA    = "0x76e98cce95f3ba992c2ee25cef25f756495147608a3da3aa2e5ca43109fe77cc"
PROJECT_REVIEW_SCHEMA = "0xebbf697d5d3ca4b53579917ffc3597fb8d1a85b8c6ca10ec10039709903b9277"
REVIEWER_ADDRESS      = "0x621477dBA416E12df7FF0d48E14c4D20DC85D7D9"
DATA_EXPORT_JSON      = "rpgf3_applicant_data.json"


def fetch_attestations(schema_id, time_created_after=0):
    """
    Generalized function to fetch attestations for a given schema ID.
    """

    url = 'https://optimism.easscan.org/graphql'
    query_limit = 100

    query = '''
    query Attestations($schemaId: StringFilter!, $skip: Int!, $take: Int!, $timeCreatedAfter: IntFilter) {
        attestations(where: {schemaId: $schemaId, timeCreated: $timeCreatedAfter}, take: $take, skip: $skip) {
            id
            attester
            recipient
            refUID
            revocable
            revocationTime
            expirationTime
            timeCreated 
            decodedDataJson    
        }
    }
    '''
    
    variables = {
        "schemaId": {
            "equals": schema_id
        },
        "skip": 0,
        "take": query_limit,
        "timeCreatedAfter": {
            "gt": time_created_after
        },
    }

    headers = {
        'Content-Type': 'application/json',
    }

    all_attestations = []

    while True:
        payload = {
            'query': query,
            'variables': variables
        }

        try:
            response = requests.post(url, headers=headers, data=json.dumps(payload))
            response.raise_for_status()

            data = response.json()
            attestations = data.get('data', {}).get('attestations', [])
            all_attestations.extend(attestations)

            if len(attestations) < query_limit:
                break

            variables["skip"] += query_limit

        except (requests.exceptions.RequestException, json.JSONDecodeError) as e:
            print(f"Failed to fetch attestations for {schema_id}: {str(e)}")
            break

    print(f"Total attestations for Schema ID {schema_id}: {len(all_attestations)}")
    return all_attestations


def fetch_project_applications(app_schema_id):
    """
    Fetch all project applications from EAS and decode the data.
    """

    project_apps = fetch_attestations(app_schema_id)
    project_data = []
    for a in project_apps:
        decoded_data = json.loads(a["decodedDataJson"])
        project_data.append({
            "id": a["id"],
            "attester": a["attester"],
            "timeCreated": a["timeCreated"],
            "data": decoded_data
        })
    return project_data


def fetch_approved_project_ids(approval_schema_id, approver_address):
    """
    Fetch all approved project IDs from EAS.
    """

    approved_projects = fetch_attestations(approval_schema_id)
    approved_project_ids = []
    rejected_project_ids = []
    for a in approved_projects:
        if a["attester"] != approver_address:
            continue
        data = json.loads(a["decodedDataJson"])
        if data[0]['value']['value'] == True:
            approved_project_ids.append(a["refUID"])
        else:
            rejected_project_ids.append(a["refUID"])
    for rejected_id in rejected_project_ids:        
        if rejected_id in approved_project_ids:
            approved_project_ids.remove(rejected_id)
    return approved_project_ids


def fetch_json_data(url):
    """
    Fetch JSON data from a URL.
    """
 
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.json()
    except (requests.exceptions.RequestException, json.JSONDecodeError) as e:
        print(f"Error fetching JSON data from URL: {url}. Error: {str(e)}")
        return None


def request_json_data(project_attestations):
    """
    Request and insert the JSON data for each project application.
    """

    for a in project_attestations:
        url = a["data"][2]["value"]["value"]
        if isinstance(url, str) and ".json" in url:
            json_data = fetch_json_data(url)
            if json_data:
                a.update({"json_data": json_data})


def main():

    project_apps = fetch_project_applications(PROJECT_APP_SCHEMA)
    approved_project_ids = fetch_approved_project_ids(PROJECT_REVIEW_SCHEMA, REVIEWER_ADDRESS)
    approved_apps = [project for project in project_apps if project["id"] in approved_project_ids]
    print(f"\nTotal approved projects: {len(approved_apps)}")

    print("\nRequesting JSON data for approved projects (this may take a minute)...")
    request_json_data(approved_apps)

    with open(DATA_EXPORT_JSON, "w") as json_file:
        json.dump(approved_apps, json_file, indent=4)
    print(f"\nExported data to {DATA_EXPORT_JSON}.")


if __name__ == "__main__":
    main()