150 lines
4.6 KiB
Python
150 lines
4.6 KiB
Python
"""Кодове на извлечения: PREFIX_0001_SEC_0001 и проверка на номерацията."""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
WIPED_HASH = "0" * 64
|
||
_CODE_RE = re.compile(r"^(.+)_(\d{4})_SEC_(\d{4})$")
|
||
|
||
|
||
def make_code(prefix: str, file_index: int, sec_index: int) -> str:
|
||
return f"{prefix}_{file_index:04d}_SEC_{sec_index:04d}"
|
||
|
||
|
||
def parse_code(code: str) -> Optional[tuple[str, int, int]]:
|
||
m = _CODE_RE.match(str(code or "").strip())
|
||
if not m:
|
||
return None
|
||
return m.group(1), int(m.group(2)), int(m.group(3))
|
||
|
||
|
||
def source_basename(path_str: str) -> str:
|
||
return Path(str(path_str or "").replace("\\", "/")).name
|
||
|
||
|
||
def source_identity(path: Path, input_dir: Optional[Path] = None) -> str:
|
||
"""Стабилен ключ за файла: относителен път в input_dir, иначе basename."""
|
||
if input_dir:
|
||
try:
|
||
return str(path.resolve().relative_to(input_dir.resolve())).replace("\\", "/")
|
||
except ValueError:
|
||
pass
|
||
return path.name
|
||
|
||
|
||
def numbering_report(codes: list[str]) -> dict:
|
||
"""Проверява уникална последователна SEC номерация за един файл."""
|
||
parsed: list[tuple[str, str, int, int]] = []
|
||
issues: list[str] = []
|
||
for c in codes:
|
||
p = parse_code(c)
|
||
if not p:
|
||
issues.append(f"невалиден код: {c}")
|
||
continue
|
||
parsed.append((c, p[0], p[1], p[2]))
|
||
if not codes:
|
||
return {
|
||
"count": 0,
|
||
"ok": True,
|
||
"file_index": None,
|
||
"first": None,
|
||
"last": None,
|
||
"codes": [],
|
||
"issues": [],
|
||
}
|
||
indexes = {p[2] for p in parsed}
|
||
if len(indexes) > 1:
|
||
issues.append(
|
||
"различни file_index в кодовете: " + ", ".join(str(i) for i in sorted(indexes))
|
||
)
|
||
file_index = min(indexes) if indexes else None
|
||
secs = [p[3] for p in parsed]
|
||
tally: dict[int, int] = {}
|
||
for s in secs:
|
||
tally[s] = tally.get(s, 0) + 1
|
||
dups = sorted(s for s, n in tally.items() if n > 1)
|
||
if dups:
|
||
issues.append("дублирани SEC: " + ", ".join(f"{s:04d}" for s in dups[:8]))
|
||
n = len(parsed)
|
||
expected = list(range(1, n + 1))
|
||
actual = sorted(secs)
|
||
if actual != expected:
|
||
missing = [x for x in expected if x not in tally]
|
||
if missing:
|
||
issues.append("липсващи SEC: " + ", ".join(f"{s:04d}" for s in missing[:8]))
|
||
elif actual:
|
||
issues.append(f"номерацията не е 1…{n:04d}")
|
||
codes_sorted = [p[0] for p in sorted(parsed, key=lambda x: (x[2], x[3], x[0]))]
|
||
return {
|
||
"count": len(codes),
|
||
"ok": not issues,
|
||
"file_index": file_index,
|
||
"first": codes_sorted[0] if codes_sorted else None,
|
||
"last": codes_sorted[-1] if codes_sorted else None,
|
||
"codes": codes_sorted,
|
||
"issues": issues,
|
||
}
|
||
|
||
|
||
def remove_section_outputs(
|
||
output_dir: Path,
|
||
codes: list[str],
|
||
output_paths: Optional[list[str]] = None,
|
||
) -> int:
|
||
removed = 0
|
||
images_dir = output_dir / "images"
|
||
for code in codes:
|
||
if not code:
|
||
continue
|
||
local_txt = output_dir / f"{code}.txt"
|
||
try:
|
||
if local_txt.exists():
|
||
local_txt.unlink()
|
||
removed += 1
|
||
except Exception:
|
||
pass
|
||
if images_dir.is_dir():
|
||
for img in images_dir.glob(f"{code}_*"):
|
||
try:
|
||
img.unlink()
|
||
removed += 1
|
||
except Exception:
|
||
pass
|
||
for op in output_paths or []:
|
||
if not op:
|
||
continue
|
||
try:
|
||
opath = Path(op)
|
||
stem = Path(str(op).replace("\\", "/")).stem
|
||
local_txt = output_dir / f"{stem}.txt"
|
||
if opath.exists():
|
||
try:
|
||
if not local_txt.exists() or opath.resolve() != local_txt.resolve():
|
||
opath.unlink()
|
||
removed += 1
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return removed
|
||
|
||
|
||
def file_result(
|
||
rel: str,
|
||
file_index: Optional[int],
|
||
codes: list[str],
|
||
saved: int = 0,
|
||
skipped: bool = False,
|
||
) -> dict:
|
||
report = numbering_report(codes)
|
||
if file_index is not None:
|
||
report["file_index"] = file_index
|
||
return {
|
||
"file": rel,
|
||
"saved": saved,
|
||
"skipped": skipped,
|
||
**report,
|
||
}
|