path: root/.github/scripts/analytics/export_issues_to_ydb.py

#!/usr/bin/env python3

import os
import ydb
import configparser
import time
import json
from datetime import datetime, timezone, timedelta
import requests
from typing import List, Dict, Any, Optional

# Configuration
ORG_NAME = 'ydb-platform'
REPO_NAME = 'ydb'
PROJECT_ID = None #'45'  # Optional: set to None to skip project data

# Load YDB configuration
dir = os.path.dirname(__file__)
config = configparser.ConfigParser()
config_file_path = f"{dir}/../../config/ydb_qa_db.ini"
config.read(config_file_path)

DATABASE_ENDPOINT = config["QA_DB"]["DATABASE_ENDPOINT"]
DATABASE_PATH = config["QA_DB"]["DATABASE_PATH"]

def run_query(query: str, variables: Optional[Dict] = None) -> Dict[str, Any]:
    """Execute GraphQL query against GitHub API"""
    GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
    HEADERS = {"Authorization": f"Bearer {GITHUB_TOKEN}", "Content-Type": "application/json"}
    
    request = requests.post(
        'https://api.github.com/graphql', 
        json={'query': query, 'variables': variables}, 
        headers=HEADERS
    )
    
    if request.status_code == 200:
        response = request.json()
        if 'errors' in response:
            for error in response['errors']:
                print(f"GraphQL Error: {error.get('message', 'Unknown error')}")
                raise Exception(f"GraphQL Error: {error.get('message', 'Unknown error')}")
        return response
    else:
        raise Exception(f"Query failed with status {request.status_code}: {request.text}")

def get_last_update_time(driver, table_path: str) -> Optional[datetime]:
    """Get the latest updated_at timestamp from existing records"""
    try:
        session = driver.table_client.session().create()
        with session.transaction() as tx:
            result = tx.execute(
                f"SELECT MAX(updated_at) as max_updated_at FROM `{table_path}`",
                commit_tx=True,
                settings=ydb.BaseRequestSettings().with_timeout(30)
            )
            rows = result[0].rows
            if rows and rows[0]['max_updated_at']:
                return rows[0]['max_updated_at']
            return None

    except Exception as e:
        print(f"Warning: Could not get last update time: {e}")
        return None



def fetch_repository_issues(org_name: str = ORG_NAME, repo_name: str = REPO_NAME, since: Optional[datetime] = None) -> List[Dict[str, Any]]:
    """Fetch all issues from GitHub repository with comprehensive information"""
    if since:
        print(f"Fetching issues updated since {since.isoformat()} from repository {org_name}/{repo_name}...")
    else:
        print(f"Fetching all issues from repository {org_name}/{repo_name}...")
    start_time = time.time()
    
    issues = []
    has_next_page = True
    end_cursor = "null"
    
    # Convert datetime to GitHub API format if needed
    since_filter = ""
    if since:
        since_str = since.strftime('%Y-%m-%dT%H:%M:%SZ')
        since_filter = f', filterBy: {{since: "{since_str}"}}'
    
    repository_issues_query = """
    {
      organization(login: "%s") {
        repository(name: "%s") {
          issues(first: 100, after: %s, orderBy: {field: UPDATED_AT, direction: DESC}%s) {
            nodes {
              id
              number
              title
              url
              state
              body
              bodyText
              createdAt
              updatedAt
              closedAt
              author {
                login
                url
              }
              assignees(first: 10) {
                nodes {
                  login
                  url
                }
              }
              labels(first: 20) {
                nodes {
                  id
                  name
                  color
                  description
                }
              }
              milestone {
                id
                title
                url
                state
                dueOn
              }
              reactions {
                totalCount
              }
              comments {
                totalCount
              }
              repository {
                id
                name
                url
              }
              participants(first: 10) {
                totalCount
              }
            }
            pageInfo {
              hasNextPage
              endCursor
            }
          }
        }
      }
    }
    """
    
    total_fetched = 0
    while has_next_page:
        query = repository_issues_query % (org_name, repo_name, end_cursor, since_filter)
        result = run_query(query)
        
        if result and 'data' in result:
            repository_issues = result['data']['organization']['repository']['issues']
            current_batch = repository_issues['nodes']
            
            issues.extend(current_batch)
            total_fetched += len(current_batch)
            
            print(f"Fetched {len(current_batch)} issues from repository (total: {total_fetched})")
            
            page_info = repository_issues['pageInfo']
            has_next_page = page_info['hasNextPage']
            end_cursor = f'"{page_info["endCursor"]}"' if page_info['endCursor'] else "null"
        else:
            has_next_page = False
    
    elapsed = time.time() - start_time
    print(f"Fetched {len(issues)} issues total (took {elapsed:.2f}s)")
    return issues

