jze9/kadastr18
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Приём документов → реестр + журнал

Browse Source 
Внесение данных из файлов приёма в реестр невостребованных документов
и заполнение журнала. Дата внесения — текущая; журнал заполняется чисто
(без строк-примеров шаблона).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jze9
2026-06-26 03:11:18 +05:00
commit b5f7060a1f
20 changed files with 1571 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
.gitignore vendored Normal file
View File

@@ -0,0 +1,35 @@
# ── Персональные данные граждан — НЕ публиковать ───────────────────────
# Реестр невостребованных документов (ФИО, адреса, кадастровые номера)
2026_ЕДИНЫЙ РЕЕСТР НЕВОСТРЕБОВАННЫХ ДОКУМЕНТОВ*.ods
# Журнал внесения и его заполненная версия
Журнал внесения*.xlsx
# Папка приёма с исходными файлами по гражданам
прием/
# Любые результаты работы программы
*_ЗАПОЛНЕНО.*
# Внутреннее задание (ТЗ)
*.docx
# ── Python ─────────────────────────────────────────────────────────────
__pycache__/
*.py[cod]
*.egg-info/
.pytest_cache/
.mypy_cache/
.ruff_cache/
# Виртуальные окружения
.venv/
venv/
env/
ENV/
# ── ОС / IDE ───────────────────────────────────────────────────────────
.DS_Store
Thumbs.db
.idea/
.vscode/
*.swp
# Временные lock-файлы LibreOffice
.~lock.*#

80
README.md Normal file
View File

@@ -0,0 +1,80 @@
# Приём документов → реестр + журнал
Программа вносит данные о принятых документах из файлов в папке `прием`
в файл **2026_ЕДИНЫЙ РЕЕСТР НЕВОСТРЕБОВАННЫХ ДОКУМЕНТОВ** на соответствующие
листы (последующими записями) и заполняет **Журнал внесения**.
Работает на чистых Python-библиотеках (`odfpy`, `openpyxl`, `xlrd`),
**LibreOffice не требуется**. Оформление реестра (стили ячеек, формат дат,
прочие листы) сохраняется: новые строки копируют стиль последней строки листа.
## Запуск
Графический интерфейс (выбор файлов мышью, кнопка «Внести данные»):
```bash
pip install -r requirements.txt
python3 gui.py
```
Или из командной строки (по умолчанию берёт файлы из текущей папки):
```bash
python3 run.py
```
Оригиналы не изменяются — результат пишется рядом:
* `2026_ЕДИНЫЙ РЕЕСТР НЕВОСТРЕБОВАННЫХ ДОКУМЕНТОВ._ЗАПОЛНЕНО.ods`
* `Журнал внесения_ЗАПОЛНЕНО.xlsx`
Скрипт идемпотентен: всегда читает оригиналы, повторный запуск не задваивает данные.
## Правило сопоставления
Файл `DD.MM.YYYY_74_NN.<ext>` из папки `ТОk` → лист `ТОk_NN` реестра.
По заданию: **«внести все строки, начиная со второй»**. Поэтому пропускается
только шапка (строка 1), а каждая последующая **непустая** строка вносится как
отдельная запись — без фильтрации по № п/п. № п/п в реестре проставляется
заново (сквозной, продолжает нумерацию листа), исходный № отбрасывается.
Колонки источника B…M переносятся **позиционно** в одноимённые колонки листа.
Значения ячеек очищаются от внешних пробелов (единообразно для всех форматов).
## Данные правее колонки M
В реестре 13 колонок (A…M). Если в источнике есть данные в 14-й колонке
(напр. «закладная погашена» в `16.04.2026_74_33`), отдельной колонки для них
нет — они добавляются в «Примечание» той же строки. Все такие случаи
перечисляются в `Pipeline().run().notes` (информационно).
## Известное ограничение
Сопоставление колонок — **позиционное** (колонка N источника → колонка N листа).
Это верно для текущего шаблона приёма. Если появится файл с другим порядком
или набором колонок (сдвиг влево), значения попадут не в свои поля, и это
**не** будет обнаружено автоматически — сдвиг вправо/лишние колонки ловятся
предупреждением, сдвиг влево — нет. При смене шаблона нужна проверка по
заголовкам.
## Структура
```
gui.py десктоп-интерфейс (tkinter)
run.py точка входа из командной строки
priem_register/
config.py пути, константы, раскладка колонок A..M
models.py Record, SourceFile, JournalEntry
mapping.py поиск файлов приёма и привязка к листам
readers/ чтение источников по форматам
base.py SourceReader: пропуск шапки, все непустые строки, нарезка колонок
xls_reader.py .xls (xlrd; даты-сериалы Excel)
xlsx_reader.py .xlsx (openpyxl)
ods_reader.py .ods (odfpy; разворот повторов)
factory.py выбор читателя по расширению
writers/
register_writer.py дозапись строк в .ods с копированием стилей
journal_writer.py заполнение журнала .xlsx
pipeline.py оркестратор
```

286
gui.py Normal file
View File

@@ -0,0 +1,286 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Десктоп-интерфейс для внесения принятых документов в реестр.
Запуск: python3 gui.py
Окно позволяет выбрать реестр, журнал и папку «прием», задать дату внесения
и нажать «Внести данные». Обработка идёт в фоновом потоке, поэтому окно не
зависает; ход работы и аномалии видны в логе. Результат пишется рядом с
исходными файлами с суффиксом «_ЗАПОЛНЕНО» — оригиналы не изменяются.
"""
from __future__ import annotations
import datetime
import logging
import queue
import threading
import tkinter as tk
from pathlib import Path
from tkinter import filedialog, messagebox, ttk
from priem_register import Pipeline, config
log = logging.getLogger("priem")
PAD = 8
def derive_output(path: str) -> Path:
"""Путь результата: рядом с исходным, с суффиксом «_ЗАПОЛНЕНО»."""
p = Path(path)
return p.with_name(p.stem + "_ЗАПОЛНЕНО" + p.suffix)
class QueueHandler(logging.Handler):
"""Перекладывает записи лога в очередь — читает их главный поток GUI."""
def __init__(self, q: "queue.Queue"):
super().__init__()
self.q = q
def emit(self, record: logging.LogRecord) -> None:
self.q.put(("log", record.levelno, self.format(record)))
class App(tk.Tk):
def __init__(self):
super().__init__()
self.title("Приём документов → реестр")
self.geometry("780x580")
self.minsize(680, 500)
self.msg_queue: "queue.Queue" = queue.Queue()
self.worker: threading.Thread | None = None
self.register_var = tk.StringVar(value=str(config.REGISTER_SRC))
self.journal_var = tk.StringVar(value=str(config.JOURNAL_SRC))
self.priem_var = tk.StringVar(value=str(config.PRIEM_DIR))
self.date_var = tk.StringVar(value=config.ENTRY_DATE.strftime("%d.%m.%Y"))
self._build_form()
self._build_log()
self._attach_logging()
self.after(120, self._poll)
# --- построение интерфейса ------------------------------------------
def _build_form(self) -> None:
frm = ttk.Frame(self, padding=PAD)
frm.pack(fill="x")
frm.columnconfigure(1, weight=1)
self._file_row(frm, 0, "Реестр (.ods):", self.register_var,
[("Таблицы ODF", "*.ods"), ("Все файлы", "*.*")])
self._file_row(frm, 1, "Журнал (.xlsx):", self.journal_var,
[("Книги Excel", "*.xlsx"), ("Все файлы", "*.*")])
self._dir_row(frm, 2, "Папка «прием»:", self.priem_var)
ttk.Label(frm, text="Дата внесения:").grid(
row=3, column=0, sticky="w", pady=4)
ttk.Entry(frm, textvariable=self.date_var, width=14).grid(
row=3, column=1, sticky="w", pady=4)
ttk.Label(frm, text="дд.мм.гггг", foreground="#888").grid(
row=3, column=2, sticky="w")
bar = ttk.Frame(self, padding=(PAD, 0))
bar.pack(fill="x")
self.run_btn = ttk.Button(bar, text="Внести данные", command=self._start)
self.run_btn.pack(side="left", pady=PAD)
self.progress = ttk.Progressbar(bar, mode="indeterminate", length=180)
self.status = ttk.Label(bar, text="Готово к работе", foreground="#444")
self.status.pack(side="right")
def _file_row(self, frm, row, label, var, patterns) -> None:
ttk.Label(frm, text=label).grid(row=row, column=0, sticky="w", pady=4)
ttk.Entry(frm, textvariable=var).grid(row=row, column=1, sticky="ew", pady=4)
ttk.Button(frm, text="Обзор…",
command=lambda: self._pick_file(var, patterns)).grid(
row=row, column=2, padx=(PAD, 0))
def _dir_row(self, frm, row, label, var) -> None:
ttk.Label(frm, text=label).grid(row=row, column=0, sticky="w", pady=4)
ttk.Entry(frm, textvariable=var).grid(row=row, column=1, sticky="ew", pady=4)
ttk.Button(frm, text="Обзор…",
command=lambda: self._pick_dir(var)).grid(
row=row, column=2, padx=(PAD, 0))
def _build_log(self) -> None:
frm = ttk.Frame(self, padding=PAD)
frm.pack(fill="both", expand=True)
ttk.Label(frm, text="Ход работы:").pack(anchor="w")
self.log_text = tk.Text(frm, height=16, wrap="word", state="disabled",
font=("monospace", 10))
scroll = ttk.Scrollbar(frm, command=self.log_text.yview)
self.log_text.configure(yscrollcommand=scroll.set)
self.log_text.pack(side="left", fill="both", expand=True)
scroll.pack(side="right", fill="y")
self.log_text.tag_configure("warn", foreground="#b35900")
self.log_text.tag_configure("err", foreground="#c0392b")
self.log_text.tag_configure("ok", foreground="#1e7d34")
def _attach_logging(self) -> None:
handler = QueueHandler(self.msg_queue)
handler.setFormatter(logging.Formatter("%(message)s"))
log.setLevel(logging.INFO)
log.addHandler(handler)
# --- выбор файлов ---------------------------------------------------
def _pick_file(self, var, patterns) -> None:
start = Path(var.get()).parent if var.get() else Path.cwd()
path = filedialog.askopenfilename(initialdir=str(start), filetypes=patterns)
if path:
var.set(path)
def _pick_dir(self, var) -> None:
start = var.get() or str(Path.cwd())
path = filedialog.askdirectory(initialdir=start)
if path:
var.set(path)
# --- запуск обработки -----------------------------------------------
def _start(self) -> None:
if self.worker and self.worker.is_alive():
return
try:
params = self._validate()
except ValueError as exc:
messagebox.showerror("Проверьте поля", str(exc))
return
self._clear_log()
self.run_btn.configure(state="disabled")
self.progress.pack(side="left", padx=PAD)
self.progress.start(12)
self.status.configure(text="Обработка…", foreground="#444")
self.worker = threading.Thread(target=self._work, args=(params,), daemon=True)
self.worker.start()
def _validate(self) -> dict:
register = self.register_var.get().strip()
journal = self.journal_var.get().strip()
priem = self.priem_var.get().strip()
if not (register and journal and priem):
raise ValueError("Заполните все три пути: реестр, журнал и папку «прием».")
if not Path(register).is_file():
raise ValueError("Файл реестра не найден:\n%s" % register)
if not Path(journal).is_file():
raise ValueError("Файл журнала не найден:\n%s" % journal)
if not Path(priem).is_dir():
raise ValueError("Папка «прием» не найдена:\n%s" % priem)
raw = self.date_var.get().strip()
try:
date = datetime.datetime.strptime(raw, "%d.%m.%Y").date()
except ValueError:
raise ValueError("Дата должна быть в формате дд.мм.гггг, например 08.06.2026.")
return {
"register_src": register, "journal_src": journal,
"priem_dir": priem, "entry_date": date,
"register_out": derive_output(register),
"journal_out": derive_output(journal),
}
def _work(self, params: dict) -> None:
"""Выполняется в фоновом потоке. Связь с GUI — только через очередь."""
try:
pipeline = Pipeline(
register_src=params["register_src"],
register_out=params["register_out"],
journal_src=params["journal_src"],
journal_out=params["journal_out"],
priem_dir=params["priem_dir"],
entry_date=params["entry_date"],
)
result = pipeline.run()
self.msg_queue.put(("done", {
"total": result.total,
"files": len(result.entries),
"notes": list(result.notes),
"warnings": list(result.warnings),
"register_out": str(params["register_out"]),
"journal_out": str(params["journal_out"]),
}))
except Exception as exc: # noqa: BLE001 — показываем пользователю
log.exception("Сбой обработки")
self.msg_queue.put(("error", str(exc)))
# --- обновление GUI из очереди --------------------------------------
def _poll(self) -> None:
try:
while True:
kind, *payload = self.msg_queue.get_nowait()
if kind == "log":
levelno, text = payload
tag = "warn" if levelno >= logging.WARNING else None
self._append(text, tag)
elif kind == "done":
self._on_done(payload[0])
elif kind == "error":
self._on_error(payload[0])
except queue.Empty:
pass
self.after(120, self._poll)
def _on_done(self, info: dict) -> None:
self._finish()
n_warn = len(info["warnings"])
n_note = len(info["notes"])
if n_warn:
self.status.configure(text="Готово, требуют внимания: %d" % n_warn,
foreground="#b35900")
self._append("\nВыполнено. %d строк не привязаны — проверьте выше."
% n_warn, "warn")
else:
self.status.configure(text="Готово", foreground="#1e7d34")
self._append("\nВыполнено. Все данные внесены.", "ok")
if n_note:
self._append("Присоединено заметок к записям: %d (см. «•» выше)." % n_note)
self._append(
"\nРеестр: %s\nЖурнал: %s" % (info["register_out"], info["journal_out"]))
summary = ("Внесено записей: %d\nОбработано файлов: %d\n"
"Присоединено заметок: %d\nНе привязано строк: %d\n\n"
"Реестр: %s\nЖурнал: %s") % (
info["total"], info["files"], n_note, n_warn,
info["register_out"], info["journal_out"])
if n_warn:
summary += ("\n\n%d строк не удалось привязать к записи "
"(см. лог) — проверьте вручную." % n_warn)
messagebox.showwarning("Готово (есть непривязанные строки)", summary)
else:
messagebox.showinfo("Готово", summary)
def _on_error(self, message: str) -> None:
self._finish()
self.status.configure(text="Ошибка", foreground="#c0392b")
self._append("\nОШИБКА: %s" % message, "err")
messagebox.showerror("Ошибка обработки", message)
def _finish(self) -> None:
self.progress.stop()
self.progress.pack_forget()
self.run_btn.configure(state="normal")
# --- вывод в лог ----------------------------------------------------
def _append(self, text: str, tag: str | None = None) -> None:
self.log_text.configure(state="normal")
self.log_text.insert("end", text + "\n", (tag,) if tag else ())
self.log_text.see("end")
self.log_text.configure(state="disabled")
def _clear_log(self) -> None:
self.log_text.configure(state="normal")
self.log_text.delete("1.0", "end")
self.log_text.configure(state="disabled")
def main() -> None:
App().mainloop()
if __name__ == "__main__":
main()

5
priem_register/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
# -*- coding: utf-8 -*-
"""Приём документов: внесение в реестр и заполнение журнала."""
from .pipeline import Pipeline, Result
__all__ = ["Pipeline", "Result"]

49
priem_register/config.py Normal file
View File

@@ -0,0 +1,49 @@
# -*- coding: utf-8 -*-
"""Конфигурация: пути, константы, раскладка колонок."""
from __future__ import annotations
import datetime
import re
from pathlib import Path
# Корень проекта (на уровень выше пакета).
BASE_DIR = Path(__file__).resolve().parent.parent
# Исходный реестр и файл-результат (оригинал не перезаписываем).
REGISTER_SRC = BASE_DIR / "2026_ЕДИНЫЙ РЕЕСТР НЕВОСТРЕБОВАННЫХ ДОКУМЕНТОВ..ods"
REGISTER_OUT = BASE_DIR / "2026_ЕДИНЫЙ РЕЕСТР НЕВОСТРЕБОВАННЫХ ДОКУМЕНТОВ._ЗАПОЛНЕНО.ods"
# Журнал внесения.
JOURNAL_SRC = BASE_DIR / "Журнал внесения.xlsx"
JOURNAL_OUT = BASE_DIR / "Журнал внесения_ЗАПОЛНЕНО.xlsx"
# Папка с файлами приёма.
PRIEM_DIR = BASE_DIR / "прием"
# Дата внесения, которая проставляется в журнал (по умолчанию — сегодня).
ENTRY_DATE = datetime.date.today()
# Раскладка листа реестра: 13 колонок A..M.
# 0:A № п/п (нумеруется заново, сквозная)
# 1:B Адрес хранения
# 2:C Рег. номер дела
# 3:D Объект недвижимости
# 4:E Кол-во документов
# 5:F Дата получения
# 6:G Направление (филиал субъекта РФ)
# 7:H Получение (филиал субъекта РФ)
# 8:I Направление (тер. подразделение)
# 9:J Получение (тер. подразделение)
# 10:K Дата выдачи
# 11:L Примечание
# 12:M Номер реестра
REGISTER_COLUMNS = 13
# Колонка № п/п в источнике (пропускается, заменяется сквозной нумерацией).
SOURCE_NUMBER_COL = 0
# Шаблон имени файла приёма: DD.MM.YYYY_74_NN.<ext>
FILENAME_RE = re.compile(r"^(\d{2})\.(\d{2})\.(\d{4})_74_(\d+)\.(ods|xls|xlsx)$", re.I)
# Номер ТО берётся из пути .../ТОk/...
TO_DIR_RE = re.compile(r"[/\\]ТО(\d+)[/\\]")

44
priem_register/mapping.py Normal file
View File

@@ -0,0 +1,44 @@
# -*- coding: utf-8 -*-
"""Поиск файлов приёма и сопоставление их с листами реестра.
Правило: файл DD.MM.YYYY_74_NN.<ext> из папки ТОk -> лист ТОk_NN.
"""
from __future__ import annotations
import datetime
from pathlib import Path
from typing import List
from .config import FILENAME_RE, PRIEM_DIR, TO_DIR_RE
from .models import SourceFile
def discover_sources(priem_dir: Path = PRIEM_DIR) -> List[SourceFile]:
priem_dir = Path(priem_dir)
if not priem_dir.is_dir():
raise FileNotFoundError("Папка приёма не найдена: %s" % priem_dir)
sources: List[SourceFile] = []
for path in sorted(priem_dir.rglob("*")):
if not path.is_file():
continue
match = FILENAME_RE.match(path.name)
if not match:
continue
dd, mm, yyyy, nn, _ext = match.groups()
to_match = TO_DIR_RE.search(str(path))
if not to_match:
raise ValueError("Не удалось определить ТО для файла: %s" % path)
try:
date = datetime.date(int(yyyy), int(mm), int(dd))
except ValueError as exc:
raise ValueError(
"Некорректная дата в имени файла %s: %s" % (path.name, exc)) from exc
sources.append(SourceFile(
path=path,
date=date,
sheet="ТО%s_%s" % (to_match.group(1), nn),
name=path.stem,
))
# порядок внесения: по листу, затем по дате (последующие записи)
sources.sort(key=lambda s: (s.sheet, s.date))
return sources

58
priem_register/models.py Normal file
View File

@@ -0,0 +1,58 @@
# -*- coding: utf-8 -*-
"""Доменные модели."""
from __future__ import annotations
import datetime
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional, Union
# Значение ячейки: число, дата, строка либо пусто.
CellValue = Optional[Union[str, int, float, datetime.date]]
@dataclass
class Record:
"""Одна принятая запись — значения колонок B..M (без № п/п)."""
values: List[CellValue] = field(default_factory=list)
def value(self, register_col: int) -> CellValue:
"""Значение для колонки реестра по её индексу (1=B .. 12=M)."""
idx = register_col - 1
if 0 <= idx < len(self.values):
return self.values[idx]
return None
@dataclass
class ReadResult:
"""Результат чтения источника.
records — записи для внесения в реестр;
notes — что было присоединено к записям (информационно, не ошибки);
warnings — данные, которые присоединить не к чему (требуют внимания).
"""
records: List["Record"] = field(default_factory=list)
notes: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list)
@dataclass
class SourceFile:
"""Файл приёма и его привязка к листу реестра."""
path: Path
date: datetime.date
sheet: str # имя листа в реестре, напр. "ТО1_25"
name: str # имя файла без расширения, напр. "09.04.2026_74_25"
@dataclass
class JournalEntry:
"""Строка журнала внесения."""
date: datetime.date
file_name: str
count: int

97
priem_register/pipeline.py Normal file
View File

@@ -0,0 +1,97 @@
# -*- coding: utf-8 -*-
"""Оркестратор: связывает чтение источников, запись реестра и журнала."""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import List
from . import config
from .mapping import discover_sources
from .models import JournalEntry
from .readers import get_reader
from .writers import JournalWriter, RegisterWriter
log = logging.getLogger("priem")
@dataclass
class Result:
entries: List[JournalEntry]
notes: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list)
@property
def total(self) -> int:
return sum(e.count for e in self.entries)
class Pipeline:
def __init__(
self,
register_src: Path = config.REGISTER_SRC,
register_out: Path = config.REGISTER_OUT,
journal_src: Path = config.JOURNAL_SRC,
journal_out: Path = config.JOURNAL_OUT,
priem_dir: Path = config.PRIEM_DIR,
entry_date=config.ENTRY_DATE,
):
self.register_src = Path(register_src)
self.register_out = Path(register_out)
self.journal_src = Path(journal_src)
self.journal_out = Path(journal_out)
self.priem_dir = Path(priem_dir)
self.entry_date = entry_date
if self.register_out == self.register_src:
raise ValueError("register_out совпадает с оригиналом — он был бы перезаписан")
if self.journal_out == self.journal_src:
raise ValueError("journal_out совпадает с оригиналом — он был бы перезаписан")
def run(self) -> Result:
sources = discover_sources(self.priem_dir)
log.info("Найдено файлов приёма: %d", len(sources))
register = RegisterWriter(self.register_src)
entries: List[JournalEntry] = []
notes: List[str] = []
warnings: List[str] = []
for src in sources:
if not register.has_sheet(src.sheet):
raise ValueError(
"В реестре нет листа '%s' (файл %s)" % (src.sheet, src.name))
try:
read = get_reader(src.path).read()
except Exception as exc:
raise RuntimeError(
"Не удалось прочитать файл %s: %s" % (src.path.name, exc)) from exc
count = register.append(src.sheet, read.records)
entries.append(JournalEntry(self.entry_date, src.name, count))
log.info(" %-22s -> %-8s : %3d записей", src.name, src.sheet, count)
for n in read.notes:
notes.append("%s: %s" % (src.name, n))
log.info(" • %s", n)
for w in read.warnings:
warnings.append("%s: %s" % (src.name, w))
log.warning(" ⚠ %s", w)
register.save(self.register_out)
log.info("Реестр сохранён: %s", self.register_out.name)
journal = JournalWriter(self.journal_src)
journal.write(entries)
journal.save(self.journal_out)
log.info("Журнал сохранён: %s", self.journal_out.name)
result = Result(entries, notes, warnings)
log.info("ИТОГО записей: %d", result.total)
if notes:
log.info("Присоединено заметок: %d (данные не потеряны, см. выше)",
len(notes))
if warnings:
log.warning("ВНИМАНИЕ: %d строк НЕ удалось привязать — проверьте вручную:",
len(warnings))
for w in warnings:
log.warning(" • %s", w)
return result

6
priem_register/readers/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
"""Читатели исходных файлов приёма."""
from .base import SourceReader
from .factory import get_reader
__all__ = ["SourceReader", "get_reader"]

86
priem_register/readers/base.py Normal file
View File

@@ -0,0 +1,86 @@
# -*- coding: utf-8 -*-
"""Базовый класс читателя источника."""
from __future__ import annotations
import abc
import datetime
from pathlib import Path
from typing import List
from ..config import REGISTER_COLUMNS, SOURCE_NUMBER_COL
from ..models import CellValue, ReadResult, Record
# Позиция колонки «Примечание» (L) в раскладке реестра (A=0 … M=12).
NOTE_COL = 11
class SourceReader(abc.ABC):
"""Читает файл приёма и возвращает записи для внесения.
По заданию: «внести все строки, начиная со второй». Поэтому пропускается
только шапка (строка 1), а каждая последующая непустая строка вносится как
отдельная запись — без фильтрации по № п/п. № п/п в реестре назначается
заново (сквозной), исходный № отбрасывается.
В реестре 13 колонок (A..M). Если в источнике есть данные правее колонки M
(например, безымянная 14-я колонка), переносить их в отдельную колонку
некуда — они добавляются в «Примечание» той же строки, а факт фиксируется
в ``notes`` (информационно).
Подклассы реализуют только :meth:`_rows` — выдачу «сырых» строк таблицы
(список типизированных значений ячеек) первого листа.
"""
def __init__(self, path: Path):
self.path = Path(path)
@abc.abstractmethod
def _rows(self) -> List[List[CellValue]]:
"""Сырые строки первого листа, типизированные значения ячеек."""
def read(self) -> ReadResult:
result = ReadResult()
for index, row in enumerate(self._rows()):
if index == 0:
continue # шапка (строка 1)
if not self._has_content(row):
continue # пустые строки не вносим
record = Record(values=list(row[SOURCE_NUMBER_COL + 1: REGISTER_COLUMNS]))
self._fold_overflow(row, record, result, index + 1)
result.records.append(record)
return result
def _fold_overflow(self, row: List[CellValue], record: Record,
result: ReadResult, human_row: int) -> None:
overflow = [v for i, v in enumerate(row)
if i >= REGISTER_COLUMNS and v not in (None, "")]
for value in overflow:
_merge_cell(record, NOTE_COL, value)
if overflow:
result.notes.append(
"строка %d: данные за колонкой M добавлены в «Примечание»: %s"
% (human_row, ", ".join(_as_text(v) for v in overflow)))
@staticmethod
def _has_content(row: List[CellValue]) -> bool:
return any(v not in (None, "") for v in row)
def _merge_cell(record: Record, position: int, value: CellValue) -> None:
"""Записать значение в колонку записи; занятое поле дополнить через «; »."""
idx = position - 1
while len(record.values) <= idx:
record.values.append(None)
current = record.values[idx]
if current in (None, ""):
record.values[idx] = value
else:
record.values[idx] = "%s; %s" % (_as_text(current), _as_text(value))
def _as_text(value: CellValue) -> str:
if isinstance(value, datetime.date): # покрывает и datetime
return value.strftime("%d.%m.%Y")
if isinstance(value, float) and value.is_integer():
return str(int(value))
return str(value)

24
priem_register/readers/factory.py Normal file
View File

@@ -0,0 +1,24 @@
# -*- coding: utf-8 -*-
"""Выбор читателя по расширению файла."""
from __future__ import annotations
from pathlib import Path
from .base import SourceReader
from .ods_reader import OdsReader
from .xls_reader import XlsReader
from .xlsx_reader import XlsxReader
_READERS = {
".ods": OdsReader,
".xls": XlsReader,
".xlsx": XlsxReader,
}
def get_reader(path: Path) -> SourceReader:
ext = Path(path).suffix.lower()
try:
return _READERS[ext](path)
except KeyError:
raise ValueError("Неподдерживаемый формат: %s" % ext)

86
priem_register/readers/ods_reader.py Normal file
View File

@@ -0,0 +1,86 @@
# -*- coding: utf-8 -*-
"""Читатель .ods (odfpy). Разворачивает повторяющиеся ячейки/строки."""
from __future__ import annotations
import datetime
from typing import List
from odf.opendocument import load
from odf.table import Table, TableCell, TableRow
from odf.teletype import extractText
from odf.text import P
from ..models import CellValue
from .base import SourceReader
# Пределы на «бесконечные» повторения пустого хвоста таблицы.
_MAX_COL_REPEAT = 200
_MAX_ROW_REPEAT = 200
class OdsReader(SourceReader):
def _rows(self) -> List[List[CellValue]]:
doc = load(str(self.path))
table = doc.spreadsheet.getElementsByType(Table)[0]
rows: List[List[CellValue]] = []
for tr in table.getElementsByType(TableRow):
row = self._row_values(tr)
row_rep = _repeat(tr, "numberrowsrepeated", _MAX_ROW_REPEAT)
for _ in range(row_rep):
rows.append(list(row))
_trim_trailing_empty(rows)
return rows
@staticmethod
def _row_values(tr: TableRow) -> List[CellValue]:
values: List[CellValue] = []
for cell in tr.getElementsByType(TableCell):
col_rep = _repeat(cell, "numbercolumnsrepeated", _MAX_COL_REPEAT)
value = _cell_value(cell)
values.extend([value] * col_rep)
return values
def _repeat(element, attr: str, limit: int) -> int:
raw = element.getAttribute(attr)
n = int(raw) if raw else 1
# огромные повторения — это пустой хвост листа, не размножаем
return n if n <= limit else 1
def _cell_value(cell: TableCell) -> CellValue:
vtype = cell.getAttribute("valuetype")
if vtype in ("float", "currency", "percentage"):
raw = cell.getAttribute("value")
if raw is not None:
num = float(raw)
return int(num) if num.is_integer() else num
if vtype == "date":
raw = cell.getAttribute("datevalue")
if raw:
return _parse_date(raw)
text = _cell_text(cell).strip()
return text or None
def _cell_text(cell: TableCell) -> str:
# extractText корректно собирает текст из <text:span>, <text:s> (кратные
# пробелы), <text:tab> и <text:line-break> — ручной обход childNodes их терял.
return "\n".join(extractText(p) for p in cell.getElementsByType(P))
def _parse_date(raw: str):
for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"):
try:
dt = datetime.datetime.strptime(raw, fmt)
if dt.hour == dt.minute == dt.second == 0:
return dt.date()
return dt
except ValueError:
continue
return raw
def _trim_trailing_empty(rows: List[List[CellValue]]) -> None:
while rows and all(v in (None, "") for v in rows[-1]):
rows.pop()

44
priem_register/readers/xls_reader.py Normal file
View File

@@ -0,0 +1,44 @@
# -*- coding: utf-8 -*-
"""Читатель .xls (xlrd). Корректно разбирает даты-сериалы Excel."""
from __future__ import annotations
from typing import List
import xlrd
from ..models import CellValue
from .base import SourceReader
class XlsReader(SourceReader):
def _rows(self) -> List[List[CellValue]]:
book = xlrd.open_workbook(self.path)
try:
sheet = book.sheet_by_index(0)
rows: List[List[CellValue]] = []
for r in range(sheet.nrows):
row: List[CellValue] = []
for c in range(sheet.ncols):
row.append(self._cell(book, sheet.cell(r, c)))
rows.append(row)
return rows
finally:
book.release_resources()
@staticmethod
def _cell(book, cell) -> CellValue:
t = cell.ctype
if t == xlrd.XL_CELL_EMPTY or t == xlrd.XL_CELL_BLANK:
return None
if t == xlrd.XL_CELL_DATE:
dt = xlrd.xldate_as_datetime(cell.value, book.datemode)
# дата без времени -> date
if dt.hour == dt.minute == dt.second == 0:
return dt.date()
return dt
if t == xlrd.XL_CELL_NUMBER:
return cell.value
if t == xlrd.XL_CELL_BOOLEAN:
return bool(cell.value)
text = str(cell.value).strip()
return text or None

30
priem_register/readers/xlsx_reader.py Normal file
View File

@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""Читатель .xlsx (openpyxl)."""
from __future__ import annotations
from typing import List
import openpyxl
from ..models import CellValue
from .base import SourceReader
class XlsxReader(SourceReader):
def _rows(self) -> List[List[CellValue]]:
wb = openpyxl.load_workbook(self.path, data_only=True, read_only=True)
try:
ws = wb.worksheets[0]
rows: List[List[CellValue]] = []
for row in ws.iter_rows(values_only=True):
rows.append([_norm(v) for v in row])
return rows
finally:
wb.close()
def _norm(value: CellValue) -> CellValue:
if isinstance(value, str):
value = value.strip()
return value or None
return value

