Sync InvenTree part links to Bambuddy archives

src/bambuddy_inventree_sync/bambuddy.py

@@ -54,6 +54,10 @@ class BambuddyClient:
             filename=f"bambuddy-archive-{archive_id}-thumbnail.{extension}",
         )
 
+    async def update_archive_external_url(self, archive_id: int, external_url: str) -> Archive:
+        data = await self._request("PATCH", f"/archives/{archive_id}", json={"external_url": external_url})
+        return Archive.model_validate(data)
+
     async def list_archives(
         self,
         *,

src/bambuddy_inventree_sync/config.py

@@ -13,6 +13,7 @@ class Settings(BaseSettings):
     bambuddy_api_key: str
 
     inventree_base_url: str
+    inventree_web_url: str | None = None
     inventree_token: str
     inventree_part_category_id: int
     inventree_stock_location_id: int
@@ -23,6 +24,8 @@ class Settings(BaseSettings):
     sync_success_only: bool = True
     sync_part_images: bool = True
     overwrite_part_images: bool = False
+    sync_archive_external_link: bool = True
+    overwrite_archive_external_link: bool = False
     default_stock_quantity: Annotated[float, Field(gt=0)] = 1
     inventree_stock_status: int = 10
     part_ipn_prefix: str = "BMB"
@@ -33,9 +36,11 @@ class Settings(BaseSettings):
     http_timeout_seconds: Annotated[int, Field(ge=1)] = 30
     data_dir: Path = Path("/data")
 
-    @field_validator("bambuddy_base_url", "inventree_base_url")
+    @field_validator("bambuddy_base_url", "inventree_base_url", "inventree_web_url")
     @classmethod
-    def strip_url(cls, value: str) -> str:
+    def strip_url(cls, value: str | None) -> str | None:
+        if value is None or not value.strip():
+            return None
         return value.rstrip("/")
 
     @field_validator("part_ipn_prefix")
@@ -53,6 +58,13 @@ class Settings(BaseSettings):
         fields = [field.strip() for field in self.part_key_fields.split(",") if field.strip()]
         return fields or ["filename", "name"]
 
+    @property
+    def inventree_browser_url(self) -> str:
+        base_url = (self.inventree_web_url or self.inventree_base_url).rstrip("/")
+        if base_url.endswith("/api"):
+            return base_url[:-4]
+        return base_url
+
 
 @lru_cache
 def get_settings() -> Settings:

src/bambuddy_inventree_sync/models.py

@@ -27,6 +27,7 @@ class Archive(BaseModel):
     quantity: float | None = None
     object_count: int | None = None
     cost: float | None = None
+    external_url: str | None = None
     notes: str | None = None
     tags: list[str] | None = None
 

src/bambuddy_inventree_sync/sync.py

@@ -133,6 +133,7 @@ class ArchiveSyncService:
         part_id = await self._get_or_create_part(part_key=part_key, ipn=ipn, name=name, description=description)
         await self._sync_part_parameters(part_id=part_id, archive=archive)
         await self._sync_part_image(part_id=part_id, archive=archive)
+        await self._sync_archive_external_link(part_id=part_id, archive=archive)
         batch = self.inventree.batch_for_archive(archive.id)
         existing_stock = await self.inventree.find_stock_by_batch(part_id=part_id, batch=batch)
 
@@ -228,6 +229,23 @@ class ArchiveSyncService:
         )
         logger.info("Synced Bambuddy archive %s thumbnail to InvenTree part %s", archive.id, part_id)
 
+    async def _sync_archive_external_link(self, *, part_id: int, archive: Archive) -> None:
+        if not self.settings.sync_archive_external_link:
+            return
+
+        part_url = self.inventree_part_url(part_id)
+        if archive.external_url and not self.settings.overwrite_archive_external_link:
+            if archive.external_url == part_url:
+                return
+            if not archive.external_url.startswith(self.settings.inventree_browser_url):
+                return
+
+        await self.bambuddy.update_archive_external_url(archive.id, part_url)
+        logger.info("Synced InvenTree part %s link to Bambuddy archive %s", part_id, archive.id)
+
+    def inventree_part_url(self, part_id: int) -> str:
+        return f"{self.settings.inventree_browser_url}/web/part/{part_id}/"
+
     def part_parameters_for_archive(self, archive: Archive) -> list[dict[str, str | float | None]]:
         parameters: list[dict[str, str | float | None]] = []