def get_project_fields_for_issues(org_name: str, project_id: str, issue_numbers: List[int]) -> Dict[int, Dict[str, Any]]:
    """Get project fields for specific issues from GitHub project"""
    if not project_id:
        return {}
    
    print(f"Fetching project fields for {len(issue_numbers)} issues from project {project_id}...")
    start_time = time.time()
    
    project_fields = {}
    has_next_page = True
    end_cursor = "null"
    
    project_issues_query = """
    {
      organization(login: "%s") {
        projectV2(number: %s) {
          id
          title
          url
          items(first: 1000, after: %s) {
            nodes {
              id
              content {
                ... on Issue {
                  number
                }
              }
              fieldValues(first: 20) {
                nodes {
                  ... on ProjectV2ItemFieldSingleSelectValue {
                    field {
                      ... on ProjectV2SingleSelectField {
                        id
                        name
                      }
                    }
                    name
                    id
                    updatedAt
                  }
                  ... on ProjectV2ItemFieldTextValue {
                    field {
                      ... on ProjectV2Field {
                        id
                        name
                      }
                    }
                    text
                    id
                    updatedAt
                    creator {
                      login
                    }
                  }
                  ... on ProjectV2ItemFieldMilestoneValue {
                    field {
                      ... on ProjectV2Field {
                        id
                        name
                      }
                    }
                    milestone {
                      id
                      title
                    }
                  }
                  ... on ProjectV2ItemFieldDateValue {
                    field {
                      ... on ProjectV2Field {
                        id
                        name
                      }
                    }
                    date
                    updatedAt
                  }
                  ... on ProjectV2ItemFieldNumberValue {
                    field {
                      ... on ProjectV2Field {
                        id
                        name
                      }
                    }
                    number
                    updatedAt
                  }
                }
              }
            }
            pageInfo {
              hasNextPage
              endCursor
            }
          }
        }
      }
    }
    """
    
    issue_numbers_set = set(issue_numbers)
    
    while has_next_page:
        query = project_issues_query % (org_name, project_id, end_cursor)
        result = run_query(query)
        
        if result and 'data' in result:
            project_items = result['data']['organization']['projectV2']['items']
            current_batch = project_items['nodes']
            
            for item in current_batch:
                content = item.get('content')
                if content and content.get('number') in issue_numbers_set:
                    issue_number = content['number']
                    
                    # Extract project field values
                    fields = {}
                    field_values = item.get('fieldValues', {}).get('nodes', [])
                    
                    for field_value in field_values:
                        field_name = field_value.get('field', {}).get('name', '')
                        if field_name:
                            if 'name' in field_value:  # SingleSelect
                                fields[field_name.lower()] = field_value.get('name')
                            elif 'text' in field_value:  # Text
                                fields[field_name.lower()] = field_value.get('text')
                            elif 'number' in field_value:  # Number
                                fields[field_name.lower()] = field_value.get('number')
                            elif 'date' in field_value:  # Date
                                fields[field_name.lower()] = field_value.get('date')
                            elif 'milestone' in field_value:  # Milestone
                                milestone = field_value.get('milestone', {})
                                fields[field_name.lower()] = milestone.get('title') if milestone else None
                    
                    project_fields[issue_number] = fields
            
            page_info = project_items['pageInfo']
            has_next_page = page_info['hasNextPage']
            end_cursor = f'"{page_info["endCursor"]}"' if page_info['endCursor'] else "null"
        else:
            has_next_page = False
    
    elapsed = time.time() - start_time
    print(f"Fetched project fields for {len(project_fields)} issues (took {elapsed:.2f}s)")
    return project_fields

