ready to deploy

.gitea/workflows/deploy.yml

@@ -0,0 +1,40 @@
+name: Build and Deploy
+
+on:
+  push:
+    branches:
+      - prod
+
+jobs:
+  build_and_deploy:
+    runs-on: docker
+    
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v4
+
+      - name: Define Image Name
+        id: image_vars
+        run: |
+          IMAGE_TAG="${{ GITEA_BRANCH }}-${{ GITEA_SHA::7 }}"
+          FULL_IMAGE_NAME="${{ GITEA_REPO_NAME }}:${IMAGE_TAG}"
+          echo "::set-output name=FULL_IMAGE_NAME::${FULL_IMAGE_NAME}"
+      
+      - name: Build Docker Image
+        run: |
+          docker buildx build --load --platform linux/amd64 \
+            -t "${{ steps.image_vars.outputs.FULL_IMAGE_NAME }}" .
+
+      # --- ШАГ ВЕРИФИКАЦИИ УДАЛЕН ---
+            
+      - name: Set up SSH
+        run: |
+          mkdir -p ~/.ssh
+          echo "${{ secrets.PROD_SSH_KEY }}" > ~/.ssh/id_ed25519
+          chmod 600 ~/.ssh/id_ed25519
+          
+      - name: Save, Transfer and Deploy Image
+        run: |
+          IMAGE_TO_DEPLOY="${{ steps.image_vars.outputs.FULL_IMAGE_NAME }}"
+          echo "Transferring image ${IMAGE_TO_DEPLOY} to production..."
+          docker save "${IMAGE_TO_DEPLOY}" | ssh -o StrictHostKeyChecking=no deployer@YOUR_PROD_SERVER_IP "/home/deployer/deploy.sh '${IMAGE_TO_DEPLOY}'"

ftp_parser.py

@@ -1,9 +1,10 @@
-# ftp_parser.py
 
 import os
 import json
 import logging
 from typing import List, Dict
+from datetime import datetime, timedelta
+from dateutil import parser
 
 log = logging.getLogger(__name__)
 
@@ -11,6 +12,7 @@ def process_json_files(directory: str) -> List[Dict]:
     """
     Сканирует директорию, читает все .json файлы и извлекает из них данные
     о фискальных регистраторах в стандартизированном формате.
+    Файлы, старше 21 дня (по полю 'current_time'), игнорируются.
 
     :param directory: Путь к директории с JSON файлами.
     :return: Список словарей, где каждый словарь представляет один ФР.
@@ -22,6 +24,9 @@ def process_json_files(directory: str) -> List[Dict]:
     all_fr_data = []
     log.info(f"Начинаю обработку JSON файлов из директории: {directory}")
 
+    # Пороговая дата: всё что старше, считаем неактуальным
+    freshness_threshold = datetime.now() - timedelta(days=21)
+
     for filename in os.listdir(directory):
         if filename.lower().endswith('.json'):
             file_path = os.path.join(directory, filename)
@@ -29,6 +34,26 @@ def process_json_files(directory: str) -> List[Dict]:
                 with open(file_path, 'r', encoding='utf-8') as f:
                     data = json.load(f)
 
+                current_time_str = data.get('current_time')
+                if not current_time_str:
+                    log.warning(f"Пропущен файл {filename}: отсутствует поле 'current_time' для проверки актуальности.")
+                    continue
+
+                try:
+                    file_datetime = parser.parse(current_time_str)
+                    # Приводим к aware-объекту с локальной таймзоной, если он naive, для корректного сравнения
+                    if file_datetime.tzinfo is None:
+                        file_datetime = file_datetime.astimezone()
+
+                    # Сравниваем с пороговым значением (пороговое значение тоже делаем aware)
+                    if file_datetime < freshness_threshold.astimezone(file_datetime.tzinfo):
+                        log.warning(f"Пропущен файл {filename}: данные неактуальны (старше 21 дня). "
+                                    f"Дата файла: {file_datetime.strftime('%Y-%m-%d')}")
+                        continue
+                except parser.ParserError:
+                    log.error(f"Не удалось распознать дату в поле 'current_time' в файле {filename}: '{current_time_str}'")
+                    continue
+
                 if 'serialNumber' not in data or not data['serialNumber']:
                     log.warning(f"Пропущен файл {filename}: отсутствует serialNumber.")
                     continue

sync_logic.py

@@ -3,7 +3,7 @@
 import re
 import logging
 from typing import Dict, List, Optional, Any
-from datetime import datetime
+from datetime import datetime, timedelta
 from dateutil import parser
 
 # Импортируем наши модули
@@ -172,8 +172,8 @@ class Synchronizer:
 
     def _update_existing_frs(self):
         """
-        Находит и обновляет ФР с отличающимися датами окончания ФН,
+        Находит и обновляет ФР с отличающимися датами окончания ФН.
-        воспроизводя проверенную логику в более производительном виде.
+        Пропускает обновление, если разница в датах более 10 лет.
         """
         log.info("Поиск ФР для обновления...")
         
@@ -199,6 +199,9 @@ class Synchronizer:
         records_to_check = self.db._execute_query(query, fetch='all')
         update_counter = 0
 
+        #Порог для сравнения дат
+        archive_delta = timedelta(days=365*10)
+
         for rec in records_to_check:
             try:
                 # 2. Приводим даты к одному "знаменателю" - объектам datetime
@@ -214,12 +217,18 @@ class Synchronizer:
                 # 3. Сравниваем даты. Если они различаются, готовим обновление.
                 #    Проверяем на неравенство. Величина различия не важна.
                 if pos_date != sd_date:
+                    # Проверяем, что разница не больше 10 лет
+                    if abs(pos_date - sd_date) > archive_delta:
+                        log.warning(f"Пропуск сравнения для S/N {rec['serialNumber']}: разница в датах больше 10 лет."
+                                    f"FTP: {pos_date.date()}, SD: {sd_date.date()}.")
+                        continue
+
                     log.info(f"Найдено расхождение в дате для S/N {rec['serialNumber']} (UUID: {rec['sd_uuid']}). "
                              f"FTP: {pos_date}, SD: {sd_date}. Подготовка к обновлению.")
 
                     # 4. Формируем данные для обновления
                     
-                    # 5. Проверяем, закончился ли ФН
+                    # Проверяем, закончился ли ФН
                     if rec['pos_rnm'] == '0000000000000000':
                         legal_name = 'ЗАКОНЧИЛСЯ ФН'
                     else: