sguid_tool/processor.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import date, datetime
import os
from pathlib import Path

import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from werkzeug.datastructures import FileStorage
from werkzeug.utils import secure_filename


HEADER_FILL = PatternFill(fill_type="solid", fgColor="D9EAF4")
HEADER_FONT = Font(bold=True)
CENTER_ALIGN = Alignment(horizontal="center", vertical="center")

SUPPORTED_EXTENSIONS = {".xlsx", ".csv"}
ALL_SITE_LABEL = "全站点"
INCLUDE_RAW_SHEET = os.environ.get("MRO_INCLUDE_RAW_SHEET", "1") != "0"
FAST_EXCEL_WRITER = os.environ.get("MRO_FAST_EXCEL_WRITER", "1") != "0"

LETTER_INDEX = {
    "A": 0,
    "B": 1,
    "C": 2,
    "D": 3,
    "E": 4,
    "F": 5,
    "G": 6,
    "H": 7,
    "I": 8,
    "J": 9,
    "K": 10,
    "L": 11,
    "M": 12,
    "N": 13,
    "O": 14,
    "P": 15,
    "Q": 16,
    "R": 17,
    "S": 18,
    "T": 19,
    "U": 20,
    "V": 21,
    "W": 22,
    "X": 23,
    "Y": 24,
    "Z": 25,
    "AA": 26,
    "AB": 27,
}


@dataclass(frozen=True)
class GroupingConfig:
    key: str
    display_name: str
    column_index: int
    source_description: str
    include_product_line_column: bool


GROUPING_CONFIGS = [
    GroupingConfig("SGUID", "SGUID", LETTER_INDEX["B"], "B 列 sguid", True),
    GroupingConfig("SKUID", "SKUID", LETTER_INDEX["A"], "A 列 skuid", True),
    GroupingConfig("PRODUCT_LINE", "产品线", LETTER_INDEX["E"], "E 列 产品线", False),
]

COMMON_REQUIRED_COLUMNS = {
    "负责人": LETTER_INDEX["D"],
    "产品线": LETTER_INDEX["E"],
    "channel": LETTER_INDEX["G"],
    "pl_date": LETTER_INDEX["I"],
    "sold": LETTER_INDEX["K"],
    "GMV": LETTER_INDEX["L"],
    "头程": LETTER_INDEX["V"],
    "尾程": LETTER_INDEX["W"],
    "PL": LETTER_INDEX["AB"],
}

PLTYPE_FALLBACK_INDEX = LETTER_INDEX["M"]
PLTYPE_FIELD_MAP = {
    "normal": "normal pl",
    "ads": "ADS",
    "refund": "refund",
    "amazon_storagefee": "仓储费",
    "age_storagefee": "超龄仓储费",
}
METRIC_COLUMNS = ["PL", "normal pl", "ADS", "refund", "仓储费", "超龄仓储费", "头程", "尾程"]
PERCENTAGE_COLUMNS = [f"{metric}占比" for metric in METRIC_COLUMNS]
NUMERIC_COLUMNS = ["sold", "GMV", *METRIC_COLUMNS]


class ProcessingError(Exception):
    pass


@dataclass
class ProcessingResult:
    file_path: Path
    counts: dict[str, int]
    result_type: str


def process_upload(uploaded_file: FileStorage, upload_dir: Path, output_dir: Path) -> ProcessingResult:
    extension = Path(uploaded_file.filename).suffix.lower()
    if extension not in SUPPORTED_EXTENSIONS:
        raise ProcessingError("仅支持上传 .xlsx 或 .csv 文件。")

    safe_name = secure_filename(uploaded_file.filename) or f"upload{extension}"
    upload_path = upload_dir / safe_name
    uploaded_file.save(upload_path)

    try:
        return process_saved_upload(upload_path=upload_path, output_dir=output_dir)
    finally:
        upload_path.unlink(missing_ok=True)


