682 lines
23 KiB
Python
682 lines
23 KiB
Python
"""
|
|
Data ingestion pipeline for the agricultural treatment assistant system.
|
|
|
|
Reads local product profile JSON files and label-chunk JSONL files, then loads
|
|
them into the configured backends:
|
|
|
|
- **Weaviate** (``ProductProfile``) — semantic ``retrieval_summary`` for profiles
|
|
- **SQL Server** — ``products``, ``product_uses`` (profiles), ``label_chunks`` (chunks)
|
|
|
|
The pipeline is idempotent: a hash manifest (``.processed_manifest.json``)
|
|
tracks which files have already been ingested (and their content hash), so
|
|
re-running the script skips unchanged files unless ``--force`` is passed.
|
|
|
|
Usage:
|
|
python ingest.py
|
|
python ingest.py --profiles-dir data/productProfiles
|
|
python ingest.py --chunks-dir data/chunks
|
|
python ingest.py --file data/productProfiles/ramin_sc_0916.json
|
|
python ingest.py --chunk-file data/chunks/delan_pro_16562.jsonl
|
|
python ingest.py --force
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import uuid
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterable, Optional
|
|
|
|
import pyodbc
|
|
import weaviate
|
|
from dotenv import load_dotenv
|
|
|
|
from setup_weaviate import ensure_product_profile_collection
|
|
|
|
COLLECTION_NAME = "ProductProfile"
|
|
MANIFEST_PATH = Path(".processed_manifest.json")
|
|
DEFAULT_PROFILES_DIR = Path("data/productProfiles")
|
|
DEFAULT_CHUNKS_DIR = Path("data/chunks")
|
|
|
|
# Namespace used to derive deterministic Weaviate UUIDs from product_id, so
|
|
# re-ingesting the same product always maps to the same object.
|
|
WEAVIATE_UUID_NAMESPACE = uuid.UUID("1b7e2c3a-4f5d-4a8b-9c1e-6d2f3a8b9c0d")
|
|
|
|
ARRAY_FIELDS = (
|
|
"active_ingredients",
|
|
"frac_groups",
|
|
"target_crops",
|
|
"target_diseases",
|
|
"action_type",
|
|
)
|
|
|
|
USE_FIELDS = (
|
|
"crop",
|
|
"disease",
|
|
"setting",
|
|
"dose_min",
|
|
"dose_max",
|
|
"dose_unit",
|
|
"concentration_min",
|
|
"concentration_max",
|
|
"concentration_unit",
|
|
"treatment_interval_min_days",
|
|
"treatment_interval_max_days",
|
|
"max_treatments_per_season",
|
|
"pre_harvest_interval_days",
|
|
"spray_volume_min",
|
|
"spray_volume_max",
|
|
"growth_stage_start",
|
|
"growth_stage_end",
|
|
)
|
|
|
|
CHUNK_INSERT_COLUMNS = (
|
|
"chunk_id",
|
|
"product_id",
|
|
"product_name",
|
|
"target_crops",
|
|
"target_diseases",
|
|
"chunk_type",
|
|
"chunk_text",
|
|
)
|
|
|
|
CHUNK_ARRAY_FIELDS = ("target_crops", "target_diseases")
|
|
CHUNK_REQUIRED_KEYS = ("chunk_id", "product_id", "chunk_text")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def load_config() -> Dict[str, str]:
|
|
"""Load and validate required environment variables from .env."""
|
|
load_dotenv()
|
|
|
|
required = [
|
|
"GEMINI_API_KEY",
|
|
"SQL_DRIVER",
|
|
"SQL_SERVER",
|
|
"SQL_DATABASE",
|
|
"SQL_USERNAME",
|
|
"SQL_PASSWORD",
|
|
]
|
|
missing = [key for key in required if not os.getenv(key)]
|
|
if missing:
|
|
raise EnvironmentError(
|
|
f"Missing required environment variable(s): {', '.join(missing)}. "
|
|
"Copy .env.example to .env and fill in the values."
|
|
)
|
|
|
|
return {
|
|
"GEMINI_API_KEY": os.getenv("GEMINI_API_KEY", ""),
|
|
"WEAVIATE_HOST": os.getenv("WEAVIATE_HOST", "localhost"),
|
|
"WEAVIATE_HTTP_PORT": os.getenv("WEAVIATE_HTTP_PORT", "8080"),
|
|
"WEAVIATE_GRPC_PORT": os.getenv("WEAVIATE_GRPC_PORT", "50051"),
|
|
"SQL_DRIVER": os.getenv("SQL_DRIVER", ""),
|
|
"SQL_SERVER": os.getenv("SQL_SERVER", ""),
|
|
"SQL_DATABASE": os.getenv("SQL_DATABASE", ""),
|
|
"SQL_USERNAME": os.getenv("SQL_USERNAME", ""),
|
|
"SQL_PASSWORD": os.getenv("SQL_PASSWORD", ""),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# File discovery, hashing, and manifest tracking
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def discover_files(data_dir: Path, pattern: str) -> list[Path]:
|
|
"""Recursively find files matching pattern under data_dir, sorted for stable order."""
|
|
if not data_dir.exists():
|
|
raise FileNotFoundError(f"Data directory not found: {data_dir}")
|
|
return sorted(data_dir.rglob(pattern))
|
|
|
|
|
|
def discover_json_files(data_dir: Path) -> list[Path]:
|
|
"""Recursively find all *.json files under data_dir."""
|
|
return discover_files(data_dir, "*.json")
|
|
|
|
|
|
def discover_jsonl_files(data_dir: Path) -> list[Path]:
|
|
"""Recursively find all *.jsonl files under data_dir."""
|
|
return discover_files(data_dir, "*.jsonl")
|
|
|
|
|
|
def sha256_of(path: Path) -> str:
|
|
"""Compute the SHA-256 hex digest of a file's contents."""
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(8192), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def load_manifest(manifest_path: Path = MANIFEST_PATH) -> Dict[str, str]:
|
|
"""Load the processed-files manifest (filename -> content hash)."""
|
|
if not manifest_path.exists():
|
|
return {}
|
|
try:
|
|
with manifest_path.open("r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, OSError) as exc:
|
|
print(f"[WARN] Could not read manifest at {manifest_path}: {exc}. Starting fresh.", file=sys.stderr)
|
|
return {}
|
|
|
|
|
|
def save_manifest(manifest: Dict[str, str], manifest_path: Path = MANIFEST_PATH) -> None:
|
|
"""Persist the processed-files manifest to disk."""
|
|
with manifest_path.open("w", encoding="utf-8") as f:
|
|
json.dump(manifest, f, indent=2, sort_keys=True)
|
|
|
|
|
|
def should_process(path: Path, file_hash: str, manifest: Dict[str, str], force: bool) -> bool:
|
|
"""Determine whether a file needs (re)processing based on the manifest."""
|
|
if force:
|
|
return True
|
|
key = str(path.resolve())
|
|
return manifest.get(key) != file_hash
|
|
|
|
|
|
def collect_files_to_process(
|
|
candidates: list[Path],
|
|
manifest: Dict[str, str],
|
|
force: bool,
|
|
) -> list[tuple[Path, str]]:
|
|
"""Filter candidates through the manifest and return (path, hash) pairs to process."""
|
|
to_process: list[tuple[Path, str]] = []
|
|
for path in candidates:
|
|
try:
|
|
file_hash = sha256_of(path)
|
|
except OSError as exc:
|
|
print(f"[ERROR] Could not read {path}: {exc}", file=sys.stderr)
|
|
continue
|
|
if should_process(path, file_hash, manifest, force):
|
|
to_process.append((path, file_hash))
|
|
else:
|
|
print(f"[SKIP] {path} already processed and unchanged.")
|
|
return to_process
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def parse_profile(path: Path) -> Dict[str, Any]:
|
|
"""Load and lightly validate a product profile JSON file."""
|
|
with path.open("r", encoding="utf-8") as f:
|
|
profile = json.load(f)
|
|
|
|
required_keys = ("product_id", "product_name", "retrieval_summary")
|
|
missing = [key for key in required_keys if key not in profile]
|
|
if missing:
|
|
raise ValueError(f"Profile {path} is missing required key(s): {', '.join(missing)}")
|
|
|
|
return profile
|
|
|
|
|
|
def parse_chunks_file(path: Path) -> list[Dict[str, Any]]:
|
|
"""
|
|
Load a JSONL label-chunk file, validating each non-empty line.
|
|
|
|
Malformed lines are logged to stderr and skipped; returns all valid chunks.
|
|
"""
|
|
chunks: list[Dict[str, Any]] = []
|
|
with path.open("r", encoding="utf-8") as f:
|
|
for line_number, line in enumerate(f, start=1):
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
try:
|
|
chunk = json.loads(stripped)
|
|
except json.JSONDecodeError as exc:
|
|
print(
|
|
f"[ERROR] {path} line {line_number}: invalid JSON: {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
if not isinstance(chunk, dict):
|
|
print(
|
|
f"[ERROR] {path} line {line_number}: expected a JSON object, got {type(chunk).__name__}",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
missing = [key for key in CHUNK_REQUIRED_KEYS if key not in chunk]
|
|
if missing:
|
|
print(
|
|
f"[ERROR] {path} line {line_number}: missing required key(s): {', '.join(missing)}",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
chunks.append(chunk)
|
|
return chunks
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Connections
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def connect_weaviate(cfg: Dict[str, str]) -> weaviate.WeaviateClient:
|
|
"""Open a connection to the local Weaviate instance with the Gemini API key header."""
|
|
client = weaviate.connect_to_local(
|
|
host=cfg["WEAVIATE_HOST"],
|
|
port=int(cfg["WEAVIATE_HTTP_PORT"]),
|
|
grpc_port=int(cfg["WEAVIATE_GRPC_PORT"]),
|
|
headers={"X-Goog-Studio-Api-Key": cfg["GEMINI_API_KEY"]},
|
|
)
|
|
return client
|
|
|
|
|
|
def connect_sql(cfg: Dict[str, str]) -> pyodbc.Connection:
|
|
"""Open a connection to the Microsoft SQL Server database via pyodbc."""
|
|
conn_str = (
|
|
f"DRIVER={{{cfg['SQL_DRIVER']}}};"
|
|
f"SERVER={cfg['SQL_SERVER']};"
|
|
f"DATABASE={cfg['SQL_DATABASE']};"
|
|
f"UID={cfg['SQL_USERNAME']};"
|
|
f"PWD={cfg['SQL_PASSWORD']};"
|
|
)
|
|
conn = pyodbc.connect(conn_str, autocommit=False)
|
|
return conn
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Weaviate insertion
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upsert_weaviate(client: weaviate.WeaviateClient, profile: Dict[str, Any]) -> None:
|
|
"""Insert or replace the semantic summary object for this product in Weaviate."""
|
|
collection = client.collections.get(COLLECTION_NAME)
|
|
|
|
product_id = profile["product_id"]
|
|
object_uuid = uuid.uuid5(WEAVIATE_UUID_NAMESPACE, product_id)
|
|
|
|
properties = {
|
|
"product_id": product_id,
|
|
"product_name": profile["product_name"],
|
|
"retrieval_summary": profile["retrieval_summary"],
|
|
}
|
|
|
|
if collection.data.exists(object_uuid):
|
|
collection.data.replace(uuid=object_uuid, properties=properties)
|
|
else:
|
|
collection.data.insert(properties=properties, uuid=object_uuid)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SQL Server: products table
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upsert_product(cursor: pyodbc.Cursor, profile: Dict[str, Any]) -> None:
|
|
"""Upsert the global metadata row for this product into the products table."""
|
|
array_json = {field: json.dumps(profile.get(field, [])) for field in ARRAY_FIELDS}
|
|
|
|
params = (
|
|
profile["product_id"],
|
|
profile.get("product_name"),
|
|
profile.get("registration_number"),
|
|
profile.get("manufacturer"),
|
|
int(bool(profile.get("organic_certified"))),
|
|
profile.get("systemicity"),
|
|
int(bool(profile.get("preventive_action"))),
|
|
int(bool(profile.get("curative_action"))),
|
|
int(bool(profile.get("eradicant_action"))),
|
|
array_json["active_ingredients"],
|
|
array_json["frac_groups"],
|
|
array_json["target_crops"],
|
|
array_json["target_diseases"],
|
|
array_json["action_type"],
|
|
)
|
|
|
|
cursor.execute(
|
|
"""
|
|
IF EXISTS (SELECT 1 FROM products WHERE product_id = ?)
|
|
UPDATE products
|
|
SET product_name = ?,
|
|
registration_number = ?,
|
|
manufacturer = ?,
|
|
organic_certified = ?,
|
|
systemicity = ?,
|
|
preventive_action = ?,
|
|
curative_action = ?,
|
|
eradicant_action = ?,
|
|
active_ingredients = ?,
|
|
frac_groups = ?,
|
|
target_crops = ?,
|
|
target_diseases = ?,
|
|
action_type = ?
|
|
WHERE product_id = ?
|
|
ELSE
|
|
INSERT INTO products (
|
|
product_id, product_name, registration_number, manufacturer,
|
|
organic_certified, systemicity, preventive_action, curative_action,
|
|
eradicant_action, active_ingredients, frac_groups, target_crops,
|
|
target_diseases, action_type
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
# IF EXISTS check
|
|
profile["product_id"],
|
|
# UPDATE ... SET (13 params) + WHERE product_id
|
|
*params[1:],
|
|
profile["product_id"],
|
|
# INSERT ... VALUES (14 params)
|
|
*params,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SQL Server: product_uses table
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def replace_uses(cursor: pyodbc.Cursor, profile: Dict[str, Any]) -> None:
|
|
"""Replace all use-rule rows for this product with the ones from the profile."""
|
|
product_id = profile["product_id"]
|
|
uses: Iterable[Dict[str, Any]] = profile.get("uses", [])
|
|
|
|
cursor.execute("DELETE FROM product_uses WHERE product_id = ?", product_id)
|
|
|
|
insert_sql = f"""
|
|
INSERT INTO product_uses (
|
|
product_id, {", ".join(USE_FIELDS)}
|
|
)
|
|
VALUES (?, {", ".join(["?"] * len(USE_FIELDS))})
|
|
"""
|
|
|
|
for index, use in enumerate(uses):
|
|
try:
|
|
values = (product_id, *(use.get(field) for field in USE_FIELDS))
|
|
cursor.execute(insert_sql, values)
|
|
except pyodbc.Error as exc:
|
|
print(
|
|
f"[ERROR] Failed to insert use rule #{index} for product '{product_id}': {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SQL Server: label_chunks table
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def chunk_insert_values(chunk: Dict[str, Any]) -> tuple[Any, ...]:
|
|
"""Build parameterized insert values for a single label chunk row."""
|
|
values: list[Any] = []
|
|
for column in CHUNK_INSERT_COLUMNS:
|
|
if column in CHUNK_ARRAY_FIELDS:
|
|
values.append(json.dumps(chunk.get(column, [])))
|
|
else:
|
|
values.append(chunk.get(column))
|
|
return tuple(values)
|
|
|
|
|
|
def replace_label_chunks(
|
|
cursor: pyodbc.Cursor,
|
|
product_id: str,
|
|
chunks: Iterable[Dict[str, Any]],
|
|
) -> None:
|
|
"""Replace all label-chunk rows for a product with the provided set."""
|
|
cursor.execute("DELETE FROM label_chunks WHERE product_id = ?", product_id)
|
|
|
|
insert_sql = f"""
|
|
INSERT INTO label_chunks (
|
|
{", ".join(CHUNK_INSERT_COLUMNS)}
|
|
)
|
|
VALUES ({", ".join(["?"] * len(CHUNK_INSERT_COLUMNS))})
|
|
"""
|
|
|
|
for chunk in chunks:
|
|
chunk_id = chunk.get("chunk_id", "<unknown>")
|
|
try:
|
|
cursor.execute(insert_sql, chunk_insert_values(chunk))
|
|
except pyodbc.Error as exc:
|
|
print(
|
|
f"[ERROR] Failed to insert label chunk '{chunk_id}' for product '{product_id}': {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Orchestration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def process_profile_file(
|
|
path: Path,
|
|
weaviate_client: weaviate.WeaviateClient,
|
|
sql_conn: pyodbc.Connection,
|
|
manifest: Dict[str, str],
|
|
file_hash: str,
|
|
) -> bool:
|
|
"""
|
|
Process a single product profile JSON file across Weaviate and SQL sinks.
|
|
|
|
Returns True on full success, False if any sink failed. On SQL failure the
|
|
transaction for this file is rolled back so no partial rows are committed.
|
|
"""
|
|
try:
|
|
profile = parse_profile(path)
|
|
except (json.JSONDecodeError, ValueError, OSError) as exc:
|
|
print(f"[ERROR] Failed to parse {path}: {exc}", file=sys.stderr)
|
|
return False
|
|
|
|
product_id = profile.get("product_id", "<unknown>")
|
|
|
|
weaviate_ok = True
|
|
try:
|
|
upsert_weaviate(weaviate_client, profile)
|
|
except Exception as exc: # noqa: BLE001 - log and continue to SQL sinks
|
|
weaviate_ok = False
|
|
print(f"[ERROR] Weaviate upsert failed for '{product_id}' ({path}): {exc}", file=sys.stderr)
|
|
|
|
sql_ok = True
|
|
cursor = sql_conn.cursor()
|
|
try:
|
|
upsert_product(cursor, profile)
|
|
replace_uses(cursor, profile)
|
|
sql_conn.commit()
|
|
except pyodbc.Error as exc:
|
|
sql_ok = False
|
|
sql_conn.rollback()
|
|
print(f"[ERROR] SQL upsert failed for '{product_id}' ({path}): {exc}. Transaction rolled back.", file=sys.stderr)
|
|
finally:
|
|
cursor.close()
|
|
|
|
success = weaviate_ok and sql_ok
|
|
if success:
|
|
manifest[str(path.resolve())] = file_hash
|
|
print(f"[OK] Processed profile '{product_id}' from {path}")
|
|
else:
|
|
print(f"[SKIP] Profile '{product_id}' from {path} not fully ingested; manifest not updated.", file=sys.stderr)
|
|
|
|
return success
|
|
|
|
|
|
def process_chunks_file(
|
|
path: Path,
|
|
sql_conn: pyodbc.Connection,
|
|
manifest: Dict[str, str],
|
|
file_hash: str,
|
|
) -> bool:
|
|
"""
|
|
Process a single label-chunk JSONL file into the label_chunks table.
|
|
|
|
Returns True on success. On SQL failure the transaction is rolled back.
|
|
"""
|
|
try:
|
|
chunks = parse_chunks_file(path)
|
|
except OSError as exc:
|
|
print(f"[ERROR] Failed to read {path}: {exc}", file=sys.stderr)
|
|
return False
|
|
|
|
if not chunks:
|
|
print(f"[ERROR] No valid chunks found in {path}; manifest not updated.", file=sys.stderr)
|
|
return False
|
|
|
|
chunks_by_product: dict[str, list[Dict[str, Any]]] = defaultdict(list)
|
|
for chunk in chunks:
|
|
chunks_by_product[chunk["product_id"]].append(chunk)
|
|
|
|
product_ids = ", ".join(sorted(chunks_by_product))
|
|
cursor = sql_conn.cursor()
|
|
try:
|
|
for product_id, product_chunks in sorted(chunks_by_product.items()):
|
|
replace_label_chunks(cursor, product_id, product_chunks)
|
|
sql_conn.commit()
|
|
except pyodbc.Error as exc:
|
|
sql_conn.rollback()
|
|
print(
|
|
f"[ERROR] SQL label_chunks upsert failed for product(s) [{product_ids}] ({path}): {exc}. "
|
|
"Transaction rolled back.",
|
|
file=sys.stderr,
|
|
)
|
|
return False
|
|
finally:
|
|
cursor.close()
|
|
|
|
manifest[str(path.resolve())] = file_hash
|
|
print(f"[OK] Processed {len(chunks)} label chunk(s) for product(s) [{product_ids}] from {path}")
|
|
return True
|
|
|
|
|
|
def run(
|
|
profiles_dir: Path,
|
|
chunks_dir: Path,
|
|
single_profile: Optional[Path],
|
|
single_chunk: Optional[Path],
|
|
force: bool,
|
|
) -> int:
|
|
"""Main pipeline execution. Returns a process exit code."""
|
|
cfg = load_config()
|
|
manifest = load_manifest()
|
|
|
|
if single_profile is not None:
|
|
profile_candidates = [single_profile]
|
|
else:
|
|
profile_candidates = discover_json_files(profiles_dir)
|
|
|
|
if single_chunk is not None:
|
|
chunk_candidates = [single_chunk]
|
|
elif chunks_dir.exists():
|
|
chunk_candidates = discover_jsonl_files(chunks_dir)
|
|
else:
|
|
chunk_candidates = []
|
|
if single_profile is None:
|
|
print(f"[WARN] Chunks directory not found: {chunks_dir}; skipping chunk ingestion.")
|
|
|
|
profiles_to_process = collect_files_to_process(profile_candidates, manifest, force)
|
|
chunks_to_process = collect_files_to_process(chunk_candidates, manifest, force)
|
|
|
|
if not profiles_to_process and not chunks_to_process:
|
|
print("Nothing to process.")
|
|
return 0
|
|
|
|
weaviate_client: Optional[weaviate.WeaviateClient] = None
|
|
sql_conn: Optional[pyodbc.Connection] = None
|
|
failures = 0
|
|
|
|
try:
|
|
sql_conn = connect_sql(cfg)
|
|
|
|
if profiles_to_process:
|
|
weaviate_client = connect_weaviate(cfg)
|
|
ensure_product_profile_collection(weaviate_client)
|
|
|
|
for path, file_hash in profiles_to_process:
|
|
try:
|
|
success = process_profile_file(path, weaviate_client, sql_conn, manifest, file_hash)
|
|
except Exception as exc: # noqa: BLE001 - never let one file crash the batch
|
|
success = False
|
|
print(f"[ERROR] Unexpected failure processing profile {path}: {exc}", file=sys.stderr)
|
|
if not success:
|
|
failures += 1
|
|
|
|
for path, file_hash in chunks_to_process:
|
|
try:
|
|
success = process_chunks_file(path, sql_conn, manifest, file_hash)
|
|
except Exception as exc: # noqa: BLE001
|
|
success = False
|
|
print(f"[ERROR] Unexpected failure processing chunks {path}: {exc}", file=sys.stderr)
|
|
if not success:
|
|
failures += 1
|
|
|
|
save_manifest(manifest)
|
|
finally:
|
|
if weaviate_client is not None:
|
|
try:
|
|
weaviate_client.close()
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[WARN] Error closing Weaviate client: {exc}", file=sys.stderr)
|
|
if sql_conn is not None:
|
|
try:
|
|
sql_conn.close()
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[WARN] Error closing SQL connection: {exc}", file=sys.stderr)
|
|
|
|
total_files = len(profiles_to_process) + len(chunks_to_process)
|
|
if failures:
|
|
print(f"Completed with {failures} failure(s) out of {total_files} file(s).", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Completed successfully. {total_files} file(s) processed.")
|
|
return 0
|
|
|
|
|
|
def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Ingest product profile JSON and label-chunk JSONL files into Weaviate and SQL Server."
|
|
)
|
|
parser.add_argument(
|
|
"--profiles-dir",
|
|
type=Path,
|
|
default=DEFAULT_PROFILES_DIR,
|
|
help="Directory to scan recursively for *.json profile files (default: data/productProfiles)",
|
|
)
|
|
parser.add_argument(
|
|
"--chunks-dir",
|
|
type=Path,
|
|
default=DEFAULT_CHUNKS_DIR,
|
|
help="Directory to scan recursively for *.jsonl chunk files (default: data/chunks)",
|
|
)
|
|
parser.add_argument(
|
|
"--file",
|
|
type=Path,
|
|
default=None,
|
|
help="Process a single profile JSON file instead of scanning --profiles-dir.",
|
|
)
|
|
parser.add_argument(
|
|
"--chunk-file",
|
|
type=Path,
|
|
default=None,
|
|
help="Process a single label-chunk JSONL file instead of scanning --chunks-dir.",
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Reprocess files even if unchanged since the last successful run.",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
try:
|
|
exit_code = run(
|
|
args.profiles_dir,
|
|
args.chunks_dir,
|
|
args.file,
|
|
args.chunk_file,
|
|
args.force,
|
|
)
|
|
except (EnvironmentError, FileNotFoundError) as exc:
|
|
print(f"[FATAL] {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
sys.exit(exit_code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|