refine

msr.py

@@ -217,6 +217,7 @@ def get_duckdb_connection():
 def generate_file_path_patterns(start_date, end_date, data_dir=GHARCHIVE_DATA_DIR):
     """
     Generate file path patterns for GHArchive data in date range.
+    Only includes files that actually exist on disk.
 
     Args:
         start_date: Start datetime
@@ -224,21 +225,35 @@ def generate_file_path_patterns(start_date, end_date, data_dir=GHARCHIVE_DATA_DI
         data_dir: Directory containing GHArchive data files
 
     Returns:
-        List of file path patterns (one per day)
+        List of file path patterns (hourly JSON.gz files) that exist
     """
     file_patterns = []
+    missing_dates = set()
 
     current_date = start_date.replace(hour=0, minute=0, second=0, microsecond=0)
     end_day = end_date.replace(hour=0, minute=0, second=0, microsecond=0)
 
     while current_date <= end_day:
-        # Pattern for daily parquet file: 2024-11-15.parquet
-        pattern = os.path.join(data_dir, f"{current_date.strftime('%Y-%m-%d')}.parquet")
-        file_patterns.append(pattern)
+        # Pattern for hourly JSON.gz files: 2024-11-15-{0..23}.json.gz
+        date_has_files = False
+        for hour in range(24):
+            pattern = os.path.join(data_dir, f"{current_date.strftime('%Y-%m-%d')}-{hour}.json.gz")
+            # Only add pattern if file exists
+            if os.path.exists(pattern):
+                file_patterns.append(pattern)
+                date_has_files = True
+
+        # Track missing dates
+        if not date_has_files:
+            missing_dates.add(current_date.strftime('%Y-%m-%d'))
 
         # Move to next day
         current_date += timedelta(days=1)
 
+    # Print warning about missing dates
+    if missing_dates:
+        print(f"   Warning: Skipping {len(missing_dates)} date(s) with no data files: {', '.join(sorted(missing_dates))}")
+
     return file_patterns
 
 
@@ -288,72 +303,72 @@ def fetch_all_pr_metadata_single_query(conn, identifiers, start_date, end_date):
     # Build identifier list for IN clause
     identifier_list = ', '.join([f"'{id}'" for id in identifiers])
 
