447 lines
14 KiB
Python
447 lines
14 KiB
Python
"""
|
|
Workflow to standardize components in PartDB.
|
|
|
|
For parts in the Passives category:
|
|
- Sets the name to "Value Package" format
|
|
- Sets the description to "MPN Value Package Tolerance Voltage/Current/Power" format
|
|
- Sets the EDA value to match the component value
|
|
- Fixes names/descriptions with "/" separators
|
|
|
|
For parts in other categories:
|
|
- Checks if EDA value is set, if not sets it to the part name
|
|
- Fixes names/descriptions with "/" separators
|
|
"""
|
|
|
|
import re
|
|
from typing import Optional, List, Tuple
|
|
from tqdm import tqdm
|
|
|
|
from config import PARTDB_BASE, PARTDB_TOKEN
|
|
from apis.partdb_api import PartDB
|
|
from workflows.standardize_passives import standardize_part as standardize_passive_part
|
|
|
|
|
|
def fix_slash_separated_fields(name: str, description: str) -> Tuple[str, str]:
|
|
"""
|
|
Fix names and descriptions that have two parts separated by '/'.
|
|
For example: "ABC123/XYZ789" -> "ABC123"
|
|
|
|
Returns:
|
|
Tuple of (fixed_name, fixed_description)
|
|
"""
|
|
fixed_name = name
|
|
fixed_desc = description
|
|
|
|
# Check if name has a slash with content on both sides
|
|
if '/' in name:
|
|
parts = name.split('/')
|
|
if len(parts) == 2 and parts[0].strip() and parts[1].strip():
|
|
# Take the first part
|
|
fixed_name = parts[0].strip()
|
|
|
|
# Check if description has a slash with content on both sides
|
|
if '/' in description:
|
|
parts = description.split('/')
|
|
if len(parts) == 2 and parts[0].strip() and parts[1].strip():
|
|
# Take the first part
|
|
fixed_desc = parts[0].strip()
|
|
|
|
return fixed_name, fixed_desc
|
|
|
|
|
|
def standardize_non_passive_part(api: PartDB, part: dict, dry_run: bool = False) -> Tuple[bool, str]:
|
|
"""
|
|
Standardize a non-passive component.
|
|
|
|
For non-passives:
|
|
1. Fix slash-separated names/descriptions
|
|
2. Ensure EDA value is set (use name if not set)
|
|
|
|
Returns:
|
|
(success: bool, message: str)
|
|
"""
|
|
part_id = api._extract_id(part)
|
|
current_name = part.get("name", "")
|
|
current_desc = part.get("description", "")
|
|
current_eda = part.get("value", "")
|
|
|
|
# Fix slash-separated fields
|
|
new_name, new_desc = fix_slash_separated_fields(current_name, current_desc)
|
|
|
|
# Check if EDA value needs to be set
|
|
new_eda = current_eda
|
|
if not current_eda or current_eda.strip() == "":
|
|
new_eda = current_name
|
|
|
|
# Check what needs to be changed
|
|
changes = []
|
|
if current_name != new_name:
|
|
changes.append(f"name: '{current_name}' -> '{new_name}'")
|
|
if current_desc != new_desc:
|
|
changes.append(f"desc: '{current_desc}' -> '{new_desc}'")
|
|
if current_eda != new_eda:
|
|
changes.append(f"EDA: '{current_eda}' -> '{new_eda}'")
|
|
|
|
if not changes:
|
|
return (True, "Already correct")
|
|
|
|
if dry_run:
|
|
return (True, f"Would update: {'; '.join(changes)}")
|
|
|
|
# Apply updates
|
|
try:
|
|
# Update name and description if needed
|
|
if current_name != new_name or current_desc != new_desc:
|
|
payload = {}
|
|
if current_name != new_name:
|
|
payload["name"] = new_name
|
|
if current_desc != new_desc:
|
|
payload["description"] = new_desc
|
|
|
|
r = api._patch_merge(f"/api/parts/{part_id}", payload)
|
|
if r.status_code not in range(200, 300):
|
|
return (False, f"Failed to update name/desc: {r.status_code}")
|
|
|
|
# Update EDA value if needed
|
|
if current_eda != new_eda:
|
|
success = api.patch_eda_value(part_id, new_eda)
|
|
if not success:
|
|
return (False, "Failed to update EDA value")
|
|
|
|
return (True, f"Updated: {'; '.join(changes)}")
|
|
|
|
except Exception as e:
|
|
return (False, f"Update failed: {e}")
|
|
|
|
|
|
def get_all_parts_paginated(api: PartDB) -> List[dict]:
|
|
"""
|
|
Get all parts from PartDB using pagination.
|
|
"""
|
|
all_parts = []
|
|
page = 1
|
|
per_page = 30
|
|
|
|
print("Fetching all parts from API...")
|
|
|
|
while True:
|
|
params = {"per_page": per_page, "page": page}
|
|
print(f" Fetching page {page}...")
|
|
|
|
try:
|
|
parts = api._get("/api/parts", params=params)
|
|
|
|
if isinstance(parts, list):
|
|
if not parts:
|
|
break
|
|
all_parts.extend(parts)
|
|
print(f" Got {len(parts)} parts (total: {len(all_parts)})")
|
|
page += 1
|
|
else:
|
|
print(f" Unexpected response type: {type(parts)}")
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f" Error fetching parts: {e}")
|
|
break
|
|
|
|
print(f"Total parts fetched: {len(all_parts)}")
|
|
return all_parts
|
|
|
|
|
|
def is_part_in_passives_category(api: PartDB, part: dict, passives_category_ids: List[int]) -> bool:
|
|
"""
|
|
Check if a part belongs to the Passives category or any of its subcategories.
|
|
"""
|
|
category = part.get("category")
|
|
if not category:
|
|
return False
|
|
|
|
# Extract category ID
|
|
if isinstance(category, dict):
|
|
cat_id_str = category.get("id") or category.get("_id")
|
|
elif isinstance(category, str):
|
|
cat_id_str = category
|
|
else:
|
|
return False
|
|
|
|
# Parse the ID
|
|
try:
|
|
if isinstance(cat_id_str, str):
|
|
cat_id = int(''.join(c for c in cat_id_str if c.isdigit()))
|
|
else:
|
|
cat_id = int(cat_id_str)
|
|
|
|
return cat_id in passives_category_ids
|
|
except:
|
|
return False
|
|
|
|
|
|
def get_passives_category_ids(api: PartDB) -> List[int]:
|
|
"""
|
|
Get all category IDs for Passives and its subcategories.
|
|
"""
|
|
categories = api.list_categories()
|
|
target_cat_id = None
|
|
|
|
# Find the Passives category
|
|
for cat in categories:
|
|
name = (cat.get("name") or "").strip()
|
|
if name.lower() == "passives":
|
|
target_cat_id = api._extract_id(cat)
|
|
break
|
|
|
|
if not target_cat_id:
|
|
print("Warning: Passives category not found!")
|
|
return []
|
|
|
|
# Find all subcategories
|
|
category_ids = [target_cat_id]
|
|
|
|
def find_children(parent_id: int):
|
|
for cat in categories:
|
|
parent = cat.get("parent")
|
|
if parent:
|
|
parent_id_str = None
|
|
if isinstance(parent, dict):
|
|
parent_id_str = parent.get("id") or parent.get("_id")
|
|
elif isinstance(parent, str):
|
|
parent_id_str = parent
|
|
|
|
if parent_id_str:
|
|
try:
|
|
if isinstance(parent_id_str, str):
|
|
parent_num = int(''.join(c for c in parent_id_str if c.isdigit()))
|
|
else:
|
|
parent_num = int(parent_id_str)
|
|
|
|
if parent_num == parent_id:
|
|
child_id = api._extract_id(cat)
|
|
if child_id and child_id not in category_ids:
|
|
category_ids.append(child_id)
|
|
find_children(child_id)
|
|
except:
|
|
pass
|
|
|
|
find_children(target_cat_id)
|
|
|
|
print(f"Found Passives category with {len(category_ids)} total categories (including subcategories)")
|
|
return category_ids
|
|
|
|
|
|
def run_standardize_components(dry_run: bool = False, progress_callback=None):
|
|
"""
|
|
Main function to standardize all components.
|
|
|
|
For passives: Full standardization (value, format, etc.)
|
|
For others: Fix slashes, ensure EDA value is set
|
|
|
|
Args:
|
|
dry_run: If True, only show what would be changed without making changes
|
|
progress_callback: Optional callback function(current, total, status_text)
|
|
Returns True if operation should be cancelled
|
|
"""
|
|
print("=" * 70)
|
|
print("COMPONENT STANDARDIZATION")
|
|
print("=" * 70)
|
|
print(f"Mode: {'DRY RUN (no changes)' if dry_run else 'LIVE (making changes)'}")
|
|
print("=" * 70)
|
|
print()
|
|
|
|
# Initialize API
|
|
api = PartDB(PARTDB_BASE, PARTDB_TOKEN)
|
|
|
|
# Get passives category IDs
|
|
passives_ids = get_passives_category_ids(api)
|
|
|
|
# Get all parts
|
|
print("\nFetching all parts...")
|
|
all_parts = get_all_parts_paginated(api)
|
|
|
|
if not all_parts:
|
|
print("No parts found!")
|
|
return
|
|
|
|
# Separate passives from others
|
|
passives = []
|
|
non_passives = []
|
|
|
|
for part in all_parts:
|
|
if is_part_in_passives_category(api, part, passives_ids):
|
|
passives.append(part)
|
|
else:
|
|
non_passives.append(part)
|
|
|
|
print(f"\nFound {len(passives)} passive components")
|
|
print(f"Found {len(non_passives)} non-passive components")
|
|
print(f"Total: {len(all_parts)} parts")
|
|
print()
|
|
|
|
# Statistics
|
|
passive_success = 0
|
|
passive_already_correct = 0
|
|
passive_failed = 0
|
|
|
|
non_passive_success = 0
|
|
non_passive_already_correct = 0
|
|
non_passive_failed = 0
|
|
|
|
errors = []
|
|
|
|
# Calculate total for progress
|
|
total_parts = len(all_parts)
|
|
processed = 0
|
|
|
|
# Process passives
|
|
if passives:
|
|
print("=" * 70)
|
|
print("PROCESSING PASSIVE COMPONENTS")
|
|
print("=" * 70)
|
|
|
|
bar = tqdm(passives, desc="Passives", unit="part") if not progress_callback else None
|
|
|
|
for part in passives:
|
|
# Check for cancellation
|
|
if progress_callback:
|
|
cancelled = progress_callback(
|
|
processed, total_parts,
|
|
f"Processing passives ({processed+1}/{total_parts})..."
|
|
)
|
|
if cancelled:
|
|
print("\n⚠ Operation cancelled by user")
|
|
if bar:
|
|
bar.close()
|
|
return
|
|
|
|
part_id = api._extract_id(part)
|
|
mpn = part.get("manufacturer_product_number") or part.get("mpn") or f"ID:{part_id}"
|
|
|
|
if bar:
|
|
bar.set_postfix_str(mpn[:30])
|
|
|
|
success, message = standardize_passive_part(api, part, dry_run=dry_run)
|
|
|
|
if success:
|
|
if "Already correct" in message:
|
|
passive_already_correct += 1
|
|
else:
|
|
passive_success += 1
|
|
if bar:
|
|
tqdm.write(f"✓ {mpn}: {message}")
|
|
else:
|
|
print(f"✓ {mpn}: {message}")
|
|
else:
|
|
passive_failed += 1
|
|
errors.append((mpn, message))
|
|
if bar:
|
|
tqdm.write(f"✗ {mpn}: {message}")
|
|
else:
|
|
print(f"✗ {mpn}: {message}")
|
|
|
|
processed += 1
|
|
|
|
if bar:
|
|
bar.close()
|
|
|
|
# Process non-passives
|
|
if non_passives:
|
|
print()
|
|
print("=" * 70)
|
|
print("PROCESSING NON-PASSIVE COMPONENTS")
|
|
print("=" * 70)
|
|
|
|
bar = tqdm(non_passives, desc="Others", unit="part") if not progress_callback else None
|
|
|
|
for part in non_passives:
|
|
# Check for cancellation
|
|
if progress_callback:
|
|
cancelled = progress_callback(
|
|
processed, total_parts,
|
|
f"Processing non-passives ({processed+1}/{total_parts})..."
|
|
)
|
|
if cancelled:
|
|
print("\n⚠ Operation cancelled by user")
|
|
if bar:
|
|
bar.close()
|
|
return
|
|
|
|
part_id = api._extract_id(part)
|
|
mpn = part.get("manufacturer_product_number") or part.get("mpn") or f"ID:{part_id}"
|
|
|
|
if bar:
|
|
bar.set_postfix_str(mpn[:30])
|
|
|
|
success, message = standardize_non_passive_part(api, part, dry_run=dry_run)
|
|
|
|
if success:
|
|
if "Already correct" in message:
|
|
non_passive_already_correct += 1
|
|
else:
|
|
non_passive_success += 1
|
|
if bar:
|
|
tqdm.write(f"✓ {mpn}: {message}")
|
|
else:
|
|
print(f"✓ {mpn}: {message}")
|
|
else:
|
|
non_passive_failed += 1
|
|
errors.append((mpn, message))
|
|
if bar:
|
|
tqdm.write(f"✗ {mpn}: {message}")
|
|
else:
|
|
print(f"✗ {mpn}: {message}")
|
|
|
|
processed += 1
|
|
|
|
if bar:
|
|
bar.close()
|
|
|
|
# Final progress update
|
|
if progress_callback:
|
|
progress_callback(total_parts, total_parts, "Complete!")
|
|
|
|
# Print summary
|
|
print()
|
|
print("=" * 70)
|
|
print("SUMMARY")
|
|
print("=" * 70)
|
|
print(f"PASSIVE COMPONENTS:")
|
|
print(f" Total processed: {len(passives)}")
|
|
print(f" Already correct: {passive_already_correct}")
|
|
print(f" Successfully updated: {passive_success}")
|
|
print(f" Failed: {passive_failed}")
|
|
print()
|
|
print(f"NON-PASSIVE COMPONENTS:")
|
|
print(f" Total processed: {len(non_passives)}")
|
|
print(f" Already correct: {non_passive_already_correct}")
|
|
print(f" Successfully updated: {non_passive_success}")
|
|
print(f" Failed: {non_passive_failed}")
|
|
print()
|
|
print(f"TOTAL:")
|
|
print(f" Parts processed: {total_parts}")
|
|
print(f" Successfully updated: {passive_success + non_passive_success}")
|
|
print(f" Failed: {passive_failed + non_passive_failed}")
|
|
print("=" * 70)
|
|
|
|
if errors:
|
|
print()
|
|
print("ERRORS:")
|
|
for mpn, msg in errors[:20]: # Show first 20 errors
|
|
print(f" {mpn}: {msg}")
|
|
if len(errors) > 20:
|
|
print(f" ... and {len(errors) - 20} more")
|
|
|
|
print()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
# Check for dry-run flag
|
|
dry_run = "--dry-run" in sys.argv or "-n" in sys.argv
|
|
|
|
if dry_run:
|
|
print("Running in DRY RUN mode - no changes will be made")
|
|
print()
|
|
|
|
run_standardize_components(dry_run=dry_run)
|