From 3a997b2d2e99a525f6c18aacd919b511ee276224 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Sat, 1 Aug 2026 19:44:52 +0000 Subject: [PATCH 1/2] Update tqdm requirement from <5.0,>=4.67.3 to >=4.70.0,<5.0 Updates the requirements on [tqdm](https://github.com/tqdm/tqdm) to permit the latest version. - [Release notes](https://github.com/tqdm/tqdm/releases) - [Commits](https://github.com/tqdm/tqdm/compare/v4.67.3...v4.70.0) --- updated-dependencies: - dependency-name: tqdm dependency-version: 4.70.0 dependency-type: direct:production ... Signed-off-by: dependabot[bot] --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index f7f34b2..a3fa954 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ -tqdm>=4.67.3,<5.0 +tqdm>=4.70.0,<5.0 pytest>=9.0.3,<10.0 From 8aaf375ce620f44f294b302c50c6c0123de7d72f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 2 Aug 2026 06:55:53 +0000 Subject: [PATCH 2/2] Fix ruff linting errors (UP006, UP035, DTZ004, DTZ007, RUF015, SIM102) --- ghmap/cli.py | 38 ++++++++++++++--------------- ghmap/mapping/action_mapper.py | 31 +++++++++++------------ ghmap/mapping/activity_mapper.py | 17 +++++++------ ghmap/preprocess/event_processor.py | 37 ++++++++++++++-------------- ghmap/utils.py | 1 + tests/test_cli.py | 4 +-- 6 files changed, 64 insertions(+), 64 deletions(-) diff --git a/ghmap/cli.py b/ghmap/cli.py index 6248dcb..1307667 100644 --- a/ghmap/cli.py +++ b/ghmap/cli.py @@ -4,15 +4,14 @@ from datetime import datetime, timezone from importlib.resources import files from pathlib import Path -from typing import Dict, List, Tuple -from .preprocess.event_processor import EventProcessor from .mapping.action_mapper import ActionMapper from .mapping.activity_mapper import ActivityMapper +from .preprocess.event_processor import EventProcessor from .utils import load_json_file, save_to_jsonl_file -def extract_version_info(filename: str) -> Tuple[str, datetime]: +def extract_version_info(filename: str) -> tuple[str, datetime]: """Extract platform and version date from mapping filename. Expected format: {platform}_{type}_{date}.json @@ -30,8 +29,7 @@ def extract_version_info(filename: str) -> Tuple[str, datetime]: # Parse ISO 8601 Basic Format: YYYYMMDDTHHMMSSZ try: - version_date = datetime.strptime(version_str, '%Y%m%dT%H%M%SZ') - version_date = version_date.replace(tzinfo=timezone.utc) + version_date = datetime.strptime(version_str, '%Y%m%dT%H%M%SZ').replace(tzinfo=timezone.utc) except ValueError as e: raise ValueError( f"Invalid timestamp format: {version_str}. " @@ -41,7 +39,7 @@ def extract_version_info(filename: str) -> Tuple[str, datetime]: return platform, version_date -def find_valid_mappings(platform: str, event_date: datetime) -> Dict[str, Path]: +def find_valid_mappings(platform: str, event_date: datetime) -> dict[str, Path]: """Find the valid mapping files for a given platform and event date.""" config_dir = Path(files("ghmap").joinpath("config")) @@ -89,8 +87,8 @@ def find_valid_mappings(platform: str, event_date: datetime) -> Dict[str, Path]: def split_events_by_mapping_versions( - events: List[Dict], platform: str -) -> Dict[Tuple[datetime, datetime], List[Dict]]: + events: list[dict], platform: str +) -> dict[tuple[datetime, datetime], list[dict]]: """Split events into time periods based on available mapping versions.""" config_dir = Path(files("ghmap").joinpath("config")) @@ -114,7 +112,7 @@ def _get_version_dates(config_dir: Path, platform: str) -> set: return version_dates -def _create_time_periods(sorted_versions: List[datetime]) -> List[Tuple[datetime, datetime]]: +def _create_time_periods(sorted_versions: list[datetime]) -> list[tuple[datetime, datetime]]: """Create time periods from sorted version dates.""" periods = [] for i, start_date in enumerate(sorted_versions): @@ -128,9 +126,9 @@ def _create_time_periods(sorted_versions: List[datetime]) -> List[Tuple[datetime def _assign_events_to_periods( - events: List[Dict], - time_periods: List[Tuple[datetime, datetime]] -) -> Dict[Tuple[datetime, datetime], List[Dict]]: + events: list[dict], + time_periods: list[tuple[datetime, datetime]] +) -> dict[tuple[datetime, datetime], list[dict]]: """Assign each event to its corresponding time period.""" events_by_period = {period: [] for period in time_periods} @@ -148,7 +146,7 @@ def _assign_events_to_periods( def _parse_event_date(date_str: str | int) -> datetime | None: """Parse the event date string or timestamp into a datetime object.""" if isinstance(date_str, int): - return datetime.utcfromtimestamp(date_str / 1000) + return datetime.fromtimestamp(date_str / 1000, tz=timezone.utc) return datetime.fromisoformat(date_str.replace('Z', '+00:00')) def main(): @@ -212,7 +210,7 @@ def _parse_args() -> argparse.Namespace: return parser.parse_args() -def _process_events(args: argparse.Namespace) -> (List[Dict], List[Dict]): +def _process_events(args: argparse.Namespace) -> (list[dict], list[dict]): """Process raw events into actions and activities.""" print("Step 0: Preprocessing events...") processor = EventProcessor(args.platform, progress_bar=args.progress_bar) @@ -238,9 +236,9 @@ def _process_events(args: argparse.Namespace) -> (List[Dict], List[Dict]): def _apply_custom_mappings( - events: List[Dict], + events: list[dict], args: argparse.Namespace -) -> (List[Dict], List[Dict]): +) -> (list[dict], list[dict]): """Apply custom action and activity mappings if provided.""" print("Using custom mappings, skipping automatic mapping detection...") all_actions, all_activities = [], [] @@ -267,11 +265,11 @@ def _apply_custom_mappings( def _process_period( - period_events: List[Dict], + period_events: list[dict], period_start: datetime, period_end: datetime, args: argparse.Namespace -) -> tuple[List[Dict], List[Dict]]: +) -> tuple[list[dict], list[dict]]: """Process events for a single time period and return actions and activities.""" print(f"\nProcessing period: {period_start} to {period_end}") print(f" Events in period: {len(period_events)}") @@ -300,8 +298,8 @@ def _process_period( def _save_results( - all_actions: List[Dict], - all_activities: List[Dict], + all_actions: list[dict], + all_activities: list[dict], output_actions: str, output_activities: str ): diff --git a/ghmap/mapping/action_mapper.py b/ghmap/mapping/action_mapper.py index 4b1f339..60deca4 100644 --- a/ghmap/mapping/action_mapper.py +++ b/ghmap/mapping/action_mapper.py @@ -2,8 +2,9 @@ import json import re -from datetime import datetime -from typing import List, Dict, Any +from datetime import datetime, timezone +from typing import Any + from tqdm import tqdm @@ -18,7 +19,7 @@ class ActionMapper: # pylint: disable=too-few-public-methods progress_bar (bool): Flag to enable or disable progress bar (tqdm). """ - def __init__(self, action_mapping: Dict, progress_bar: bool = True): + def __init__(self, action_mapping: dict, progress_bar: bool = True): self.action_mapping = action_mapping parameters = action_mapping.get('parameters', {}) self.event_type_key = parameters.get('event_type_key', 'type') @@ -26,13 +27,13 @@ def __init__(self, action_mapping: Dict, progress_bar: bool = True): self.progress_bar = progress_bar @staticmethod - def _deserialize_payload(event_record: Dict) -> Dict: + def _deserialize_payload(event_record: dict) -> dict: """Deserializes the 'payload' field of the event record if it's a string.""" if isinstance(event_record['payload'], str): event_record['payload'] = json.loads(event_record['payload']) return event_record - def _convert_date_to_iso(self, event_record: Dict) -> Dict: + def _convert_date_to_iso(self, event_record: dict) -> dict: """Converts 'created_at' to ISO 8601 format if it's a Unix timestamp or string.""" created_at = event_record.get(self.created_at_key) if isinstance(created_at, str): @@ -41,10 +42,10 @@ def _convert_date_to_iso(self, event_record: Dict) -> Dict: created_at = created_at.split('.')[0] + "Z" event_record[self.created_at_key] = datetime.strptime( created_at, '%Y-%m-%dT%H:%M:%SZ' - ).strftime('%Y-%m-%dT%H:%M:%SZ') + ).replace(tzinfo=timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') elif isinstance(created_at, int): - event_record[self.created_at_key] = datetime.utcfromtimestamp( - created_at / 1000 + event_record[self.created_at_key] = datetime.fromtimestamp( + created_at / 1000, tz=timezone.utc ).strftime('%Y-%m-%dT%H:%M:%SZ') return event_record @@ -73,8 +74,8 @@ def _match_condition(event_value: Any, mapping_value: Any) -> bool: return event_value == mapping_value def _extract_attributes( - self, event_record: Dict, action_details: Dict, action_name: str - ) -> Dict: + self, event_record: dict, action_details: dict, action_name: str + ) -> dict: """Extracts attributes and common fields from the event record.""" mapped_action = {'action': action_name} @@ -91,7 +92,7 @@ def _extract_attributes( return mapped_action - def _extract_fields(self, event_record: Dict, field_mapping: Dict) -> Dict: + def _extract_fields(self, event_record: dict, field_mapping: dict) -> dict: """Extracts values from the event record using provided mappings.""" extracted_data = {} for field_key, mapping_value in field_mapping.items(): @@ -109,9 +110,9 @@ def _extract_fields(self, event_record: Dict, field_mapping: Dict) -> Dict: ) return extracted_data - def _extract_list(self, event_record: Dict, list_mapping: list) -> list: + def _extract_list(self, event_record: dict, list_mapping: list) -> list: """Extracts a list of values from the event record.""" - base_path = list_mapping[0][list(list_mapping[0].keys())[0]].split('.')[:-1] + base_path = list_mapping[0][next(iter(list_mapping[0].keys()))].split('.')[:-1] base_list = self._extract_field(event_record, base_path) if not isinstance(base_list, list): return [] @@ -124,7 +125,7 @@ def _extract_list(self, event_record: Dict, list_mapping: list) -> list: ] @staticmethod - def _extract_field(event_record: Dict, field_path: str) -> Any: + def _extract_field(event_record: dict, field_path: str) -> Any: """Extracts a value from the event record using a dotted field path.""" keys = field_path.split('.') if isinstance(field_path, str) else field_path value = event_record @@ -136,7 +137,7 @@ def _extract_field(event_record: Dict, field_path: str) -> Any: return None return value - def map(self, events: List[Dict], mapping_strategy: str = "flexible") -> List[Dict]: + def map(self, events: list[dict], mapping_strategy: str = "flexible") -> list[dict]: """ Maps events to high-level actions using mapping configuration. diff --git a/ghmap/mapping/activity_mapper.py b/ghmap/mapping/activity_mapper.py index 51ade82..45ec397 100644 --- a/ghmap/mapping/activity_mapper.py +++ b/ghmap/mapping/activity_mapper.py @@ -1,7 +1,8 @@ """Module to map GitHub actions to higher-level activities based on rules.""" from datetime import datetime, timedelta -from typing import List, Dict, Tuple, Any +from typing import Any + from tqdm import tqdm @@ -15,13 +16,13 @@ class ActivityMapper: # pylint: disable=too-few-public-methods progress_bar (bool): Flag to enable or disable progress bar (tqdm). """ - def __init__(self, activity_mapping: Dict, progress_bar: bool = True): + def __init__(self, activity_mapping: dict, progress_bar: bool = True): self.activity_mapping = self._preprocess_activities(activity_mapping) self.used_ids = set() self.progress_bar = progress_bar @staticmethod - def _preprocess_activities(activity_mapping: Dict) -> Dict: + def _preprocess_activities(activity_mapping: dict) -> dict: for activity in activity_mapping["activities"]: activity["time_window"] = timedelta( seconds=int(activity["time_window"].replace("s", "")) @@ -37,7 +38,7 @@ def _within_time_limit(start_time: str, end_time: str, time_window: timedelta) - return diff <= time_window @staticmethod - def _get_nested_value(data: Dict, field: str) -> Any: + def _get_nested_value(data: dict, field: str) -> Any: for key in field.split('.'): data = data.get(key) if data is None: @@ -45,7 +46,7 @@ def _get_nested_value(data: Dict, field: str) -> Any: return data @staticmethod - def _group_actions(actions: List[Dict]) -> Dict[Tuple[int, int], List[Dict]]: + def _group_actions(actions: list[dict]) -> dict[tuple[int, int], list[dict]]: grouped = {} for action in actions: key = (action["actor"]["id"], action["repository"]["id"]) @@ -54,7 +55,7 @@ def _group_actions(actions: List[Dict]) -> Dict[Tuple[int, int], List[Dict]]: group.sort(key=lambda x: x["date"]) return grouped - def _validate_gathered_actions(self, gathered: List[Dict], activity: Dict) -> Tuple[List[Dict], List[Dict]]: # pylint: disable=line-too-long + def _validate_gathered_actions(self, gathered: list[dict], activity: dict) -> tuple[list[dict], list[dict]]: # pylint: disable=line-too-long if len(gathered) == 1: return gathered, [] @@ -86,7 +87,7 @@ def _validate_gathered_actions(self, gathered: List[Dict], activity: Dict) -> Tu return validated, invalid - def _gather_actions(self, actions: List[Dict], start_idx: int, activity: Dict) -> Tuple[List[Dict], int, List[Dict]]: # pylint: disable=line-too-long + def _gather_actions(self, actions: list[dict], start_idx: int, activity: dict) -> tuple[list[dict], int, list[dict]]: # pylint: disable=line-too-long gathered, preserved = [], [] found_required = set() time_window = activity["time_window"] @@ -122,7 +123,7 @@ def _gather_actions(self, actions: List[Dict], start_idx: int, activity: Dict) - preserved.extend(invalid) return validated, start_idx + len(validated), preserved - def map(self, actions: List[Dict]) -> List[Dict]: + def map(self, actions: list[dict]) -> list[dict]: """Map actions to activities based on activity mapping configuration.""" grouped = self._group_actions(actions) all_mapped_activities = [] diff --git a/ghmap/preprocess/event_processor.py b/ghmap/preprocess/event_processor.py index 9c9f0f9..3ef74ac 100644 --- a/ghmap/preprocess/event_processor.py +++ b/ghmap/preprocess/event_processor.py @@ -1,8 +1,8 @@ """Preprocess module for filtering and cleaning GitHub events.""" import json import os -from datetime import datetime -from typing import List, Dict +from datetime import datetime, timezone + from tqdm import tqdm @@ -21,8 +21,8 @@ def __init__(self, platform: str = 'github', progress_bar: bool = True): def _parse_time(timestamp: str | int) -> datetime: """Converts a Unix timestamp (in milliseconds) or ISO 8601 string to a datetime object.""" if isinstance(timestamp, str): - return datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%SZ') - return datetime.utcfromtimestamp(timestamp / 1000) + return datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=timezone.utc) + return datetime.fromtimestamp(timestamp / 1000, tz=timezone.utc) @staticmethod def _calculate_time_diff(start: datetime, end: datetime) -> float: @@ -30,7 +30,7 @@ def _calculate_time_diff(start: datetime, end: datetime) -> float: return (end - start).total_seconds() @staticmethod - def _is_within_time_window(event1: Dict, event2: Dict, window: int = 2) -> bool: + def _is_within_time_window(event1: dict, event2: dict, window: int = 2) -> bool: """Checks if event2 is within a specified time window (in seconds) of event1.""" time_diff = abs(EventProcessor._calculate_time_diff( EventProcessor._parse_time(event1['created_at']), @@ -38,7 +38,7 @@ def _is_within_time_window(event1: Dict, event2: Dict, window: int = 2) -> bool: )) return time_diff <= window - def _should_keep_event(self, current_event: Dict, events: List[Dict], index: int) -> bool: + def _should_keep_event(self, current_event: dict, events: list[dict], index: int) -> bool: """Determines whether the current event should be kept based on redundant review checks.""" actor_id = current_event['actor']['id'] repo_id = current_event['repo']['id'] @@ -61,7 +61,7 @@ def _should_keep_event(self, current_event: Dict, events: List[Dict], index: int return True - def _filter_redundant_review_events(self, events: List[Dict]) -> List[Dict]: + def _filter_redundant_review_events(self, events: list[dict]) -> list[dict]: """Filters out redundant PullRequestReviewEvent events.""" filtered_events = [] combined_events = self.pending_events + events @@ -69,16 +69,15 @@ def _filter_redundant_review_events(self, events: List[Dict]) -> List[Dict]: for i, event in enumerate(combined_events): if event['type'] == "PullRequestReviewEvent" and event['id'] not in self.processed_ids: - if self._should_keep_event(event, combined_events, i): - if not ( - filtered_events and - filtered_events[-1]['type'] == "PullRequestReviewEvent" and - filtered_events[-1]['actor']['id'] == event['actor']['id'] and - filtered_events[-1]['repo']['id'] == event['repo']['id'] and - self._is_within_time_window(filtered_events[-1], event) - ): - filtered_events.append(event) - self.processed_ids.add(event['id']) + if self._should_keep_event(event, combined_events, i) and not ( + filtered_events and + filtered_events[-1]['type'] == "PullRequestReviewEvent" and + filtered_events[-1]['actor']['id'] == event['actor']['id'] and + filtered_events[-1]['repo']['id'] == event['repo']['id'] and + self._is_within_time_window(filtered_events[-1], event) + ): + filtered_events.append(event) + self.processed_ids.add(event['id']) elif event['id'] not in self.processed_ids: filtered_events.append(event) self.processed_ids.add(event['id']) @@ -88,7 +87,7 @@ def _filter_redundant_review_events(self, events: List[Dict]) -> List[Dict]: def process( self, input_path: str - ) -> List[Dict]: + ) -> list[dict]: """ Processes an input file or directory of files. Supports: @@ -123,7 +122,7 @@ def process( return all_events - def _load_events(self, path: str) -> List[Dict]: + def _load_events(self, path: str) -> list[dict]: """Loads events from JSON, JSON list, or JSON lines.""" with open(path, 'r', encoding='utf-8') as file: first_char = file.read(1) diff --git a/ghmap/utils.py b/ghmap/utils.py index 8f17360..70f8aa5 100644 --- a/ghmap/utils.py +++ b/ghmap/utils.py @@ -2,6 +2,7 @@ import json + def load_jsonl_file(file_path): """Load actions from a JSON Lines file.""" with open(file_path, 'r', encoding='utf-8') as file: diff --git a/tests/test_cli.py b/tests/test_cli.py index 4661979..d4fdff4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,9 +1,9 @@ """Test the ghmap CLI with a sample input file and expected output.""" -import subprocess import filecmp -import tempfile import os +import subprocess +import tempfile def test_ghmap_cli_on_sample():