6
priem_register/writers/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
"""Запись результатов в реестр и журнал."""
from .journal_writer import JournalWriter
from .register_writer import RegisterWriter
__all__ = ["RegisterWriter", "JournalWriter"]

52
priem_register/writers/journal_writer.py Normal file
View File

@@ -0,0 +1,52 @@
# -*- coding: utf-8 -*-
"""Заполнение журнала внесения (.xlsx) через openpyxl.
Пишет строки (Дата внесения, Файл внесения, Количество записей) сразу под
шапкой, сохраняя её оформление. Строки-примеры из шаблона (напр. «Например»)
затираются — в результате остаются только реальные записи.
"""
from __future__ import annotations
from pathlib import Path
from typing import List
import openpyxl
from ..models import JournalEntry
_DATE_FORMAT = "DD.MM.YYYY"
# Заголовок шапки в колонке A, по которому она опознаётся.
_HEADER_MARKER = "дата внесения"
class JournalWriter:
def __init__(self, path: Path):
self.wb = openpyxl.load_workbook(path)
self.ws = self.wb.active
def write(self, entries: List[JournalEntry]) -> None:
row = self._header_row() + 1
self._clear_below(row)
for entry in entries:
date_cell = self.ws.cell(row=row, column=1, value=entry.date)
date_cell.number_format = _DATE_FORMAT
self.ws.cell(row=row, column=2, value=entry.file_name)
self.ws.cell(row=row, column=3, value=entry.count)
row += 1
def save(self, path: Path) -> None:
self.wb.save(path)
def _header_row(self) -> int:
"""Строка шапки (где в колонке A «Дата внесения»); по умолчанию 1."""
for row in range(1, self.ws.max_row + 1):
a = self.ws.cell(row=row, column=1).value
if isinstance(a, str) and a.strip().lower() == _HEADER_MARKER:
return row
return 1
def _clear_below(self, start: int) -> None:
"""Очистить значения колонок A..C во всех строках от start и ниже."""
for row in range(start, self.ws.max_row + 1):
for col in (1, 2, 3):
self.ws.cell(row=row, column=col).value = None