def process_saved_upload(upload_path: Path, output_dir: Path) -> ProcessingResult:
    extension = upload_path.suffix.lower()
    try:
        source_df = read_source_file(upload_path, extension)
        validate_dataframe(source_df)
        normalized = build_normalized_rows(source_df)
        summaries = {config.key: build_summary(normalized, config) for config in GROUPING_CONFIGS}
        output_path = build_result_workbook(source_df, summaries, output_dir)
        counts = {config.key: int(summaries[config.key].shape[0]) for config in GROUPING_CONFIGS}
        return ProcessingResult(file_path=output_path, counts=counts, result_type="success")
    except ProcessingError as exc:
        error_path = build_error_workbook(str(exc), output_dir)
        return ProcessingResult(
            file_path=error_path,
            counts={config.key: 0 for config in GROUPING_CONFIGS},
            result_type="error",
        )
    except Exception as exc:
        error_path = build_error_workbook(f"系统处理失败：{exc}", output_dir)
        return ProcessingResult(
            file_path=error_path,
            counts={config.key: 0 for config in GROUPING_CONFIGS},
            result_type="error",
        )


def read_source_file(file_path: Path, extension: str) -> pd.DataFrame:
    try:
        if extension == ".xlsx":
            dataframe = read_excel_file(file_path, header=None, dtype=object)
        else:
            dataframe = _read_csv_with_fallback(file_path)
    except UnicodeDecodeError as exc:
        raise ProcessingError("文件乱码，无法识别字符编码，请检查源文件编码格式。") from exc
    except ValueError as exc:
        raise ProcessingError("文件内容为空，无法执行汇总。") from exc
    except Exception as exc:
        raise ProcessingError(f"文件读取失败，请确认文件格式正确。原始错误：{exc}") from exc

    dataframe = dataframe.dropna(how="all").reset_index(drop=True)
    if dataframe.empty:
        raise ProcessingError("空表，未读取到任何有效数据。")
    return dataframe


def read_excel_file(file_path: Path, **kwargs) -> pd.DataFrame:
    try:
        return pd.read_excel(file_path, engine="calamine", **kwargs)
    except ImportError:
        return pd.read_excel(file_path, **kwargs)


def _read_csv_with_fallback(file_path: Path) -> pd.DataFrame:
    encodings = ["utf-8-sig", "gbk", "gb18030", "utf-8"]
    last_error: Exception | None = None
    for encoding in encodings:
        try:
            return pd.read_csv(file_path, header=None, dtype=object, encoding=encoding)
        except UnicodeDecodeError as exc:
            last_error = exc
    if last_error:
        raise last_error
    raise ProcessingError("CSV 文件读取失败。")


def validate_dataframe(dataframe: pd.DataFrame) -> None:
    required_columns = {
        "SKUID": LETTER_INDEX["A"],
        "SGUID": LETTER_INDEX["B"],
        **COMMON_REQUIRED_COLUMNS,
    }
    if dataframe.shape[1] <= max(required_columns.values()):
        missing = [name for name, index in required_columns.items() if dataframe.shape[1] <= index]
        raise ProcessingError(f"缺少必要列：{'、'.join(missing)}。请检查模板列位是否完整。")

    data_rows = dataframe.iloc[1:].copy() if looks_like_header_row(dataframe.iloc[0]) else dataframe.copy()
    if data_rows.dropna(how="all").empty:
        raise ProcessingError("空表，文件中没有可处理的数据行。")


def looks_like_header_row(first_row: pd.Series) -> bool:
    tokens = {str(value).strip().lower() for value in first_row.dropna().tolist()}
    return any(token in {"sguid", "skuid", "负责人", "产品线", "channel", "pl_date", "sale", "gmv"} for token in tokens)


