CapacityReport/app/utils/file_dates.py

"""Filename date parsing and per-directory date filtering."""
from __future__ import annotations

import re
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Callable, Iterable, TypeVar


FILENAME_DATE_RE = re.compile(r"(?<!\d)(20\d{10}(?:\d{2})?)(?!\d)")
T = TypeVar("T")


@dataclass(frozen=True)
class DirectoryDateSelection:
    directory: str
    total_count: int
    selected_count: int
    skipped_count: int
    undated_count: int
    max_date: date | None = None
    start_date: date | None = None


@dataclass(frozen=True)
class FileDateRange:
    start: date
    end_exclusive: date
    timestamp_count: int

    @property
    def span_days(self) -> int:
        return max((self.end_exclusive - self.start).days, 1)

    def covered_days(self) -> list[date]:
        return [self.start + timedelta(days=offset) for offset in range(self.span_days)]

    def covers_any(self, target_days: set[date]) -> bool:
        return any(day in target_days for day in self.covered_days())

    def covers_all(self, target_days: set[date]) -> bool:
        return target_days.issubset(set(self.covered_days()))


def extract_file_date(filename: str | Path) -> date | None:
    """Extract the first timestamp date from a filename."""
    date_range = parse_file_date_range(filename)
    return date_range.start if date_range else None


def parse_file_date_range(filename: str | Path) -> FileDateRange | None:
    """Extract the covered natural date range from a filename.

    Files with one timestamp are treated as one-day data files. Files with two
    timestamps use the second timestamp as an exclusive end when it is midnight,
    matching names like xxx_202605110000_202605120000.zip.
    """
    values = FILENAME_DATE_RE.findall(Path(str(filename)).name)
    if not values:
        return None

    start_dt = parse_file_timestamp_datetime(values[0])
    if not start_dt:
        return None

    if len(values) == 1:
        return FileDateRange(
            start=start_dt.date(),
            end_exclusive=start_dt.date() + timedelta(days=1),
            timestamp_count=1,
        )

    end_dt = parse_file_timestamp_datetime(values[1])
    if not end_dt or end_dt <= start_dt:
        return FileDateRange(
            start=start_dt.date(),
            end_exclusive=start_dt.date() + timedelta(days=1),
            timestamp_count=len(values),
        )

    end_exclusive = end_dt.date()
    if end_dt.time() != datetime.min.time():
        end_exclusive += timedelta(days=1)
    if end_exclusive <= start_dt.date():
        end_exclusive = start_dt.date() + timedelta(days=1)

    return FileDateRange(
        start=start_dt.date(),
        end_exclusive=end_exclusive,
        timestamp_count=len(values),
    )


def parse_file_timestamp(value: str) -> date | None:
    parsed = parse_file_timestamp_datetime(value)
    return parsed.date() if parsed else None


def parse_file_timestamp_datetime(value: str) -> datetime | None:
    fmt = "%Y%m%d%H%M%S" if len(value) == 14 else "%Y%m%d%H%M"
    try:
        return datetime.strptime(value, fmt)
    except ValueError:
        return None


def required_week_days(week_offset: int = 0, today: date | None = None) -> list[date]:
    """Return Monday-Sunday dates for the target natural week.

    week_offset=0 means last week, -1 means the week before last.
    """
    base_day = today or date.today()
    this_monday = base_day - timedelta(days=base_day.weekday())
    target_monday = this_monday - timedelta(days=7 * (1 - week_offset))
    return [target_monday + timedelta(days=offset) for offset in range(7)]


def select_recent_items_by_directory(
    items: Iterable[T],
    *,
    parent_key: Callable[[T], str],
    name_key: Callable[[T], str],
    days: int = 7,
) -> tuple[list[T], list[DirectoryDateSelection]]:
    """Select the latest N natural days of dated files in each directory.

    If a directory has no dated files at all, all files in that directory are
    kept to preserve compatibility with legacy/manual data.
    """
    safe_days = max(int(days), 1)
    groups: dict[str, list[T]] = {}
    for item in items:
        groups.setdefault(parent_key(item), []).append(item)

    selected: list[T] = []
    summaries: list[DirectoryDateSelection] = []
    for directory, group_items in sorted(groups.items(), key=lambda pair: pair[0]):
        dated_items = [(item, extract_file_date(name_key(item))) for item in group_items]
        valid_dates = [file_date for _, file_date in dated_items if file_date]
        if not valid_dates:
            selected.extend(group_items)
            summaries.append(
                DirectoryDateSelection(
                    directory=directory,
                    total_count=len(group_items),
                    selected_count=len(group_items),
                    skipped_count=0,
                    undated_count=len(group_items),
                )
            )
            continue

        max_date = max(valid_dates)
        start_date = max_date - timedelta(days=safe_days - 1)
        group_selected = [
            item
            for item, file_date in dated_items
            if file_date is not None and start_date <= file_date <= max_date
        ]
        selected.extend(group_selected)
        summaries.append(
            DirectoryDateSelection(
                directory=directory,
                total_count=len(group_items),
                selected_count=len(group_selected),
                skipped_count=len(group_items) - len(group_selected),
                undated_count=sum(1 for _, file_date in dated_items if file_date is None),
                max_date=max_date,
                start_date=start_date,
            )
        )

    return selected, summaries