def parse_datetime(dt_str: Optional[str]) -> Optional[datetime]:
    """Parse GitHub datetime string to datetime object"""
    if not dt_str:
        return None
    try:
        # GitHub returns ISO format with Z suffix
        return datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
    except (ValueError, TypeError):
        return None

# --- branch version helpers ---
def parse_branch(label):
    if label == 'main':
        return (0, 0, 0, 0, 0)  # main — всегда минимальный
    if label.startswith('prestable-'):
        parts = label.split('-')
        nums = [int(x) for x in parts[1:] if x.isdigit()]
        while len(nums) < 3:
            nums.append(0)
        return (1, *nums, 0)  # prestable < analytics < stable
    if label.startswith('stable-'):
        parts = label.split('-')
        nums = []
        analytics = 0
        for x in parts[1:]:
            if x.isdigit():
                nums.append(int(x))
            elif x == 'analytics':
                analytics = 1
        while len(nums) < 3:
            nums.append(0)
        if analytics:
            # analytics-лейбл: всегда меньше любого stable с числовым патчем, но больше prestable
            return (2, *nums, 0)  # analytics = 2
        else:
            return (3, *nums, 1)  # обычный stable = 3, всегда больше analytics
    return (-1, 0, 0, 0, 0)  # некорректные/другие — минимальные

def get_max_branch(branch_labels):
    best = None
    best_key = (-2, 0, 0, 0, 0)  # всегда меньше любого корректного branch
    for label in branch_labels:
        key = parse_branch(label)
        if key > best_key:
            best = label
            best_key = key
    return best