def build_normalized_rows(source_df: pd.DataFrame) -> pd.DataFrame:
    data_df = source_df.iloc[1:].copy() if looks_like_header_row(source_df.iloc[0]) else source_df.copy()
    data_df = data_df.reset_index(drop=True)

    pltype_series = extract_pltype_column(data_df)
    normalized = pd.DataFrame(
        {
            "SKUID": data_df.iloc[:, LETTER_INDEX["A"]].fillna("").astype(str).str.strip(),
            "SGUID": data_df.iloc[:, LETTER_INDEX["B"]].fillna("").astype(str).str.strip(),
            "负责人": data_df.iloc[:, COMMON_REQUIRED_COLUMNS["负责人"]].fillna("").astype(str).str.strip(),
            "产品线": data_df.iloc[:, COMMON_REQUIRED_COLUMNS["产品线"]].fillna("").astype(str).str.strip(),
            "channel": data_df.iloc[:, COMMON_REQUIRED_COLUMNS["channel"]].fillna("").astype(str).str.strip(),
            "pl_date": data_df.iloc[:, COMMON_REQUIRED_COLUMNS["pl_date"]].map(normalize_pl_month),
            "sold": pd.to_numeric(data_df.iloc[:, COMMON_REQUIRED_COLUMNS["sold"]], errors="coerce").fillna(0.0),
            "GMV": pd.to_numeric(data_df.iloc[:, COMMON_REQUIRED_COLUMNS["GMV"]], errors="coerce").fillna(0.0),
            "PL": pd.to_numeric(data_df.iloc[:, COMMON_REQUIRED_COLUMNS["PL"]], errors="coerce").fillna(0.0),
            "头程": pd.to_numeric(data_df.iloc[:, COMMON_REQUIRED_COLUMNS["头程"]], errors="coerce").fillna(0.0),
            "尾程": pd.to_numeric(data_df.iloc[:, COMMON_REQUIRED_COLUMNS["尾程"]], errors="coerce").fillna(0.0),
            "pltype": pltype_series.fillna("").astype(str).str.strip().str.lower(),
        }
    )

    normalized = normalized[normalized["SKUID"].ne("") | normalized["SGUID"].ne("") | normalized["产品线"].ne("")].copy()
    if normalized.empty:
        raise ProcessingError("缺少必要列：关键分组字段数据为空，无法执行汇总。")

    return normalized.reset_index(drop=True)


def build_summary(normalized: pd.DataFrame, config: GroupingConfig) -> pd.DataFrame:
    group_label = config.display_name
    grouped_rows = normalized[normalized[group_label].ne("")].copy()
    if grouped_rows.empty:
        raise ProcessingError(f"缺少必要列：{group_label} 数据为空，无法执行汇总。")

    detail_summary = aggregate_summary(grouped_rows, group_label, include_all_sites=False)
    all_site_summary = aggregate_summary(grouped_rows, group_label, include_all_sites=True)
    summary = pd.concat([detail_summary, all_site_summary], ignore_index=True)
    summary["_channel_sort"] = summary["channel"].eq(ALL_SITE_LABEL).astype(int)
    summary = summary.sort_values(
        by=[group_label, "pl_date", "_channel_sort", "channel"],
        kind="stable",
    ).reset_index(drop=True)

    if config.key == "SKUID":
        ordered_columns = [group_label, "SGUID", "负责人", "channel", "产品线", "pl_date"]
    elif config.include_product_line_column:
        ordered_columns = [group_label, "负责人", "channel", "产品线", "pl_date"]
    else:
        ordered_columns = [group_label, "pl_date", "负责人", "channel"]
    ordered_columns.extend(
        [
            "sold",
            "GMV",
            "PL",
            "PL占比",
            "normal pl",
            "normal pl占比",
            "ADS",
            "ADS占比",
            "refund",
            "refund占比",
            "仓储费",
            "仓储费占比",
            "超龄仓储费",
            "超龄仓储费占比",
            "头程",
            "头程占比",
            "尾程",
            "尾程占比",
        ]
    )
    summary = summary[ordered_columns].copy()

    for column in NUMERIC_COLUMNS:
        if column in summary.columns:
            summary[column] = summary[column].round(2)
    for column in PERCENTAGE_COLUMNS:
        if column in summary.columns:
            summary[column] = summary[column].round(4)
    return summary


def aggregate_summary(rows: pd.DataFrame, group_label: str, include_all_sites: bool) -> pd.DataFrame:
    if include_all_sites:
        working_rows = rows.copy()
        working_rows["channel"] = ALL_SITE_LABEL
    else:
        working_rows = rows

    group_fields = [group_label, "channel", "pl_date"]
    aggregation_map: dict[str, str | callable] = {
        "负责人": "first",
        "产品线": "first",
        "sold": "sum",
        "GMV": "sum",
        "PL": "sum",
        "头程": "sum",
        "尾程": "sum",
    }
    if group_label == "SKUID":
        aggregation_map["SGUID"] = collapse_unique_text

    base_summary = working_rows.groupby(group_fields, as_index=False).agg(aggregation_map)

    pltype_summary = (
        working_rows[working_rows["pltype"].isin(PLTYPE_FIELD_MAP)]
        .assign(pl_field=working_rows["pltype"].map(PLTYPE_FIELD_MAP))
        .pivot_table(
            index=group_fields,
            columns="pl_field",
            values="PL",
            aggfunc="sum",
            fill_value=0.0,
        )
        .reset_index()
    )
    if isinstance(pltype_summary.columns, pd.MultiIndex):
        pltype_summary.columns = [
            column[-1] if isinstance(column, tuple) else column for column in pltype_summary.columns
        ]

    summary = base_summary.merge(pltype_summary, on=group_fields, how="left")
    for field_name in PLTYPE_FIELD_MAP.values():
        if field_name not in summary.columns:
            summary[field_name] = 0.0
    summary[list(PLTYPE_FIELD_MAP.values())] = summary[list(PLTYPE_FIELD_MAP.values())].fillna(0.0)

    for metric in METRIC_COLUMNS:
        ratio = pd.Series(0.0, index=summary.index, dtype=float)
        non_zero_mask = summary["GMV"] != 0
        ratio.loc[non_zero_mask] = summary.loc[non_zero_mask, metric] / summary.loc[non_zero_mask, "GMV"]
        summary[f"{metric}占比"] = ratio

    return summary


