refine

msr.py

@@ -397,7 +397,9 @@ def fetch_all_pr_metadata_streaming(conn, identifiers, start_date, end_date):
         file_patterns_sql = '[' + ', '.join([f"'{fp}'" for fp in file_patterns]) + ']'
         
         # Query for this batch
-        # Note: For PullRequestReviewEvent, we use the actor as author
+        # We need both opened and closed events:
+        # - opened events: to identify PRs created within the time frame
+        # - closed events: to determine if/when those PRs were merged
         query = f"""
         SELECT DISTINCT
             CONCAT(
@@ -405,6 +407,7 @@ def fetch_all_pr_metadata_streaming(conn, identifiers, start_date, end_date):
                 '/pull/',
                 CAST(payload.pull_request.number AS VARCHAR)
             ) as url,
+            TRY_CAST(json_extract_string(to_json(payload), '$.action') AS VARCHAR) as action,
             TRY_CAST(json_extract_string(to_json(payload), '$.pull_request.user.login') AS VARCHAR) as pr_author,
             TRY_CAST(json_extract_string(to_json(payload), '$.pull_request.created_at') AS VARCHAR) as created_at,
             TRY_CAST(json_extract_string(to_json(payload), '$.pull_request.merged_at') AS VARCHAR) as merged_at,
@@ -422,34 +425,64 @@ def fetch_all_pr_metadata_streaming(conn, identifiers, start_date, end_date):
             AND payload.pull_request.number IS NOT NULL
             AND TRY_CAST(json_extract_string(to_json(payload), '$.pull_request.created_at') AS VARCHAR) IS NOT NULL
             AND TRY_CAST(json_extract_string(to_json(payload), '$.pull_request.user.login') AS VARCHAR) IN ({identifier_list})
+            AND TRY_CAST(json_extract_string(to_json(payload), '$.action') AS VARCHAR) IN ('opened', 'closed')
         """
         
         try:
             results = conn.execute(query).fetchall()
-            batch_prs = 0
-            
-            # Add results to accumulating dictionary
+
+            # Group events by PR URL to merge opened and closed events
+            pr_events = defaultdict(lambda: {'opened': None, 'closed': None})
+
             for row in results:
                 url = row[0]
-                pr_author = row[1]
-                created_at = normalize_date_format(row[2]) if row[2] else None
-                merged_at = normalize_date_format(row[3]) if row[3] else None
-                closed_at = normalize_date_format(row[4]) if row[4] else None
-                
-                if not url:
+                action = row[1]
+                pr_author = row[2]
+                created_at = normalize_date_format(row[3]) if row[3] else None
+                merged_at = normalize_date_format(row[4]) if row[4] else None
+                closed_at = normalize_date_format(row[5]) if row[5] else None
+
+                if not url or not action:
                     continue
-                
-                pr_metadata = {
-                    'html_url': url,
+
+                event_data = {
+                    'pr_author': pr_author,
                     'created_at': created_at,
                     'merged_at': merged_at,
                     'closed_at': closed_at,
                 }
-                
+
+                pr_events[url][action] = event_data
+
+            # Only include PRs that have an 'opened' event
+            # Use closed event data (if available) to get merged_at and closed_at
+            batch_prs = 0
+            for url, events in pr_events.items():
+                if not events['opened']:
+                    # Skip PRs without an 'opened' event in this time frame
+                    continue
+
+                opened_event = events['opened']
+                closed_event = events['closed']
+
+                pr_author = opened_event['pr_author']
+                if not pr_author:
+                    continue
+
+                # Build complete PR metadata
+                # created_at comes from opened event
+                # merged_at and closed_at come from closed event (if available)
+                pr_metadata = {
+                    'html_url': url,
+                    'created_at': opened_event['created_at'],
+                    'merged_at': closed_event['merged_at'] if closed_event else None,
+                    'closed_at': closed_event['closed_at'] if closed_event else None,
+                }
+
                 metadata_by_agent[pr_author].append(pr_metadata)
                 batch_prs += 1
                 total_prs += 1
-            
+
             print(f"✓ {batch_prs} PRs found")
             
         except Exception as e: