improved typing

backend/pashub_fetcher/core_files.py

@@ -0,0 +1,13 @@
+from enum import Enum
+
+
+class CoreFiles(Enum):
+    PHOTOPACK = "Photopack"
+    SITENOTE = "SiteNote"
+    RDSAP_SITENOTE = "RdSAP_SiteNote"
+    PAS2023_VENTILATION = "PAS 2023 Ventilation Assessment Report"
+    PAS2023_CONDITION = "PAS 2023 Condition Report"
+    PAS_SIGNIFICANCE = "PAS Significance"
+    PAR_PHOTOPACK = "PAR Photo Pack"
+    PAS2023_PROPERTY = "PAS 2023 Property Assessment Report"
+    PAS2023_OCCUPANCY = "PAS 2023 Occupancy Assessment Report"

backend/pashub_fetcher/cotality_client.py

@@ -0,0 +1,110 @@
+from typing import List, Optional
+
+import requests
+
+from backend.pashub_fetcher.evidence_file_data import EvidenceFileData
+from backend.pashub_fetcher.evidence_metadata import EvidenceMetadata
+
+
+class CotalityClient:
+    def __init__(self, token: str):
+        self.token = token
+        self.company_id = "cb5249e2-8f31-4ef4-aefd-08ddaccb1fa2"
+        self.base = "https://pashub.net/api"
+
+        self.session = requests.Session()
+        self.session.headers.update(
+            {
+                "Authorization": f"Bearer {self.token}",
+                "Accept": "application/json",
+            }
+        )
+
+    def get_core_envidence_files_by_job_id(self, job_id: str) -> List[str]:
+        # url = f"{self.base}/jobs/{job_id}/evidence"
+
+        raise NotImplementedError
+
+    def get_evidence_files_by_uprn(self, uprn: str) -> List[str]:
+        """
+        Download evidence files for the most recent job for a UPRN.
+        Returns a list of saved filenames.
+        """
+
+        job_id: Optional[str] = self._get_latest_job_id(uprn)
+        if not job_id:
+            return []
+
+        evidence_list: List[EvidenceFileData] = self._get_evidence_list(job_id)
+        if not evidence_list:
+            return []
+
+        saved_files: List[str] = []
+
+        for evidence in evidence_list:
+            evidence_id = evidence.file_id
+            if not evidence_id:
+                continue
+
+            metadata: EvidenceMetadata = self._get_evidence_metadata(
+                job_id, evidence_id
+            )
+
+            download_url: str = self._build_download_url(metadata, evidence.file_id)
+            file_name = evidence.file_name
+
+            self._download_file(download_url, file_name)
+            saved_files.append(file_name)
+
+        return saved_files
+
+    def _get_latest_job_id(self, uprn: str) -> Optional[str]:
+        url = f"{self.base}/jobs"
+        params = {
+            "pageIndex": 0,
+            "pageSize": 20,
+            "orderBy": "createdUtc",
+            "orderDesc": "true",
+            "addressUprn": uprn,
+            "companyId": self.company_id,
+        }
+
+        r = self.session.get(url, params=params)
+        r.raise_for_status()
+
+        jobs = r.json().get("results", [])
+        return jobs[0]["id"] if jobs else None
+
+    def _get_evidence_list(self, job_id: str) -> List[EvidenceFileData]:
+        url = f"{self.base}/jobs/{job_id}/evidence"
+
+        r = self.session.get(url)
+        r.raise_for_status()
+
+        results = r.json().get("results", [])
+
+        return [EvidenceFileData.from_api(item) for item in results]
+
+    def _get_evidence_metadata(self, job_id: str, evidence_id: str) -> EvidenceMetadata:
+        url = f"{self.base}/jobs/{job_id}/evidenceMetadata"
+        params = {"evidenceIds": evidence_id}
+
+        r = self.session.get(url, params=params)
+        r.raise_for_status()
+
+        return EvidenceMetadata.from_api(r.json())
+
+    def _build_download_url(self, metadata: EvidenceMetadata, file_id: str) -> str:
+        container = metadata.container_name
+        blob_uri = metadata.blob_uri
+
+        base, sas = blob_uri.split("?", 1)
+
+        return f"{base}{container}/{file_id}?{sas}"
+
+    def _download_file(self, url: str, file_name: str) -> None:
+        r = self.session.get(url)
+        r.raise_for_status()
+
+        with open(file_name, "wb") as f:
+            f.write(r.content)

backend/pashub_fetcher/evidence_file_data.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+
+
+@dataclass
+class EvidenceFileData:
+    file_id: str
+    file_name: str
+    created_utc: str
+    file_size: int
+    file_extension: str
+
+    evidence_category: Optional[str] = None
+
+    @classmethod
+    def from_api(cls, data: Dict[str, Any]) -> EvidenceFileData:
+        return cls(
+            file_id=data["fileId"],
+            file_name=data["fileName"],
+            created_utc=data["createdUtc"],
+            file_size=data["fileSize"],
+            file_extension=data["fileExtension"],
+            evidence_category=data.get("evidenceCategory"),
+        )

backend/pashub_fetcher/evidence_metadata.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Any, Dict
+
+
+@dataclass
+class EvidenceMetadata:
+    container_name: str
+    blob_uri: str
+
+    @classmethod
+    def from_api(cls, data: Dict[str, Any]) -> EvidenceMetadata:
+        return cls(
+            container_name=data["containerName"],
+            blob_uri=data["blobUri"],
+        )

