anafapi/tests/test_anafoauth.py

'''Created 6 Jul 2022 Levi'''
# coding: UTF-8
import platform
import unittest
import os
from datetime import datetime

import ujson
from lxml import etree
from lxml.etree import LxmlSyntaxError

from anafapi.anafoauth import Anafoauth, Efactoauth
from anafapi import utils
from anafapi.en16931 import Invoice

if platform.system() == 'Linux':
    token_file_path = '/home/deeejas/git/anafapi/anafapi/{}'
elif platform.system() == 'Windows':
    token_file_path = os.path.join(os.path.expanduser('~'), 'git/anafapi/anafapi/{}')

class oauthTest(unittest.TestCase):
    anoauth = Anafoauth(
        # client_id='16cbef8754bef0bcf015529353427e8a7e3ee71d56879565',
        # client_secret='c6aa66d89002f36461f93facd990e6bf5bc706e1df897e8a7e3ee71d56879565',
        client_id='312afcb54fae740eaa721aa7364b7e8a7e3ee71d52220c66',
        client_secret='3eba7d34e1c516403033bcdf22619ab1942bec0bb0ef7e8a7e3ee71d52220c66',
        # redirect_uri='https://www.anaf.ro/oauth2callback',
        redirect_uri='https://oauth.pstmn.io/v1/browser-callback',
    )

    def test_is_valid(self):
        self.anoauth.load_token_file(token_file_path.format('efact.json'))
        # print(self.anoauth.to)
        print(self.anoauth.is_valid)

    def test_jwt_props(self):
        print(datetime.fromtimestamp(self.anoauth.jwt_props('../anafapi/efact_jwt.json')['exp']).strftime('%d-%m-%Y %H:%M:%S'))

    def test_browser_redirect(self):
        print(self.anoauth._browser_redirect())

    def test_get_token(self):
        self.anoauth.get_token()
        self.anoauth.save_token(token_file_path.format('efact.json'))

    def test_get_jwt_token(self):
        # self.anoauth.get_jwt_token()
        self.anoauth.get_jwt_token_requests()
        # self.anoauth.save_jwt_token(token_file_path.format('efact_jwt.json'))

    def test_revive_token(self):
        self.anoauth.load_token_file(token_file_path.format('efact.json'))
        self.anoauth.revive_token()
        self.anoauth.save_token(token_file_path.format('efact_refreshed.json'))