def extract_pltype_column(data_df: pd.DataFrame) -> pd.Series:
    text_scan_columns = [
        index
        for index in range(min(data_df.shape[1], LETTER_INDEX["AB"] + 1))
        if data_df.iloc[:, index].dtype == object or str(data_df.iloc[:, index].dtype) == "object"
    ]
    keywords = set(PLTYPE_FIELD_MAP.keys())

    for index in text_scan_columns:
        series = data_df.iloc[:, index].astype("string").fillna("").str.strip().str.lower()
        values = set(series[series.ne("")].head(200).tolist())
        if sum(keyword in values for keyword in keywords) >= 2:
            return series

    if data_df.shape[1] > PLTYPE_FALLBACK_INDEX:
        return data_df.iloc[:, PLTYPE_FALLBACK_INDEX].astype("string").fillna("").str.strip().str.lower()

    raise ProcessingError("缺少必要列：无法识别 pltype 列，无法计算 normal pl、ADS、refund、仓储费。")


def collapse_unique_text(series: pd.Series) -> str:
    values = [str(value).strip() for value in series.fillna("") if str(value).strip()]
    return " / ".join(dict.fromkeys(values))


def normalize_pl_month(value) -> str:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return ""
    if isinstance(value, pd.Timestamp):
        return value.strftime("%Y-%m")
    if isinstance(value, datetime):
        return value.strftime("%Y-%m")
    if isinstance(value, date):
        return value.strftime("%Y-%m")

    if isinstance(value, (int, float)) and not pd.isna(value) and value > 1000:
        try:
            excel_base = pd.Timestamp("1899-12-30")
            return (excel_base + pd.to_timedelta(float(value), unit="D")).strftime("%Y-%m")
        except Exception:
            pass

    text = str(value).strip()
    if not text:
        return ""

    parsed = pd.to_datetime(text, errors="coerce")
    if not pd.isna(parsed):
        return parsed.strftime("%Y-%m")
    return text[:7] if len(text) >= 7 else text


def build_result_workbook(source_df: pd.DataFrame, summaries: dict[str, pd.DataFrame], output_dir: Path) -> Path:
    if FAST_EXCEL_WRITER and not INCLUDE_RAW_SHEET:
        return build_summary_only_workbook(summaries, output_dir)

    workbook = Workbook()
    workbook.remove(workbook.active)

    if INCLUDE_RAW_SHEET:
        raw_sheet = workbook.create_sheet("原始数据")
        write_dataframe(raw_sheet, source_df, include_header=False)

    for config in GROUPING_CONFIGS:
        summary_sheet = workbook.create_sheet(f"{config.display_name}汇总")
        write_dataframe(
            summary_sheet,
            summaries[config.key],
            percentage_columns=PERCENTAGE_COLUMNS,
            include_header=True,
        )

    guide_sheet = workbook.create_sheet("字段说明")
    guide_sheet.append(["汇总sheet", "输出字段", "来源列 / 规则", "说明", "是否输出占比"])
    for config in GROUPING_CONFIGS:
        for row in build_field_guide_rows(config):
            guide_sheet.append(list(row))
    style_sheet(guide_sheet)

    output_path = next_available_filename(output_dir, build_daily_name("汇总"))
    workbook.save(output_path)
    return output_path


