RSHB/DKB_parsers
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: DKB parsers service

Browse Source 
FastAPI service that consumes NATS `files.uploaded` events and parses
uploaded documents from MinIO/S3 into legal entity, company group and
RF office schemas.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-09-10 18:05:48 +04:00
commit 9283ded5ae
19 changed files with 812 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
.gitignore vendored Normal file
View File

@ -0,0 +1,35 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
dist/
*.egg-info/
.eggs/
# Virtual environments
.venv/
venv/
env/
ENV/
# Environment / secrets
.env
.env.*
!.env.example
# Tooling caches
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
htmlcov/
# IDE
.idea/
.vscode/
# OS
.DS_Store

58
main.py Normal file
View File

@ -0,0 +1,58 @@
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from miniopy_async import Minio
from src.core.acync_cache import AsyncInMemoryCache
from src.core.poll_worker import PollWorker
from src.core.config import settings
from src.core.clients import S3Storage
from nats.aio.client import Client as NATS
import asyncio
from miniopy_async.error import S3Error
import json
from src.core.nats_callback import build_file_handler
from src.core.nats_callback_pdf import build_file_handler_pdf
@asynccontextmanager
async def lifespan(app):
s3_client = Minio(settings.MINIO_URL, settings.MINIO_ADMIN_LOGIN, settings.MINIO_ADMIN_PASSWORD, secure=False)
storage = S3Storage(client=s3_client, bucket_name=settings.MINIO_BUCKET)
app.state.storage = storage
nc = NATS()
await nc.connect("nats://localhost:4222", max_reconnect_attempts=5)
# cache = AsyncInMemoryCache()
# poller = PollWorker(nc, storage, cache)
# await poller.start()
handler_cb = build_file_handler(nc, storage, bucket_name=settings.MINIO_BUCKET)
# handler_cb = build_file_handler_pdf(nc, storage, settings.MINIO_BUCKET, cache)
sub = await nc.subscribe('files.uploaded', queue='xlsxparsing', cb=handler_cb)
app.state.nats = nc
app.state.nats_subs = [sub]
try:
yield
finally:
# await poller.stop()
for sub in app.state.nats_subs:
await sub.unsubscribe()
await nc.close()
app = FastAPI(lifespan=lifespan)
@app.get('/healthcheck')
async def health():
return {
'status': 'alive'
}
app.add_middleware(
CORSMiddleware,
allow_credentials=['*'], allow_headers=['*'], allow_methods=['*'], allow_origins=['*']
)

43
src/core/acync_cache.py Normal file
View File

@ -0,0 +1,43 @@
import asyncio
from datetime import datetime, timedelta
import uuid
from .types import FileDictItem
class AsyncInMemoryCache:
def __init__(self):
self.mapping: dict[str, FileDictItem] = {}
self._lock = asyncio.Lock()
async def add(self, job_id: uuid.UUID, file_id: uuid.UUID) -> FileDictItem:
now = datetime.now()
expiration_date = datetime.now() + timedelta(minutes=5)
new_item = {
'started_at': now,
'expires_at': expiration_date,
'file_id': str(file_id),
'job_id': str(job_id)
}
async with self._lock:
self.mapping[str(job_id)] = new_item
print("ADDED IN CACHE FROM OUTSIDE")
return new_item
async def get(self, job_id: uuid.UUID):
async with self._lock:
return self.mapping.get(str(job_id))
async def remove(self, source_link: str) -> None:
async with self._lock:
self._items.pop(source_link, None)
async def get_active_items(self) -> list[FileDictItem]:
print("ACTIVE ITEMS")
print(self.mapping)
async with self._lock:
return [
item
for item in self.mapping.values()
]

14
src/core/clients.py Normal file
View File

@ -0,0 +1,14 @@
from miniopy_async import Minio
from io import BytesIO
class S3Storage:
def __init__(self, client: Minio, bucket_name: str):
self.client = client
self.bucket = bucket_name
async def put_object(self, obj_name, file_data):
try:
stream = BytesIO(file_data)
await self.client.put_object(self.bucket, obj_name, stream, length=len(file_data))
except:
raise

12
src/core/config.py Normal file
View File

@ -0,0 +1,12 @@
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
DATABASE_URL: str
MINIO_URL: str
MINIO_ADMIN_LOGIN: str
MINIO_ADMIN_PASSWORD: str
MINIO_BUCKET: str
NATS_URL: str
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
settings = Settings()

100
src/core/nats_callback.py Normal file
View File