class efactTest(unittest.TestCase):
    efactoauth = Efactoauth(
        # client_id='16cbef8754bef0bcf015529353427e8a7e3ee71d56879565',
        # client_secret='c6aa66d89002f36461f93facd990e6bf5bc706e1df897e8a7e3ee71d56879565',
        client_id='312afcb54fae740eaa721aa7364b7e8a7e3ee71d52220c66',
        client_secret='3eba7d34e1c516403033bcdf22619ab1942bec0bb0ef7e8a7e3ee71d52220c66',
        # redirect_uri='https://www.anaf.ro/oauth2callback',
        redirect_uri='https://oauth.pstmn.io/v1/browser-callback'
    )
    efactoauth.load_token_file(token_file_path.format('efact_jwt.json'))
    efactoauth.load_session()

    def test_to_erp(self):
        # self.efactoauth.getxml_from_zip(path='../efact/3016038776.zip', dest_f='')
        # latest = self.efactoauth.get_latest_invoice('efact')
        inv_files = utils.get_not_booked('17259191')
        # inv_files = [r'C:\Users\Levi\git\anafapi\tests\17259191\FACTURA PRIMITA\2024_05\23\16702141\4290867417.xml',
        #
        #             ]
        for latest in inv_files:
            if 'booked' not in latest:
                inv = self.efactoauth.parse_inv(invoice=latest)
                print(inv.seller_party.name, inv.invoice_id, inv.issue_date.date())
                # for line in inv.lines:
                #     print(line.tax)
                self.efactoauth.addinvoice_tonextup(invoice=inv,
                                                    kind='stock',
                                                    template=32,
                                                    exp_plan=363,
                                                    warehouse_code=23,
                                                    generic=True,
                                                    generic_value='2209',
                                                    send=True)
                os.rename(latest, latest.replace('.xml'.lower(), '_booked.xml'))
            # os.remove(latest)

    def test_validate_invoice(self):
        # print(os.listdir('c:/Users/DECEL/FACT1/MZK/2023.01.03/'))
        for root, _dir, files in os.walk(r'c:\Users\levi\FACT1\MZK\2024_05_22'):
            for file in files:
                # print(file)
                self.efactoauth.validate_invoice(os.path.join(root, file))
        # self.efactoauth.validate_invoice('c:/Users/levi/FACT1/MZK/2024_01_23/MZK241157.xml')

    def test_add_invoice(self):
        # self.efactoauth.validate_invoice('C:/Users/DECEL/FACT1/MZK/2023_06/MZK2321602.xml', env='prod')
        self.efactoauth.add_invoice('17259191', r'C:/Users/DECEL/FACT1/MZK/2023_06/MZK2321602.xml', env='prod')

    def test_bulk_add_invoice(self):
        for root, _, files in os.walk(r'f:/DECEL/EFACT/KINVEST/2024_01/'):
            for file in files:
                if file.endswith('.xml'):
                    inv_file = os.path.join(root, file)
                    self.efactoauth.add_invoice('31434328', inv_file, env='prod')

    def test_transform_to_pdf(self):
        self.efactoauth.transform_to_pdf(r'c:/Users/DECEL/FACT1/MZK/2023_01_03/MZK2414.xml')

    def test_bulk_transform(self):
        for root, _, files in os.walk(r'c:\Users\levi\git\anafapi\tests\17259191\FACTURA PRIMITA\2024_03\20\40628013'):
            for file in files:
                if file.endswith('.xml'):
                    inv_file = os.path.join(root, file)
                    self.efactoauth.transform_to_pdf(inv_file, no_validation=True)

    def test_get_all_messages(self, vat_id, days, root_path, env, filter_, extract):
        self.efactoauth.get_all_messages(vat_id, days=days, env=env, root_path=root_path, filter_=filter_, extract=extract)

    def test_pretty_messages(self):
        self.efactoauth.pretty_messages('17259191', env='prod')

    def test_download_invoice(self):
        self.efactoauth.download_invoice('3105899770', path=r'C:/Users/DECEL/FACT1/MZK/2023_06', env='prod')

    def test_getxml(self):
        files = [file for file in os.listdir(r'C:\Users\Levi\git\anafapi\tests\17259191\FACTURA PRIMITA\2024_05\24\9845734') if '.zip' in file]
        # print(files)
        for file in files:
            self.efactoauth.getxml_from_zip(path=os.path.join(r'C:\Users\Levi\git\anafapi\tests\17259191\FACTURA PRIMITA\2024_05\24\9845734', file), dest_f=r'C:\Users\Levi\git\anafapi\tests\17259191\FACTURA PRIMITA\2024_05\24\9845734')

    def test_partner_code(self):
        with open(r'C:/Users/Levi/git/ciel_rest_api/parts.json') as partner_file:
            partners = ujson.load(partner_file)
        invoice = Invoice.from_xml(r'C:\Users\Levi\git\anafapi\tests\17259191\FACTURA PRIMITA\2024_04\08\16702141\4238612905_booked.xml')
        tax_code = invoice.seller_party.tax_scheme_id.replace('ro'.upper(), '')
        for partner in partners:
            if tax_code == partner['TaxCode']:
                print(partner['Code'], partner['Name'])
                break

    def test_verify_invoice(self):
        self.efactoauth.verify_invoice('4075641152', env='prod')

    def test_isvalid(self):
        self.efactoauth.load_token_file(token_file_path.format(r'efact.json'))
        print(self.efactoauth.is_valid)

    def test_get_latest_invoice(self):
        latest = self.efactoauth.get_latest_invoice('efact')
        print(latest)
        # os.replace(latest, latest.replace('.xml'.lower(), '_booked.xml'))

    def test_rename_all(self, dir_path, kind: str):  # @UndefinedVariable
        if kind == 'seller':
            account = 'AccountingCustomerParty'
            # print(os.path.basename(dir_path))
            # dir_path = os.path.join(dir_path, os.path.basename(dir_path))
        else:
            account = 'AccountingSupplierParty'
            # next_subfolder = next(os.walk(dir_path))[1][0]
        for root_f, _dirs, files in os.walk(dir_path):
            for file in files:
                # print(root_f.split('/'))
                # print(os.path.basename(dir_path))
                # print(root_f, file)
                if file.endswith('.xml') and file.count('_') < 2 and 'cif error' not in root_f:
                    # print(os.path.join(root_f, file))
                    try:
                        tree = etree.parse(os.path.join(root_f, file))
                        root = tree.getroot()
                        inv_id = root.xpath('//cbc:ID', namespaces={'cbc': 'urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2'})
                        inv_id = inv_id[0].text # noqa
                        parties = root.xpath('//cbc:RegistrationName', namespaces={'cbc': 'urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2'})
                        for partie in parties: # noqa
                            parent_p_type = partie.getparent().getparent().getparent() # noqa
                            if etree.QName(parent_p_type).localname == account:  # @UndefinedVariable
                                # print(partie.text)
                                file_partie = partie.text # noqa

                        is_date = root.xpath('//cbc:IssueDate', namespaces={'cbc': 'urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2'})
                        is_date = is_date[0].text # noqa

                        partie_ids = root.xpath('//cbc:CompanyID', namespaces={'cbc': 'urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2'})
                        for partie_id in partie_ids: # noqa
                            parent_p_type = partie_id.getparent().getparent().getparent() # noqa
                            parent_t_type = partie_id.getparent() # noqa
                            # print(parent_t_type)
                            if etree.QName(parent_t_type).localname == 'PartyTaxScheme' and etree.QName(parent_p_type).localname == account:  # @UndefinedVariable
                                # print(partie_id.text)
                                file_partn_id = partie_id.text # noqa
                        new_name = '{}_{}_{}_{}.xml'.format(file.replace('.xml', ''), inv_id, file_partie, is_date)
                        current_path = os.path.join(root_f, file)
                        new_path = os.path.join(root_f, new_name)
                        #
                        # !This will rename all files
                        #
                        os.rename(current_path, new_path)
                        print(new_name)
                    except LxmlSyntaxError as exc_:
                        print(f'{root_f}/{file} not valid')
                        print(str(exc_))
        else:
            print('Nothing to rename')

    def test_parse_inv(self):
        # inv = self.efactoauth.parse_inv('C:/Users/Levi/git/anafapi/tests/17259191/8262600/FACTURA PRIMITA/01_2024/31/4154212674_booked.xml')
        # print(self.efactoauth._get_series(inv.invoice_id))
        inv_files = utils.get_not_booked('17259191')
        print(inv_files)


if __name__ == "__main__":
    # unittest.main()
    # efactTest().test_add_invoice()
    efactTest().test_get_all_messages('17259191', 2, '17259191', 'prod', 'P', True)
    # efactTest().test_rename_all(os.path.join(os.path.expanduser('~'), 'git/anafapi/tests/17259191/FACTURA TRIMISA'), kind='seller')
    # efactTest().test_rename_all(os.path.join(os.path.expanduser('~'), 'git/anafapi/tests/17259191/FACTURA PRIMITA'), kind='buyer')
    # anoauth.get_token()