dqa/extensions/date_parser.py - master - nolizard/DQA

dqa/extensions/date_parser.py
288 строк10 KB
Mikhail Nyrkov
Initial commit
14 апр 2026, 19:01
14 апр 2026, 19:018911266
import re
from typing import Dict, List, Optional, Tuple
import warnings

import pandas as pd


def column_data_type_is_date_like(date_col: pd.Series) -> bool:
    """Checks that all values in a column are of type datetime."""
    return pd.api.types.is_datetime64_any_dtype(date_col)


def parse_date_col(date_col: pd.Series) -> pd.Series:
    """Returns a column with string to datetime conversions."""
    try:
        return pd.to_datetime(date_col, dayfirst=False, format='mixed')
    except ValueError:
        ...
    try:
        return pd.to_datetime(date_col, dayfirst=True, format='mixed')
    except ValueError:
        ...
    return parse_dates(date_col)


def parse_dates(
    date_column: pd.Series,
    date_formats: Optional[List[str]] = None,
    user_formats: Optional[List[str]] = None,
    user_systems: Optional[List[str]] = None,
) -> pd.Series:
    """Converts dates from string-type to datetime type
    using a list of formats in the order given.
    Dates that do not fit any format are converted to null.

    :param date_column: Series with string-type dates.
    :type date_column: pd.Series
    :param date_formats: list of date formats of string-type.
    :type date_formats: Optional[List[str]]
    :param user_formats: formats taken from user manually.
    :type user_formats: List
    :param user_systems: formats taken from user by system.
    :type user_systems: List
    :return: Series with datetime-type dates.
    :rtype: pd.Series
    """
    if date_column is None:
        raise AttributeError("Expected type pd.Series, not NoneType")
    if not date_formats:
        date_formats, _ = search_format(date_column)
    if user_formats:
        date_formats = __union_lists(user_formats, date_formats)
    if user_systems:
        system_formats = [__SYSTEM_FORMATS[system] for system in user_systems]
        date_formats = __union_lists(system_formats, date_formats)

    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        try:
            # весь столбец за раз
            parsed_dates = pd.to_datetime(date_column, format="mixed", errors="coerce")
        except Exception as e:
            parsed_dates = pd.to_datetime(date_column, errors="coerce")


    if date_formats:
        for date_format in date_formats:
          # NaN
            mask = parsed_dates.isna()
            if not mask.any():
                break
            parsed_dates[mask] = pd.to_datetime(
                date_column[mask], format=date_format, errors="coerce"
            )

    return parsed_dates


# TODO: поместить реальный список форматов для АС
# Обновленный словарь системных форматов
__SYSTEM_FORMATS = {
    "АС 1": "%Y-%d-%m %H:%M:%S%z",
    "АС 2": "%d.%m.%Y",
    "АС 3": "%Y.%m.%d",
    "ISO": "%Y-%m-%dT%H:%M:%S%z",
    "RFC 2822": "%a, %d %b %Y %H:%M:%S %z",
    "US Military": "%d %b %Y %H:%MZ",
    "ODBC": "%Y-%m-%d %H:%M:%S.%f",
    "Unix": "%s",
    "European Time": "%d.%m.%Y %H:%M:%S",
    "Chinese Standard": "%Y年%m月%d日",
    "Log Format": "%d/%b/%Y:%H:%M:%S %z"
}


def search_format(date_column: pd.Series) -> Tuple[List[str],
                                                   Dict[str, Dict[str, int]]]:
    """Return a sorted by occurrence list of timestamp formats
    encountered in the column and the overall rating of
    timestamp formats.
    """
    format_rating = {}
    for timestamp in date_column:
        timestamp_formats = __get_timestamp_formats(timestamp)

        if not timestamp_formats:
            continue  # TODO: keep date in a list of exceptions

        probably = __timestamp_format_defined_probably(timestamp_formats)

        format_rating = __add_format_to_rating(timestamp_formats[0],
                                               format_rating, int(probably))
        if probably:
            format_rating = __add_format_to_rating(timestamp_formats[1],
                                                   format_rating, 0)

    formats = sorted(format_rating.items(),
                     key=lambda x: x[1]['probably'],
                     reverse=True)
    # TODO: remove format_rating from returns
    return [pattern for pattern, rating in formats], format_rating


def __get_timestamp_formats(timestamp: Optional[str]) -> Optional[List[str]]:
    delimiters = re.findall(r'[^\d:AMPamp]', str(timestamp))
    timestamp_parts = re.split(r'[^\d:AMPamp]', str(timestamp))

    if (pd.isna(timestamp)
            or __timestamp_cannot_be_parsed(delimiters, timestamp_parts)):
        return

    date_format_parts_list = __get_date_format_parts_list(timestamp_parts[:3])

    time_format = ''
    if __timestamp_has_time(timestamp_parts):
        time_format = __get_time_format(timestamp_parts[3:])

    if (not date_format_parts_list
            or __timestamp_has_time(timestamp_parts) and not time_format):
        return

    timestamp_formats = []
    for date_format_parts in date_format_parts_list:
        date_format = __get_date_format(date_format_parts, delimiters)
        if time_format:
            timestamp_formats.append(f'{date_format} {time_format}')
        else:
            timestamp_formats.append(date_format)

    return timestamp_formats


def __timestamp_cannot_be_parsed(delimiters: List,
                                 timestamp_parts: List) -> bool:
    """Checks that the number of delimiters and timestamp parts is
    not enough to recognize the time format."""
    return len(delimiters) < 2 or len(timestamp_parts) < 3 \
        or not any(timestamp_parts)


def __get_date_format_parts_list(
        date_parts: List[str]) -> Optional[List[Tuple[str, str, str]]]:
    """Returns a list of date format parts based on the order of date parts."""
    for order in __SURE_DATES:
        day, month, year = order
        if __sure_date(date_parts[day], date_parts[month], date_parts[year]):
            return __SURE_DATES[order]

    for order in __PROB_DATES:
        day, month, year = order
        if __prob_date(date_parts[day], date_parts[month], date_parts[year]):
            return __PROB_DATES[order]


__SURE_DATES = {(0, 1, 2): [('%d', '%m', '%Y')],
                (0, 2, 1): [('%d', '%Y', '%m')],
                (1, 0, 2): [('%m', '%d', '%Y')],
                (1, 2, 0): [('%Y', '%d', '%m')],
                (2, 0, 1): [('%m', '%Y', '%d')],
                (2, 1, 0): [('%Y', '%m', '%d')]}


def __sure_date(day: str, month: str, year: str) -> bool:
    """Checks that date can be parsed exactly."""
    return 31 >= int(day) > 12 >= int(month) and len(year) == 4


__PROB_DATES = {(0, 1, 2): [('%d', '%m', '%Y'), ('%m', '%d', '%Y')],
                (0, 2, 1): [('%d', '%Y', '%m'), ('%m', '%Y', '%d')],
                (2, 1, 0): [('%Y', '%m', '%d'), ('%Y', '%d', '%m')]}


def __prob_date(day: str, month: str, year: str) -> bool:
    """Checks that date may not be exactly parsed."""
    return int(day) <= 12 and int(month) <= 12 and len(year) == 4


def __timestamp_has_time(timestamp_parts: List[str]) -> bool:
    return len(timestamp_parts) > 3


def __get_time_format(time_parts: List[str]) -> Optional[str]:
    """Returns time format formed based on time parts."""
    if __time_is_in_12_hour_format(time_parts):
        return __get_12_hour_time_format(time_parts)

    if __time_format_has_utc_or_microseconds(time_parts):
        return __get_time_format_with_utc_or_microseconds(time_parts)

    if __time_format_is_simple(time_parts):
        return __get_simple_time_format(time_parts)


def __time_is_in_12_hour_format(time_parts: List[str]) -> bool:
    return time_parts[-1] in ['am', 'pm', 'AM', 'PM']


def __get_12_hour_time_format(time_parts: List[str]) -> Optional[str]:
    """Returns 12-hour time format  based on the number of time parts."""
    postfix = '%P' if time_parts[-1] in ['AM', 'PM'] else '%p'
    if len(time_parts) == 3:
        return '%I:%M:%S.%f ' + postfix
    if len(time_parts) == 2:
        if len(time_parts[0].split(':')) == 3:
            return '%I:%M:%S ' + postfix
        elif len(time_parts[0].split(':')) == 2:
            return '%I:%M ' + postfix


def __time_format_has_utc_or_microseconds(time_parts: List[str]) -> bool:
    """Checks that time format has microseconds or timezone."""
    return len(time_parts) == 2


def __get_time_format_with_utc_or_microseconds(
        time_parts: List[str]) -> Optional[str]:
    """Returns time format with utc or microseconds based on
    the number of time elements.
    """
    if len(time_parts[0].split(':')) == 3:
        if ':' in time_parts[1]:
            return '%H:%M:%S:%z'
        return '%H:%M:%S.%f'


def __time_format_is_simple(time_parts: List[str]) -> bool:
    """Checks that time format has no additional attributes."""
    return len(time_parts) == 1


def __get_simple_time_format(time_parts: List[str]) -> Optional[str]:
    """Returns time format without additional attributes based on
    the number of time elements.
    """
    if len(time_parts[0].split(':')) == 3:
        return '%H:%M:%S'
    if len(time_parts[0].split(':')) == 2:
        return '%H:%M'


def __get_date_format(date_parts: Tuple[str, str, str], delimiters) -> str:
    """Returns date format formed from date format parts and delimiters."""
    return (f'{date_parts[0]}{delimiters[0]}'
            f'{date_parts[1]}{delimiters[1]}'
            f'{date_parts[2]}')


def __timestamp_format_defined_probably(timestamp_formats):
    """Checks that date format is not defined exactly."""
    return len(timestamp_formats) == 2


def __add_format_to_rating(timestamp_format: str,
                           format_rating: Dict[str, Dict[str, int]],
                           probably: int) -> Dict[str, Dict[str, int]]:
    """Adds a format score to the overall rating of formats."""
    if timestamp_format in format_rating:
        format_rating[timestamp_format]['sure'] += probably
        format_rating[timestamp_format]['probably'] += 1
    else:
        format_rating[timestamp_format] = {'sure': probably, 'probably': 1}
    return format_rating


def __union_lists(f1: List, f2: List) -> List[str]:
    """Unions lists of formats (elements of f1 have higher priority)."""
    f1.extend(x for x in f2 if x not in f1)
    return f1