137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
# import mysql.connector
|
|
# from mysql.connector import Error
|
|
# import config
|
|
# import openpyxl
|
|
# import os
|
|
# import re
|
|
# import ast
|
|
# from datetime import datetime
|
|
|
|
|
|
# class ContractorInfo:
|
|
# ID = ""
|
|
# contInfo = None
|
|
# def __init__(self, id):
|
|
# self.ID = id
|
|
# print(id)
|
|
# self.fetchData()
|
|
|
|
# def fetchData(self):
|
|
# try:
|
|
# connection = config.get_db_connection()
|
|
# cursor = connection.cursor(dictionary=True, buffered=True)
|
|
# print("here", flush=True)
|
|
|
|
# cursor.callproc('GetContractorInfoById', [self.ID])
|
|
# for result in cursor.stored_results():
|
|
# self.contInfo = result.fetchone()
|
|
|
|
# print(self.contInfo,flush=True)
|
|
# finally:
|
|
# cursor.close()
|
|
# connection.close()
|
|
|
|
# def fetchalldata(self):
|
|
|
|
# try:
|
|
# connection = config.get_db_connection()
|
|
# cursor = connection.cursor(dictionary=True, buffered=True)
|
|
# print("here", flush=True)
|
|
|
|
|
|
# # ---------------- Hold Types ----------------
|
|
# cursor = connection.cursor(dictionary=True)
|
|
|
|
# cursor.callproc('GetHoldTypesByContractor', [self.ID])
|
|
|
|
# hold_types = []
|
|
# for result in cursor.stored_results():
|
|
# hold_types = result.fetchall()
|
|
# hold_type_map = {ht['hold_type_id']: ht['hold_type'] for ht in hold_types}
|
|
|
|
# # ---------------- Invoices ----------------
|
|
# cursor = connection.cursor(dictionary=True)
|
|
|
|
# cursor.callproc('GetInvoicesByContractor', [self.ID])
|
|
|
|
# invoices = []
|
|
# for result in cursor.stored_results():
|
|
# invoices = result.fetchall()
|
|
|
|
# # Remove duplicate invoices
|
|
# invoice_ids_seen = set()
|
|
# unique_invoices = []
|
|
# for inv in invoices:
|
|
# if inv["Invoice_Id"] not in invoice_ids_seen:
|
|
# invoice_ids_seen.add(inv["Invoice_Id"])
|
|
# unique_invoices.append(inv)
|
|
# invoices = unique_invoices
|
|
|
|
# finally:
|
|
# cursor.close()
|
|
# connection.close()
|
|
|
|
|
|
from mysql.connector import Error
|
|
import config
|
|
from datetime import datetime
|
|
|
|
|
|
class ContractorInfo:
|
|
def __init__(self, contractor_id):
|
|
self.ID = contractor_id
|
|
self.contInfo = None
|
|
self.fetchData()
|
|
|
|
def fetchData(self):
|
|
"""Fetch basic contractor info by ID."""
|
|
try:
|
|
connection = config.get_db_connection()
|
|
with connection.cursor(dictionary=True, buffered=True) as cursor:
|
|
cursor.callproc('GetContractorInfoById', [self.ID])
|
|
# Get the first result set
|
|
for result in cursor.stored_results():
|
|
self.contInfo = result.fetchone()
|
|
except Error as e:
|
|
print(f"Error fetching contractor info: {e}")
|
|
finally:
|
|
if connection.is_connected():
|
|
connection.close()
|
|
|
|
def fetchalldata(self):
|
|
"""Fetch hold types and invoices for contractor."""
|
|
data = {}
|
|
try:
|
|
connection = config.get_db_connection()
|
|
with connection.cursor(dictionary=True, buffered=True) as cursor:
|
|
# Fetch Hold Types
|
|
cursor.callproc('GetHoldTypesByContractor', [self.ID])
|
|
hold_types = []
|
|
for result in cursor.stored_results():
|
|
hold_types = result.fetchall()
|
|
hold_type_map = {ht['hold_type_id']: ht['hold_type'] for ht in hold_types}
|
|
data['hold_types'] = hold_type_map
|
|
|
|
# Fetch Invoices
|
|
cursor.callproc('GetInvoicesByContractor', [self.ID])
|
|
invoices = []
|
|
for result in cursor.stored_results():
|
|
invoices = result.fetchall()
|
|
|
|
# Remove duplicate invoices
|
|
seen_ids = set()
|
|
unique_invoices = []
|
|
for inv in invoices:
|
|
if inv['Invoice_Id'] not in seen_ids:
|
|
seen_ids.add(inv['Invoice_Id'])
|
|
unique_invoices.append(inv)
|
|
data['invoices'] = unique_invoices
|
|
|
|
except Error as e:
|
|
print(f"Error fetching contractor data: {e}")
|
|
finally:
|
|
if connection.is_connected():
|
|
connection.close()
|
|
|
|
return data
|