@ -0,0 +1,100 @@
import asyncio
import json
import io
import pandas as pd
from datetime import datetime, timedelta
from nats.aio.client import Client as NATS
from miniopy_async.error import S3Error
from .clients import S3Storage
from .acync_cache import AsyncInMemoryCache
from .types import FileDictItem
def create_json_file(job_id, leg_entity_id, stored_file_id):
file_data = {
'recognition_job_id': job_id,
'legal_entity_id': leg_entity_id,
'stored_file_id': stored_file_id,
'period_quarter': 'Q3',
'period_year': 2024,
'rows': [
{
'account_code': '70.01',
'account_root': '70',
'counterparty_name': 'ООО Компания',
# 'counterparty_id': 'c161c259-3a80-45aa-af7b-33cd5a3d2731',
'counterparty_inn': '34146448377',
'opening_debit': 45.87,
'opening_credit': 65.98,
'turnover_debit': 32.70,
'turnover_credit': 643.98,
'closing_debit': 43.87,
'closing_credit': 45.59
},
{
'account_code': '70.02',
'account_root': '70',
'counterparty_name': 'ООО Компания',
# 'counterparty_id': 'c161c259-3a80-45aa-af7b-33cd5a3d2731',
'counterparty_inn': '34146448377',
'opening_debit': 5.87,
'opening_credit': 62.38,
'turnover_debit': 31.40,
'turnover_credit': 63.98,
'closing_debit': 4.83,
'closing_credit': 4.59
}
]
}
return json.dumps(file_data).encode('utf-8')
def build_file_handler(nc: NATS, storage: S3Storage, bucket_name: str):
async def handler(msg):
print("GOT MESSAGE")
print(msg)
await asyncio.sleep(3)
response_payload = {
'status': None,
'data': None
}
try:
payload = json.loads(msg.data.decode('utf-8'))
# print("PAYLOAD", payload)
s3_key = payload['s3_key']
if not(s3_key.lower().endswith('xlsx') or s3_key.lower().endswith('xls')):
return
file_id = payload['file_id']
legal_entity_id = payload['legal_entity_id']
response = await storage.client.get_object(bucket_name, s3_key) # заменить на storage.get_object('gdffd')
file = await response.content.read()
bio = io.BytesIO(file)
df = pd.read_excel(bio)
result_bytes = create_json_file(payload['job_id'], legal_entity_id, file_id)
await storage.put_object(f'{file_id}.json', file_data=result_bytes)
response_payload['status'] = 'success'
response_payload['data'] = {
# 'result_s3_path': f'{file_id}.json',
'result_s3_path': None,
**payload
}
# storage.put_object()
# file
# print("RESPONSE", response)
except S3Error:
print("NO S3 FILE")
response_payload['status'] = 'error'
response_payload['data'] = {
**payload
}
await nc.publish('files.parsed', json.dumps(response_payload).encode('utf-8'))
return handler

143
src/core/nats_callback_pdf.py Normal file
View File

@ -0,0 +1,143 @@
import asyncio
import json
import io
import pandas as pd
from datetime import datetime, timedelta
from nats.aio.client import Client as NATS
from miniopy_async.error import S3Error
from .clients import S3Storage
from .acync_cache import AsyncInMemoryCache
from .types import FileDictItem
def create_json_file(job_id, file_id):
file_data = {
"job_id": job_id,
"files": [
{
'file_uid': file_id,
"error": {
"code": "",
"message": ""
},
"data_ocr": {
"label": {
'form_name': 's'
},
"organization": {},
"inn": {},
"units": {},
"date": {'form_name': '01.06.2026', 'confidence': 100.0},
"table_structure": [
{
"label": {
"form_name": "Бухгалтерский баланс",
"confidence": 100.0
},
"rows": [
{
"row_index": 0,
"cells": [
{
"column_index": 0,
"value": "Наименование показателя 2",
"confidence": 100.0
},
{
"column_index": 1,
"value": "Код",
"confidence": 100.0
},
{
"column_index": 2,
"value": "За первое полугодие",
"confidence": 100.0
}
]
},
{
"row_index": 1,
"cells": [
{
"column_index": 0,
"value": "Наименование показателя 2",
"confidence": 100.0
},
{
"column_index": 1,
"value": "1100",
"confidence": 100.0
},
{
"column_index": 2,
"value": 984,
"confidence": 100.0
}
]
}
]
}
]
}
}
]
}
def build_file_handler_pdf(nc: NATS, storage: S3Storage, bucket_name: str, cache: AsyncInMemoryCache):
async def handler(msg):
print("GOT MESSAGE")
print(msg)
await asyncio.sleep(3)
response_payload = {
'status': None,
'data': None
}
try:
payload = json.loads(msg.data.decode('utf-8'))
print("PAYLOAD", payload)
s3_key = payload['s3_key']
if (s3_key.lower().endswith('xlsx') or s3_key.lower().endswith('xls')):
return
file_id = payload['file_id']
job_id = payload['job_id']
legal_entity_id = payload['legal_entity_id']
response = await storage.client.get_object(bucket_name, s3_key) # заменить на storage.get_object('gdffd')
file = await response.content.read()
with open(f'/home/sokratmillman/projects/work-test-processor/files/input/{s3_key}', 'wb') as f:
f.write(file)
await cache.add(str(job_id), str(file_id))
print("ADDED to CACHE", str(job_id), str(file_id))
# bio = io.BytesIO(file)
# df = pd.read_excel(bio)
# result_bytes = create_json_file(job_id)
# await storage.put_object(f'{file_id}.json', file_data=result_bytes)
# response_payload['status'] = 'success'
# response_payload['data'] = {
# 'result_s3_path': f'{file_id}.json',
# **payload
# }
# storage.put_object()
# file
# print("RESPONSE", response)
except S3Error:
print("NO S3 FILE")
response_payload['status'] = 'error'
response_payload['data'] = {
**payload
}
await nc.publish('files.parsed', json.dumps(response_payload).encode('utf-8'))
return handler