def transform_issues_for_ydb(issues: List[Dict[str, Any]], project_fields: Optional[Dict[int, Dict[str, Any]]] = None) -> List[Dict[str, Any]]:
    """Transform GitHub issues data for YDB storage"""
    print("Transforming issues data for YDB...")
    start_time = time.time()
    
    if project_fields is None:
        project_fields = {}
    
    transformed_issues = []
    
    for issue in issues:
        # Get project fields for this issue if available
        issue_number = issue.get('number')
        issue_project_fields = project_fields.get(issue_number, {}) if issue_number else {}
        
        # Extract labels
        labels = []
        branch_labels = []
        env = None
        priority = None
        area = None
        for label in issue.get('labels', {}).get('nodes', []):
            name = label.get('name', '')
            labels.append({
                'name': name,
                'color': label.get('color', ''),
                'description': label.get('description', '')
            })
            # branch detection (main, stable-*, prestable-*)
            if name == 'main' or name.startswith('stable-') or name.startswith('prestable-'):
                branch_labels.append(name)
            # env detection
            if name.startswith('env:'):
                env = name
            # priority detection
            if name.startswith('prio:'):
                priority = name
            # area detection
            if name.startswith('area/'):
                area = name
        branch = ';'.join(branch_labels) if branch_labels else None
        max_branch = get_max_branch(branch_labels) if branch_labels else None
        info = {'branch': branch, 'max_branch': max_branch, 'env': env, 'priority': priority, 'area': area}
        # Issue type from project fields
        issue_type = issue_project_fields.get('type') or issue_project_fields.get('Type')
        
        # Extract assignees
        assignees = []
        for assignee in issue.get('assignees', {}).get('nodes', []):
            assignees.append({
                'login': assignee.get('login', ''),
                'url': assignee.get('url', '')
            })
        
        # Extract milestone
        milestone = issue.get('milestone')
        milestone_info = None
        if milestone:
            milestone_info = {
                'title': milestone.get('title', ''),
                'url': milestone.get('url', ''),
                'state': milestone.get('state', ''),
                'due_on': milestone.get('dueOn')
            }
        
        # Extract author
        author = issue.get('author', {})
        author_info = {
            'login': author.get('login', '') if author else '',
            'url': author.get('url', '') if author else ''
        }
        
        # Parse timestamps
        created_at = parse_datetime(issue.get('createdAt'))
        updated_at = parse_datetime(issue.get('updatedAt'))
        closed_at = parse_datetime(issue.get('closedAt'))
        now = datetime.now(timezone.utc)
        
        is_in_project = bool(issue_project_fields)
        
        # Calculate time-based metrics
        days_since_created = 0
        days_since_updated = 0
        time_to_close_hours = 0
        
        if created_at:
            days_since_created = (now - created_at).days
        if updated_at:
            days_since_updated = (now - updated_at).days
        if closed_at and created_at:
            time_to_close_hours = int((closed_at - created_at).total_seconds() / 3600)
        
        # Build the record
        issue_record = {
            # Primary identifiers
            'project_item_id': f"repo-{issue.get('number', 0)}",
            'issue_id': issue.get('id', ''),
            'issue_number': issue.get('number', 0),
            
            # Core issue data
            'title': issue.get('title', ''),
            'url': issue.get('url', ''),
            'state': issue.get('state', ''),
            'body': issue.get('body', '') or '',
            'body_text': issue.get('bodyText', ''),
            
            # Time dimensions
            'created_at': created_at,
            'updated_at': updated_at,
            'closed_at': closed_at,
            'created_date': created_at.date() if created_at else None,
            'updated_date': updated_at.date() if updated_at else None,
            
            # User dimensions
            'author_login': author_info['login'],
            'author_url': author_info['url'],
            
            
            # Repository dimensions
            'repository_name': issue.get('repository', {}).get('name', ''),
            'repository_url': issue.get('repository', {}).get('url', ''),
            
            # Project dimensions
            'project_status': issue_project_fields.get('status'),
            'project_owner': issue_project_fields.get('owner'),
            'project_priority': issue_project_fields.get('priority'),
            'is_in_project': is_in_project,
            
            # Time-based metrics
            'days_since_created': days_since_created,
            'days_since_updated': days_since_updated,
            'time_to_close_hours': time_to_close_hours,
            
            # Complex data
            'assignees': json.dumps(assignees) if assignees else None,
            'labels': json.dumps(labels) if labels else None,
            'milestone': json.dumps(milestone_info) if milestone_info else None,
            'project_fields': json.dumps(issue_project_fields) if issue_project_fields else None,
            'info': json.dumps(info) if any(info.values()) else None,
            'issue_type': issue_type,
            
            # System fields
            'exported_at': now
        }
        
        transformed_issues.append(issue_record)
    
    elapsed = time.time() - start_time
    print(f"Transformed {len(transformed_issues)} issues (took {elapsed:.2f}s)")
    return transformed_issues

def check_table_exists(session, table_path: str) -> bool:
    """Check if table exists in YDB"""
    try:
        session.describe_table(table_path)
        return True
    except ydb.SchemeError:
        return False
    except Exception as e:
        print(f"Unexpected error while checking table: {e}")
        return False