scripts/download_cotality_evidence.py

@@ -1,7 +1,7 @@
 import requests
 import json
 
-TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Ik1EUTRNRU5GUTBVNU9FUXpOelk1TVRFME0wUkdOMFpFUkRoR1JVVkJNVGMxT1RFNFJERXlPQSJ9.eyJodHRwOi8vZW1haWwiOiJzZWJhc3RpYW5Ab3Ntb3Npcy1hY2QuY29tIiwiaHR0cDovL2NsdWsudG9rZW4vbGFzdFBhc3N3b3JkQ2hhbmdlIjoiMjAyNS0wOC0yNlQwOTo1NDoyNi4zMjZaIiwiaHR0cDovL2NsdWsudG9rZW4vY29ubmVjdGlvbiI6ImVUZWNoSUQiLCJodHRwOi8vY2x1ay50b2tlbi9zdHJhdGVneSI6ImF1dGgwIiwiaHR0cDovL2NsdWsudG9rZW4vc3RyYXRlZ3lUeXBlIjoiZGF0YWJhc2UiLCJpc3MiOiJodHRwczovL2V0ZWNoaWQuZXUuYXV0aDAuY29tLyIsInN1YiI6ImF1dGgwfDY4YWQ4NDUyZDI2YzI1ZmMyMzkwZmYxYSIsImF1ZCI6WyJodHRwczovL3Bhc2h1Yi5hcGkuZXRlY2gubmV0IiwiaHR0cHM6Ly9ldGVjaGlkLmV1LmF1dGgwLmNvbS91c2VyaW5mbyJdLCJpYXQiOjE3NzMyMzc4MjQsImV4cCI6MTc3MzI0NTAyNCwic2NvcGUiOiJvcGVuaWQiLCJhenAiOiJEaVp6d3VVaTVkVmozOXR3NG00bWZ6emZvRm5MdmVLZyJ9.mkkxeZiD_ByHY4TJKpLQ-trmeGs15s0ekL6u1n-ek9j-EzNyf6qalEHCyHf8gzdNhU_vay96bIOMRHp4vXFaLqSANwKZayIS3EoA_b9-u2FAZpooxEvReAMNJGoZ6WLD01AQXWv-l7ww1ZqAnQzw0moL_Oma6hVmA5oa-RJKJ3MerS7e0Wei97Db48E140-EAbQf2iPcKYYtCNRA4il6n8DFiqGeoUMGo99jkR1ceZAvMpOAj8RhKX-4qSiDfX6yXUS2G96U5m7S_GWI-DEj5TazkN10Af3TyOY3EVjmZoJcRpiAR4cFmlfcTydjrShU03DWmPZm1QItf2McxfCpNA"
+TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Ik1EUTRNRU5GUTBVNU9FUXpOelk1TVRFME0wUkdOMFpFUkRoR1JVVkJNVGMxT1RFNFJERXlPQSJ9.eyJodHRwOi8vZW1haWwiOiJzZWJhc3RpYW5Ab3Ntb3Npcy1hY2QuY29tIiwiaHR0cDovL2NsdWsudG9rZW4vbGFzdFBhc3N3b3JkQ2hhbmdlIjoiMjAyNS0wOC0yNlQwOTo1NDoyNi4zMjZaIiwiaHR0cDovL2NsdWsudG9rZW4vY29ubmVjdGlvbiI6ImVUZWNoSUQiLCJodHRwOi8vY2x1ay50b2tlbi9zdHJhdGVneSI6ImF1dGgwIiwiaHR0cDovL2NsdWsudG9rZW4vc3RyYXRlZ3lUeXBlIjoiZGF0YWJhc2UiLCJpc3MiOiJodHRwczovL2V0ZWNoaWQuZXUuYXV0aDAuY29tLyIsInN1YiI6ImF1dGgwfDY4YWQ4NDUyZDI2YzI1ZmMyMzkwZmYxYSIsImF1ZCI6WyJodHRwczovL3Bhc2h1Yi5hcGkuZXRlY2gubmV0IiwiaHR0cHM6Ly9ldGVjaGlkLmV1LmF1dGgwLmNvbS91c2VyaW5mbyJdLCJpYXQiOjE3NzQyNzg3NjIsImV4cCI6MTc3NDI4NTk2Miwic2NvcGUiOiJvcGVuaWQiLCJhenAiOiJEaVp6d3VVaTVkVmozOXR3NG00bWZ6emZvRm5MdmVLZyJ9.ESIbau52J7KXL22tM8GlO9eV0f0pCOFdoQGL2YcjsTEcSeucHBuI9lHXT2dNJn0E8qlgafjazaMkoMs2g0TiTUUZU6XsKqKpUAJy4kk-qKp53V5az7e2MG9uDSa5bB1vWsQQw37zaNVQ0FQkpYHSiFeGoBh1PjuKwCpLjbl94bx7S4bQKaJSZRUj5TS75k6HnSOhUtN9LYLMPRoLty7TwqFLDxgj8Ixl_ddEF3C3Y6Mcxa5UF57BNTnFXmLefqsryex0XV4b5Btu4W5wZ4bjhX2M7PSXbk4lTv1YZdQxWLpzvNpEVnFueawtqedGYipqH1v4bg99YUnXDbajd2SSVQ"
 
 base = "https://pashub.net/api"