feat(etl): извлечение по форматам (агент B) — docx/xlsx/xls/pdf
- единый грид-экстрактор: поиск шапки, роли колонок, секции, N тарифов - отсев колонок-индексов и не-цен; название по доле кириллицы - PDF: реконструкция таблицы по координатам слов (нет линий → extract_tables пуст) - детектор битого текстового слоя → пометка scan_pdf для Vision - проверено на реальном архиве: 6/8 клиник нативно, 15599 позиций; 2 файла → Vision Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,58 @@
|
||||
"""Диспетчер извлечения: путь к файлу → ParsedDocument.
|
||||
|
||||
Определяет формат, читает файл в сетки и прогоняет их через общий грид-экстрактор.
|
||||
PDF с битым текстовым слоем помечается как scan_pdf — его добирает путь через Vision.
|
||||
Любой сбой чтения логируется, но не роняет обработку остального архива.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
|
||||
from contracts.models import ParsedDocument # noqa: E402
|
||||
from etl.extractors import readers # noqa: E402
|
||||
from etl.extractors.grid import extract_rows_from_grid # noqa: E402
|
||||
|
||||
|
||||
def extract(path: str) -> ParsedDocument:
|
||||
"""Разобрать один файл прайса в структурированный ParsedDocument."""
|
||||
name = Path(path).name
|
||||
fmt = readers.detect_format(path)
|
||||
doc = ParsedDocument(file_name=name, file_format=fmt, partner_name=readers.partner_from_name(name))
|
||||
|
||||
year = readers.year_from_name(name)
|
||||
if year:
|
||||
doc.effective_date = date(year, 1, 1)
|
||||
|
||||
try:
|
||||
if fmt == "xlsx":
|
||||
grids, text = readers.xlsx_grids(path)
|
||||
elif fmt == "xls":
|
||||
grids, text = readers.xls_grids(path)
|
||||
elif fmt == "docx":
|
||||
grids, text = readers.docx_grids(path)
|
||||
elif fmt == "pdf":
|
||||
grids, text = readers.pdf_grids(path)
|
||||
if readers.is_garbled(text):
|
||||
doc.file_format = "scan_pdf"
|
||||
doc.parse_log.append("битый текстовый слой — требуется распознавание (Vision)")
|
||||
else:
|
||||
doc.parse_log.append(f"неизвестный формат: {fmt}")
|
||||
return doc
|
||||
except Exception as exc: # noqa: BLE001 — сбой чтения логируем, конвейер не роняем
|
||||
doc.parse_log.append(f"ошибка чтения: {type(exc).__name__}: {exc}")
|
||||
return doc
|
||||
|
||||
doc.raw_content = text[:200_000]
|
||||
for grid in grids:
|
||||
try:
|
||||
doc.rows.extend(extract_rows_from_grid(grid))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
doc.parse_log.append(f"ошибка разбора таблицы: {type(exc).__name__}: {exc}")
|
||||
|
||||
if not doc.rows and doc.file_format != "scan_pdf":
|
||||
doc.parse_log.append("позиции не извлечены — возможно, нужен Vision или иной парсер")
|
||||
return doc
|
||||
@@ -0,0 +1,127 @@
|
||||
"""Эвристики извлечения, общие для всех форматов.
|
||||
|
||||
Все табличные источники (xlsx, xls, docx, таблицы из PDF) сводятся к «сетке» —
|
||||
списку строк из строковых ячеек — и обрабатываются единым кодом. Здесь собраны
|
||||
функции, понимающие реальные особенности казахстанских прайсов: разрядные пробелы
|
||||
в числах, многотарифные колонки цен, строки-заголовки секций и шапку не в первой
|
||||
строке таблицы.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
Grid = list[list[str]]
|
||||
|
||||
# Ячейка-цена целиком: «14 400», «12600.0», «3 980», «9000 тг», «от 12000».
|
||||
# Требуем, чтобы вся ячейка была числом (плюс валюта или «от»), иначе цифра внутри
|
||||
# текста («Раздел 1.Консультации») ошибочно сошла бы за цену.
|
||||
_PRICE_CELL_RE = re.compile(
|
||||
r"^(?:от\s+)?(\d[\d .,]*\d|\d)\s*(?:тг|тенге|₸|kzt|руб|\$)?\.?$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
# Похоже на код услуги: «U1.1», «A02.004.000», «1.0», «В02.110.002».
|
||||
_CODE_RE = re.compile(r"^[A-ZА-Я]?\.?\d+([.\-]\d+)*[A-ZА-Я]?$")
|
||||
|
||||
_NAME_KEYS = ("наименование", "услуг", "название", "перечень")
|
||||
_PRICE_KEYS = ("цена", "стоимость", "тариф", "тенге")
|
||||
_UNIT_KEYS = ("ед.", "единица", "ед изм", "ед. изм")
|
||||
_CODE_KEYS = ("код", "№", "тарификатор")
|
||||
|
||||
# Заголовок ценовой колонки → тип тарифа. Порядок важен: частные случаи раньше общих.
|
||||
_TIER_PATTERNS: list[tuple[str, tuple[str, ...]]] = [
|
||||
("nonresident", ("нерезидент", "без гражданства", "иностран")),
|
||||
("far", ("дальнего", "дальнее")),
|
||||
("cis", ("снг", "ближнего", "оралман")),
|
||||
("insurance", ("страхов",)),
|
||||
("partner", ("партнер", "партнён", "партнёр")),
|
||||
("resident", ("резидент", "республики казахстан", " рк", "граждан республики")),
|
||||
]
|
||||
|
||||
|
||||
def clean(text) -> str:
|
||||
"""Строковое значение ячейки без хвостовых пробелов и переводов строк."""
|
||||
if text is None:
|
||||
return ""
|
||||
return re.sub(r"\s+", " ", str(text)).strip()
|
||||
|
||||
|
||||
def parse_price(text) -> float | None:
|
||||
"""Вытащить цену из ячейки, если ВСЯ ячейка — число (плюс валюта или «от»).
|
||||
|
||||
Возвращает положительный float или None. Намеренно строгая: «Раздел 1.…» или
|
||||
«приём (30 минут)» ценой не считаются — это защищает детектор секций.
|
||||
"""
|
||||
s = clean(text)
|
||||
if not s:
|
||||
return None
|
||||
m = _PRICE_CELL_RE.match(s)
|
||||
if not m:
|
||||
return None
|
||||
raw = m.group(1).replace(" ", "")
|
||||
# Есть и точка, и запятая — запятая разрядная; иначе запятая = десятичная.
|
||||
if "," in raw and "." in raw:
|
||||
raw = raw.replace(",", "")
|
||||
else:
|
||||
raw = raw.replace(",", ".")
|
||||
try:
|
||||
value = float(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
return value if value > 0 else None
|
||||
|
||||
|
||||
def looks_like_code(text) -> bool:
|
||||
"""Похоже ли значение на код услуги (внутренний или тарификатора)."""
|
||||
s = clean(text)
|
||||
return bool(s) and len(s) <= 16 and bool(_CODE_RE.match(s))
|
||||
|
||||
|
||||
def classify_price_tier(header_text) -> str | None:
|
||||
"""По тексту заголовка ценовой колонки определить тип тарифа."""
|
||||
h = clean(header_text).lower()
|
||||
if not h:
|
||||
return None
|
||||
for tier, keys in _TIER_PATTERNS:
|
||||
if any(k in h for k in keys):
|
||||
return tier
|
||||
if any(k in h for k in _PRICE_KEYS):
|
||||
return "resident" # обобщённая «Цена»/«Стоимость» — основной тариф
|
||||
return None
|
||||
|
||||
|
||||
def _has_key(cells: list[str], keys: tuple[str, ...]) -> bool:
|
||||
joined = " | ".join(c.lower() for c in cells)
|
||||
return any(k in joined for k in keys)
|
||||
|
||||
|
||||
def find_header_row(grid: Grid, scan: int = 30) -> int:
|
||||
"""Найти индекс строки заголовков (она не всегда первая).
|
||||
|
||||
Заголовок — строка, где есть и колонка названия услуги, и хотя бы одна
|
||||
ценовая колонка. Если явной шапки нет, берётся первая строка, под которой
|
||||
идут пары «текст + число».
|
||||
"""
|
||||
best_idx, best_score = -1, 0
|
||||
for i, row in enumerate(grid[:scan]):
|
||||
cells = [clean(c) for c in row]
|
||||
if not any(cells):
|
||||
continue
|
||||
score = 0
|
||||
if _has_key(cells, _NAME_KEYS):
|
||||
score += 2
|
||||
score += sum(1 for c in cells if classify_price_tier(c))
|
||||
if _has_key(cells, _UNIT_KEYS):
|
||||
score += 1
|
||||
if score > best_score:
|
||||
best_idx, best_score = i, score
|
||||
if best_idx >= 0 and best_score >= 2:
|
||||
return best_idx
|
||||
# Фолбэк: первая строка, под которой есть пара «непустой текст + число».
|
||||
for i in range(min(scan, len(grid))):
|
||||
for j in range(i + 1, min(i + 6, len(grid))):
|
||||
cells = [clean(c) for c in grid[j]]
|
||||
texts = [c for c in cells if c and not parse_price(c)]
|
||||
nums = [c for c in cells if parse_price(c)]
|
||||
if texts and nums:
|
||||
return i
|
||||
return 0
|
||||
@@ -0,0 +1,174 @@
|
||||
"""Извлечение позиций прайса из «сетки» строк.
|
||||
|
||||
Алгоритм: найти строку заголовков → определить роли колонок (название, код,
|
||||
единица, ценовые тарифы) → пройти строки, отделяя заголовки секций от позиций.
|
||||
Один и тот же код работает для xlsx, xls, docx и таблиц, восстановленных из PDF.
|
||||
|
||||
Распознавание ролей колонок учитывает реальные ловушки прайсов:
|
||||
- колонка «№» (последовательные 1, 2, 3…) — это индекс, а не цена;
|
||||
- настоящая цена медуслуги — сотни и тысячи тенге, а не мелкое число;
|
||||
- название услуги — самая «кириллическая» колонка, код — цифро-точечная.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import statistics
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
|
||||
from contracts.models import RawRow # noqa: E402
|
||||
from etl.extractors.common import ( # noqa: E402
|
||||
_NAME_KEYS,
|
||||
_UNIT_KEYS,
|
||||
Grid,
|
||||
classify_price_tier,
|
||||
clean,
|
||||
find_header_row,
|
||||
looks_like_code,
|
||||
parse_price,
|
||||
)
|
||||
|
||||
_TIER_ORDER = ["resident", "nonresident", "cis", "far", "insurance", "partner"]
|
||||
_CODE_HEADER_KEYS = ("код", "тарификатор")
|
||||
_INDEX_HEADER_KEYS = ("№", "п/п", "n п")
|
||||
_MIN_REAL_PRICE = 100 # медуслуга дешевле сотни тенге не бывает — отсекает индексы и мелочь
|
||||
|
||||
|
||||
def _col_values(sample: list[list[str]], c: int) -> list[str]:
|
||||
return [r[c] for r in sample if c < len(r) and r[c]]
|
||||
|
||||
|
||||
def _col_prices(sample: list[list[str]], c: int) -> list[float]:
|
||||
return [p for r in sample if c < len(r) and (p := parse_price(r[c]))]
|
||||
|
||||
|
||||
def _cyrillic_score(sample: list[list[str]], c: int) -> float:
|
||||
"""Средняя доля кириллических букв в колонке — признак колонки названий."""
|
||||
total = 0
|
||||
for r in sample:
|
||||
if c < len(r):
|
||||
total += sum(1 for ch in r[c] if "а" <= ch.lower() <= "я" or ch.lower() == "ё")
|
||||
return total / max(1, len(sample))
|
||||
|
||||
|
||||
def _is_index_column(values: list[float]) -> bool:
|
||||
"""Колонка-индекс: целые, монотонные, шаг около 1, старт у единицы."""
|
||||
if len(values) < 5:
|
||||
return False
|
||||
integers = [v for v in values if abs(v - round(v)) < 1e-9]
|
||||
if len(integers) < 0.9 * len(values):
|
||||
return False
|
||||
ordered = sorted(round(v) for v in values)
|
||||
steps = [ordered[i + 1] - ordered[i] for i in range(len(ordered) - 1)]
|
||||
return ordered[0] <= 3 and statistics.mean(steps) < 1.5
|
||||
|
||||
|
||||
def _assign_columns(grid: Grid, header_idx: int):
|
||||
"""Определить роли колонок по заголовку и выборке строк под ним."""
|
||||
header = [clean(c) for c in grid[header_idx]]
|
||||
ncol = max((len(r) for r in grid), default=0)
|
||||
header += [""] * (ncol - len(header))
|
||||
sample = [[clean(c) for c in r] for r in grid[header_idx + 1 : header_idx + 60]]
|
||||
|
||||
# --- ценовые колонки ---
|
||||
price_cols: dict[int, str | None] = {}
|
||||
for c in range(ncol):
|
||||
head = header[c].lower()
|
||||
if any(k in head for k in _INDEX_HEADER_KEYS) or any(k in head for k in _CODE_HEADER_KEYS):
|
||||
continue # «№» и «Код» — не цены
|
||||
prices = _col_prices(sample, c)
|
||||
nonempty = _col_values(sample, c)
|
||||
ratio = len(prices) / len(nonempty) if nonempty else 0
|
||||
if not prices or _is_index_column(prices):
|
||||
continue
|
||||
tier = classify_price_tier(header[c])
|
||||
if tier:
|
||||
price_cols[c] = tier
|
||||
elif ratio >= 0.6 and statistics.median(prices) >= _MIN_REAL_PRICE:
|
||||
price_cols[c] = None # числовая колонка без явного тарифа — назначим по позиции
|
||||
|
||||
used = {t for t in price_cols.values() if t}
|
||||
fallback = [t for t in _TIER_ORDER if t not in used]
|
||||
for c in sorted(price_cols):
|
||||
if price_cols[c] is None:
|
||||
price_cols[c] = fallback.pop(0) if fallback else f"extra_{c}"
|
||||
|
||||
# --- колонка названия: явный заголовок, иначе самая кириллическая ---
|
||||
name_col = next(
|
||||
(c for c in range(ncol) if c not in price_cols and any(k in header[c].lower() for k in _NAME_KEYS)),
|
||||
None,
|
||||
)
|
||||
if name_col is None:
|
||||
candidates = [c for c in range(ncol) if c not in price_cols]
|
||||
name_col = max(candidates, key=lambda c: _cyrillic_score(sample, c)) if candidates else 0
|
||||
|
||||
# --- колонка кода: заголовок «Код»/«тарификатор» или код-подобные значения ---
|
||||
code_col = next(
|
||||
(
|
||||
c
|
||||
for c in range(ncol)
|
||||
if c != name_col and c not in price_cols and any(k in header[c].lower() for k in _CODE_HEADER_KEYS)
|
||||
),
|
||||
None,
|
||||
)
|
||||
if code_col is None:
|
||||
for c in range(ncol):
|
||||
if c == name_col or c in price_cols:
|
||||
continue
|
||||
vals = _col_values(sample, c)
|
||||
if vals and sum(1 for v in vals if looks_like_code(v)) >= max(2, 0.5 * len(vals)):
|
||||
code_col = c
|
||||
break
|
||||
|
||||
unit_col = next(
|
||||
(c for c in range(ncol) if c not in price_cols and any(k in header[c].lower() for k in _UNIT_KEYS)),
|
||||
None,
|
||||
)
|
||||
return name_col, code_col, unit_col, price_cols
|
||||
|
||||
|
||||
def extract_rows_from_grid(grid: Grid) -> list[RawRow]:
|
||||
"""Превратить сетку в список позиций прайса."""
|
||||
grid = [[clean(c) for c in r] for r in grid if any(clean(c) for c in r)]
|
||||
if len(grid) < 2:
|
||||
return []
|
||||
|
||||
header_idx = find_header_row(grid)
|
||||
name_col, code_col, unit_col, price_cols = _assign_columns(grid, header_idx)
|
||||
|
||||
rows: list[RawRow] = []
|
||||
section: str | None = None
|
||||
for r in grid[header_idx + 1 :]:
|
||||
nonempty = [c for c in r if c]
|
||||
if not nonempty:
|
||||
continue
|
||||
|
||||
prices: dict[str, float] = {}
|
||||
for c, tier in price_cols.items():
|
||||
if c < len(r):
|
||||
price = parse_price(r[c])
|
||||
if price:
|
||||
prices[tier] = price
|
||||
|
||||
# Заголовок секции: одна осмысленная ячейка и ни одной цены.
|
||||
if not prices and len(nonempty) == 1:
|
||||
section = nonempty[0]
|
||||
continue
|
||||
|
||||
name = r[name_col] if name_col < len(r) else ""
|
||||
if not name or not prices:
|
||||
continue # без названия или без цены это не позиция прайса
|
||||
|
||||
rows.append(
|
||||
RawRow(
|
||||
service_name_raw=name,
|
||||
service_code_source=(
|
||||
r[code_col] if code_col is not None and code_col < len(r) and r[code_col] else None
|
||||
),
|
||||
prices=prices,
|
||||
unit=(r[unit_col] if unit_col is not None and unit_col < len(r) and r[unit_col] else None),
|
||||
section=section,
|
||||
)
|
||||
)
|
||||
return rows
|
||||
@@ -0,0 +1,136 @@
|
||||
"""Чтение исходных файлов в «сетки» строк плюс метаданные документа.
|
||||
|
||||
Каждый читатель возвращает (список сеток, сырой текст). Несколько сеток бывает у
|
||||
многолистовых книг и многотабличных PDF. Сырой текст нужен для аудита и для
|
||||
детектора битого текстового слоя.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from etl.extractors.common import Grid, clean
|
||||
|
||||
|
||||
def year_from_name(name: str) -> int | None:
|
||||
"""Год прайса из имени файла («Клиника 2 прайс 2025 год» → 2025)."""
|
||||
m = re.search(r"(20\d{2})", name)
|
||||
return int(m.group(1)) if m else None
|
||||
|
||||
|
||||
def partner_from_name(name: str) -> str:
|
||||
"""Название клиники из имени файла, очищенное от слов «прайс», года и т. п."""
|
||||
base = Path(name).stem
|
||||
base = re.sub(r"(?i)\b(прайс|price|прейскурант|год)\b", " ", base)
|
||||
base = re.sub(r"20\d{2}", " ", base)
|
||||
base = re.sub(r"[_\-]+", " ", base)
|
||||
return re.sub(r"\s+", " ", base).strip()
|
||||
|
||||
|
||||
def detect_format(path: str) -> str:
|
||||
ext = Path(path).suffix.lower()
|
||||
return {".xlsx": "xlsx", ".xls": "xls", ".docx": "docx", ".pdf": "pdf"}.get(ext, ext.lstrip("."))
|
||||
|
||||
|
||||
def docx_grids(path: str) -> tuple[list[Grid], str]:
|
||||
import docx
|
||||
|
||||
document = docx.Document(path)
|
||||
grids = [[[clean(cell.text) for cell in row.cells] for row in table.rows] for table in document.tables]
|
||||
text = "\n".join(p.text for p in document.paragraphs)
|
||||
return grids, text
|
||||
|
||||
|
||||
def xlsx_grids(path: str) -> tuple[list[Grid], str]:
|
||||
from openpyxl import load_workbook
|
||||
|
||||
workbook = load_workbook(path, read_only=True, data_only=True)
|
||||
grids = []
|
||||
for sheet in workbook.worksheets:
|
||||
grid = [[clean(c) for c in row] for row in sheet.iter_rows(values_only=True)]
|
||||
if any(any(row) for row in grid):
|
||||
grids.append(grid)
|
||||
return grids, ""
|
||||
|
||||
|
||||
def xls_grids(path: str) -> tuple[list[Grid], str]:
|
||||
import xlrd
|
||||
|
||||
book = xlrd.open_workbook(path)
|
||||
grids = []
|
||||
for sheet in book.sheets():
|
||||
grid = [
|
||||
[clean(sheet.cell_value(r, c)) for c in range(sheet.ncols)] for r in range(sheet.nrows)
|
||||
]
|
||||
if any(any(row) for row in grid):
|
||||
grids.append(grid)
|
||||
return grids, ""
|
||||
|
||||
|
||||
def _line_to_cells(words: list[dict], x_gap: float) -> list[str]:
|
||||
"""Слова одной строки → ячейки: большой горизонтальный разрыв = граница колонки."""
|
||||
words.sort(key=lambda w: w["x0"])
|
||||
cells = [words[0]["text"]]
|
||||
right = words[0]["x1"]
|
||||
for w in words[1:]:
|
||||
if w["x0"] - right > x_gap:
|
||||
cells.append(w["text"])
|
||||
else:
|
||||
cells[-1] += " " + w["text"]
|
||||
right = w["x1"]
|
||||
return [clean(c) for c in cells]
|
||||
|
||||
|
||||
def _page_to_grid(page, y_tol: float = 3.0, x_gap: float = 18.0) -> Grid:
|
||||
"""Восстановить табличную сетку страницы PDF по координатам слов.
|
||||
|
||||
У многих прайсов нет линий таблицы, поэтому pdfplumber.extract_tables ничего не
|
||||
находит. Здесь слова группируются в строки по вертикали (top), а внутри строки
|
||||
режутся на ячейки по горизонтальным разрывам — так колонки цен (резидент/
|
||||
нерезидент/…) не сливаются в одну.
|
||||
"""
|
||||
words = page.extract_words(use_text_flow=False, keep_blank_chars=False)
|
||||
if not words:
|
||||
return []
|
||||
rows: Grid = []
|
||||
line: list[dict] = []
|
||||
current_band: int | None = None
|
||||
for w in sorted(words, key=lambda w: (round(w["top"] / y_tol), w["x0"])):
|
||||
band = round(w["top"] / y_tol)
|
||||
if current_band is None or band == current_band:
|
||||
line.append(w)
|
||||
else:
|
||||
rows.append(_line_to_cells(line, x_gap))
|
||||
line = [w]
|
||||
current_band = band
|
||||
if line:
|
||||
rows.append(_line_to_cells(line, x_gap))
|
||||
return rows
|
||||
|
||||
|
||||
def pdf_grids(path: str) -> tuple[list[Grid], str]:
|
||||
"""Все страницы PDF собираются в одну сетку — так шапка распознаётся один раз."""
|
||||
import pdfplumber
|
||||
|
||||
grid: Grid = []
|
||||
texts: list[str] = []
|
||||
with pdfplumber.open(path) as pdf:
|
||||
for page in pdf.pages:
|
||||
texts.append(page.extract_text() or "")
|
||||
grid.extend(_page_to_grid(page))
|
||||
return [grid], "\n".join(texts)
|
||||
|
||||
|
||||
def is_garbled(text: str, min_letters: int = 200) -> bool:
|
||||
"""Битый текстовый слой: мало кириллицы или много латиницы-мусора.
|
||||
|
||||
Русский прайс почти не содержит латинских букв (кроме редких брендов), поэтому
|
||||
их обилие — верный признак испорченной кодировки. На таком документе нативный
|
||||
парсинг бессмыслен, его добирает путь через Vision.
|
||||
"""
|
||||
letters = [ch for ch in text if ch.isalpha()]
|
||||
if len(letters) < min_letters:
|
||||
return False
|
||||
cyrillic = sum(1 for ch in letters if "а" <= ch.lower() <= "я" or ch.lower() == "ё")
|
||||
latin = sum(1 for ch in letters if "a" <= ch.lower() <= "z")
|
||||
return cyrillic / len(letters) < 0.55 or latin / len(letters) > 0.18
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Прогон извлечения по всему архиву: сколько позиций берёт каждый формат."""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(REPO))
|
||||
|
||||
from etl.extractors import extract
|
||||
|
||||
files = sorted((REPO / "data/raw").glob("*"))
|
||||
total = 0
|
||||
for path in files:
|
||||
doc = extract(str(path))
|
||||
total += len(doc.rows)
|
||||
tiers = sorted({t for row in doc.rows for t in row.prices})
|
||||
print(f"\n{'=' * 68}")
|
||||
print(f"{path.name} [{doc.file_format}] партнёр='{doc.partner_name}' дата={doc.effective_date}")
|
||||
print(f" позиций: {len(doc.rows)} | тарифы: {tiers}")
|
||||
if doc.parse_log:
|
||||
print(f" лог: {doc.parse_log[:2]}")
|
||||
for row in doc.rows[:3]:
|
||||
print(f" код={row.service_code_source!r} «{row.service_name_raw[:40]}» {row.prices} секц={row.section!r}")
|
||||
|
||||
print(f"\n{'=' * 68}\nИТОГО позиций по архиву: {total}")
|
||||
Reference in New Issue
Block a user