-    # Build comprehensive query with CTEs using parameterized file lists (Parquet optimized)
+    # Build comprehensive query with CTEs using parameterized file lists (JSON.gz format)
     query = f"""
     WITH review_events AS (
       -- Get all review events for ALL agents
       SELECT
-        payload.pull_request.html_url as url,
+        TRY_CAST(json_extract_string(payload, '$.pull_request.html_url') AS VARCHAR) as url,
         COALESCE(
-          payload.review.submitted_at,
-          CAST(created_at AS VARCHAR)
+          TRY_CAST(json_extract_string(payload, '$.review.submitted_at') AS VARCHAR),
+          TRY_CAST(created_at AS VARCHAR)
         ) as reviewed_at,
-        actor.login as reviewer,
-        repo.name as repo_name,
-        CAST(payload.pull_request.number AS INTEGER) as pr_number
-      FROM read_parquet($review_patterns, union_by_name=true, filename=true)
+        TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) as reviewer,
+        TRY_CAST(json_extract_string(repo, '$.name') AS VARCHAR) as repo_name,
+        TRY_CAST(json_extract_string(payload, '$.pull_request.number') AS INTEGER) as pr_number
+      FROM read_json($review_patterns, union_by_name=true, filename=true, compression='gzip', format='newline_delimited', ignore_errors=true, maximum_object_size=2147483648)
       WHERE
-        type = 'PullRequestReviewEvent'
-        AND actor.login IN ({identifier_list})
-        AND payload.pull_request.html_url IS NOT NULL
+        TRY_CAST(type AS VARCHAR) = 'PullRequestReviewEvent'
+        AND TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) IN ({identifier_list})
+        AND json_extract_string(payload, '$.pull_request.html_url') IS NOT NULL
 
       UNION ALL
 
       -- Get PR comments (IssueCommentEvent on PRs)
       SELECT
-        payload.issue.html_url as url,
-        CAST(created_at AS VARCHAR) as reviewed_at,
-        actor.login as reviewer,
-        repo.name as repo_name,
-        CAST(payload.issue.number AS INTEGER) as pr_number
-      FROM read_parquet($review_patterns, union_by_name=true, filename=true)
+        TRY_CAST(json_extract_string(payload, '$.issue.html_url') AS VARCHAR) as url,
+        TRY_CAST(created_at AS VARCHAR) as reviewed_at,
+        TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) as reviewer,
+        TRY_CAST(json_extract_string(repo, '$.name') AS VARCHAR) as repo_name,
+        TRY_CAST(json_extract_string(payload, '$.issue.number') AS INTEGER) as pr_number
+      FROM read_json($review_patterns, union_by_name=true, filename=true, compression='gzip', format='newline_delimited', ignore_errors=true, maximum_object_size=2147483648)
       WHERE
-        type = 'IssueCommentEvent'
-        AND actor.login IN ({identifier_list})
-        AND payload.issue.pull_request.url IS NOT NULL
-        AND payload.issue.html_url IS NOT NULL
+        TRY_CAST(type AS VARCHAR) = 'IssueCommentEvent'
+        AND TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) IN ({identifier_list})
+        AND json_extract_string(payload, '$.issue.pull_request.url') IS NOT NULL
+        AND json_extract_string(payload, '$.issue.html_url') IS NOT NULL
 
       UNION ALL
 
       -- Get review comments (PullRequestReviewCommentEvent)
       SELECT
-        payload.pull_request.html_url as url,
-        CAST(created_at AS VARCHAR) as reviewed_at,
-        actor.login as reviewer,
-        repo.name as repo_name,
-        CAST(payload.pull_request.number AS INTEGER) as pr_number
-      FROM read_parquet($review_patterns, union_by_name=true, filename=true)
+        TRY_CAST(json_extract_string(payload, '$.pull_request.html_url') AS VARCHAR) as url,
+        TRY_CAST(created_at AS VARCHAR) as reviewed_at,
+        TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) as reviewer,
+        TRY_CAST(json_extract_string(repo, '$.name') AS VARCHAR) as repo_name,
+        TRY_CAST(json_extract_string(payload, '$.pull_request.number') AS INTEGER) as pr_number
+      FROM read_json($review_patterns, union_by_name=true, filename=true, compression='gzip', format='newline_delimited', ignore_errors=true, maximum_object_size=2147483648)
       WHERE
-        type = 'PullRequestReviewCommentEvent'
-        AND actor.login IN ({identifier_list})
-        AND payload.pull_request.html_url IS NOT NULL
+        TRY_CAST(type AS VARCHAR) = 'PullRequestReviewCommentEvent'
+        AND TRY_CAST(json_extract_string(actor, '$.login') AS VARCHAR) IN ({identifier_list})
+        AND json_extract_string(payload, '$.pull_request.html_url') IS NOT NULL
     ),
 
     pr_status AS (
       -- Get merge/close status for those PRs
       SELECT
-        payload.pull_request.html_url as url,
-        CAST(payload.pull_request.merged AS BOOLEAN) as is_merged,
-        payload.pull_request.merged_at as merged_at,
-        payload.pull_request.closed_at as closed_at,
+        TRY_CAST(json_extract_string(payload, '$.pull_request.html_url') AS VARCHAR) as url,
+        TRY_CAST(json_extract_string(payload, '$.pull_request.merged') AS BOOLEAN) as is_merged,
+        TRY_CAST(json_extract_string(payload, '$.pull_request.merged_at') AS VARCHAR) as merged_at,
+        TRY_CAST(json_extract_string(payload, '$.pull_request.closed_at') AS VARCHAR) as closed_at,
         created_at,
-        ROW_NUMBER() OVER (PARTITION BY payload.pull_request.html_url ORDER BY created_at DESC) as rn
-      FROM read_parquet($status_patterns, union_by_name=true, filename=true)
+        ROW_NUMBER() OVER (PARTITION BY json_extract_string(payload, '$.pull_request.html_url') ORDER BY created_at DESC) as rn
+      FROM read_json($status_patterns, union_by_name=true, filename=true, compression='gzip', format='newline_delimited', ignore_errors=true, maximum_object_size=2147483648)
       WHERE
-        type = 'PullRequestEvent'
-        AND payload.action = 'closed'
-        AND payload.pull_request.html_url IS NOT NULL
-        AND payload.pull_request.html_url IN (
+        TRY_CAST(type AS VARCHAR) = 'PullRequestEvent'
+        AND TRY_CAST(json_extract_string(payload, '$.action') AS VARCHAR) = 'closed'
+        AND json_extract_string(payload, '$.pull_request.html_url') IS NOT NULL
+        AND json_extract_string(payload, '$.pull_request.html_url') IN (
           SELECT DISTINCT url FROM review_events
         )
     )
@@ -603,11 +618,9 @@ def batch_upload_review_metadata(all_metadata):
                         commit_message=f"Update {repo_path}",
                         max_retries=MAX_RETRIES
                     ):
-                        print(" ")
                         agent_success += 1
                         success_count += 1
                     else:
-                        print(" ")
                         agent_error += 1
                         error_count += 1
 
@@ -927,7 +940,6 @@ def save_leaderboard_data_to_hf(leaderboard_dict, monthly_metrics):
                 repo_id=LEADERBOARD_REPO,
                 repo_type="dataset"
             )
-            print(" ")
             print(f"Saved leaderboard data to HuggingFace: {filename}")
             return True
         finally:
@@ -936,7 +948,6 @@ def save_leaderboard_data_to_hf(leaderboard_dict, monthly_metrics):
                 os.remove(filename)
 
     except Exception as e:
-        print(f" ")
         print(f"Error saving leaderboard data: {str(e)}")
         import traceback
         traceback.print_exc()