73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import datetime
|
||
|
||
FILE_NAME_PATTERNS = (
|
||
re.compile(r"^SKO115FZ_\d{2}_[A-Za-zА-Яа-я0-9]{9}_\d{8}_[ХX]\d{5}\.xml$"),
|
||
)
|
||
|
||
DEFAULT_OUTPUT_DIR_NAME = "Out"
|
||
DEFAULT_ERROR_DIR_NAME = "Err"
|
||
DEFAULT_PROCESSED_DIR_NAME = "Processed"
|
||
|
||
REPORT_FILE_TEMPLATE = "report_{date_key}_{run_number}.xlsx"
|
||
PROTOCOL_FILE_TEMPLATE = "protocol_{date_key}_{run_number}.xlsx"
|
||
PROCESS_LOG_FILE_TEMPLATE = "process_{date_key}_{run_number}.log"
|
||
|
||
|
||
def validate_file_name(file_name: str) -> bool:
|
||
return any(pattern.match(file_name) for pattern in FILE_NAME_PATTERNS)
|
||
|
||
|
||
def build_date_parts(now: datetime | None = None) -> tuple[str, str, str]:
|
||
current = now or datetime.now()
|
||
year = current.strftime("%Y")
|
||
month_year = current.strftime("%m_%Y")
|
||
date_key = current.strftime("%Y%m%d")
|
||
return year, month_year, date_key
|
||
|
||
|
||
def build_out_dir(base_dir: str, now: datetime | None = None) -> str:
|
||
year, month_year, date_key = build_date_parts(now)
|
||
return join_path(base_dir, DEFAULT_OUTPUT_DIR_NAME, year, month_year, date_key)
|
||
|
||
|
||
def build_run_dir(base_dir: str, run_number: int, now: datetime | None = None) -> str:
|
||
out_dir = build_out_dir(base_dir, now)
|
||
return join_path(out_dir, str(run_number))
|
||
|
||
|
||
def build_error_dir(base_dir: str) -> str:
|
||
return join_path(base_dir, DEFAULT_ERROR_DIR_NAME)
|
||
|
||
|
||
def build_processed_dir(base_dir: str) -> str:
|
||
return join_path(base_dir, DEFAULT_PROCESSED_DIR_NAME)
|
||
|
||
|
||
def build_report_file_name(date_key: str, run_number: int) -> str:
|
||
return REPORT_FILE_TEMPLATE.format(date_key=date_key, run_number=run_number)
|
||
|
||
|
||
def build_protocol_file_name(date_key: str, run_number: int) -> str:
|
||
return PROTOCOL_FILE_TEMPLATE.format(date_key=date_key, run_number=run_number)
|
||
|
||
|
||
def build_process_log_file_name(date_key: str, run_number: int) -> str:
|
||
return PROCESS_LOG_FILE_TEMPLATE.format(date_key=date_key, run_number=run_number)
|
||
|
||
|
||
def join_path(*parts: str) -> str:
|
||
cleaned = [part.strip("/\\") for part in parts if part.strip("/\\")]
|
||
return "\\".join(cleaned)
|
||
|
||
|
||
def get_parent_dir(path: str) -> str:
|
||
normalized = path.rstrip("/\\")
|
||
if "\\" in normalized:
|
||
return normalized.rsplit("\\", maxsplit=1)[0]
|
||
if "/" in normalized:
|
||
return normalized.rsplit("/", maxsplit=1)[0]
|
||
return normalized
|