107
src/core/poll_worker.py Normal file
View File

@ -0,0 +1,107 @@
import asyncio
import json
import os
from datetime import datetime, timedelta
from nats.aio.client import Client as NATS
from .clients import S3Storage
from .acync_cache import AsyncInMemoryCache
from .types import FileDictItem
class PollWorker:
def __init__(self, nats_client: NATS, storage: S3Storage, tracker: AsyncInMemoryCache):
self.task: asyncio.Task | None = None
self.stop_event = asyncio.Event()
self.nats_client = nats_client
self.tracker = tracker
self.interval = 5
self.storage = storage
async def start(self):
self.stop_event.clear()
self.task = asyncio.create_task(self.run())
async def stop(self):
self.stop_event.set()
if self.task:
self.task.cancel()
try:
await self.task
except asyncio.CancelledError:
pass
async def run(self):
# logger.info("Pending publisher started")
print("PUBLISHER STARTED")
while not self.stop_event.is_set():
try:
await self.poll_once()
except Exception:
# logger.exception("Poll iteration failed")
pass
try:
await asyncio.wait_for(
self.stop_event.wait(),
timeout=self.interval,
)
except asyncio.TimeoutError:
pass
# logger.info("Pending publisher stopped")
print("PUBLISHER STOPPED")
async def poll_once(self):
print("POLLING FILES")
try:
jobs: list[FileDictItem] = await self.tracker.get_active_items()
print("GOT ")
except Exception as e:
print("FAILED TO CLAIN ROWS")
print(e)
all_current_files = os.listdir('/home/sokratmillman/projects/work-test-processor/files/output/')
print("ALL FILES", all_current_files)
for job in jobs:
print("ITERATE OVER FILES")
search_fname = f"{job['file_id']}.json"
print("SEARCH NAME", search_fname)
now = datetime.now()
try:
if (job['expires_at'] + timedelta(seconds=2*self.interval) > now):
try:
await self.publish_nats_expired(job['file_id'])
await self.tracker.remove(job['job_id'])
print("PUBLISHED TIMEOUT ERROR")
except Exception as e:
print("ERROR IN TIMEOUT HANDLIING")
except Exception as e:
print("ERROR", e)
print("BEFORE TRY")
print("JOB", job)
try:
print("PUBLISHING")
# await self.publish_to_nats(job["row_payload"])
print("PUBLISHED SUCESSFULLY")
except Exception as e:
# logger.exception("Failed to publish row id=%s", row_id)
print("FAILED TO PUBLISH", job['job_id'])
print(e)
# await self.mark_queued(job.job_id)
async def publish_to_nats(self, payload):
await self.nats_client.publish('files.parsed', json.dumps(payload).encode("utf-8"))
async def publish_nats_expired(self, job_id):
error_payload = {
'status': 'error',
'error_message': 'timeout error'
}
await self.nats_client.publish('files.parsed', json.dumps(error_payload).encode("utf-8"))

12
src/core/types.py Normal file
View File

@ -0,0 +1,12 @@
from dataclasses import dataclass
from datetime import datetime
@dataclass
class FileDictItem:
started_at: datetime
expires_at: datetime
file_id: str
job_id: str
# @dataclass
# class OsvRow:

3
src/schemas/__init__.py Normal file
View File

@ -0,0 +1,3 @@
from .legal_entity import LegalEntityPostRequest, LegalEntityPostResponse
from .company_group import CompanyGroupPostRequest, CompanyGroupPostResponse
from .rf_office import RfOfficePostRequest, RfOfficePostResponse

19
src/schemas/company_group.py Normal file
View File

@ -0,0 +1,19 @@
from pydantic import BaseModel, ConfigDict
import uuid
from datetime import datetime, date
class CompanyGroupPostRequest(BaseModel):
crm_id: str
name: str
base_regional_office_id: uuid.UUID
updated_by: str
class CompanyGroupPostResponse(BaseModel):
id: uuid.UUID
crm_id: str
name: str
base_regional_office_id: uuid.UUID
valid_from: datetime
updated_by: str
created_at: datetime
model_config = ConfigDict(from_attributes=True)

65
src/schemas/enums.py Normal file
View File

@ -0,0 +1,65 @@
from enum import StrEnum
class LegalEntityType(StrEnum):
IP = 'ИП'
UL = 'ЮЛ'
class RelatedGroupKind(StrEnum):
GSK = 'gsk'
GSZ = 'gsz'
class AccountKind(StrEnum):
A = 'A'
P = 'P'
AP = 'AP'
class DocumentType(StrEnum):
BUH_OT = "Бухгалтерская отчётность"
OSV = "ОСВ"
OFR = "ОФР"
OTHER = "Прочий документ"
class FileType(StrEnum):
XLSX = "XLSX"
PDF = "PDF"
PNG = "PNG"
JPG = "JPG"
class RecognitionPipeline(StrEnum):
TABULAR = "tabular"
IMAGE = "image"
class PeriodQuarter(StrEnum):
Q1 = "Q1"
Q2 = "Q2"
Q3 = "Q3"
Q4 = "Q4"
class FileStatus(StrEnum):
UPLOADED = "Загружен"
PROCESSING = "Обработка"
CHECK="Требуется проверка"
ERROR="Ошибка"
class RecognitionStatus(StrEnum):
QUEUED = "queued"
PROCESSING = "processing"
DONE = "done"
ERROR = "error"
class CounterpartyType(StrEnum):
GK = "ГК"
GSK = "ГСК"
EXTERNAL = "Внешний"
class StatementType(StrEnum):
BALANCE = "balance"
INCOME = "income"
class FslRowKind(StrEnum):
SECTION = "section"
ROW = "row"
SUB = "sub"
TOTAL = "total"
BALANCE = "balance"

25
src/schemas/legal_entity.py Normal file
View File

@ -0,0 +1,25 @@
from pydantic import BaseModel, ConfigDict, Field
import uuid
from datetime import datetime, date
from .enums import LegalEntityType
class LegalEntityPostRequest(BaseModel):
inn: str = Field(min_length=10, max_length=12, pattern=r'^\d+$')
entity_type: LegalEntityType | None
name: str
company_group_id: uuid.UUID
regional_office_id: uuid.UUID
close_date: date
updated_by: str
class LegalEntityPostResponse(BaseModel):
id: uuid.UUID
inn: str
entity_type: LegalEntityType | None
name: str
company_group_id: uuid.UUID
regional_office_id: uuid.UUID
close_date: date
updated_by: str
created_at: datetime
model_config = ConfigDict(from_attributes=True)

12
src/schemas/rf_office.py Normal file
View File

@ -0,0 +1,12 @@
from pydantic import BaseModel, ConfigDict
import uuid
class RfOfficePostRequest(BaseModel):
name: str
is_active: bool = True
class RfOfficePostResponse(BaseModel):
id: uuid.UUID
name: str
is_active: bool
model_config = ConfigDict(from_attributes=True)

4
src/services/__init__.py Normal file
View File

@ -0,0 +1,4 @@
from .company_group_service import CompanyGroupService
from .file_service import FileService
from .legal_entity_service import LegalEntityService
from .rf_office_service import RfOfficeService

30
src/services/company_group_service.py Normal file
View File

