diff --git a/backend/pashub_fetcher/core_files.py b/backend/pashub_fetcher/core_files.py new file mode 100644 index 00000000..82637f70 --- /dev/null +++ b/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" diff --git a/backend/pashub_fetcher/cotality_client.py b/backend/pashub_fetcher/cotality_client.py new file mode 100644 index 00000000..9deda776 --- /dev/null +++ b/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) diff --git a/backend/pashub_fetcher/evidence_file_data.py b/backend/pashub_fetcher/evidence_file_data.py new file mode 100644 index 00000000..8ecc2441 --- /dev/null +++ b/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"), + ) diff --git a/backend/pashub_fetcher/evidence_metadata.py b/backend/pashub_fetcher/evidence_metadata.py new file mode 100644 index 00000000..e3a9536e --- /dev/null +++ b/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"], + ) diff --git a/scripts/download_cotality_evidence.py b/scripts/download_cotality_evidence.py index 43f9afea..93148a3a 100644 --- a/scripts/download_cotality_evidence.py +++ b/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"