Correcting replication and no_files flag

.gitea/workflows/build.yaml

@@ -0,0 +1,23 @@
+name: Build xmlrpcserver image
+run-name: Build xmlrpcserver image
+on: [push]
+
+jobs:
+  Build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: https://github.com/actions/checkout@v4
+      - name: Set up Docker Buildx
+        uses: https://github.com/docker/setup-buildx-action@v3
+        with:
+          config-inline: |
+            [registry."10.10.8.83:32000"]
+              http = true
+              insecure = true
+      - name: Build and push Docker image
+        uses: https://github.com/docker/build-push-action@v5
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          tags: "10.10.8.83:32000/xmlrpc:latest"

Dockerfile

@@ -0,0 +1,14 @@
+FROM reg.ivazh.ru/infra-oodb
+WORKDIR /app
+COPY . ./
+RUN cd deps/pygost-5.12/ && \
+    python3 setup.py install && \
+    cd ../.. && \
+    pip3 install -r requirements.txt && \
+    mkdir -p /opt/tnt/bin && \
+    ln -s /usr/bin/python3 /opt/tnt/bin/python3
+ENV LD_LIBRARY_PATH "/app"
+ENV PYTHONPATH "${PYTHONPATH}:/app"
+EXPOSE 9000
+EXPOSE 8000
+CMD ["python3", "main.py"]

config.py

@@ -1,8 +1,8 @@
 
 class Config:
-    ret_path: str = 'http://10.10.8.81:9000/'
+    ret_path: str = 'http://10.10.8.83:32200/'
     self_bnd: str = 'bnd127'
-    enserver: str = 'http://127.0.0.1:7000/xmlrpc'
+    enserver: str = 'http://10.10.8.83:32210/xmlrpc'
     remote_bnd: str = 'bnd128'
 
     pg_host: str = '10.10.8.83'
@@ -12,7 +12,7 @@ class Config:
     pg_password: str = 'Root12345678'
 
     oodb_host: str = '10.10.8.83'
-    oodb_port: int = 32110
+    oodb_port: int = 32100
     oodb_dbname: str = 'db'
     oodb_username: str = 'postgres'
     oodb_passwd: str = 'Root12345678'
@@ -21,7 +21,19 @@ class Config:
     rabbit_conn: str = 'amqp://user:password@10.10.8.83:31005/%2f'
     rabbit_queue: str = 'ipd'
 
+    ws_rabbit_params: dict = {
+        'host': '10.10.8.83',
+        'port': 31005,
+        'exchange': 'ipd',
+        'user': 'user',
+        'password': 'password',
+    }
+
     s3_endpoint: str = 'http://10.10.8.83:31006'
     s3_key_id: str = 's57'
     s3_access_key: str = 'd9MMinLF3U8TLSj'
     s3_bucket: str = 'files'
+
+    gql_url: str = 'https://gql.ivazh.ru/graphql'
+    gql_download: str = 'https://gql.ivazh.ru/item/{key}'
+    gql_schema: str = 'pdim'

config/response2.json

@@ -0,0 +1,12 @@
+{
+  "filestorage": {
+    "type": "s3",
+    "endpoint": "http://10.10.8.83:31006",
+    "key_id": "s57",
+    "access_key": "d9MMinLF3U8TLSj",
+    "download_path": "/tmp"
+  },
+  "file_code": "c1000",
+  "name_code": "c122",
+  "use_version": true
+}

config/workspaces.json

@@ -0,0 +1,167 @@
+{
+  "databases": {
+    "oodb_git": {
+      "host": "10.10.8.83",
+      "port": 32100,
+      "database": "db",
+      "user": "postgres",
+      "password": "Root12345678"
+    }
+  },
+  "workspaces": {
+    "documents_src": {
+      "type": "documents",
+      "group": "src",
+      "database": "oodb_git",
+      "schema": "documents_src",
+      "alias": "Документы исходная"
+    },
+    "documents_standard": {
+      "type": "documents",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "documents_standard",
+      "alias": "Документы эталон"
+    },
+    "documents_standard_pub": {
+      "type": "documents",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "documents_standard",
+      "alias": "Документы публичная"
+    },
+    "ood": {
+      "type": "npd",
+      "group": "src",
+      "database": "oodb_git",
+      "schema": "ood",
+      "alias": "ООБД исходные НПД",
+      "map_service": "VUE_APP_GISAIS_URL:/styles/ood/style.json"
+    },
+    "oodb": {
+      "type": "oodb",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "kartap",
+      "alias": "ООДБ эталон",
+      "map_service": "VUE_APP_GISAIS_URL:/styles/oodb/style.json"
+    },
+    "oodb_standard": {
+      "type": "oodb",
+      "group": "forming_standard",
+      "database": "oodb_git",
+      "schema": "kartap",
+      "alias": "ООДБ эталон",
+      "map_service": "VUE_APP_GISAIS_URL:/styles/oodb/style.json"
+    },
+    "oodb_working": {
+      "type": "oodb",
+      "group": "forming_work",
+      "database": "oodb_git",
+      "schema": "kartap",
+      "alias": "ООДБ рабочая",
+      "map_service": "VUE_APP_GISAIS_URL:/styles/oodb_tech/style.json"
+    },
+    "oodb_pub": {
+      "type": "oodb",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "kartap",
+      "alias": "ООБД публичная",
+      "map_service": "VUE_APP_GISAIS_URL:/styles/oodb/style.json"
+    },
+    "regions": {
+      "type": "regions",
+      "database": "oodb_git",
+      "schema": "regions_hard",
+      "alias": "Регионы",
+      "map_service": "VUE_APP_GISAIS_URL_GK:/styles/regions/style.json"
+    },
+    "regions_contour": {
+      "type": "regions",
+      "database": "oodb_git",
+      "schema": "regions_hard",
+      "alias": "Регионы",
+      "map_service": "VUE_APP_GISAIS_URL_GK:/styles/regions_contour/style.json"
+    },
+    "npd_9": {
+      "type": "npd",
+      "database": "oodb_git",
+      "schema": "npd_9",
+      "alias": "НПД 9.0"
+    },
+    "npd": {
+      "type": "npd",
+      "database": "oodb_git",
+      "schema": "initial",
+      "alias": "НПД 9.0"
+    },
+    "npd_831": {
+      "type": "npd",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "npd_831",
+      "alias": "НПД 8.31"
+    },
+    "updater_test": {
+      "type": "npd",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "npd_831_test",
+      "alias": "НПД 8.31 публичная"
+    },
+    "lukoil": {
+      "type": "oodb",
+      "database": "oodb_git",
+      "schema": "lukoil",
+      "alias": "ЛУКОЙЛ",
+      "map_service": "VUE_APP_GISAIS_URL_GK:/styles/lukoil/style.json"
+    },
+    "geocover": {
+      "type": "ecpz",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "coverage",
+      "alias": "ЕЦПЗ"
+    },
+    "geocover_test": {
+      "type": "ecpz",
+      "database": "oodb_git",
+      "schema": "coverage",
+      "alias": "ЕЦПЗ тест"
+    },
+    "gcmr": {
+      "type": "gcmr",
+      "group": "order",
+      "database": "oodb_git",
+      "schema": "gcmr",
+      "alias": "ГЦМР"
+    },
+    "orders": {
+      "type": "system",
+      "database": "oodb_git",
+      "schema": "orders",
+      "alias": "Заказы"
+    },
+    "ilo": {
+      "type": "system",
+      "database": "oodb_git",
+      "schema": "ilo",
+      "alias": "ИЛО"
+    },
+    "raz_sgok": {
+      "type": "raz_sgok",
+      "database": "razsgok",
+      "schema": "razsgok",
+      "alias": "СГОК",
+      "map_service": "VUE_APP_GISAIS_URL_GK:/styles/raz_sgok/style.json"
+    },
+    "raz_vtu": {
+      "type": "raz_vtu",
+      "database": "razvtu",
+      "schema": "razvtu",
+      "alias": "ВТУ",
+      "map_service": "VUE_APP_GISAIS_URL_GK:/styles/raz_vtu/style.json"
+    }
+  }
+} 

