|
import argparse |
|
import requests |
|
import hmac |
|
import datetime |
|
import json |
|
from base64 import b64encode, b64decode |
|
from urllib.parse import urlencode |
|
import logging |
|
import sys |
|
import os |
|
|
|
# Configure logging |
|
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') |
|
|
|
# Replace with your actual API credentials |
|
API_KEY = '...' |
|
API_ID = '...' |
|
|
|
API_BASE_URL = 'https://aktiva.merit.ee/api/v2/' |
|
API_GET_INVOICES_URL = API_BASE_URL + 'getinvoices2' |
|
API_GET_INVOICE_URL = API_BASE_URL + 'getinvoice' |
|
|
|
def calculate_signature(timestamp): |
|
data_string = API_ID + timestamp |
|
hash_obj = hmac.new(key=API_KEY.encode('ascii'), msg=data_string.encode('ascii'), digestmod='sha256') |
|
return b64encode(hash_obj.digest()) |
|
|
|
def get_query_params(): |
|
timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S') |
|
signature = calculate_signature(timestamp) |
|
return { |
|
'ApiId': API_ID, |
|
'timestamp': timestamp, |
|
'signature': signature, |
|
} |
|
|
|
def read_invoice_numbers(filename): |
|
"""Read invoice numbers from a file, one per line.""" |
|
if not os.path.isfile(filename): |
|
logging.error(f"File not found: {filename}") |
|
raise FileNotFoundError(f"File not found: {filename}") |
|
try: |
|
with open(filename, 'r') as f: |
|
invoice_numbers = [line.strip() for line in f if line.strip()] |
|
if not invoice_numbers: |
|
logging.error("No invoice numbers found in the file.") |
|
raise ValueError("No invoice numbers found in the file.") |
|
return invoice_numbers |
|
except Exception as e: |
|
logging.error(f"Error reading file {filename}: {e}") |
|
raise |
|
|
|
def get_invoice_ids_by_numbers(invoice_numbers): |
|
"""Retrieve invoice IDs for the given invoice numbers using the v2 API.""" |
|
invoice_ids = {} |
|
for invoice_no in invoice_numbers: |
|
query_params = get_query_params() |
|
payload = { |
|
"InvNo": invoice_no |
|
} |
|
headers = {'Content-Type': 'application/json'} |
|
try: |
|
response = requests.post(API_GET_INVOICES_URL + '?' + urlencode(query_params), json=payload, headers=headers) |
|
response.raise_for_status() |
|
response_content = response.text.strip() |
|
try: |
|
invoices = json.loads(response_content) |
|
except json.JSONDecodeError as e: |
|
logging.error(f"Error decoding JSON response for invoice {invoice_no}: {e}") |
|
logging.error(f"Response content: {response_content}") |
|
raise |
|
if not isinstance(invoices, list): |
|
logging.error(f"Unexpected response structure for invoice {invoice_no}: {invoices}") |
|
raise ValueError("Invalid response structure") |
|
found = False |
|
for invoice in invoices: |
|
if not isinstance(invoice, dict): |
|
logging.error(f"Invalid invoice data for invoice {invoice_no}: {invoice}") |
|
raise ValueError("Invalid invoice data") |
|
retrieved_invoice_no = invoice.get('InvoiceNo') |
|
invoice_id = invoice.get('SIHId') |
|
if retrieved_invoice_no == invoice_no: |
|
invoice_ids[invoice_no] = invoice_id |
|
logging.info(f"Found invoice ID {invoice_id} for invoice number {invoice_no}") |
|
found = True |
|
break |
|
if not found: |
|
logging.error(f"Invoice number {invoice_no} not found.") |
|
raise ValueError(f"Invoice number {invoice_no} not found.") |
|
except requests.exceptions.HTTPError as http_err: |
|
logging.error(f"HTTP error occurred while fetching invoice {invoice_no}: {http_err}") |
|
logging.error(f"Response content: {response.text}") |
|
raise |
|
except Exception as err: |
|
logging.error(f"An error occurred while fetching invoice {invoice_no}: {err}") |
|
raise |
|
return invoice_ids |
|
|
|
def get_invoice_details(invoice_id, add_attachment=False): |
|
"""Retrieve the invoice details, optionally including the attachment.""" |
|
query_params = get_query_params() |
|
payload = { |
|
"Id": invoice_id, |
|
"AddAttachment": add_attachment |
|
} |
|
headers = {'Content-Type': 'application/json'} |
|
try: |
|
response = requests.post(API_GET_INVOICE_URL + '?' + urlencode(query_params), json=payload, headers=headers) |
|
response.raise_for_status() |
|
data = response.json() |
|
return data |
|
except requests.exceptions.HTTPError as http_err: |
|
logging.error(f"HTTP error occurred while fetching invoice {invoice_id}: {http_err}") |
|
logging.error(f"Response content: {response.text}") |
|
raise |
|
except Exception as err: |
|
logging.error(f"An error occurred while fetching invoice {invoice_id}: {err}") |
|
raise |
|
|
|
def download_attachments(invoice_numbers): |
|
invoice_ids = get_invoice_ids_by_numbers(invoice_numbers) |
|
for invoice_no, invoice_id in invoice_ids.items(): |
|
logging.info(f"Processing invoice {invoice_no} with ID {invoice_id}") |
|
data = get_invoice_details(invoice_id, add_attachment=True) |
|
# Check if Attachment is present |
|
attachment = data.get('Attachment') |
|
if not attachment: |
|
logging.error(f"No attachment found for invoice {invoice_no}") |
|
raise ValueError(f"No attachment found for invoice {invoice_no}") |
|
filename = attachment.get('Filename') |
|
file_content = attachment.get('FileContent') # base64 encoded |
|
if not filename or not file_content: |
|
logging.error(f"Invalid attachment data for invoice {invoice_no}") |
|
raise ValueError(f"Invalid attachment data for invoice {invoice_no}") |
|
# Save the PDF |
|
try: |
|
pdf_bytes = b64decode(file_content) |
|
# Derive filename from invoice number |
|
filename = f"{invoice_no}.pdf" |
|
with open(filename, 'wb') as pdf_file: |
|
pdf_file.write(pdf_bytes) |
|
logging.info(f"PDF saved as {filename}") |
|
except Exception as e: |
|
logging.error(f"Error saving PDF {filename}: {e}") |
|
raise |
|
|
|
def check_payments(invoice_numbers): |
|
invoice_ids = get_invoice_ids_by_numbers(invoice_numbers) |
|
paid_invoices = [] |
|
unpaid_invoices = [] |
|
for invoice_no, invoice_id in invoice_ids.items(): |
|
logging.info(f"Processing invoice {invoice_no} with ID {invoice_id}") |
|
data = get_invoice_details(invoice_id) |
|
header = data.get('Header') |
|
if not header: |
|
logging.error(f"Invoice data missing for invoice {invoice_no}") |
|
raise ValueError(f"Invoice data missing for invoice {invoice_no}") |
|
is_paid = header.get('Paid') |
|
paid_amount = header.get('PaidAmount') |
|
if paid_amount is None: |
|
logging.error(f"Paid amount missing for invoice {invoice_no}") |
|
raise ValueError(f"Paid amount missing for invoice {invoice_no}") |
|
if is_paid: |
|
paid_invoices.append({'InvoiceNo': invoice_no, 'PaidAmount': paid_amount}) |
|
else: |
|
unpaid_invoices.append({'InvoiceNo': invoice_no, 'PaidAmount': paid_amount}) |
|
# Sort invoices by PaidAmount in descending order |
|
paid_invoices.sort(key=lambda x: x['PaidAmount'], reverse=True) |
|
unpaid_invoices.sort(key=lambda x: x['PaidAmount'], reverse=True) |
|
# Output the results |
|
print("\nPaid Invoices:") |
|
for invoice in paid_invoices: |
|
print(f"Invoice {invoice['InvoiceNo']}: Paid Amount {invoice['PaidAmount']}") |
|
print("\nUnpaid Invoices:") |
|
for invoice in unpaid_invoices: |
|
print(f"Invoice {invoice['InvoiceNo']}: Paid Amount {invoice['PaidAmount']}") |
|
|
|
def check_lineitem_price(invoice_numbers, target_price): |
|
invoice_ids = get_invoice_ids_by_numbers(invoice_numbers) |
|
mismatched_items = [] |
|
for invoice_no, invoice_id in invoice_ids.items(): |
|
logging.info(f"Processing invoice {invoice_no} with ID {invoice_id}") |
|
data = get_invoice_details(invoice_id) |
|
lines = data.get('Lines') |
|
if lines is None: |
|
logging.error(f"Line items missing for invoice {invoice_no}") |
|
raise ValueError(f"Line items missing for invoice {invoice_no}") |
|
for line in lines: |
|
line_description = line.get('Description', 'N/A') |
|
price = line.get('Price') |
|
if price is None: |
|
logging.error(f"Price missing in one of the lines for invoice {invoice_no}") |
|
raise ValueError(f"Price missing in one of the lines for invoice {invoice_no}") |
|
if abs(price - target_price) > 1e-6: |
|
mismatched_items.append({ |
|
'InvoiceNo': invoice_no, |
|
'LineDescription': line_description, |
|
'Price': price |
|
}) |
|
# Output the mismatched items |
|
if mismatched_items: |
|
print("\nLine items with price mismatch:") |
|
for item in mismatched_items: |
|
print(f"Invoice {item['InvoiceNo']}, Line Item: {item['LineDescription']}, Price: {item['Price']}") |
|
else: |
|
print("\nAll line items have the price matching the specified amount.") |
|
|
|
def main(): |
|
parser = argparse.ArgumentParser(description='Interact with Merit API for invoice data.') |
|
parser.add_argument('action', choices=['download-attachments', 'check-payments', 'check-lineitem-price'], help='Action to perform.') |
|
parser.add_argument('filename', help='Path to a file containing invoice numbers (one per line).') |
|
parser.add_argument('--price', type=float, help='Price to compare line item prices against (required for check-lineitem-price).') |
|
args = parser.parse_args() |
|
|
|
# Read invoice numbers from file |
|
try: |
|
invoice_numbers = read_invoice_numbers(args.filename) |
|
except Exception as e: |
|
logging.error(f"Failed to read invoice numbers: {e}") |
|
sys.exit(1) |
|
|
|
# Perform the action |
|
try: |
|
if args.action == 'download-attachments': |
|
download_attachments(invoice_numbers) |
|
elif args.action == 'check-payments': |
|
check_payments(invoice_numbers) |
|
elif args.action == 'check-lineitem-price': |
|
if args.price is None: |
|
logging.error("Price argument is required for check-lineitem-price action.") |
|
sys.exit(1) |
|
check_lineitem_price(invoice_numbers, args.price) |
|
except Exception as e: |
|
logging.error(f"An error occurred during '{args.action}' action: {e}") |
|
sys.exit(1) |
|
|
|
if __name__ == '__main__': |
|
main() |