def build_summary_only_workbook(summaries: dict[str, pd.DataFrame], output_dir: Path) -> Path:
    output_path = next_available_filename(output_dir, build_daily_name("汇总"))
    guide_rows: list[tuple[str, str, str, str, str]] = []
    for config in GROUPING_CONFIGS:
        guide_rows.extend(build_field_guide_rows(config))

    guide_df = pd.DataFrame(guide_rows, columns=["汇总sheet", "输出字段", "来源列 / 规则", "说明", "是否输出占比"])

    try:
        with pd.ExcelWriter(output_path, engine="xlsxwriter") as writer:
            workbook = writer.book
            header_format = workbook.add_format({"bold": True, "bg_color": "#D9EAF4", "align": "center", "valign": "vcenter"})
            number_format = workbook.add_format({"num_format": "0.00"})
            percent_format = workbook.add_format({"num_format": "0.00%"})

            for config in GROUPING_CONFIGS:
                dataframe = summaries[config.key]
                sheet_name = f"{config.display_name}汇总"
                dataframe.to_excel(writer, sheet_name=sheet_name, index=False)
                worksheet = writer.sheets[sheet_name]
                format_xlsxwriter_sheet(worksheet, dataframe, header_format, number_format, percent_format)

            guide_df.to_excel(writer, sheet_name="字段说明", index=False)
            format_xlsxwriter_sheet(writer.sheets["字段说明"], guide_df, header_format, number_format, percent_format)
    except ImportError:
        workbook = Workbook()
        workbook.remove(workbook.active)
        for config in GROUPING_CONFIGS:
            summary_sheet = workbook.create_sheet(f"{config.display_name}汇总")
            write_dataframe(
                summary_sheet,
                summaries[config.key],
                percentage_columns=PERCENTAGE_COLUMNS,
                include_header=True,
            )
        guide_sheet = workbook.create_sheet("字段说明")
        guide_sheet.append(list(guide_df.columns))
        for row in guide_df.itertuples(index=False, name=None):
            guide_sheet.append(list(row))
        style_sheet(guide_sheet, dataframe=guide_df)
        workbook.save(output_path)

    return output_path


def format_xlsxwriter_sheet(worksheet, dataframe: pd.DataFrame, header_format, number_format, percent_format) -> None:
    for column_index, column_name in enumerate(dataframe.columns):
        width = calculate_column_width(dataframe[column_name], column_name)
        cell_format = None
        if column_name in PERCENTAGE_COLUMNS:
            cell_format = percent_format
        elif column_name in NUMERIC_COLUMNS:
            cell_format = number_format
        worksheet.set_column(column_index, column_index, width, cell_format)
        worksheet.write(0, column_index, column_name, header_format)


def calculate_column_width(series: pd.Series, column_name: object) -> int:
    sample = series.fillna("").head(500)
    data_width = 0 if sample.empty else int(sample.astype(str).map(len).max())
    return min(max(len(str(column_name)), data_width) + 4, 24)


def build_field_guide_rows(config: GroupingConfig) -> list[tuple[str, str, str, str, str]]:
    key = config.display_name
    rows = [
        (f"{key}汇总", key, config.source_description, f"按 {key} 维度分组汇总，输出字段名统一为 {key}", "否"),
        (f"{key}汇总", "负责人", "D 列", f"每个 {key} 取第一条", "否"),
        (f"{key}汇总", "channel", "G 列 channel", f"同一 {key} 在不同 channel 下分开统计，并额外追加一行“全站点”合计", "否"),
    ]
    if config.key == "SKUID":
        rows.insert(1, (f"{key}汇总", "SGUID", "B 列 sguid", "显示原始数据中该 skuid 对应的 sguid；如存在多个则按去重后拼接", "否"))

    if config.include_product_line_column:
        rows.append((f"{key}汇总", "产品线", "E 列", f"每个 {key} 取第一条", "否"))
        rows.append((f"{key}汇总", "pl_date", "I 列 pl_date", f"按 pl_date 的月份汇总 {key} 各项数据", "否"))
    else:
        rows.append((f"{key}汇总", "pl_date", "I 列 pl_date", f"按 pl_date 的月份汇总 {key} 各项数据", "否"))

    rows.extend(
        [
            (f"{key}汇总", "sold", "K 列 sold", f"按 {key} 求和", "否"),
            (f"{key}汇总", "GMV", "L 列 sale", f"按 {key} 求和；占比分母", "否"),
            (f"{key}汇总", "PL", "AB 列 pl", f"按 {key} 求和", "是"),
            (f"{key}汇总", "normal pl", "pltype = normal 时，对 pl 求和", f"按 {key} 求和", "是"),
            (f"{key}汇总", "ADS", "pltype = Ads 时，对 pl 求和", f"按 {key} 求和", "是"),
            (f"{key}汇总", "refund", "pltype = refund 时，对 pl 求和", f"按 {key} 求和", "是"),
            (f"{key}汇总", "仓储费", "pltype = amazon_storageFee 时，对 pl 求和", f"按 {key} 求和", "是"),
            (f"{key}汇总", "超龄仓储费", "pltype = age_storageFee 时，对 pl 求和", f"按 {key} 求和", "是"),
            (f"{key}汇总", "头程", "V 列 movefee", f"按 {key} 求和", "是"),
            (f"{key}汇总", "尾程", "W 列", f"按 {key} 求和", "是"),
        ]
    )
    return rows