180
priem_register/writers/register_writer.py Normal file
View File

@@ -0,0 +1,180 @@
# -*- coding: utf-8 -*-
"""Запись принятых документов в реестр (.ods) через odfpy.
Новые строки дописываются сразу после последней нумерованной строки листа.
Стиль каждой ячейки (границы, формат даты/числа) и стиль строки копируются
из этой последней строки-образца, поэтому оформление реестра сохраняется.
Остальные листы, формулы и стили документа odfpy не трогает.
"""
from __future__ import annotations
import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from odf.opendocument import load
from odf.table import Table, TableCell, TableRow
from odf.teletype import extractText
from odf.text import P
from ..config import REGISTER_COLUMNS
from ..models import CellValue, Record
# Повторения больше этого — «бесконечный» пустой хвост листа, не данные.
_MAX_REPEAT = 200
class RegisterWriter:
def __init__(self, path: Path):
self.doc = load(str(path))
self._tables: Dict[str, Table] = {
t.getAttribute("name"): t
for t in self.doc.spreadsheet.getElementsByType(Table)
}
def has_sheet(self, name: str) -> bool:
return name in self._tables
def append(self, sheet: str, records: List[Record]) -> int:
"""Дописать записи на лист. Возвращает число добавленных строк."""
table = self._tables[sheet]
template, last_number = self._find_last_numbered_row(table)
col_styles = self._column_styles(template)
row_style = template.getAttribute("stylename")
anchor = template
number = last_number
for record in records:
number += 1
row = self._build_row(record, number, row_style, col_styles)
table.insertBefore(row, anchor.nextSibling)
anchor = row
return len(records)
def save(self, path: Path) -> None:
self.doc.save(str(path))
# --- поиск строки-образца -------------------------------------------
def _find_last_numbered_row(self, table: Table) -> Tuple[TableRow, int]:
last_row: Optional[TableRow] = None
last_number = 0
for tr in table.getElementsByType(TableRow):
if _row_repeat(tr) > 1:
continue # повторяющийся пустой блок
cells = tr.getElementsByType(TableCell)
if not cells:
continue
number = _as_int(cells[0])
if number is not None:
last_row, last_number = tr, number
if last_row is None:
raise ValueError("На листе нет нумерованных строк: %s"
% table.getAttribute("name"))
return last_row, last_number
@staticmethod
def _column_styles(template: TableRow) -> Dict[int, Optional[str]]:
"""Карта {индекс колонки 0..12 -> имя стиля ячейки} из образца."""
styles: Dict[int, Optional[str]] = {}
col = 0
for cell in template.getElementsByType(TableCell):
rep = _col_repeat(cell)
style = cell.getAttribute("stylename")
for _ in range(rep):
if col < REGISTER_COLUMNS:
styles[col] = style
col += 1
if col >= REGISTER_COLUMNS:
break
return styles
# --- построение строки ----------------------------------------------
def _build_row(self, record: Record, number: int,
row_style: Optional[str],
col_styles: Dict[int, Optional[str]]) -> TableRow:
row = TableRow(stylename=row_style) if row_style else TableRow()
for col in range(REGISTER_COLUMNS):
style = col_styles.get(col)
value = number if col == 0 else record.value(col)
row.addElement(_make_cell(value, style))
return row
# --- фабрика ячеек по типу значения -------------------------------------
def _make_cell(value: CellValue, style: Optional[str]) -> TableCell:
if value is None or value == "":
return _cell(style)
if isinstance(value, bool):
return _text_cell(str(value), style)
if isinstance(value, datetime.datetime):
return _date_cell(value, style, with_time=True)
if isinstance(value, datetime.date):
return _date_cell(value, style, with_time=False)
if isinstance(value, (int, float)):
return _number_cell(value, style)
return _text_cell(str(value), style)
def _cell(style: Optional[str]) -> TableCell:
return TableCell(stylename=style) if style else TableCell()
def _text_cell(text: str, style: Optional[str]) -> TableCell:
cell = TableCell(valuetype="string", stylename=style) if style \
else TableCell(valuetype="string")
for line in text.split("\n"):
cell.addElement(P(text=line))
return cell
def _number_cell(num, style: Optional[str]) -> TableCell:
text = _number_text(num)
kw = dict(valuetype="float", value=text)
if style:
kw["stylename"] = style
cell = TableCell(**kw)
cell.addElement(P(text=text))
return cell
def _date_cell(value, style: Optional[str], with_time: bool) -> TableCell:
iso = value.strftime("%Y-%m-%dT%H:%M:%S" if with_time else "%Y-%m-%d")
display = value.strftime("%d.%m.%Y")
kw = dict(valuetype="date", datevalue=iso)
if style:
kw["stylename"] = style
cell = TableCell(**kw)
cell.addElement(P(text=display))
return cell
# --- мелкие помощники ---------------------------------------------------
def _number_text(num) -> str:
if isinstance(num, float) and num.is_integer():
return str(int(num))
return str(num)
def _row_repeat(tr: TableRow) -> int:
raw = tr.getAttribute("numberrowsrepeated")
n = int(raw) if raw else 1
return n if n <= _MAX_REPEAT else _MAX_REPEAT + 1
def _col_repeat(cell: TableCell) -> int:
raw = cell.getAttribute("numbercolumnsrepeated")
n = int(raw) if raw else 1
return n if n <= _MAX_REPEAT else 1
def _as_int(cell: TableCell) -> Optional[int]:
if cell.getAttribute("valuetype") == "float":
raw = cell.getAttribute("value")
if raw:
return int(float(raw))
text = "".join(extractText(p) for p in cell.getElementsByType(P)).strip()
return int(text) if text.isdigit() else None

3
requirements.txt Normal file
View File

@@ -0,0 +1,3 @@
odfpy>=1.4
openpyxl>=3.1
xlrd>=2.0

27
run.py Normal file
View File

@@ -0,0 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Точка входа: внести данные приёма в реестр и заполнить журнал.
python3 run.py
Зависимости: odfpy, openpyxl, xlrd (pip install odfpy openpyxl xlrd)
Оригинальные файлы не изменяются — результат пишется в *_ЗАПОЛНЕНО.
"""
import logging
import sys
from priem_register import Pipeline
def main() -> int:
logging.basicConfig(level=logging.INFO, format="%(message)s")
try:
Pipeline().run()
except Exception as exc: # noqa: BLE001 - короткое сообщение для пользователя
logging.error("Ошибка: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())

373
tests/test_edge_cases.py Normal file
View File

@@ -0,0 +1,373 @@
# -*- coding: utf-8 -*-
"""Тесты граничных случаев и обработки исключений.
Запуск: pytest -q (из корня проекта)
"""
from __future__ import annotations
import datetime
import sys
from pathlib import Path
import pytest
from odf.opendocument import OpenDocumentSpreadsheet, load
from odf.style import Style, TableRowProperties
from odf.table import Table, TableCell, TableRow
from odf.text import P
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from priem_register.mapping import discover_sources # noqa: E402
from priem_register.models import Record # noqa: E402
from priem_register.readers import get_reader # noqa: E402
from priem_register.readers.base import SourceReader # noqa: E402
from priem_register.writers import JournalWriter, RegisterWriter # noqa: E402
# --- Record -------------------------------------------------------------
def test_record_value_out_of_range():
r = Record(values=["a", "b"])
assert r.value(1) == "a" # колонка B
assert r.value(2) == "b" # колонка C
assert r.value(12) is None # колонка M отсутствует -> None
assert r.value(99) is None
# --- get_reader ---------------------------------------------------------
def test_get_reader_unknown_extension(tmp_path):
bad = tmp_path / "file.txt"
bad.write_text("x")
with pytest.raises(ValueError):
get_reader(bad)
# --- mapping ------------------------------------------------------------
def test_discover_missing_dir(tmp_path):
with pytest.raises(FileNotFoundError):
discover_sources(tmp_path / "нет")
def test_discover_empty_dir(tmp_path):
assert discover_sources(tmp_path) == []
def test_discover_ignores_non_matching(tmp_path):
(tmp_path / "ТО1").mkdir()
(tmp_path / "ТО1" / "заметка.txt").write_text("x")
(tmp_path / "ТО1" / "отчёт_74_99.xlsx").write_bytes(b"") # нет даты
(tmp_path / "ТО1" / "09.04.2026_99_25.docx").write_bytes(b"") # чужой формат
assert discover_sources(tmp_path) == []
def test_discover_invalid_date_raises_clear_error(tmp_path):
(tmp_path / "ТО1").mkdir()
(tmp_path / "ТО1" / "32.04.2026_74_99.xlsx").write_bytes(b"") # дня 32 нет
with pytest.raises(ValueError, match="Некорректная дата"):
discover_sources(tmp_path)
def test_discover_matches_and_sorts(tmp_path):
d = tmp_path / "ТО2"
d.mkdir()
(d / "21.04.2026_74_03.xlsx").write_bytes(b"")
(d / "03.04.2026_74_03.xlsx").write_bytes(b"")
found = discover_sources(tmp_path)
assert [s.name for s in found] == ["03.04.2026_74_03", "21.04.2026_74_03"]
assert all(s.sheet == "ТО2_03" for s in found)
assert found[0].date < found[1].date
def test_discover_file_without_to_dir(tmp_path):
(tmp_path / "09.04.2026_74_25.ods").write_bytes(b"")
with pytest.raises(ValueError):
discover_sources(tmp_path)
# --- RegisterWriter на синтетическом .ods -------------------------------
def _make_register(tmp_path, sheet_name="ТО1_25", numbers=(1, 2),
trailing_block=True):
doc = OpenDocumentSpreadsheet()
rs = Style(name="ro1", family="table-row")
rs.addElement(TableRowProperties(rowheight="0.45cm"))
doc.automaticstyles.addElement(rs)
table = Table(name=sheet_name)
# шапка
head = TableRow()
head.addElement(_p_cell("№ п/п"))
table.addElement(head)
# строки данных
for n in numbers:
row = TableRow(stylename="ro1")
row.addElement(_float_cell(n))
for _ in range(12):
row.addElement(TableCell())
table.addElement(row)
if trailing_block:
empty = TableRow(numberrowsrepeated="100")
empty.addElement(TableCell(numbercolumnsrepeated="13"))
table.addElement(empty)
doc.spreadsheet.addElement(table)
path = tmp_path / "reg.ods"
doc.save(str(path))
return path
def _p_cell(text):
c = TableCell(valuetype="string")
c.addElement(P(text=text))
return c
def _float_cell(n):
c = TableCell(valuetype="float", value=str(n))
c.addElement(P(text=str(n)))
return c
def _read_numbers(path, sheet):
doc = load(str(path))
table = [t for t in doc.spreadsheet.getElementsByType(Table)
if t.getAttribute("name") == sheet][0]
nums = []
for tr in table.getElementsByType(TableRow):
if tr.getAttribute("numberrowsrepeated"):
continue
cells = tr.getElementsByType(TableCell)
if cells and cells[0].getAttribute("value"):
nums.append(int(float(cells[0].getAttribute("value"))))
return nums
def test_register_missing_sheet(tmp_path):
path = _make_register(tmp_path)
w = RegisterWriter(path)
assert w.has_sheet("ТО1_25")
assert not w.has_sheet("НЕТ")
def test_register_no_numbered_rows(tmp_path):
doc = OpenDocumentSpreadsheet()
table = Table(name="ТО1_25")
head = TableRow()
head.addElement(_p_cell("№ п/п"))
table.addElement(head)
doc.spreadsheet.addElement(table)
path = tmp_path / "empty.ods"
doc.save(str(path))
with pytest.raises(ValueError):
RegisterWriter(path).append("ТО1_25", [Record(["a"])])
def test_register_append_continues_numbering(tmp_path):
path = _make_register(tmp_path, numbers=(1, 2, 3))
w = RegisterWriter(path)
n = w.append("ТО1_25", [Record(["x"]), Record(["y"])])
out = tmp_path / "out.ods"
w.save(out)
assert n == 2
assert _read_numbers(out, "ТО1_25") == [1, 2, 3, 4, 5]
def test_register_append_empty_records(tmp_path):
path = _make_register(tmp_path, numbers=(1, 2))
w = RegisterWriter(path)
assert w.append("ТО1_25", []) == 0
out = tmp_path / "out.ods"
w.save(out)
assert _read_numbers(out, "ТО1_25") == [1, 2]
def test_register_template_is_last_element(tmp_path):
# нет хвостового пустого блока -> образец является последним элементом
path = _make_register(tmp_path, numbers=(1, 2), trailing_block=False)
w = RegisterWriter(path)
w.append("ТО1_25", [Record(["x"]), Record(["y"])])
out = tmp_path / "out.ods"
w.save(out)
assert _read_numbers(out, "ТО1_25") == [1, 2, 3, 4]
def test_register_preserves_order_and_types(tmp_path):
path = _make_register(tmp_path, numbers=(10,))
w = RegisterWriter(path)
rec = Record(["Адрес", "MFC-1", "объект", 5, datetime.date(2026, 4, 15)])
w.append("ТО1_25", [rec])
out = tmp_path / "out.ods"
w.save(out)
doc = load(str(out))
table = [t for t in doc.spreadsheet.getElementsByType(Table)
if t.getAttribute("name") == "ТО1_25"][0]
last = [tr for tr in table.getElementsByType(TableRow)
if not tr.getAttribute("numberrowsrepeated")][-1]
cells = last.getElementsByType(TableCell)
assert cells[0].getAttribute("value") == "11" # № продолжен
assert cells[4].getAttribute("valuetype") == "float" # кол-во
assert cells[5].getAttribute("valuetype") == "date" # дата
assert cells[5].getAttribute("datevalue") == "2026-04-15"
# --- JournalWriter ------------------------------------------------------
def test_journal_writes_under_header_and_clears_examples(tmp_path):
import openpyxl
wb = openpyxl.Workbook()
ws = wb.active
ws.append(["Дата внесения", "Файл внесения", "Количество записей"])
ws.append(["Например", None, None])
ws.append([datetime.date(2026, 4, 22), "пример_файл", 99])
path = tmp_path / "j.xlsx"
wb.save(path)
from priem_register.models import JournalEntry
w = JournalWriter(path)
w.write([JournalEntry(datetime.date(2026, 6, 8), "файл_1", 5)])
out = tmp_path / "jout.xlsx"
w.save(out)
wb2 = openpyxl.load_workbook(out)
ws2 = wb2.active
# шапка на месте
assert ws2.cell(row=1, column=1).value == "Дата внесения"
# реальная запись сразу под шапкой (строка 2), а не после примера
assert ws2.cell(row=2, column=2).value == "файл_1"
assert ws2.cell(row=2, column=3).value == 5
# строки-примеры затёрты
assert ws2.cell(row=3, column=2).value is None
assert ws2.cell(row=3, column=3).value is None
# --- аномалии чтения: данные не теряются молча ---------------------------
class _FakeReader(SourceReader):
def __init__(self, rows):
super().__init__("fake.xlsx")
self._fake = rows
def _rows(self):
return self._fake
def test_reader_enters_all_rows_from_second():
# задание: «внести все строки, начиная со второй» — без фильтра по №
rows = [
["№ п/п", "Адрес"], # шапка — пропускается
[1, "Адрес", "MFC-1"], # запись с №
[None, None, None, None, None, None, None, None, None, None,
"08.04", "покупатель"], # строка без № — тоже запись
]
res = _FakeReader(rows).read()
assert len(res.records) == 2
assert res.records[0].value(2) == "MFC-1"
assert res.records[1].value(10) == "08.04" # Дата выдачи
assert res.records[1].value(11) == "покупатель" # Примечание
assert res.warnings == []
assert res.notes == []
def test_reader_extra_columns_merged_into_note():
rows = [
["№ п/п"],
[1] + ["v%d" % i for i in range(1, 12)] + ["реестр", "ХВОСТ"], # 14 колонок
]
res = _FakeReader(rows).read()
assert len(res.records) == 1
assert res.records[0].value(12) == "реестр" # колонка M на месте
assert res.records[0].value(11).endswith("ХВОСТ") # N ушёл в Примечание
assert res.warnings == []
assert len(res.notes) == 1
def test_reader_skips_header_and_empty_rows():
rows = [
["№ п/п"], # шапка
[1, "a"], # запись
[None, None, None], # пусто — пропускается
["", "", ""], # пусто — пропускается
[2, "b"], # запись
]
res = _FakeReader(rows).read()
assert len(res.records) == 2
assert res.warnings == []
assert res.notes == []
def test_reader_header_not_flagged():
rows = [["№ п/п", "Адрес", "Рег"]] # только шапка
res = _FakeReader(rows).read()
assert res.records == []
assert res.warnings == []
# --- OdsReader: текст внутри span/кратные пробелы не теряется ------------
def test_ods_reader_extracts_span_and_spaces(tmp_path):
from odf.opendocument import OpenDocumentSpreadsheet
from odf.table import Table as OTable, TableCell as OCell, TableRow as ORow
from odf.text import P as OP, S as OS, Span as OSpan
from priem_register.readers.ods_reader import OdsReader
doc = OpenDocumentSpreadsheet()
table = OTable(name="ТО1_25")
# шапка
h = ORow(); hc = OCell(valuetype="string"); hc.addElement(OP(text="№ п/п"))
h.addElement(hc); table.addElement(h)
# строка данных: объект с span и тройным пробелом
r = ORow()
c0 = OCell(valuetype="float", value="1"); c0.addElement(OP(text="1"))
r.addElement(c0)
p = OP(text="74:25:1 ")
p.addElement(OS(c=3)) # три неразрывных пробела ODF
p.addElement(OSpan(text="ЗЛАТОУСТ")) # текст внутри span
cell = OCell(valuetype="string"); cell.addElement(p)
r.addElement(cell)
table.addElement(r)
doc.spreadsheet.addElement(table)
path = tmp_path / "spans.ods"
doc.save(str(path))
res = OdsReader(path).read()
assert len(res.records) == 1
assert res.records[0].value(1) == "74:25:1 ЗЛАТОУСТ" # span + 3 пробела (колонка B)
# --- интеграционные тесты на реальных данных ----------------------------
def test_full_pipeline_real_data(tmp_path):
from priem_register import Pipeline
reg_out = tmp_path / "reg_out.ods"
jour_out = tmp_path / "jour_out.xlsx"
p = Pipeline(register_out=reg_out, journal_out=jour_out)
res = p.run()
assert res.total == 523
assert len(res.entries) == 12
assert reg_out.exists() and jour_out.exists()
assert all(e.count > 0 for e in res.entries)
assert res.warnings == []
# все непустые строки внесены (включая «уведомление»/«покупатель» как записи)
counts = {e.file_name: e.count for e in res.entries}
assert counts["09.04.2026_74_34"] == 66 # 62 + 4×уведомление
assert counts["17.04.2026_74_25"] == 46 # 45 + покупатель
# единственная заметка — данные за колонкой M (закладная) в «Примечание»
assert len(res.notes) == 1
assert "закладная" in res.notes[0]
def test_pipeline_idempotent(tmp_path):
from priem_register import Pipeline
totals = []
for i in range(2):
p = Pipeline(register_out=tmp_path / ("r%d.ods" % i),
journal_out=tmp_path / ("j%d.xlsx" % i))
totals.append(p.run().total)
assert totals[0] == totals[1] == 523
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-q"]))