80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.domain.validation import parse_scoring_model
|
|
from app.persistence.db import SessionLocal
|
|
from app.persistence.models import ScoringModelRecord, SourceRecord
|
|
from app.services.catalog_service import ensure_product_category, ensure_source_category
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
RULES = ROOT / "rules" / "generic-v1.json"
|
|
|
|
|
|
def run_bootstrap(session: Session) -> None:
|
|
_seed_source(session)
|
|
_seed_scoring_model(session)
|
|
ensure_product_category(session, "notebook", "Notebooks")
|
|
ensure_source_category(
|
|
session,
|
|
source_id="geizhals",
|
|
external_category_key="nb",
|
|
product_category_id="notebook",
|
|
name="Notebooks",
|
|
source_url="https://geizhals.de/?cat=nb",
|
|
)
|
|
session.commit()
|
|
|
|
|
|
def _seed_source(session: Session) -> None:
|
|
_upsert_source(session, "manual", "Manual entry", "manual")
|
|
_upsert_source(session, "geizhals", "Geizhals", "geizhals")
|
|
|
|
|
|
def _upsert_source(session: Session, source_id: str, name: str, adapter_type: str) -> None:
|
|
existing = session.get(SourceRecord, source_id)
|
|
if existing is None:
|
|
session.add(SourceRecord(id=source_id, name=name, adapter_type=adapter_type))
|
|
return
|
|
if existing.adapter_type != adapter_type:
|
|
raise SystemExit(f"bootstrap conflict: source '{source_id}' has a different adapter_type")
|
|
|
|
|
|
def _seed_scoring_model(session: Session) -> None:
|
|
definition = json.loads(RULES.read_text(encoding="utf-8"))
|
|
model = parse_scoring_model(definition)
|
|
existing = session.get(ScoringModelRecord, model.version)
|
|
if existing is None:
|
|
session.add(
|
|
ScoringModelRecord(
|
|
version=model.version,
|
|
algorithm_id=model.algorithm_id,
|
|
definition=model.definition,
|
|
definition_hash=model.definition_hash,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
)
|
|
return
|
|
if existing.definition_hash != model.definition_hash:
|
|
raise SystemExit(
|
|
f"bootstrap conflict: scoring model {model.version} hash "
|
|
f"{existing.definition_hash} != {model.definition_hash}"
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
session = SessionLocal()
|
|
try:
|
|
run_bootstrap(session)
|
|
print("bootstrap ok")
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|