rbd-app/src/utils.py

import os
import re
from datetime import datetime, timedelta, timezone
from typing import Optional

import pytz
from dateutil.relativedelta import relativedelta

from src.config import RELIABILITY_SERVICE_API, TIMEZONE
import hvac
from typing import Optional, Dict, List

def parse_relative_expression(date_str: str) -> Optional[datetime]:
    """
    Parse relative date expressions using T (days), M (months), and Y (years)
    Returns tuple of (datetime, type_description) or None if not a relative date
    """
    pattern = r"^([HTMY])([+-]\d+)?$"
    match = re.match(pattern, date_str)

    if not match:
        return None

    unit, offset = match.groups()
    offset = int(offset) if offset else 0
    # Use UTC timezone for consistency
    jakarta_tz = pytz.timezone("Asia/Jakarta")
    today = datetime.now(jakarta_tz)
    if unit == "H":
        # For hours, keep minutes and seconds
        result_time = today + timedelta(hours=offset)
        return result_time
    elif unit == "T":
        return today + timedelta(days=offset)
    elif unit == "M":
        return today + relativedelta(months=offset)
    elif unit == "Y":
        return today + relativedelta(years=offset)


def parse_date_string(date_str: str) -> Optional[datetime]:
    """
    Parse date strings in various formats including relative expressions
    Returns tuple of (datetime, type)
    """
    # Try parsing as relative expression first
    relative_result = parse_relative_expression(date_str)
    if relative_result:
        return relative_result

    # Try different date formats
    date_formats = [
        ("%Y-%m-%d", "iso"),  # 2024-11-08
        ("%Y/%m/%d", "slash"),  # 2024/11/08
        ("%d-%m-%Y", "european"),  # 08-11-2024
        ("%d/%m/%Y", "european_slash"),  # 08/11/2024
        ("%Y.%m.%d", "dot"),  # 2024.11.08
        ("%d.%m.%Y", "european_dot"),  # 08.11.2024
    ]

    for fmt, type_name in date_formats:
        try:
            # Parse the date and set it to start of day in UTC
            dt = datetime.strptime(date_str, fmt)
            dt = dt.replace(
                hour=0,
                minute=0,
                second=0,
                microsecond=0,
                tzinfo=pytz.timezone("Asia/Jakarta"),
            )
            return dt
        except ValueError:
            continue

    raise ValueError(
        "Invalid date format. Supported formats:\n"
        "Relative formats:\n"
        "- T (days): T, T-n, T+n\n"
        "- M (months): M, M-1, M+2\n"
        "- Y (years): Y, Y-1, Y+1\n"
        "Regular formats:\n"
        "- YYYY-MM-DD\n"
        "- YYYY/MM/DD\n"
        "- DD-MM-YYYY\n"
        "- DD/MM/YYYY\n"
        "- YYYY.MM.DD\n"
        "- DD.MM.YYYY"
    )


def time_now():
    return datetime.now(pytz.timezone(TIMEZONE))


import requests


def get_latest_numOfFail(location_tag, token) -> float:
    today = datetime.today().strftime("%Y-%m-%d")
    url_today = f"{RELIABILITY_SERVICE_API}/main/number-of-failures/{location_tag}/2016-01-01/{today}"

    try:
        response = requests.get(
            url_today,
            {
                "Content-Type": "application/json",
                "Authorization": f"Bearer {token}",
            },
        )
        data = response.json()

        raise Exception(data)

        latest_num = data["data"][-1]["num_fail"]

        raise Exception(latest_num)

        if not latest_num:
            latest_num = 0

        return latest_num
    except requests.exceptions.RequestException as e:
        print(f"Error fetching data: {e}")
        return 0



PASTEBIN_API_KEY = 'G9uCTzoI1CJWCupt5lcZRpijondt5jwA'  # Replace this with your actual API key

def save_to_pastebin(data, title="Result Log", expire_date="1H"):
    url = "https://pastebin.com/api/api_post.php"
    payload = {
        'api_dev_key': PASTEBIN_API_KEY,
        'api_option': 'paste',
        'api_paste_code': data,
        'api_paste_name': title,
        'api_paste_expire_date': expire_date,  # Example: 10M, 1H, 1D, N (never)
    }
    response = requests.post(url, data=payload)
    if response.status_code == 200:
        return response.text  # This will be the paste URL
    else:
        return f"Error: {response.status_code} - {response.text}"


def sanitize_filename(filename: str) -> str:
    """
    Sanitize the filename to ensure it is safe.
    - Remove path info.
    - Remove unsafe characters.
    - Limit length.
    """
    if not filename:
        raise ValueError("Filename cannot be empty")
        
    # Get the basename (remove any path)
    filename = os.path.basename(filename)
    
    # Remove control characters and non-printable characters
    filename = re.sub(r'[\x00-\x1f\x7f]', '', filename)

    # remove potential $( ) and ${ }
    filename = re.sub(r'\$[\(\{].*?[\)\}]', '', filename)
    
    # remove any remaining $( or ${
    filename = filename.replace('$(', '').replace('${', '')

    # Allow alphanumeric, underscore, hyphen, space, and dots
    # Remove other potentially dangerous characters.
    filename = re.sub(r'[^a-zA-Z0-9_\-\.\ ]', '', filename)
    
    # Remove consecutive dots to prevent directory traversal attempts like '..'
    filename = re.sub(r'\.{2,}', '.', filename)

    # Ensure filename is not practically empty after sanitization
    if not filename.strip() or filename.strip().replace('.', '') == '':
         raise ValueError("Filename invalid after sanitization")

    # Limit length (e.g. 200 chars)
    if len(filename) > 200:
        base, ext = os.path.splitext(filename)
        # Preserve extension if possible
        if len(ext) < 20: 
             filename = base[:(200-len(ext))] + ext
        else:
             filename = filename[:200]
        
    return filename.strip()


def get_vault_secrets(
    vault_url: str,
    role_id: str,
    secret_id: str,
    secret_path: str,
    secret_keys_to_be_returned: List[str],
    mount_point: str = "secret"
) -> Optional[Dict[str, str]]:

    try:
        client = hvac.Client(url=vault_url, verify=False)

        # Login using AppRole
        client.auth.approle.login(
            role_id=role_id,
            secret_id=secret_id
        )

        if not client.is_authenticated():
            raise Exception("Vault authentication failed")

        # Read secret
        response = client.secrets.kv.v2.read_secret_version(
            path=secret_path,
            mount_point=mount_point
        )

        secret_data = response["data"]["data"]

        # Filter only requested keys
        result = {}

        for key in secret_keys_to_be_returned:
            if key not in secret_data:
                raise KeyError(f"Key '{key}' not found in secret")
            result[key] = secret_data[key]

        return result

    except Exception as e:
        print(f"Error retrieving secret from Vault: {str(e)}")
        return None