def create_issues_table(session, table_path: str):
    """Create issues table in YDB optimized for BI"""
    print(f"Creating BI-optimized table: {table_path}")
    start_time = time.time()
    
    try:
        session.execute_scheme(f"""
            CREATE TABLE IF NOT EXISTS `{table_path}` (
                -- Primary identifiers
                `project_item_id` Utf8 NOT NULL,
                `issue_id` Utf8 NOT NULL,
                `issue_number` Uint64 NOT NULL,
                
                -- Core issue data
                `title` Utf8,
                `url` Utf8,
                `state` Utf8,
                `body` Utf8,
                `body_text` Utf8,
                
                -- Time dimensions for BI (partitioning keys)
                `created_at` Timestamp NOT NULL,
                `updated_at` Timestamp,
                `closed_at` Timestamp,
                `created_date` Date NOT NULL,  -- Extracted date for better partitioning
                `updated_date` Date NOT NULL,  -- Extracted date for better partitioning
                
                -- User dimensions
                `author_login` Utf8,
                `author_url` Utf8,
                
                
                -- Repository dimensions
                `repository_name` Utf8,
                `repository_url` Utf8,
                
                -- Project dimensions (nullable for issues not in project)
                `project_status` Utf8,
                `project_owner` Utf8,
                `project_priority` Utf8,
                `is_in_project` Int NOT NULL,  -- Boolean flag for faster filtering
                
                -- Time-based metrics
                `days_since_created` Uint64,  -- Days since creation
                `days_since_updated` Uint64,  -- Days since last update
                `time_to_close_hours` Uint64,  -- Time to close in hours (if closed)
                
                -- Complex data (keep as JSON for detailed analysis)
                `assignees` Json,
                `labels` Json,
                `milestone` Json,
                `project_fields` Json,
                `info` Json,
                `issue_type` Utf8,
                
                -- System fields
                `exported_at` Timestamp NOT NULL,
                
                PRIMARY KEY (`created_date`, `issue_number`, `project_item_id`)
            )
            PARTITION BY HASH(`created_date`)
            WITH (
                STORE = COLUMN,
                
                AUTO_PARTITIONING_BY_SIZE = ENABLED,
                AUTO_PARTITIONING_PARTITION_SIZE_MB = 2048,
                AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = 4
            )
        """)
        
        elapsed = time.time() - start_time
        print(f"BI-optimized table created successfully (took {elapsed:.2f}s)")
        
    except Exception as e:
        elapsed = time.time() - start_time
        print(f"Error creating BI-optimized table: {e} (took {elapsed:.2f}s)")
        raise

def bulk_upsert_issues(table_client, table_path: str, issues: List[Dict[str, Any]]):
    """Bulk upsert issues into YDB table optimized for BI"""
    print(f"Bulk upserting {len(issues)} issues to BI-optimized table {table_path}")
    start_time = time.time()
    
    column_types = (
        ydb.BulkUpsertColumns()
        # Primary identifiers
        .add_column("project_item_id", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("issue_id", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("issue_number", ydb.OptionalType(ydb.PrimitiveType.Uint64))
        
        # Core issue data
        .add_column("title", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("url", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("state", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("body", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("body_text", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        
        # Time dimensions
        .add_column("created_at", ydb.OptionalType(ydb.PrimitiveType.Timestamp))
        .add_column("updated_at", ydb.OptionalType(ydb.PrimitiveType.Timestamp))
        .add_column("closed_at", ydb.OptionalType(ydb.PrimitiveType.Timestamp))
        .add_column("created_date", ydb.OptionalType(ydb.PrimitiveType.Date))
        .add_column("updated_date", ydb.OptionalType(ydb.PrimitiveType.Date))
        
        # User dimensions
        .add_column("author_login", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("author_url", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        
        
        # Repository dimensions
        .add_column("repository_name", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("repository_url", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        
        # Project dimensions
        .add_column("project_status", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("project_owner", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("project_priority", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        .add_column("is_in_project", ydb.OptionalType(ydb.PrimitiveType.Int32))
        
        # Time-based metrics
        .add_column("days_since_created", ydb.OptionalType(ydb.PrimitiveType.Uint64))
        .add_column("days_since_updated", ydb.OptionalType(ydb.PrimitiveType.Uint64))
        .add_column("time_to_close_hours", ydb.OptionalType(ydb.PrimitiveType.Uint64))
        
        # Complex data
        .add_column("assignees", ydb.OptionalType(ydb.PrimitiveType.Json))
        .add_column("labels", ydb.OptionalType(ydb.PrimitiveType.Json))
        .add_column("milestone", ydb.OptionalType(ydb.PrimitiveType.Json))
        .add_column("project_fields", ydb.OptionalType(ydb.PrimitiveType.Json))
        .add_column("info", ydb.OptionalType(ydb.PrimitiveType.Json))
        .add_column("issue_type", ydb.OptionalType(ydb.PrimitiveType.Utf8))
        
        # System fields
        .add_column("exported_at", ydb.OptionalType(ydb.PrimitiveType.Timestamp))
    )
    
    table_client.bulk_upsert(table_path, issues, column_types)
    
    elapsed = time.time() - start_time
    print(f"BI-optimized bulk upsert completed (took {elapsed:.2f}s)")

def main():
    """Main function to export GitHub issues to YDB"""
    print("Starting GitHub issues export to YDB")
    script_start_time = time.time()
    
    # Check environment variables
    required_vars = ["GITHUB_TOKEN", "CI_YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS"]
    for var in required_vars:
        if var not in os.environ:
            print(f"Error: Environment variable {var} is missing")
            return 1
    
    # Set up YDB credentials
    os.environ["YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS"] = os.environ[
        "CI_YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS"
    ]
    
    table_path = "github_data/issues"

    full_table_path = f"{DATABASE_PATH}/{table_path}"
    batch_size = 100
    
    try:
        # Connect to YDB and process data
        with ydb.Driver(
            endpoint=DATABASE_ENDPOINT,
            database=DATABASE_PATH,
            credentials=ydb.credentials_from_env_variables()
        ) as driver:
            driver.wait(timeout=10, fail_fast=True)
            
            with ydb.SessionPool(driver) as pool:
                # Check and create table if needed
                def check_and_create_table(session):
                    create_issues_table(session, table_path)
                    return True
                
                pool.retry_operation_sync(check_and_create_table)
                
                # Check if this is an incremental update
                last_update_time = get_last_update_time(driver, table_path)
                
                if last_update_time:
                    print(f"Incremental update: fetching issues updated since {last_update_time.isoformat()}")
                    # Add a small buffer to avoid missing issues due to timing issues
                    since_time = last_update_time - timedelta(minutes=5)
                else:
                    print("Full export: fetching all issues")
                    since_time = None
                
                # Fetch issues from GitHub
                issues = fetch_repository_issues(ORG_NAME, REPO_NAME, since_time)
                
                if not issues:
                    print("No issues fetched from GitHub")
                    return 0
                
                # Get project fields if PROJECT_ID is specified
                project_fields = {}
                if PROJECT_ID:
                    issue_numbers = []
                    for issue in issues:
                        number = issue.get('number')
                        if number is not None and isinstance(number, int):
                            issue_numbers.append(number)
                    if issue_numbers:
                        project_fields = get_project_fields_for_issues(ORG_NAME, PROJECT_ID, issue_numbers)
                
                # Transform issues for YDB
                transformed_issues = transform_issues_for_ydb(issues, project_fields)
                
                # Upsert issues in batches
                print(f"Uploading {len(transformed_issues)} issues in batches of {batch_size}")
                upload_start_time = time.time()
                
                for start in range(0, len(transformed_issues), batch_size):
                    batch_start_time = time.time()
                    batch_issues = transformed_issues[start:start + batch_size]
                    
                    print(f"Uploading batch {start//batch_size + 1}: issues {start+1}-{start+len(batch_issues)}")
                    bulk_upsert_issues(driver.table_client, full_table_path, batch_issues)
                    
                    batch_elapsed = time.time() - batch_start_time
                    print(f"Batch completed (took {batch_elapsed:.2f}s)")
                
                upload_elapsed = time.time() - upload_start_time
                print(f"All issues uploaded (total upload time: {upload_elapsed:.2f}s)")
        
        script_elapsed = time.time() - script_start_time
        print(f"Script completed successfully (total time: {script_elapsed:.2f}s)")
        
    except Exception as e:
        print(f"Error during execution: {e}")
        return 1
    
    return 0

if __name__ == "__main__":
    exit(main())