Arsham/product_profile_ingestion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor ingestion logic to improve clarity and consistency. Updated JSON key mappings for chunk data and modified SQL column references in the upsert and insert functions. Revised README to reflect changes in primary key naming for label chunks.

Browse Source
This commit is contained in:
Arsham Mirehvandi 2026-09-30 09:20:28 +02:00
parent 4eec89aa9d
commit a68fee9025
2 changed files with 25 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -117,7 +117,7 @@ Each `.jsonl` file contains one JSON object per line. See `data/chunks/delan_pro
| JSON field | SQL column | Notes | | JSON field | SQL column | Notes |
|------------|------------|-------| |------------|------------|-------|
| `chunk_id` | `chunk_id` | Primary key | | `chunk_id` | `label_chunk_id` | Primary key |
| `product_id` | `product_id` | Foreign key to `products` | | `product_id` | `product_id` | Foreign key to `products` |
| `product_name` | `product_name` | | | `product_name` | `product_name` | |
| `target_crops` | `target_crops` | Stored as JSON string via `json.dumps()` | | `target_crops` | `target_crops` | Stored as JSON string via `json.dumps()` |

45
ingest.py
View File

@ -47,6 +47,7 @@ DEFAULT_CHUNKS_DIR = Path("data/chunks")
# re-ingesting the same product always maps to the same object. # re-ingesting the same product always maps to the same object.
WEAVIATE_UUID_NAMESPACE = uuid.UUID("1b7e2c3a-4f5d-4a8b-9c1e-6d2f3a8b9c0d") WEAVIATE_UUID_NAMESPACE = uuid.UUID("1b7e2c3a-4f5d-4a8b-9c1e-6d2f3a8b9c0d")
# JSON keys for array-valued profile fields. ``frac_groups`` is stored in SQL as ``frac_codes``.
ARRAY_FIELDS = ( ARRAY_FIELDS = (
"active_ingredients", "active_ingredients",
"frac_groups", "frac_groups",
@ -75,14 +76,15 @@ USE_FIELDS = (
"growth_stage_end", "growth_stage_end",
) )
CHUNK_INSERT_COLUMNS = ( # (SQL column, JSON key). JSON ``chunk_id`` is stored as ``label_chunk_id``.
"chunk_id", CHUNK_COLUMN_SOURCES = (
"product_id", ("label_chunk_id", "chunk_id"),
"product_name", ("product_id", "product_id"),
"target_crops", ("product_name", "product_name"),
"target_diseases", ("target_crops", "target_crops"),
"chunk_type", ("target_diseases", "target_diseases"),
"chunk_text", ("chunk_type", "chunk_type"),
("chunk_text", "chunk_text"),
) )
CHUNK_ARRAY_FIELDS = ("target_crops", "target_diseases") CHUNK_ARRAY_FIELDS = ("target_crops", "target_diseases")
@ -338,13 +340,13 @@ def upsert_product(cursor: pyodbc.Cursor, profile: Dict[str, Any]) -> None:
SET product_name = ?, SET product_name = ?,
registration_number = ?, registration_number = ?,
manufacturer = ?, manufacturer = ?,
organic_certified = ?, is_organic_certified = ?,
systemicity = ?, systemicity = ?,
preventive_action = ?, has_preventive_action = ?,
curative_action = ?, has_curative_action = ?,
eradicant_action = ?, has_eradicant_action = ?,
active_ingredients = ?, active_ingredients = ?,
frac_groups = ?, frac_codes = ?,
target_crops = ?, target_crops = ?,
target_diseases = ?, target_diseases = ?,
action_type = ? action_type = ?
@ -352,8 +354,8 @@ def upsert_product(cursor: pyodbc.Cursor, profile: Dict[str, Any]) -> None:
ELSE ELSE
INSERT INTO products ( INSERT INTO products (
product_id, product_name, registration_number, manufacturer, product_id, product_name, registration_number, manufacturer,
organic_certified, systemicity, preventive_action, curative_action, is_organic_certified, systemicity, has_preventive_action, has_curative_action,
eradicant_action, active_ingredients, frac_groups, target_crops, has_eradicant_action, active_ingredients, frac_codes, target_crops,
target_diseases, action_type target_diseases, action_type
) )
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
@ -405,11 +407,11 @@ def replace_uses(cursor: pyodbc.Cursor, profile: Dict[str, Any]) -> None:
def chunk_insert_values(chunk: Dict[str, Any]) -> tuple[Any, ...]: def chunk_insert_values(chunk: Dict[str, Any]) -> tuple[Any, ...]:
"""Build parameterized insert values for a single label chunk row.""" """Build parameterized insert values for a single label chunk row."""
values: list[Any] = [] values: list[Any] = []
for column in CHUNK_INSERT_COLUMNS: for _column, source_key in CHUNK_COLUMN_SOURCES:
if column in CHUNK_ARRAY_FIELDS: if source_key in CHUNK_ARRAY_FIELDS:
values.append(json.dumps(chunk.get(column, []))) values.append(json.dumps(chunk.get(source_key, [])))
else: else:
values.append(chunk.get(column)) values.append(chunk.get(source_key))
return tuple(values) return tuple(values)
@ -421,11 +423,12 @@ def replace_label_chunks(
"""Replace all label-chunk rows for a product with the provided set.""" """Replace all label-chunk rows for a product with the provided set."""
cursor.execute("DELETE FROM label_chunks WHERE product_id = ?", product_id) cursor.execute("DELETE FROM label_chunks WHERE product_id = ?", product_id)
insert_columns = [column for column, _source_key in CHUNK_COLUMN_SOURCES]
insert_sql = f""" insert_sql = f"""
INSERT INTO label_chunks ( INSERT INTO label_chunks (
{", ".join(CHUNK_INSERT_COLUMNS)} {", ".join(insert_columns)}
) )
VALUES ({", ".join(["?"] * len(CHUNK_INSERT_COLUMNS))}) VALUES ({", ".join(["?"] * len(insert_columns))})
""" """
for chunk in chunks: for chunk in chunks: