feat(fns): парсер ФНС бухгалтерской отчетности

- Модели FinancialReport и FinancialReportLine
- FNSExcelParser для файлов fin_{id}_{ogrn}.xlsx
- FNSReportService с дедупликацией по хешу файла
- Celery задачи для мониторинга папки (каждые 5 мин)
- API: POST /fns/upload/, GET /fns/reports/
- Django admin интеграция
- 25 unit-тестов
This commit is contained in:
2026-02-01 14:44:19 +01:00
parent eb0d6f2600
commit cd0e21350b
17 changed files with 1537 additions and 10 deletions

View File

@@ -0,0 +1,10 @@
"""
Парсер бухгалтерской отчетности ФНС.
Обрабатывает Excel файлы формата fin_{external_id}_{ogrn}.xlsx.
"""
from apps.parsers.clients.fns.parser import FNSExcelParser
from apps.parsers.clients.fns.schemas import ParsedReport, ReportLine
__all__ = ["FNSExcelParser", "ParsedReport", "ReportLine"]

View File

@@ -0,0 +1,208 @@
"""
Парсер Excel файлов бухгалтерской отчетности ФНС.
"""
import logging
import re
from pathlib import Path
import openpyxl
from apps.parsers.clients.fns.schemas import ParsedReport, ReportLine
logger = logging.getLogger(__name__)
class FNSParserError(Exception):
"""Ошибка парсинга файла ФНС."""
pass
class FNSExcelParser:
"""
Парсер Excel файлов бухгалтерской отчетности.
Обрабатывает файлы формата fin_{external_id}_{ogrn}.xlsx.
Извлекает данные из форм №1, №2, №3, №4, №6.
"""
FILENAME_PATTERN = re.compile(r"^fin_(\d+)_(\d{13,15})\.xlsx$")
FORM_MARKERS = {
"Форма №1": "1",
"Форма №2": "2",
"Форма №3": "3",
"Форма №4": "4",
"Форма №6": "6",
}
@classmethod
def parse_filename(cls, filename: str) -> tuple[str, str]:
"""
Извлекает external_id и ogrn из имени файла.
Args:
filename: Имя файла (например: fin_0000605_1027700169089.xlsx)
Returns:
Кортеж (external_id, ogrn)
Raises:
FNSParserError: Если имя файла не соответствует формату
"""
match = cls.FILENAME_PATTERN.match(filename)
if not match:
raise FNSParserError(
f"Имя файла не соответствует формату "
f"fin_{{id}}_{{ogrn}}.xlsx: {filename}"
)
return match.group(1), match.group(2)
@classmethod
def parse_file(cls, file_path: Path | str) -> ParsedReport:
"""
Парсит Excel файл и возвращает структурированные данные.
Args:
file_path: Путь к файлу
Returns:
ParsedReport с данными отчетности
Raises:
FNSParserError: При ошибке парсинга
"""
file_path = Path(file_path)
if not file_path.exists():
raise FNSParserError(f"Файл не найден: {file_path}")
external_id, ogrn = cls.parse_filename(file_path.name)
logger.info(
"Парсинг файла %s (external_id=%s, ogrn=%s)",
file_path.name,
external_id,
ogrn,
)
try:
workbook = openpyxl.load_workbook(file_path, data_only=True)
except Exception as e:
raise FNSParserError(f"Ошибка открытия файла: {e}") from e
lines: list[ReportLine] = []
for sheet_name in workbook.sheetnames:
sheet = workbook[sheet_name]
sheet_lines = cls._parse_sheet(sheet)
lines.extend(sheet_lines)
logger.debug("Лист '%s': извлечено %d строк", sheet_name, len(sheet_lines))
workbook.close()
logger.info(
"Файл %s обработан: %d строк, годы %s, формы %s",
file_path.name,
len(lines),
sorted({line.year for line in lines}),
sorted({line.form_code for line in lines}),
)
return ParsedReport(external_id=external_id, ogrn=ogrn, lines=lines)
@classmethod
def _parse_sheet(cls, sheet) -> list[ReportLine]: # noqa: C901
"""Парсит один лист Excel."""
lines: list[ReportLine] = []
current_form: str | None = None
years: list[int] = []
header_row_found = False
for row in sheet.iter_rows(values_only=True):
if not any(row):
continue
first_cell = str(row[0]).strip() if row[0] else ""
# Ищем маркер формы
for marker, form_code in cls.FORM_MARKERS.items():
if first_cell.startswith(marker):
current_form = form_code
years = cls._extract_years_from_row(row)
header_row_found = False
logger.debug("Найдена форма %s, годы: %s", form_code, years)
break
# Пропускаем заголовочную строку с "Код", "Начало", "Конец"
is_header = len(row) > 1 and row[1] == "Код"
if current_form and not header_row_found and is_header:
header_row_found = True
continue
# Парсим строки данных
if current_form and header_row_found and years:
line_name = first_cell
line_code = str(row[1]).strip() if row[1] else ""
# Пропускаем строки без кода или заголовки секций
if not line_code or not line_code.isdigit():
continue
# Извлекаем значения по годам
for year_idx, year in enumerate(years):
col_start = 2 + year_idx * 2 # Начало: 2, 4, 6, 8
col_end = 3 + year_idx * 2 # Конец: 3, 5, 7, 9
period_start = cls._parse_value(
row[col_start] if col_start < len(row) else None
)
period_end = cls._parse_value(
row[col_end] if col_end < len(row) else None
)
# Добавляем строку только если есть хотя бы одно значение
if period_start is not None or period_end is not None:
lines.append(
ReportLine(
form_code=current_form,
line_code=line_code,
line_name=line_name,
year=year,
period_start=period_start,
period_end=period_end,
)
)
return lines
@classmethod
def _extract_years_from_row(cls, row: tuple) -> list[int]:
"""Извлекает годы из строки заголовка формы."""
years = []
for cell in row:
if cell is None:
continue
try:
value = int(cell)
if 1990 <= value <= 2100:
years.append(value)
except (ValueError, TypeError):
continue
return sorted(set(years))
@classmethod
def _parse_value(cls, value) -> int | None:
"""Преобразует значение ячейки в int или None."""
if value is None:
return None
if isinstance(value, int | float):
return int(value)
if isinstance(value, str):
value = value.strip()
if not value or value == "-":
return None
try:
return int(float(value.replace(",", ".").replace(" ", "")))
except ValueError:
return None
return None

View File

@@ -0,0 +1,61 @@
"""
Схемы данных для парсера бухгалтерской отчетности ФНС.
"""
from dataclasses import dataclass, field
@dataclass
class ReportLine:
"""
Строка бухгалтерской отчетности.
Attributes:
form_code: Код формы (1, 2, 3, 4, 6)
line_code: Код строки (например: 1100, 2110)
line_name: Наименование строки
year: Отчетный год
period_start: Значение на начало периода (тыс. руб.), None если пусто
period_end: Значение на конец периода (тыс. руб.), None если пусто
"""
form_code: str
line_code: str
line_name: str
year: int
period_start: int | None = None
period_end: int | None = None
@dataclass
class ParsedReport:
"""
Результат парсинга Excel файла отчетности.
Attributes:
external_id: Внешний ID из имени файла
ogrn: ОГРН организации
lines: Список строк отчетности
"""
external_id: str
ogrn: str
lines: list[ReportLine] = field(default_factory=list)
@property
def years(self) -> set[int]:
"""Получить все годы, представленные в отчете."""
return {line.year for line in self.lines}
@property
def forms(self) -> set[str]:
"""Получить все формы, представленные в отчете."""
return {line.form_code for line in self.lines}
def get_lines_by_form(self, form_code: str) -> list[ReportLine]:
"""Получить строки по коду формы."""
return [line for line in self.lines if line.form_code == form_code]
def get_lines_by_year(self, year: int) -> list[ReportLine]:
"""Получить строки по году."""
return [line for line in self.lines if line.year == year]

View File

@@ -457,9 +457,7 @@ class ZakupkiClient:
if progress_callback:
progress_callback(95, f"Загружено {len(all_procurements)} закупок")
logger.info(
"Total fetched %d procurements via HTTP", len(all_procurements)
)
logger.info("Total fetched %d procurements via HTTP", len(all_procurements))
return all_procurements
def _discover_data_files(