def build_error_workbook(message: str, output_dir: Path) -> Path:
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "错误说明"
    sheet.append(["项", "内容"])
    sheet.append(["错误类型", "数据校验失败"])
    sheet.append(["错误原因", message])
    sheet.append(["建议修正方向", "请检查文件格式、必要列位、数据是否为空，以及 pltype 列是否可识别。"])
    style_sheet(sheet)

    output_path = next_available_filename(output_dir, build_daily_name("错误说明"))
    workbook.save(output_path)
    return output_path


def write_dataframe(
    worksheet,
    dataframe: pd.DataFrame,
    percentage_columns: list[str] | None = None,
    include_header: bool = True,
) -> None:
    if include_header:
        worksheet.append(list(dataframe.columns))
    for row in dataframe.fillna("").itertuples(index=False, name=None):
        worksheet.append(list(row))

    apply_number_formats(worksheet, include_header, percentage_columns)
    style_sheet(worksheet, dataframe=dataframe, has_header=include_header)


def apply_number_formats(worksheet, include_header: bool, percentage_columns: list[str] | None) -> None:
    if not include_header:
        return

    header_map = {cell.value: cell.column for cell in worksheet[1]}
    if percentage_columns:
        for column_name in percentage_columns:
            column_index = header_map.get(column_name)
            if not column_index:
                continue
            for row in range(2, worksheet.max_row + 1):
                worksheet.cell(row=row, column=column_index).number_format = "0.00%"

    for column_name in NUMERIC_COLUMNS:
        column_index = header_map.get(column_name)
        if not column_index:
            continue
        for row in range(2, worksheet.max_row + 1):
            worksheet.cell(row=row, column=column_index).number_format = "0.00"


def style_sheet(worksheet, dataframe: pd.DataFrame | None = None, has_header: bool = True) -> None:
    if worksheet.max_row == 0:
        return
    if has_header:
        for cell in worksheet[1]:
            cell.fill = HEADER_FILL
            cell.font = HEADER_FONT
            cell.alignment = CENTER_ALIGN

    widths = calculate_column_widths(dataframe, has_header)
    for column_index, width in enumerate(widths, start=1):
        worksheet.column_dimensions[get_column_letter(column_index)].width = width


def calculate_column_widths(dataframe: pd.DataFrame | None, has_header: bool) -> list[int]:
    if dataframe is None:
        return []

    sample = dataframe.fillna("")
    sample = sample.head(500)
    widths: list[int] = []
    for column_name in dataframe.columns:
        header_width = len(str(column_name)) if has_header else 0
        if sample.empty:
            data_width = 0
        else:
            data_width = sample[column_name].astype(str).map(len).max()
        widths.append(min(max(header_width, data_width) + 4, 24))
    return widths


def build_daily_name(suffix: str) -> str:
    now = datetime.now()
    return f"{now.year}.{now.month}.{now.day} {suffix}.xlsx"


def next_available_filename(output_dir: Path, base_name: str) -> Path:
    base_path = output_dir / base_name
    if not base_path.exists():
        return base_path

    stem = base_path.stem
    suffix = base_path.suffix
    counter = 2
    while True:
        candidate = output_dir / f"{stem}({counter}){suffix}"
        if not candidate.exists():
            return candidate
        counter += 1