Spaces:
Sleeping
Sleeping
Download src/storage/csv_storage.py from gouriikarus3d/Catalog-AI: direct link, hf CLI and curl.
- Browser
- Download file 5.32 kB
-
https://huggingface.co/spaces/gouriikarus3d/Catalog-AI/resolve/main/src/storage/csv_storage.py
- Command line
-
hf download hf://spaces/gouriikarus3d/Catalog-AI/src/storage/csv_storage.py
-
curl -L -o csv_storage.py https://huggingface.co/spaces/gouriikarus3d/Catalog-AI/resolve/main/src/storage/csv_storage.py
5.32 kB
| import csv | |
| import io | |
| from typing import Dict, List, Union | |
| class CSVStorage: | |
| """Save catalog data as CSV (in-memory).""" | |
| def catalog_to_rows(catalog: Union[Dict, List]) -> List[List[str]]: | |
| """ | |
| Convert catalog to CSV rows, handling both dict and list formats. | |
| Each row is: [Category, Component, Reference/Image] | |
| """ | |
| rows = [] | |
| # Handle list format (array of products) | |
| if isinstance(catalog, list): | |
| for product in catalog: | |
| product_name = product.get("product_name") or product.get("name", "Unknown Product") | |
| product_url = product.get("url", "") | |
| specifications = product.get("specifications", {}) | |
| # Base product row | |
| rows.append([ | |
| f"Base Model({product_name})", | |
| product_name, | |
| product_url | |
| ]) | |
| # Specifications | |
| if specifications: | |
| rows.append(["", "", ""]) # Empty row | |
| rows.append(["Specifications", "", ""]) | |
| for spec_key, spec_value in specifications.items(): | |
| rows.append([ | |
| "", | |
| f"{spec_key}: {spec_value}", | |
| "" | |
| ]) | |
| rows.append(["", "", ""]) # Empty row | |
| # Customizations | |
| customizations = product.get("customizations", {}) | |
| if isinstance(customizations, dict): | |
| for category, options in customizations.items(): | |
| if not options: | |
| continue | |
| # First option | |
| rows.append([ | |
| category, | |
| options[0].get("label", ""), | |
| options[0].get("image", "") | |
| ]) | |
| # Remaining options | |
| for opt in options[1:]: | |
| rows.append([ | |
| "", | |
| opt.get("label", ""), | |
| opt.get("image", "") | |
| ]) | |
| # Handle dict format (keyed by product ID) | |
| elif isinstance(catalog, dict): | |
| for product_id, product in catalog.items(): | |
| product_name = product.get("product_name") or product.get("name", "Unknown Product") | |
| product_url = product.get("url", "") | |
| specifications = product.get("specifications", {}) | |
| # Base product row | |
| rows.append([ | |
| f"Base Model({product_name})", | |
| product_name, | |
| product_url | |
| ]) | |
| # Specifications | |
| if specifications: | |
| rows.append(["", "", ""]) # Empty row | |
| rows.append(["Specifications", "", ""]) | |
| for spec_key, spec_value in specifications.items(): | |
| rows.append([ | |
| "", | |
| f"{spec_key}: {spec_value}", | |
| "" | |
| ]) | |
| rows.append(["", "", ""]) # Empty row | |
| # Customizations | |
| customizations = product.get("customizations", {}) | |
| if isinstance(customizations, dict): | |
| for category, options in customizations.items(): | |
| if not options: | |
| continue | |
| # First option | |
| rows.append([ | |
| category, | |
| options[0].get("label", ""), | |
| options[0].get("image", "") | |
| ]) | |
| # Remaining options | |
| for opt in options[1:]: | |
| rows.append([ | |
| "", | |
| opt.get("label", ""), | |
| opt.get("image", "") | |
| ]) | |
| return rows | |
| def to_csv_string(catalog: Union[Dict, List]) -> str: | |
| """ | |
| Convert catalog to a CSV string. | |
| Returns a CSV-formatted string suitable for download or API response. | |
| """ | |
| output = io.StringIO() | |
| writer = csv.writer(output) | |
| # Header row | |
| writer.writerow(["Categories", "Component", "References"]) | |
| # Data rows | |
| writer.writerows(CSVStorage.catalog_to_rows(catalog)) | |
| return output.getvalue() | |
| def save_simple(catalog: Union[Dict, List], file_path: str = None) -> str: | |
| """ | |
| Backward-compatible method. | |
| If file_path is provided, writes CSV to file and returns the path. | |
| Otherwise, returns CSV string (same as `to_csv_string`). | |
| """ | |
| csv_str = CSVStorage.to_csv_string(catalog) | |
| if file_path: | |
| with open(file_path, 'w', encoding='utf-8', newline='') as f: | |
| f.write(csv_str) | |
| return file_path | |
| return csv_str | |