ksef_app/app/models/invoice.py

import enum

from app.extensions import db
from app.models.base import TimestampMixin


invoice_tags = db.Table(
    'invoice_tags',
    db.Column('invoice_id', db.Integer, db.ForeignKey('invoice.id'), primary_key=True),
    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'), primary_key=True),
)


class InvoiceType(enum.Enum):
    PURCHASE = 'purchase'
    SALE = 'sale'
    CORRECTION = 'correction'


class InvoiceStatus(enum.Enum):
    NEW = 'new'
    READ = 'read'
    ACCOUNTED = 'accounted'
    SENT = 'sent'
    ARCHIVED = 'archived'
    NEEDS_ATTENTION = 'needs_attention'
    ERROR = 'error'



INVOICE_TYPE_LABELS = {
    InvoiceType.PURCHASE: 'Zakupowa',
    InvoiceType.SALE: 'Sprzedażowa',
    InvoiceType.CORRECTION: 'Korekta',
}

INVOICE_STATUS_LABELS = {
    InvoiceStatus.NEW: 'Nowa',
    InvoiceStatus.READ: 'Odczytana',
    InvoiceStatus.ACCOUNTED: 'Zaksięgowana',
    InvoiceStatus.SENT: 'Wysłana',
    InvoiceStatus.ARCHIVED: 'Archiwalna',
    InvoiceStatus.NEEDS_ATTENTION: 'Do księgowania',
    InvoiceStatus.ERROR: 'Błąd',
}

ISSUED_STATUS_LABELS = {
    'draft': 'Robocza',
    'pending': 'Oczekuje wysyłki',
    'issued': 'Wysłana do KSeF',
    'received': 'Odebrana',
    'read': 'Odczytana',
    'accounted': 'Zaksięgowana',
    'queued': 'W kolejce',
    'error': 'Błąd',
    'sent': 'Wysłana',
    'needs_attention': 'Do księgowania',
    'issued_mock': 'Wysłana testowo',
}


class Invoice(TimestampMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    company_id = db.Column(db.Integer, db.ForeignKey('company.id'), index=True)
    ksef_number = db.Column(db.String(128), nullable=False, index=True)
    invoice_number = db.Column(db.String(128), nullable=False, index=True)

    contractor_name = db.Column(db.String(255), nullable=False, index=True)
    contractor_nip = db.Column(db.String(32), index=True)
    contractor_regon = db.Column(db.String(32), index=True)
    contractor_address = db.Column(db.String(512))

    issue_date = db.Column(db.Date, nullable=False, index=True)
    received_date = db.Column(db.Date, index=True)
    fetched_at = db.Column(db.DateTime, index=True)

    net_amount = db.Column(db.Numeric(12, 2), nullable=False, default=0)
    vat_amount = db.Column(db.Numeric(12, 2), nullable=False, default=0)
    gross_amount = db.Column(db.Numeric(12, 2), nullable=False, default=0)
    split_payment = db.Column(db.Boolean, default=False, nullable=False)
    currency = db.Column(db.String(8), default='PLN')
    seller_bank_account = db.Column(db.String(64), default='')

    invoice_type = db.Column(db.Enum(InvoiceType), nullable=False, default=InvoiceType.PURCHASE)
    status = db.Column(db.Enum(InvoiceStatus), nullable=False, default=InvoiceStatus.NEW)

    xml_path = db.Column(db.String(512))
    pdf_path = db.Column(db.String(512))
    html_preview = db.Column(db.Text)
    internal_note = db.Column(db.Text)
    source_hash = db.Column(db.String(128))
    read_at = db.Column(db.DateTime)
    last_synced_at = db.Column(db.DateTime)

    external_metadata = db.Column(db.JSON, default=dict)

    is_unread = db.Column(db.Boolean, default=True, nullable=False)
    pinned = db.Column(db.Boolean, default=False, nullable=False)
    queue_accounting = db.Column(db.Boolean, default=False, nullable=False)
    source = db.Column(db.String(32), default='ksef', nullable=False)

    customer_id = db.Column(db.Integer, db.ForeignKey('customer.id'), index=True)

    issued_to_ksef_at = db.Column(db.DateTime)
    issued_status = db.Column(db.String(32), default='received', nullable=False)

    tags = db.relationship(
        'Tag',
        secondary=invoice_tags,
        lazy='joined',
        backref=db.backref('invoices', lazy='dynamic'),
    )
    sync_events = db.relationship(
        'SyncEvent',
        backref='invoice',
        lazy='dynamic',
        cascade='all, delete-orphan',
    )
    mail_deliveries = db.relationship(
        'MailDelivery',
        backref='invoice',
        lazy='dynamic',
        cascade='all, delete-orphan',
    )
    notifications = db.relationship(
        'NotificationLog',
        backref='invoice',
        lazy='dynamic',
        cascade='all, delete-orphan',
    )
    company = db.relationship('Company', backref=db.backref('invoices', lazy='dynamic'))
    customer = db.relationship('Customer', backref=db.backref('invoices', lazy='dynamic'))

    __table_args__ = (
        db.UniqueConstraint('company_id', 'ksef_number', name='uq_invoice_company_ksef'),
    )

    @property
    def month_key(self):
        return f'{self.issue_date.year}-{self.issue_date.month:02d}'

    @property
    def invoice_type_label(self):
        return INVOICE_TYPE_LABELS.get(self.invoice_type, getattr(self.invoice_type, 'value', self.invoice_type))

    @property
    def status_label(self):
        return INVOICE_STATUS_LABELS.get(self.status, getattr(self.status, 'value', self.status))

    @property
    def issued_status_label(self):
        return ISSUED_STATUS_LABELS.get(self.issued_status, self.issued_status)


class Tag(TimestampMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, nullable=False)
    color = db.Column(db.String(32), default='secondary')


class SyncEvent(TimestampMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    invoice_id = db.Column(db.Integer, db.ForeignKey('invoice.id'), nullable=False)
    status = db.Column(db.String(32), nullable=False)
    message = db.Column(db.Text)
    source = db.Column(db.String(32), default='ksef')


class MailDelivery(TimestampMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    invoice_id = db.Column(db.Integer, db.ForeignKey('invoice.id'), nullable=False)
    recipient = db.Column(db.String(255), nullable=False)
    status = db.Column(db.String(32), default='queued')
    subject = db.Column(db.String(255))
    error_message = db.Column(db.Text)
    sent_at = db.Column(db.DateTime)


class NotificationLog(TimestampMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    invoice_id = db.Column(db.Integer, db.ForeignKey('invoice.id'))
    channel = db.Column(db.String(32), nullable=False)
    status = db.Column(db.String(32), default='queued')
    message = db.Column(db.Text)
    sent_at = db.Column(db.DateTime)