db.py

@@ -2,6 +2,7 @@ from datetime import datetime
 from typing import List, Optional
 from sqlalchemy import create_engine, String, select, ForeignKey, Enum
 from sqlalchemy.orm import Session, DeclarativeBase, Mapped, mapped_column, relationship
+from config import Config
 
 
 def tow(day: int, hour: int, minute: int):
@@ -27,6 +28,10 @@ class User(Base):
         back_populates='user', cascade='all, delete-orphan'
     )
 
+    income_branches: Mapped[List['IncomeBranch']] = relationship(
+        back_populates='user', cascade='all, delete-orphan'
+    )
+
     schedule: Mapped[List['Schedule']] = relationship(
         back_populates='user', cascade='all, delete-orphan'
     )
@@ -49,6 +54,7 @@ class User(Base):
             'profiles': [x.to_dict() for x in self.profiles],
             'schedule': [x.to_dict() for x in self.schedule],
             'queue': [x.to_dict() for x in self.queue],
+            'income_branches': [x.to_dict() for x in self.income_branches],
         }
 
     def is_active_now(self):
@@ -69,17 +75,41 @@ class Profile(Base):
     id: Mapped[int] = mapped_column(primary_key=True)
     user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
     scheme: Mapped[str]
-    json: Mapped[str]
+    branch: Mapped[str] = mapped_column(String, nullable=True)
+    json: Mapped[str] = mapped_column(String, nullable=True)
+    no_files: Mapped[bool]
 
     user: Mapped['User'] = relationship(back_populates='profiles')
 
     def to_dict(self) -> dict:
         return {
             'id': self.id,
+            'scheme': self.scheme,
+            'branch': self.branch,
             'json': self.json,
         }
 
 
+class IncomeBranch(Base):
+    __tablename__ = 'income_branches'
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
+    scheme: Mapped[str]
+    branch: Mapped[str]
+    local_scheme: Mapped[str]
+
+    user: Mapped['User'] = relationship(back_populates='income_branches')
+
+    def to_dict(self) -> dict:
+        return {
+            'id': self.id,
+            'scheme': self.scheme,
+            'branch': self.branch,
+            'local_scheme': self.local_scheme,
+        }
+
+
 class Schedule(Base):
     __tablename__ = 'schedule'
 
@@ -133,4 +163,4 @@ class Schemas(Base):
 
 
 def connect_db():
-    return create_engine("postgresql+psycopg://postgres:Root12345678@10.10.8.83:32101/db")
+    return create_engine(f"postgresql+psycopg://{Config.pg_username}:{Config.pg_password}@{Config.pg_host}:{Config.pg_port}/{Config.pg_dbname}")

infra.py

@@ -0,0 +1,9 @@
+from libcommon import *
+from libdatabase import *
+from libgeodata import *
+from libgeodriver import *
+from libgeodesy import *
+from libgeom import *
+from libipdutilities import *
+from liboodriver import *
+

main.py

@@ -1,7 +1,7 @@
-import asyncio
 from tempfile import TemporaryDirectory
+from typing import Optional, Any
+
 import pika
-import sys
 import threading
 import time
 from queue import Queue
@@ -11,31 +11,43 @@ from uuid import uuid4, UUID
 from xmlrpc.server import SimpleXMLRPCServer
 from xmlrpc.client import ServerProxy
 import logging
-import os
+import io
+import zlib
 import os.path
 import requests
-from reqs.graphql import get_catalog, get_object
+from reqs_graphql import get_catalog, get_object
 from pygost import gost34112012256
 import xml.etree.ElementTree as ET
 from reqs.request_xml_service import RequestXmlService
 import zipfile
-from infra import *
 from config import Config
 from zip import Zip
 import boto3
 import db
 from sqlalchemy.orm import Session
-from fastapi import FastAPI, Response, Form, UploadFile, File
+from fastapi import FastAPI, Response, Form, UploadFile, File, Request
+from fastapi.middleware.cors import CORSMiddleware
 import uvicorn
-from typing import Annotated
+from typing_extensions import Annotated
+import pathlib
+from infra import *
+from shutil import make_archive
 
 
 NEW_REPLICATION_REQUEST = 99
+NEW_COMMIT_REQUEST = 98
+NEW_COMMIT_RESPONSE = 1098
+NEW_DATA_REQUEST = 97
+NEW_DATA_RESPONSE = 1097
 
 
 tasks = Queue()
 connected = set()
 
+connection: Optional[pika.BlockingConnection] = None
+channel: Optional[Any] = None
+server: Optional[SimpleXMLRPCServer] = None
+
 logger = logging.getLogger('xmlrpcserver')
 
 
