Initial Codes
This commit is contained in:
+364
@@ -0,0 +1,364 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
QBO Entity ID Lookup Script
|
||||
============================
|
||||
This script connects to QuickBooks Online and retrieves entity IDs
|
||||
for Accounts, Customers, Vendors, Items, Classes, Departments, etc.
|
||||
|
||||
Usage:
|
||||
python get_qbo_ids.py <entity_type> [--search NAME] [--export FILE]
|
||||
|
||||
Entity Types:
|
||||
account, customer, vendor, item, class, department, term, paymentmethod
|
||||
|
||||
Examples:
|
||||
python get_qbo_ids.py account # List all accounts
|
||||
python get_qbo_ids.py customer # List all customers
|
||||
python get_qbo_ids.py vendor --search "ABC" # Search vendors
|
||||
python get_qbo_ids.py item --export items.csv # Export items to CSV
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
import csv
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
# Add parent directory to path to import from src
|
||||
script_dir = Path(__file__).parent
|
||||
sys.path.insert(0, str(script_dir))
|
||||
|
||||
from src.config.settings import Settings
|
||||
from src.api.qbo_client import QBOClient
|
||||
|
||||
|
||||
# Entity configurations
|
||||
ENTITY_CONFIG = {
|
||||
'account': {
|
||||
'query_name': 'Account',
|
||||
'display_name': 'Accounts',
|
||||
'columns': ['Id', 'Name', 'AccountType', 'AccountSubType', 'AcctNum', 'CurrentBalance', 'Active'],
|
||||
'search_field': 'Name',
|
||||
'order_by': 'Name',
|
||||
},
|
||||
'customer': {
|
||||
'query_name': 'Customer',
|
||||
'display_name': 'Customers',
|
||||
'columns': ['Id', 'DisplayName', 'CompanyName', 'PrimaryEmailAddr.Address', 'PrimaryPhone.FreeFormNumber', 'Balance', 'Active'],
|
||||
'search_field': 'DisplayName',
|
||||
'order_by': 'DisplayName',
|
||||
},
|
||||
'vendor': {
|
||||
'query_name': 'Vendor',
|
||||
'display_name': 'Vendors',
|
||||
'columns': ['Id', 'DisplayName', 'CompanyName', 'PrimaryEmailAddr.Address', 'PrimaryPhone.FreeFormNumber', 'Balance', 'Active'],
|
||||
'search_field': 'DisplayName',
|
||||
'order_by': 'DisplayName',
|
||||
},
|
||||
'item': {
|
||||
'query_name': 'Item',
|
||||
'display_name': 'Items/Products',
|
||||
'columns': ['Id', 'Name', 'Type', 'Description', 'UnitPrice', 'Active'],
|
||||
'search_field': 'Name',
|
||||
'order_by': 'Name',
|
||||
},
|
||||
'class': {
|
||||
'query_name': 'Class',
|
||||
'display_name': 'Classes',
|
||||
'columns': ['Id', 'Name', 'FullyQualifiedName', 'Active'],
|
||||
'search_field': 'Name',
|
||||
'order_by': 'Name',
|
||||
},
|
||||
'department': {
|
||||
'query_name': 'Department',
|
||||
'display_name': 'Departments',
|
||||
'columns': ['Id', 'Name', 'FullyQualifiedName', 'Active'],
|
||||
'search_field': 'Name',
|
||||
'order_by': 'Name',
|
||||
},
|
||||
'term': {
|
||||
'query_name': 'Term',
|
||||
'display_name': 'Payment Terms',
|
||||
'columns': ['Id', 'Name', 'DueDays', 'Active'],
|
||||
'search_field': 'Name',
|
||||
'order_by': 'Name',
|
||||
},
|
||||
'paymentmethod': {
|
||||
'query_name': 'PaymentMethod',
|
||||
'display_name': 'Payment Methods',
|
||||
'columns': ['Id', 'Name', 'Type', 'Active'],
|
||||
'search_field': 'Name',
|
||||
'order_by': 'Name',
|
||||
},
|
||||
'employee': {
|
||||
'query_name': 'Employee',
|
||||
'display_name': 'Employees',
|
||||
'columns': ['Id', 'DisplayName', 'GivenName', 'FamilyName', 'PrimaryEmailAddr.Address', 'Active'],
|
||||
'search_field': 'DisplayName',
|
||||
'order_by': 'DisplayName',
|
||||
},
|
||||
'taxcode': {
|
||||
'query_name': 'TaxCode',
|
||||
'display_name': 'Tax Codes',
|
||||
'columns': ['Id', 'Name', 'Description', 'Active'],
|
||||
'search_field': 'Name',
|
||||
'order_by': 'Name',
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def get_nested_value(obj, key_path):
|
||||
"""Get nested value from dictionary using dot notation."""
|
||||
keys = key_path.split('.')
|
||||
value = obj
|
||||
for key in keys:
|
||||
if isinstance(value, dict):
|
||||
value = value.get(key)
|
||||
else:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def fetch_entities(client, entity_type, search_term=None, max_results=1000):
|
||||
"""
|
||||
Fetch entities from QBO.
|
||||
|
||||
Args:
|
||||
client: QBOClient instance
|
||||
entity_type: Type of entity to fetch
|
||||
search_term: Optional search term
|
||||
max_results: Maximum results to return
|
||||
|
||||
Returns:
|
||||
List of entity dictionaries
|
||||
"""
|
||||
config = ENTITY_CONFIG.get(entity_type)
|
||||
if not config:
|
||||
raise ValueError(f"Unknown entity type: {entity_type}")
|
||||
|
||||
query_name = config['query_name']
|
||||
search_field = config['search_field']
|
||||
order_by = config['order_by']
|
||||
|
||||
# Build query
|
||||
query = f"SELECT * FROM {query_name}"
|
||||
|
||||
if search_term:
|
||||
query += f" WHERE {search_field} LIKE '%{search_term}%'"
|
||||
|
||||
query += f" ORDERBY {order_by} MAXRESULTS {max_results}"
|
||||
|
||||
# Execute query
|
||||
response = client._make_request('GET', 'query', params={'query': query})
|
||||
|
||||
if response and 'QueryResponse' in response:
|
||||
return response['QueryResponse'].get(query_name, [])
|
||||
|
||||
return []
|
||||
|
||||
|
||||
def format_table(entities, columns):
|
||||
"""Format entities as a readable table."""
|
||||
if not entities:
|
||||
return "No records found."
|
||||
|
||||
# Calculate column widths
|
||||
widths = {}
|
||||
for col in columns:
|
||||
col_name = col.split('.')[-1] # Use last part of nested path as header
|
||||
values = [str(get_nested_value(e, col) or '') for e in entities]
|
||||
widths[col] = max(len(col_name), min(40, max(len(v) for v in values) if values else 0))
|
||||
|
||||
# Build header
|
||||
headers = [col.split('.')[-1] for col in columns]
|
||||
header_row = " | ".join(h.ljust(widths[columns[i]]) for i, h in enumerate(headers))
|
||||
separator = "-+-".join("-" * widths[col] for col in columns)
|
||||
|
||||
lines = [header_row, separator]
|
||||
|
||||
# Build data rows
|
||||
for entity in entities:
|
||||
row = []
|
||||
for col in columns:
|
||||
value = get_nested_value(entity, col)
|
||||
if value is None:
|
||||
value = '-'
|
||||
elif isinstance(value, bool):
|
||||
value = 'Yes' if value else 'No'
|
||||
elif isinstance(value, (int, float)) and 'Balance' in col or 'Price' in col:
|
||||
value = f"${value:,.2f}"
|
||||
else:
|
||||
value = str(value)[:40]
|
||||
row.append(value.ljust(widths[col]))
|
||||
lines.append(" | ".join(row))
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def export_to_csv(entities, columns, filepath):
|
||||
"""Export entities to CSV file."""
|
||||
with open(filepath, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.writer(f)
|
||||
|
||||
# Write header
|
||||
headers = [col.split('.')[-1] for col in columns]
|
||||
writer.writerow(headers)
|
||||
|
||||
# Write data
|
||||
for entity in entities:
|
||||
row = [get_nested_value(entity, col) or '' for col in columns]
|
||||
writer.writerow(row)
|
||||
|
||||
print(f"Exported {len(entities)} records to {filepath}")
|
||||
|
||||
|
||||
def export_to_json(entities, filepath):
|
||||
"""Export entities to JSON file."""
|
||||
with open(filepath, 'w', encoding='utf-8') as f:
|
||||
json.dump(entities, f, indent=2)
|
||||
|
||||
print(f"Exported {len(entities)} records to {filepath}")
|
||||
|
||||
|
||||
def print_id_mapping(entities, entity_type):
|
||||
"""Print a simple ID to Name mapping."""
|
||||
config = ENTITY_CONFIG.get(entity_type)
|
||||
name_field = config['search_field'] if config else 'Name'
|
||||
|
||||
print("\nID Mapping (for copy/paste):")
|
||||
print("-" * 50)
|
||||
for entity in entities:
|
||||
entity_id = entity.get('Id')
|
||||
name = get_nested_value(entity, name_field) or entity.get('Name', 'Unknown')
|
||||
print(f"{entity_id}: {name}")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description='Fetch Entity IDs from QuickBooks Online',
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Entity Types:
|
||||
account - Chart of Accounts
|
||||
customer - Customers
|
||||
vendor - Vendors
|
||||
item - Products and Services
|
||||
class - Classes (for tracking)
|
||||
department - Departments/Locations
|
||||
term - Payment Terms
|
||||
paymentmethod - Payment Methods
|
||||
employee - Employees
|
||||
taxcode - Tax Codes
|
||||
|
||||
Examples:
|
||||
%(prog)s account # List all accounts
|
||||
%(prog)s customer --search "John" # Search customers
|
||||
%(prog)s vendor --export vendors.csv # Export to CSV
|
||||
%(prog)s item --json # Output as JSON
|
||||
%(prog)s account --type Bank # Filter accounts by type
|
||||
"""
|
||||
)
|
||||
|
||||
parser.add_argument('entity_type',
|
||||
choices=list(ENTITY_CONFIG.keys()),
|
||||
help='Type of entity to fetch')
|
||||
parser.add_argument('--search', '-s', dest='search_term',
|
||||
help='Search by name')
|
||||
parser.add_argument('--type', '-t', dest='filter_type',
|
||||
help='Filter by type (for accounts)')
|
||||
parser.add_argument('--export', '-e', dest='export_file',
|
||||
help='Export to file (CSV or JSON)')
|
||||
parser.add_argument('--json', action='store_true',
|
||||
help='Output as JSON')
|
||||
parser.add_argument('--ids-only', action='store_true',
|
||||
help='Show only ID:Name mapping')
|
||||
parser.add_argument('--max', type=int, default=1000,
|
||||
help='Maximum results to fetch (default: 1000)')
|
||||
parser.add_argument('--verbose', '-v', action='store_true',
|
||||
help='Show detailed information')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
config = ENTITY_CONFIG[args.entity_type]
|
||||
|
||||
# Initialize settings and client
|
||||
print("=" * 60)
|
||||
print(f"QBO {config['display_name']} Lookup")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
settings = Settings()
|
||||
credentials = settings.get_credentials()
|
||||
|
||||
if not credentials.access_token:
|
||||
print("\nError: Not connected to QuickBooks Online.")
|
||||
print("Please connect via the web application first.")
|
||||
sys.exit(1)
|
||||
|
||||
client = QBOClient(credentials)
|
||||
|
||||
if not client.is_authenticated:
|
||||
print("\nError: Authentication failed or token expired.")
|
||||
print("Please reconnect via the web application.")
|
||||
sys.exit(1)
|
||||
|
||||
# Get company info
|
||||
company_info = client.get_company_info()
|
||||
print(f"\nConnected to: {company_info.get('CompanyName', 'Unknown')}")
|
||||
print(f"Environment: {credentials.environment}")
|
||||
print("-" * 60)
|
||||
|
||||
# Fetch entities
|
||||
print(f"\nFetching {config['display_name'].lower()}...")
|
||||
if args.search_term:
|
||||
print(f" Search term: {args.search_term}")
|
||||
|
||||
entities = fetch_entities(client, args.entity_type, args.search_term, args.max)
|
||||
|
||||
# Filter by type if specified (for accounts)
|
||||
if args.filter_type and args.entity_type == 'account':
|
||||
entities = [e for e in entities if e.get('AccountType') == args.filter_type]
|
||||
|
||||
print(f"\nFound {len(entities)} record(s)")
|
||||
print("-" * 60)
|
||||
|
||||
if not entities:
|
||||
print("No records match your criteria.")
|
||||
sys.exit(0)
|
||||
|
||||
# Output results
|
||||
if args.json:
|
||||
print(json.dumps(entities, indent=2))
|
||||
elif args.ids_only:
|
||||
print_id_mapping(entities, args.entity_type)
|
||||
elif args.verbose:
|
||||
for i, entity in enumerate(entities, 1):
|
||||
print(f"\n[{i}] {get_nested_value(entity, config['search_field']) or entity.get('Name', 'Unknown')}")
|
||||
for key, value in entity.items():
|
||||
if not key.startswith('Meta') and value is not None:
|
||||
print(f" {key}: {value}")
|
||||
else:
|
||||
print(format_table(entities, config['columns']))
|
||||
|
||||
# Export if requested
|
||||
if args.export_file:
|
||||
filepath = args.export_file
|
||||
if filepath.endswith('.json'):
|
||||
export_to_json(entities, filepath)
|
||||
else:
|
||||
export_to_csv(entities, config['columns'], filepath)
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("Done!")
|
||||
|
||||
except Exception as e:
|
||||
print(f"\nError: {str(e)}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user