feat: 导入 CapacityReport 初始源码

app/utils/file_dates.py

@@ -0,0 +1,174 @@
+"""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