/
alsoalgo
/
CTF
Обзор
Документация
Войти
/
alsoalgo
/
CTF
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/models/application.py
292 строки
12 KB
alsoalgo
some improvements
14 дек 2025, 22:23
14 дек 2025, 22:23
37d2ec0
Код
Авторство
О чём код?
""" Модель заявки (Application) для CTF системы """ from datetime import datetime from typing import Optional, List, Dict, Any from enum import Enum import uuid class ApplicationStatus(Enum): """Статусы заявки""" NEW = "new" # Новая заявка IN_PROGRESS = "in_progress" # В обработке PENDING_DOCUMENTS = "pending_documents" # Ожидает документы UNDER_REVIEW = "under_review" # На проверке APPROVED = "approved" # Одобрено REJECTED = "rejected" # Отклонено CLOSED = "closed" # Закрыто class Application: """ Модель заявки на сделку CTF. Содержит всю информацию о заявке, связанных документах и статусе. """ def __init__( self, application_id: Optional[str] = None, status: ApplicationStatus = ApplicationStatus.NEW, created_at: Optional[datetime] = None, updated_at: Optional[datetime] = None, supplier: Optional[str] = None, off_taker: Optional[str] = None, product: Optional[str] = None, quantity: Optional[float] = None, amount: Optional[float] = None, currency: Optional[str] = None, contract_date: Optional[datetime] = None, delivery_date: Optional[datetime] = None, payment_terms: Optional[str] = None, documents: Optional[List[Dict[str, Any]]] = None, metadata: Optional[Dict[str, Any]] = None ): """ Инициализация заявки. :param application_id: Уникальный ID заявки :param status: Статус заявки :param created_at: Дата создания :param updated_at: Дата последнего обновления :param supplier: Поставщик :param off_taker: Покупатель :param product: Товар/продукт :param quantity: Количество :param amount: Сумма :param currency: Валюта :param contract_date: Дата контракта :param delivery_date: Дата поставки :param payment_terms: Условия оплаты :param documents: Список связанных документов :param metadata: Дополнительные метаданные """ self.application_id = application_id or str(uuid.uuid4()) self.status = status self.created_at = created_at or datetime.now() self.updated_at = updated_at or datetime.now() # Основные поля заявки self.supplier = supplier self.off_taker = off_taker self.product = product self.quantity = quantity self.amount = amount self.currency = currency or "USD" self.contract_date = contract_date self.delivery_date = delivery_date self.payment_terms = payment_terms # Документы self.documents = documents or [] # Метаданные (для хранения дополнительной информации) self.metadata = metadata or {} # История изменений self.history: List[Dict[str, Any]] = [] self._add_history_entry("created", "Application created") def update_status(self, new_status: ApplicationStatus, reason: Optional[str] = None): """ Обновляет статус заявки и добавляет запись в историю. :param new_status: Новый статус :param reason: Причина изменения статуса """ old_status = self.status # Убеждаемся, что old_status - это ApplicationStatus enum if isinstance(old_status, str): try: old_status = ApplicationStatus(old_status) except ValueError: old_status = ApplicationStatus.NEW # Убеждаемся, что new_status - это ApplicationStatus enum if isinstance(new_status, str): try: new_status = ApplicationStatus(new_status) except ValueError: new_status = ApplicationStatus.NEW self.status = new_status self.updated_at = datetime.now() reason_text = reason or f"Status changed from {old_status.value} to {new_status.value}" self._add_history_entry("status_changed", reason_text, { "old_status": old_status.value, "new_status": new_status.value }) def add_document(self, document_name: str, document_path: str, document_type: str, metadata: Optional[Dict] = None): """ Добавляет документ к заявке. :param document_name: Имя документа :param document_path: Путь к файлу :param document_type: Тип документа (invoice, contract, insurance, etc.) :param metadata: Дополнительные метаданные документа """ doc = { "name": document_name, "path": document_path, "type": document_type, "added_at": datetime.now().isoformat(), "metadata": metadata or {} } self.documents.append(doc) self.updated_at = datetime.now() self._add_history_entry("document_added", f"Document added: {document_name} ({document_type})") def update_field(self, field_name: str, value: Any, source: Optional[str] = None): """ Обновляет поле заявки. :param field_name: Имя поля :param value: Новое значение :param source: Источник данных (имя документа или "manual") """ if hasattr(self, field_name): old_value = getattr(self, field_name) setattr(self, field_name, value) self.updated_at = datetime.now() source_text = f" from {source}" if source else "" self._add_history_entry("field_updated", f"Field {field_name} updated{source_text}", { "field": field_name, "old_value": str(old_value), "new_value": str(value), "source": source }) def _add_history_entry(self, action: str, description: str, details: Optional[Dict] = None): """ Добавляет запись в историю изменений. :param action: Тип действия :param description: Описание действия :param details: Дополнительные детали """ entry = { "timestamp": datetime.now().isoformat(), "action": action, "description": description, "details": details or {} } self.history.append(entry) def to_dict(self) -> Dict[str, Any]: """ Преобразует заявку в словарь для сериализации. :return: Словарь с данными заявки """ # Сериализуем metadata, преобразуя DataFrame в словари serializable_metadata = {} if self.metadata: for key, value in self.metadata.items(): if hasattr(value, 'to_dict'): # pandas DataFrame try: # Преобразуем DataFrame в словарь if hasattr(value, 'index') and hasattr(value.index, 'names'): # MultiIndex - преобразуем в список словарей serializable_metadata[key] = { '_type': 'dataframe', 'data': value.to_dict(orient='records'), 'index': [str(idx) for idx in value.index] } else: serializable_metadata[key] = { '_type': 'dataframe', 'data': value.to_dict(orient='records') } except Exception: # Если не удалось сериализовать, пропускаем serializable_metadata[key] = None elif isinstance(value, dict): # Рекурсивно обрабатываем вложенные словари serializable_metadata[key] = self._serialize_dict(value) else: serializable_metadata[key] = value return { "application_id": self.application_id, "status": self.status.value, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), "supplier": self.supplier, "off_taker": self.off_taker, "product": self.product, "quantity": self.quantity, "amount": self.amount, "currency": self.currency, "contract_date": self.contract_date.isoformat() if self.contract_date else None, "delivery_date": self.delivery_date.isoformat() if self.delivery_date else None, "payment_terms": self.payment_terms, "documents": self.documents, "metadata": serializable_metadata, "history": self.history } def _serialize_dict(self, d: Dict) -> Dict: """Рекурсивно сериализует словарь, преобразуя DataFrame""" result = {} for key, value in d.items(): if hasattr(value, 'to_dict'): # pandas DataFrame try: if hasattr(value, 'index') and hasattr(value.index, 'names'): result[key] = { '_type': 'dataframe', 'data': value.to_dict(orient='records'), 'index': [str(idx) for idx in value.index] } else: result[key] = { '_type': 'dataframe', 'data': value.to_dict(orient='records') } except Exception: result[key] = None elif isinstance(value, dict): result[key] = self._serialize_dict(value) else: result[key] = value return result @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Application': """ Создает заявку из словаря. :param data: Словарь с данными заявки :return: Экземпляр Application """ # Парсим даты created_at = datetime.fromisoformat(data.get("created_at")) if data.get("created_at") else None updated_at = datetime.fromisoformat(data.get("updated_at")) if data.get("updated_at") else None contract_date = datetime.fromisoformat(data["contract_date"]) if data.get("contract_date") else None delivery_date = datetime.fromisoformat(data["delivery_date"]) if data.get("delivery_date") else None # Парсим статус status = ApplicationStatus(data.get("status", "new")) return cls( application_id=data.get("application_id"), status=status, created_at=created_at, updated_at=updated_at, supplier=data.get("supplier"), off_taker=data.get("off_taker"), product=data.get("product"), quantity=data.get("quantity"), amount=data.get("amount"), currency=data.get("currency"), contract_date=contract_date, delivery_date=delivery_date, payment_terms=data.get("payment_terms"), documents=data.get("documents", []), metadata=data.get("metadata", {}) ) def __repr__(self): return f"Application(id={self.application_id}, status={self.status.value}, supplier={self.supplier}, off_taker={self.off_taker})"