@@ -59,6 +71,24 @@ def upload_file(filename: str, key: str, bucket: str):
         client.put_object(Body=f.read(), Bucket=bucket, Key=key)
 
 
+def get_branch(bndname: str, scheme: str):
+    conn = db.connect_db()
+    with Session(conn) as session:
+        item = session.query(db.IncomeBranch).filter_by(scheme=scheme).join(db.User).filter_by(bndname=bndname).one_or_none()
+        if item:
+            return item.branch, item.local_scheme
+        return None, None
+
+
+def get_profile(bndname: str, scheme: str):
+    conn = db.connect_db()
+    with Session(conn) as session:
+        item = session.query(db.Profile).filter_by(scheme=scheme).join(db.User).filter_by(bndname=bndname).one_or_none()
+        if item:
+            return item
+        return None
+
+
 def run_tasks():
     logger.debug('Task thread started.')
     while True:
@@ -68,29 +98,181 @@ def run_tasks():
 
 def replication_task():
     while True:
-        print()
         conn = db.connect_db()
-        with (Session(conn) as session):
-            for bndname in connected:
-                for item in session.query(db.Queue).join(db.Queue.user).filter_by(bndname=bndname).all():
-                    if item.user.is_active_now():
+        # logging.warning(connected)
+        with Session(conn) as session:
+            for item in session.query(db.Queue).join(db.Queue.user).all():
+                bndname = item.user.bndname
+                if item.user.is_active_now() and bndname != Config.self_bnd:
+                    if item.user.newbnd:
                         replication(bndname, item.commit_id, item.schema)
-                        session.delete(item)
+                    else:
+                        replication_old(bndname, item.commit_id, item.schema)
+                    session.delete(item)
             session.commit()
         time.sleep(60)
 
 
+def process_chart(ws, chart, chart_class):
+    for x in chart.get('properties') or {}:
+        attr = ws.attribute(x)
+        if not attr:
+            continue
+        if str(attr.type()) == 'AT_Domain':
+            dom = attr.domain()
+            key = str(chart['properties'][x])
+            chart['properties'][x] = variantToString(dom.value(variantFromString(key)))
+        if x == 'c103':
+            chart['properties'][x] = 'Несекретно'
+    chart['class'] = chart_class['name'].split('_')[-1]
+    return chart
+
+
+def crc32(filename, chunk_size=65536):
+    """Compute the CRC-32 checksum of the contents of the given filename"""
+    with open(filename, "rb") as f:
+        checksum = 0
+        while chunk := f.read(chunk_size):
+            checksum = zlib.crc32(chunk, checksum)
+        return "%08X" % (checksum & 0xFFFFFFFF)
+
+
+def send_object_replication(bndname, ws, ids):
+    qu = GroupQuery(Envelope())
+    query = GroupQuery(Envelope.world())
+    query.setUids(ids)
+    uids = []
+    ws.load(query, uids)
+    res = json.loads(ws.dataToJson())
+    charts = [process_chart(ws, f, x) for x in res for f in x['features']]
+
+    for chart in charts:
+        logger.warning('\n')
+        date = datetime.datetime.now()
+        rxmls = RequestXmlService()
+        res_id = uuid4().hex
+        res = rxmls.get_request_document(res_id, None)
+        res.set('replication_package', '1')
+        res.set('replication_version', date.strftime('%Y%m%d%H%M%S'))
+        res.set('user_permit', 'AA0AA00020200726D3E75C80B713A7A3D3E75C80B713A7A363F7CB889AA3F520')
+        rxmls.set_result(res, 0, '')
+        print(chart)
+        properties = chart.get('properties') or {}
+        c1000 = properties.get('c1000')
+        if not c1000:
+            logger.warning(f'No file for {chart["uid"].replace("-", "")}')
+            continue
+        z = Zip()
+        xml_objects = ET.SubElement(res, 'objects')
+        created_date = datetime.datetime.fromisoformat(chart.get('date_created'))
+        xml_chart = ET.SubElement(res, 'chart', {
+            'Class': chart['class'],
+            'ID': chart['uid'].replace('-', ''),
+            'Name': c1000[0]['fileName'],
+            'Type': 'OOD',
+            'metadata_version': '1',
+            'source': Config.self_bnd,
+            'system_date': str(created_date.timestamp()),
+        })
+        xml_version = ET.SubElement(xml_objects, 'version', {
+            'object_id': chart['uid'].replace('-', ''),
+            'source': Config.self_bnd,
+            'system_date': str(created_date.timestamp()),
+            'version': '1.0',
+            'version_id': chart['uid'].replace('-', ''),
+        })
+        total_size = 0
+        for file in c1000:
+            directory = os.path.join(z.dirname, f'maps/{res_id}/ENC_ROOT/{c1000[0]["fileName"]}')
+            fp = os.path.join(directory, file['fileName'])
+            if not os.path.exists(directory):
+                os.makedirs(directory)
+            download_file(file['key'], Config.s3_bucket, fp)
+            size = os.stat(fp).st_size
+            _crc32 = crc32(fp)
+            ET.SubElement(xml_version, 'file', {
+                'cell_file': 'false',
+                'crc32': _crc32,
+                'crc32_enc': _crc32,
+                'file_id': file['key'].replace('-', ''),
+                'file_ref': os.path.join(f'maps/{res_id}/ENC_ROOT', file['fileName']),
+                'file_size': str(size),
+                'file_size_enc': str(size),
+            })
+            xml_version.set('crc32', _crc32)
+            xml_version.set('crc32', _crc32)
+            total_size += size
+        xml_version.set('size', str(total_size))
+        xml_version.set('crc32', str(total_size))
+
+        xml_tags = ET.SubElement(res, 'tags')
+        xml_archs = ET.SubElement(res, 'archs')
+        xml_arch = ET.SubElement(xml_archs, 'arch', {
+            'obj_id': chart['uid'].replace('-', ''),
+            'ver_cl': '8.31',
+            'ver_id': chart['uid'].replace('-', ''),
+        })
+        for attribute in properties:
+            if attribute.startswith('c'):
+                ET.SubElement(xml_chart, 'Attribute', {
+                    'name': attribute.replace('_', '.'),
+                    'value': str(chart['properties'][attribute]),
+                })
+                ET.SubElement(xml_arch, 'attr', {
+                    'code': attribute.replace('_', '.'),
+                    'name': attribute.replace('_', '.'),
+                    'value': str(chart['properties'][attribute]),
+                })
+        params = {
+            'from': f'tcp://{Config.self_bnd}',
+            'to': f'tcp://{bndname}',
+            'ts_added': date.timestamp(),
+            'user_id': '0',
+            'query_type': NEW_REPLICATION_REQUEST,
+            'query_data': ET.tostring(res, encoding='unicode', xml_declaration=True),
+        }
+        filepath = z.pack()
+        response_files = [{'name': os.path.basename(filepath), 'url': filepath, 'size': os.path.getsize(filepath)}]
+        logger.warning(response_files)
+        logging.debug('Send replication package')
+        proxy = ServerProxy(Config.enserver)
+        try:
+            proxy.send(params, response_files, Config.ret_path)
+        except:
+            logger.error('Error sending')
+
+
+def replication_old(bnd_name: str, commit_id: str, schema: str):
+    logger.warning('Start replication')
+    if schema != Config.oodb_schema:
+        return
+    con = OOConnectionParams(schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
+                             Config.oodb_username, Config.oodb_passwd, schema)
+    ws = OODBWorkspace.ws(schema)
+    if not ws.isInit():
+        res = ws.init(con)
+        logger.warning(res)
+    created, updated, _ = ws.changes(commit_id)
+    ids = list(set(created) | set(updated))
+    send_object_replication(bnd_name, ws, ids)
+    ws.clearData(True)
+    logger.warning('Replication to old bnd is sent')
+
+
 def replication(bnd_name: str, commit_id: str, schema: str):
+    logging.warning(f'{bnd_name} {commit_id} {schema}')
     date = datetime.datetime.now()
     rxmls = RequestXmlService()
     res_id = uuid4().hex
 
+    profile = get_profile(bndname=bnd_name, scheme=schema)
+
     res = rxmls.get_request_document(res_id, None)
     rxmls.set_result(res, 0, '')
     ET.SubElement(res, 'replication', {'id': commit_id, 'scheme': schema})
     response_params = {
         'from': f'tcp://{Config.self_bnd}',
-        'to': f'tcp://{Config.remote_bnd}',
+        'to': f'tcp://{bnd_name}',
         'ts_added': date.timestamp(),
         'user_id': '0',
         'query_type': NEW_REPLICATION_REQUEST,
@@ -136,20 +318,23 @@ def replication(bnd_name: str, commit_id: str, schema: str):
     qu.setUids(created)
     ws.load(qu, uids)
     exported_files = []
-    for feature_uid in uids:
-        feature = ws.featureByUid(feature_uid)
-        if not feature:
-            continue
-        for attr in feature.attributes('c1000'):
-            updated_files.append(variantToFileValue(attr.val()))
-            # updated_files.append(feature_uid)
-    for x in updated_files:
-        exported_files.append({
-            'key': x.key,
-            'bucket': x.bucket,
-            'filename': x.fileName,
-        })
-        download_file(x.key, x.bucket, os.path.join(z.dirname, x.key))
+    if not profile or not profile.no_files:
+        for feature_uid in uids:
+            feature = ws.featureByUid(feature_uid)
+            if not feature:
+                continue
+            for attr in feature.attributes('c1000'):
+                updated_files.append(variantToFileValue(attr.val()))
+                # updated_files.append(feature_uid)
+        for x in updated_files:
+            exported_files.append({
+                'key': x.key,
+                'bucket': x.bucket,
+                'filename': x.fileName,
+            })
+            fp = os.path.join(z.dirname, x.key)
+            os.makedirs(os.path.dirname(fp), exist_ok=True)
+            download_file(x.key, x.bucket, fp)
     with open(os.path.join(z.dirname, 'export_files.json'), 'w') as f:
         f.write(json.dumps(exported_files))
     ws.clearData(True)
@@ -167,18 +352,26 @@ def replication(bnd_name: str, commit_id: str, schema: str):
 
 def pika_callback(ch, method, properties, body):
     commit_info = json.loads(body)
+    logging.warning(commit_info)
     schema = commit_info.get('schema') or Config.oodb_schema
     commit = commit_info['commit']
     conn = db.connect_db()
     with Session(conn) as session:
-        for user in session.query(db.User).filter(db.User.active == True, db.User.upstream == False).all():
-            item = db.Queue(user_id=user.id, commit_id=commit, schema=schema)
-            session.add(item)
+        for user in session.query(db.User).filter(db.User.active == True).all():
+            if user.bndname == Config.self_bnd:
+                continue
+            profiles = {x.scheme: x.to_dict() for x in user.profiles}
+            if len(profiles) == 0 or schema in profiles:
+                item = db.Queue(user_id=user.id, commit_id=commit, schema=schema)
+                logging.warning(item)
+                session.add(item)
         session.commit()
     ch.basic_ack(delivery_tag=method.delivery_tag)
 
 
 def pika_task():
+    global connection
+    global channel
     connection = pika.BlockingConnection(pika.URLParameters(Config.rabbit_conn))
     channel = connection.channel()
     channel.basic_consume(queue=Config.rabbit_queue, on_message_callback=pika_callback)
@@ -186,6 +379,7 @@ def pika_task():
     channel.start_consuming()
 
 
+
 def list_contents(dir_name):
     logger.warning('list_contents(%s)', dir_name)
     return os.listdir(dir_name)
@@ -285,7 +479,7 @@ def get_objects(params, files, url):
         for file in obj['properties'].get('c1000', []):
             if not main_filename:
                 main_filename = file['fileName']
-            res = requests.get(f'https://gql.ivazh.ru/item/{file["key"]}')
+            res = requests.get(Config.gql_download, params={'item_id': file["key"]})
             zipf.writestr(f'{main_filename}/{file["fileName"]}', res.content)
     zipf.close()
     response_files = [{'name': filename, 'url': filepath, 'size': os.path.getsize(filepath)}]
@@ -345,35 +539,228 @@ def get_metadata(params, files, url):
 
 
 def put_object(params, files, url):
-    pass
+    date = datetime.datetime.now()
+    req = ET.fromstring(params['query_data'])
+    obj = req.find('chart')
+    class_id = obj.get('Class')
+    con = OOConnectionParams(Config.oodb_schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
+                             Config.oodb_username, Config.oodb_passwd, Config.oodb_schema)
+    ws = OODBWorkspace.ws(Config.oodb_schema)
+    if not ws.isInit():
+        res = ws.init(con)
+        logger.warning(res)
+    logging.info(class_id)
+    fc = ws.featureClass(class_id)
+    feature = fc.createFeature('')
+    geom = Polygon.fromExtent(Envelope(0.0, 0.0, 1.0, 1.0, SRFactory.PZ9011()))
+    res = feature.setGeometry(geom)
+    for attr in obj.findall('Attribute'):
+        name = attr.get('name')
+        value = attr.get('value')
+        res &= feature.addAttribute(name, variantFromString(value))
+
+    assert len(files) == 1
+    file = files[0]
+    dir = TemporaryDirectory()
+    with zipfile.ZipFile(file, 'r') as zip_ref:
+        zip_ref.extractall(dir.name)
+
+    fp = pathlib.Path(dir.name)
+    for item in fp.glob('**/*'):
+        if not item.is_file():
+            continue
+        fileVal = FileValue()
+        key = uuid4().hex
+        fileVal.fileName = variantToString(item.relative_to(dir.name))
+        fileVal.key = variantToString(key)
+        fileVal.bucket = variantToString(Config.s3_bucket)
+        res &= feature.addAttribute('c1000', variantFromFileValue(fileVal))
+        upload_file(str(item), key, Config.s3_bucket)
+
+    ws.transaction()
+    res = ws.save()
+    ws.commit(f'Putobject from {params["to"]}')
+    ws.clearData(True)
 
 
 def apply_commits(params, files, url):
-    logger.warning(params, files, url)
+    logging.warning("Apply commits")
+    logging.warning(params)
     assert len(files) == 1
     file = files[0]
-    dir = TemporaryDirectory[str]
-    with zipfile.ZipFile(file, 'r') as zip_ref:
+    dir = TemporaryDirectory()
+    r = requests.get(file['url'])
+    with zipfile.ZipFile(io.BytesIO(r.content)) as zip_ref:
         zip_ref.extractall(dir.name)
     req = ET.fromstring(params['query_data'])
     repl = req.find('replication')
     scheme = repl.get('scheme')
     commit = repl.get('id')
-    logger.warning(scheme, commit)
-    os.path.join(dir.name, 'export.o5c')
+    bnd = params['from'].replace('tcp://', '')
+    branch, new_scheme = get_branch(bnd, scheme)
+    if new_scheme:
+        scheme = new_scheme
     con = OOConnectionParams(scheme, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
                              Config.oodb_username, Config.oodb_passwd, scheme)
     ws = OODBWorkspace.ws(scheme)
+    logging.warning("Connected to schema")
+    if Config.ws_rabbit_params and not ws.hasProducer():
+        ws.setProducer(Config.ws_rabbit_params['host'], Config.ws_rabbit_params['port'],
+                       Config.ws_rabbit_params['exchange'], Config.ws_rabbit_params['user'],
+                       Config.ws_rabbit_params['password'])
+    ws.setCurrentUser(bnd)
+    logging.warning("Set branch if needed")
+    if branch:
+        logging.warning(branch)
+        ws.switchBranch(branch)
     if not ws.isInit():
         res = ws.init(con)
         logger.warning(res)
     oe = IpdExporter(ws)
-    oe.improtFromOsm(os.path.join(dir.name, 'export.o5c'))
-    oe.improtFromMbtiles(os.path.join(dir.name, 'export.mbtiles'))
-    with open('export_files.json', 'r') as f:
-        files_data = json.load(f)
-    for file_data in files_data:
-        upload_file(file_data['filename'], file_data['key'], file_data['bucket'])
+    logging.warning("Importing...")
+    if not oe.importChanges(os.path.join(dir.name, 'export.o5c'), os.path.join(dir.name, 'export.mbtiles')):
+        logging.warning(f'Error importing commit {commit}: {oe.lastError().text()}')
+    else:
+        logging.warning(f'Importing commit {commit} finished successfully')
+        with open(os.path.join(dir.name, 'export_files.json'), 'r') as f:
+            files_data = json.load(f)
+        for file_data in files_data:
+            upload_file(os.path.join(dir.name, file_data['key']), file_data['key'], file_data['bucket'])
+    logging.warning("Finished import")
+    ws.clearData(True)
+
+
+def get_commit(params, files, url):
+    date = datetime.datetime.now()
+    rxmls = RequestXmlService()
+    req = ET.fromstring(params['query_data'])
+    req_id = rxmls.get_request_uuid(req)
+    res_id = uuid4().hex
+    res_doc = rxmls.get_request_document(res_id, req_id)
+    schema = req.find('currentCommit').get('scheme')
+    con = OOConnectionParams(schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
+                             Config.oodb_username, Config.oodb_passwd, schema)
+    ws = OODBWorkspace.ws(schema)
+    if not ws.isInit():
+        res = ws.init(con)
+        logger.warning(res)
+    ET.SubElement(res_doc, 'commit', {'id': ws.currentCommit(), 'schema': schema})
+
+    response_params = {
+        'from': params['to'],
+        'to': params['from'],
+        'ts_added': date.timestamp(),
+        'user_id': '1',
+        'user_id_to': 0,
+        'query_type': NEW_COMMIT_RESPONSE,
+        'query_data': ET.tostring(res_doc, encoding='unicode', xml_declaration=True)
+    }
+    proxy = ServerProxy(url)
+    proxy.send(response_params, [], Config.ret_path)
+
+
+def query_commits(params, files, url):
+    req = ET.fromstring(params['query_data'])
+    commit_el = req.find('commit')
+    commit_id = commit_el.get('id')
+    schema = commit_el.get('schema')
+    bnd = params['from'].replace('tcp://', '')
+
+    con = OOConnectionParams(schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
+                             Config.oodb_username, Config.oodb_passwd, schema)
+    ws = OODBWorkspace.ws(schema)
+    if not ws.isInit():
+        res = ws.init(con)
+        logger.warning(res)
+    schema_commits = ws.commits(ws.branch())
+    logger.warning(schema_commits)
+    if commit_id not in schema_commits:
+        logger.warning(f'Error in commits in schema {schema}: no commit {commit_id}')
+        return
+    logger.warning(schema_commits[schema_commits.index(commit_id) + 1:])
+
+    conn = db.connect_db()
+    with Session(conn) as session:
+        for commit in schema_commits[schema_commits.index(commit_id) + 1:]:
+            for user in session.query(db.User).filter(db.User.bndname == bnd, db.User.active == True).all():
+                if user.bndname == Config.self_bnd:
+                    continue
+                profiles = {x.scheme: x.to_dict() for x in user.profiles}
+                if len(profiles) == 0 or schema in profiles:
+                    item = db.Queue(user_id=user.id, commit_id=commit, schema=schema)
+                    logging.warning(item)
+                    session.add(item)
+        session.commit()
+
+
+def get_data(params, files, url):
+    date = datetime.datetime.now()
+    rxmls = RequestXmlService()
+    req = ET.fromstring(params['query_data'])
+    req_id = rxmls.get_request_uuid(req)
+    res_id = uuid4().hex
+    res = rxmls.get_request_document(res_id, req_id)
+    rxmls.set_result(res, 0, '')
+    request_string = req.find('data').text
+
+    OODBWorkspaceFactory.init('config/workspaces.json')
+    conf = ResponseWorkerConfig('config/response2.json', 'config/workspaces.json')
+    worker = ResponseWorker(conf)
+    fn = uuid4().hex
+    dir = os.path.join(os.getcwd(), 'tmp', fn)
+    os.makedirs(dir)
+    worker.makeResponse(request_string, dir)
+    make_archive(dir, 'zip', dir, '.')
+
+    rxmls.set_result(res, 0, '')
+    response_params = {
+        'from': params['to'],
+        'to': params['from'],
+        'ts_added': date.timestamp(),
+        'user_id': '1',
+        'user_id_to': params['user_id'],
+        'query_type': NEW_DATA_RESPONSE,
+        'query_data': ET.tostring(res, encoding='unicode', xml_declaration=True)
+    }
+    filename = fn + '.zip'
+    filepath = os.path.join(os.getcwd(), 'tmp', filename)
+    response_files = [{'name': filename, 'url': filepath, 'size': os.path.getsize(filepath)}]
+    proxy = ServerProxy(url)
+    proxy.send(response_params, response_files, Config.ret_path)
+
+
+def receive_data(params, files, url):
+    req = ET.fromstring(params['query_data'])
+    commit_el = req.find('commit')
+    commit_id = commit_el.get('id')
+    schema = commit_el.get('schema')
+    bnd = params['from'].replace('tcp://', '')
+
+    con = OOConnectionParams(schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
+                             Config.oodb_username, Config.oodb_passwd, schema)
+    ws = OODBWorkspace.ws(schema)
+    if not ws.isInit():
+        res = ws.init(con)
+        logger.warning(res)
+    schema_commits = ws.commits(ws.branch())
+    logger.warning(schema_commits)
+    if commit_id not in schema_commits:
+        logger.warning(f'Error in commits in schema {schema}: no commit {commit_id}')
+        return
+    logger.warning(schema_commits[schema_commits.index(commit_id) + 1:])
+
+    conn = db.connect_db()
+    with Session(conn) as session:
+        for commit in schema_commits[schema_commits.index(commit_id) + 1:]:
+            for user in session.query(db.User).filter(db.User.bndname == bnd, db.User.active == True).all():
+                if user.bndname == Config.self_bnd:
+                    continue
+                profiles = {x.scheme: x.to_dict() for x in user.profiles}
+                if len(profiles) == 0 or schema in profiles:
+                    item = db.Queue(user_id=user.id, commit_id=commit, schema=schema)
+                    logging.warning(item)
+                    session.add(item)
+        session.commit()
 
 
 def run_task(query_type, params, files, url):
@@ -387,6 +774,14 @@ def run_task(query_type, params, files, url):
         tasks.put(lambda: put_object(params, files, url))
     if query_type == NEW_REPLICATION_REQUEST:
         tasks.put(lambda: apply_commits(params, files, url))
+    if query_type == NEW_COMMIT_REQUEST:
+        tasks.put(lambda: get_commit(params, files, url))
+    if query_type == NEW_COMMIT_RESPONSE:
+        tasks.put(lambda: query_commits(params, files, url))
+    if query_type == NEW_DATA_REQUEST:
+        tasks.put(lambda: get_data(params, files, url))
+    if query_type == NEW_DATA_RESPONSE:
+        tasks.put(lambda: receive_data(params, files, url))
 
 
 def accept(params, files, url):
@@ -417,7 +812,8 @@ def bnd_disconnected(bnd_name: str):
 
 
 def xmlrpc_task():
-    server = SimpleXMLRPCServer(('0.0.0.0', 9000), logRequests=False, allow_none=True)
+    global server
+    server = SimpleXMLRPCServer(('0.0.0.0', 9000), logRequests=True, allow_none=True)
     server.register_function(list_contents)
     server.register_function(aud_add)
     server.register_function(auth_response)
@@ -431,6 +827,12 @@ def xmlrpc_task():
 
 
 app = FastAPI()
+app.add_middleware(CORSMiddleware,
+                   allow_origins=['http://10.10.8.24:3000'],
+                   allow_credentials=True,
+                   allow_methods=['*'],
+                   allow_headers=['*']
+                   )
 
 
 @app.get("/")
@@ -438,11 +840,31 @@ async def fa_root():
     return {"message": "Hello World"}
 
 
-@app.post("/login")
-async def fa_login(response: Response):
+@app.get("/login")
+async def fa_login_get(response: Response):
     response.set_cookie(key='sessionid', value='87654321')
     response.set_cookie(key='csrftoken', value='12345678')
-    return {"login": "Ok"}
+    return {"login": "Ok", "success": True}
+
+
+@app.post("/login")
+async def fa_login(response: Response):
+    logger.warning('login action')
+    response.set_cookie(key='sessionid', value='87654321')
+    response.set_cookie(key='csrftoken', value='12345678')
+    return {"login": "Ok", "success": True}
+
+
+@app.post('/webapi')
+async def fa_webapi(request: Request):
+    body = await request.body()
+    logger.warning('webapi')
+    if 'has_perm' in body.decode('utf-8'):
+        return Response(content='<?xml version="1.0"?><methodResponse><params><param><value><struct><member><name>success</name><value><boolean>1</boolean></value></member></struct></value></param></params></methodResponse>',
+                        media_type='application/xml')
+    else:
+        return Response(content='<?xml version="1.0"?><methodResponse><params><param><value><boolean>1</boolean></value></param></params></methodResponse>',
+                        media_type='application/xml')
 
 
 @app.post("/api/easo/PutObject")
@@ -451,7 +873,7 @@ async def fa_put_object(response: Response,
                         object_file: Annotated[UploadFile, File()]
 ):
     date = datetime.datetime.now()
-    files = [{'name': object_file.filename, 'url': object_file.file.name, 'size': os.path.getsize(object_file.size)}]
+    files = [{'name': object_file.filename, 'url': object_file.file.name, 'size': object_file.size}]
     request_params = {
         'from': '',
         'to': '',
@@ -464,8 +886,67 @@ async def fa_put_object(response: Response,
     put_object(request_params, files, None)
     return {"Upload": "Ok"}
 
+@app.get("/cr")
+async def correction_replication(bnd_name: str, schema: str):
+
+    date = datetime.datetime.now()
+    rxmls = RequestXmlService()
+    res_id = uuid4().hex
+
+    res = rxmls.get_request_document(res_id, None)
+    ET.SubElement(res, 'currentCommit', {'scheme': schema})
+    params = {
+        'from': f'tcp://{Config.self_bnd}',
+        'to': f'tcp://{bnd_name}',
+        'ts_added': date.timestamp(),
+        'user_id': '0',
+        'query_type': NEW_COMMIT_REQUEST,
+        'query_data': ET.tostring(res, encoding='unicode', xml_declaration=True),
+    }
+    proxy = ServerProxy(Config.enserver)
+    try:
+        proxy.send(params, [], Config.ret_path)
+    except:
+        logger.error('Error sending')
+
+
+@app.get("/get_cr")
+async def correction_replication(bnd_name: str, schema: str):
+
+    date = datetime.datetime.now()
+    rxmls = RequestXmlService()
+    res_id = uuid4().hex
+
+    res = rxmls.get_request_document(res_id, None)
+
+    con = OOConnectionParams(schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
+                             Config.oodb_username, Config.oodb_passwd, schema)
+    ws = OODBWorkspace.ws(schema)
+    if not ws.isInit():
+        res = ws.init(con)
+        logger.warning(res)
+    ET.SubElement(res, 'commit', {'id': ws.currentCommit(), 'schema': schema})
+
+    params = {
+        'from': f'tcp://{Config.self_bnd}',
+        'to': f'tcp://{bnd_name}',
+        'ts_added': date.timestamp(),
+        'user_id': '0',
+        'query_type': NEW_COMMIT_RESPONSE,
+        'query_data': ET.tostring(res, encoding='unicode', xml_declaration=True),
+    }
+    proxy = ServerProxy(Config.enserver)
+    try:
+        proxy.send(params, [], Config.ret_path)
+    except:
+        logger.error('Error sending')
+
 
 def main():
+    global connection
+    global server
+    global channel
+
     logger.setLevel(logging.INFO)
     logger.warning('Use Control-C to exit')
 
@@ -483,9 +964,11 @@ def main():
 
     try:
         logger.warning('Start server')
-        uvicorn.run(app, host="0.0.0.0", port=9001)
+        uvicorn.run(app, host="0.0.0.0", port=8000)
     except KeyboardInterrupt:
         logger.warning('Exiting')
+    finally:
+        server.server_close()
 
 
 def vers_key(e):
@@ -493,6 +976,12 @@ def vers_key(e):
 
 
 def test():
+    # with open('/home/ashatora/catalog.json') as f:
+    #     j = json.load(f)
+    # j1 = json.loads(j[1]['data'])
+    # with open('/home/ashatora/cat.json', 'w') as f:
+    #     f.write(json.dumps(j1, indent=2, ensure_ascii=False))
+    # return
     #params = {"from": "tcp://kptsp_vb", "query_data": "<?xml version=\"1.0\" encoding=\"utf-8\"?><request><header parcel_id=\"990715ba919544a98f22cc7d3b0d9e8d\"/><getMetadataByIds><chart id=\"fc44343bd1654ee7b03ac1731567bbfd\"/></getMetadataByIds></request>", "query_type": 24, "to": "tcp://bnd127", "user_id": "3302", "ts_added": 1679825320.653038}
     #files = []
     #url = 'http://127.0.0.1:7000/xmlrpc'
@@ -500,29 +989,47 @@ def test():
     #get_metadata(params, files, url)
     #get_catalog()
     # auth_response('123', 'bnd127', False)
-    con = OOConnectionParams(Config.oodb_schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
-                             Config.oodb_username, Config.oodb_passwd, Config.oodb_schema)
-    ws = OODBWorkspace.ws(Config.oodb_schema)
-    ws.init(con)
-    created, updated, deleted = ws.changes('2dad8c8a-d2db-4074-ab7a-c01c36ada2be')
-    qu = GroupQuery(Envelope())
-    qu.setUids(updated)
-    qu.setLoadArch(True)
-    uids = []
-    ws.clearData(True)
-    ws.load(qu, uids)
-    updated_files = []
-    for feature_uid in uids:
-        not_files = True
-        vers = ws.featureVersion(feature_uid)
-        if len(vers) > 1:
-            vers.sort(key=vers_key)
-            not_files = vers[0].feature().isEqual(vers[1].feature(), ["c1000"])
-        if not not_files:
-            updated_files.append(feature_uid)
-    print(updated_files)
-    ws.clearData(True)
-    ws.close()
+    # con = OOConnectionParams(Config.oodb_schema, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
+    #                          Config.oodb_username, Config.oodb_passwd, Config.oodb_schema)
+    # ws = OODBWorkspace.ws(Config.oodb_schema)
+    # ws.init(con)
+    # print(ws.currentCommit())
+    # print(ws.commits(ws.branch()))
+    # created, updated, deleted = ws.changes('2dad8c8a-d2db-4074-ab7a-c01c36ada2be')
+    # qu = GroupQuery(Envelope())
+    # qu.setUids(updated)
+    # qu.setLoadArch(True)
+    # uids = []
+    # ws.clearData(True)
+    # ws.load(qu, uids)
+    # updated_files = []
+    # for feature_uid in uids:
+    #     not_files = True
+    #     vers = ws.featureVersion(feature_uid)
+    #     if len(vers) > 1:
+    #         vers.sort(key=vers_key)
+    #         not_files = vers[0].feature().isEqual(vers[1].feature(), ["c1000"])
+    #     if not not_files:
+    #         updated_files.append(feature_uid)
+    # print(updated_files)
+    # ws.clearData(True)
+    # ws.close()
+    #replication_old('bnd128', '23c9a275-ec0f-481a-8437-f3e41e4fe4f5', 'documents_src')
+    request_string = """
+    {
+  "type": "group_query",
+  "data_source": "npd",
+  "uids": [ "2e20130c-541a-4b9f-9efb-f2a0e8b10c33" ]
+}
+    """
+    OODBWorkspaceFactory.init('config/workspaces.json')
+    conf = ResponseWorkerConfig('config/response2.json', 'config/workspaces.json')
+    worker = ResponseWorker(conf)
+    fn = uuid4().hex
+    dir = os.path.join('tmp', fn)
+    os.makedirs(dir)
+    worker.makeResponse(request_string, dir)
+    make_archive(dir, 'zip', dir, '.')
     pass
 
 

reqs/request_xml_service.py

@@ -32,7 +32,7 @@ class RequestXmlService:
         result = header.get(attribute_name)
         return result.strip() if result else ''
 
-    def get_request_document(self, request_uuid: str, response_uuid: str | None) -> ET.Element:
+    def get_request_document(self, request_uuid: str, response_uuid) -> ET.Element:
         document = ET.Element(self.REQUEST_NODE_NAME)
         header = ET.SubElement(document, self.HEADER_NODE_NAME)
         header.set(self.REQUEST_ID_ATTRIBUTE_NAME, request_uuid.strip())
@@ -67,7 +67,7 @@ class RequestXmlService:
     def get_common_request_xml(self, request_uuid: str) -> ET.Element:
         return self.get_request_document(request_uuid, None)
 
-    def get_response_xml(self, document: ET.Element, request_uuid: str | None, response_uuid: str, result_code: int, result_message: str) -> ET.Element:
+    def get_response_xml(self, document: ET.Element, request_uuid, response_uuid: str, result_code: int, result_message: str) -> ET.Element:
         header = document.find(self.HEADER_NODE_NAME)
         header.set(self.REQUEST_ID_ATTRIBUTE_NAME, response_uuid.strip())
         header.set(self.RESPONSE_ID_ATTRIBUTE_NAME, request_uuid.strip() if request_uuid else None)
@@ -90,7 +90,7 @@ class RequestXmlService:
             raise Exception(Messages.INVALID_REQUEST_DATA)
         self.validate_common_request_xml(document)
 
-    def validate_common_request_xml(self, document: ET.Element | None):
+    def validate_common_request_xml(self, document):
         if not document:
             raise Exception(Messages.NO_REQUEST_DATA_PROVIDED.name)
         if document.tag != self.REQUEST_NODE_NAME:
@@ -102,7 +102,7 @@ class RequestXmlService:
         if not request_uuid:
             raise Exception(Messages.MISSING_ATTRIBUTE, self.REQUEST_ID_ATTRIBUTE_NAME, self.HEADER_NODE_NAME)
 
-    def get_result_code(self, document: ET.Element) -> str | None:
+    def get_result_code(self, document: ET.Element):
         result = self.get_element(document, document, self.RESULT_NODE_NAME)
         if not result:
             return None

reqs_graphql.py

@@ -1,8 +1,9 @@
 from gql import gql, Client
 from gql.transport.aiohttp import AIOHTTPTransport
+from config import Config
 
-transport = AIOHTTPTransport(url="https://gql.ivazh.ru/graphql/")
-service = 'pdim'
+transport = AIOHTTPTransport(url=Config.gql_url)
+service = Config.gql_schema
 
 
 def get_classifier():

requirements.txt

@@ -1,5 +1,8 @@
+--find-links=deps
 aiohttp==3.8.4
 aiosignal==1.3.1
+annotated-types==0.5.0
+anyio==3.7.1
 async-timeout==4.0.3
 attrs==23.1.0
 backoff==2.2.1
@@ -7,22 +10,31 @@ boto3==1.28.37
 botocore==1.31.40
 certifi==2023.7.22
 charset-normalizer==3.2.0
+click==8.1.7
+fastapi==0.103.1
 frozenlist==1.4.0
 gql==3.5.0b5
 graphql-core==3.3.0a3
+greenlet==2.0.2
+h11==0.14.0
 idna==3.4
 jmespath==1.0.1
 multidict==6.0.4
 pika==1.3.2
 pika-stubs==0.1.3
+psycopg==3.1.10
+pydantic==2.3.0
+pydantic_core==2.6.3
+pygost==5.12
 python-dateutil==2.8.2
+python-multipart==0.0.6
 requests==2.31.0
 s3transfer==0.6.2
 six==1.16.0
+sniffio==1.3.0
+SQLAlchemy==2.0.20
+starlette==0.27.0
 typing_extensions==4.7.1
-urllib3==2.0.4
+urllib3==1.26
+uvicorn==0.23.2
 yarl==1.9.2
-SQLAlchemy~=2.0.20
-fastapi=~0.103.1
-uvicorn=~0.23.2
-python-multipart=~0.0.6

test.py

@@ -0,0 +1,32 @@
+import db
+from sqlalchemy.orm import Session
+
+
+def get_branch(bndname: str, scheme: str):
+    conn = db.connect_db()
+    with Session(conn) as session:
+        item = session.query(db.IncomeBranch).filter_by(scheme=scheme).join(db.User).filter_by(bndname=bndname).one_or_none()
+        if item:
+            return item.branch
+        return None
+
+
+def is_replication_scheme(bndname: str, scheme: str):
+    conn = db.connect_db()
+    with Session(conn) as session:
+        item = session.query(db.User).filter_by(bndname=bndname).one_or_none()
+        if not item:
+            return False
+        profiles = {x.scheme: x.to_dict() for x in item.profiles}
+        if len(profiles) == 0 or scheme in profiles:
+            return True
+    return False
+
+
+def main():
+    # print(get_branch('bnd128', 'ood'))
+    print(is_replication_scheme('bnd128', 'documents_src'))
+
+
+if __name__ == '__main__':
+    main()

zip.py

@@ -4,7 +4,6 @@ from zipfile import ZipFile
 
 
 class Zip:
-    dir: TemporaryDirectory[str]
 
     def __init__(self):
         self.dir = TemporaryDirectory()
@@ -19,5 +18,6 @@ class Zip:
             for folder_name, sub_folders, file_names in os.walk(self.dir.name):
                 for filename in file_names:
                     file_path = os.path.join(folder_name, filename)
-                    zip_object.write(file_path, os.path.basename(file_path))
+                    rel_file_path = os.path.relpath(file_path, self.dir.name)
+                    zip_object.write(file_path, rel_file_path)
         return tmp_zip