@ -0,0 +1,30 @@
import uuid
from fastapi import UploadFile
from sqlalchemy.ext.asyncio import AsyncSession
from src.repositories.company_group_repo import CompanyGroupRepository
from src.common.exceptions import NotFoundError, CouldntCreateError
class CompanyGroupService:
def __init__(self, session: AsyncSession, com_group_repo: CompanyGroupRepository):
self.com_group_repo = com_group_repo
self.session = session
async def create_company_group(self, data):
try:
company_group = await self.com_group_repo.insert_one(data.model_dump())
await self.session.commit()
return company_group
except Exception as e:
await self.session.rollback()
raise
async def get_all_groups(self):
try:
company_groups = await self.com_group_repo.get_all()
if len(company_groups) == 0:
raise NotFoundError('No company groups found')
return company_groups
except Exception as e:
raise

72
src/services/file_service.py Normal file
View File

@ -0,0 +1,72 @@
import uuid
from fastapi import UploadFile
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.inspection import inspect
from src.core.clients import S3Storage
from src.models.enums import FileStatus, FileType
def get_dict(obj):
return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}
class FileService:
def __init__(self, storage: S3Storage, session: AsyncSession, file_repository, recognition_job_repo, nats_conn):
self.storage = storage
self.file_repo = file_repository
self.rec_job_repo = recognition_job_repo
self.nats_conn = nats_conn
self.session = session
async def save_file(self, file: UploadFile, legal_entity_id: uuid.UUID, quarter: str, year: int):
try:
db_payload = {
'original_name': file.filename,
'legal_entity_id': legal_entity_id,
'period_quarter': quarter,
'period_year': year,
'size_bytes': file.size,
}
file_ext = file.filename.lower().split('.')[-1]
if (file.filename.endswith('xlsx')):
db_payload['file_type'] = FileType.XLSX
elif (file.filename.endswith('pdf')):
db_payload['file_type'] = FileType.PDF
elif (file.filename.endswith('png')):
db_payload['file_type'] = FileType.PNG
elif (file.filename.endswith('jpg')):
db_payload['file_type'] = FileType.JPG
new_file = await self.file_repo.insert_one(db_payload)
s3_key = f'{str(new_file.id)}.{file_ext}'
await self.storage.put_object(s3_key, await file.read())
new_file.s3_key = s3_key
new_file.status = FileStatus.UPLOADED
rec_job_db_payload = {
'file_id': new_file.id,
'attempt': 1
}
rec_job = await self.rec_job_repo.insert_one(rec_job_db_payload)
new_file.current_recognition_job_id = rec_job.id
await self.session.commit()
print(get_dict(rec_job))
print(get_dict(new_file))
return {'file': get_dict(new_file), 'rec_job': get_dict(rec_job)}
except Exception as e:
await self.session.rollback()
print("ERROR")
print(e)
raise
async def _save_to_storage(self, file: UploadFile, file_id: uuid.UUID):
print("saving one file")
print(file)
def _generate_file_ids():
pass

28
src/services/legal_entity_service.py Normal file
View File

@ -0,0 +1,28 @@
import uuid
from fastapi import UploadFile
from sqlalchemy.ext.asyncio import AsyncSession
from src.common.exceptions import NotFoundError
from src.repositories.legal_entity_repo import LegalEntityRepository, LegalEntity
class LegalEntityService:
def __init__(self, session: AsyncSession, le_repo: LegalEntityRepository):
self.le_repo = le_repo
self.session = session
async def create_legal_entity(self, data):
try:
legal_entity: LegalEntity = await self.le_repo.insert_one(data.model_dump())
await self.session.commit()
return legal_entity
except Exception as e:
await self.session.rollback()
raise
async def get_all_offices(self):
try:
ref_offices = await self.le_repo.get_all()
if len(ref_offices) == 0:
raise NotFoundError('No legal entities found')
return ref_offices
except Exception as e:
raise

30
src/services/rf_office_service.py Normal file
View File

@ -0,0 +1,30 @@
import uuid
from fastapi import UploadFile
from sqlalchemy.ext.asyncio import AsyncSession
from src.repositories import RfRepository
from src.common.exceptions import NotFoundError, CouldntCreateError
class RfOfficeService:
def __init__(self, session: AsyncSession, ref_office_repo: RfRepository):
self.ref_office_repo = ref_office_repo
self.session = session
async def create_ref_office(self, data):
try:
ref_office = await self.ref_office_repo.insert_one(data.model_dump())
await self.session.commit()
return ref_office
except Exception as e:
await self.session.rollback()
raise
async def get_all_offices(self):
try:
ref_offices = await self.ref_office_repo.get_all()
if len(ref_offices) == 0:
raise NotFoundError('No company groups found')
return ref_offices
except Exception as e:
raise