mirror of
https://github.com/Hestia-Homes/survey-extraction.git
synced 2026-06-08 11:17:29 +00:00
103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
from sqlmodel import Field, SQLModel
|
|
from sqlalchemy import Column
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
import uuid
|
|
from pydantic import Field, field_validator, model_validator
|
|
from etl.utils.utils import get_sharepoint_path
|
|
from etl.scraper.scraper import SharePointScraper, SharePointInstaller
|
|
from etl.surveyedData.surveryedData import surveyedDataProcessor
|
|
|
|
|
|
|
|
|
|
|
|
def string_to_installer(installer):
|
|
if installer.upper() == "J & J CRUMP":
|
|
return SharePointInstaller.JJC
|
|
elif installer.upper() == "SCIS":
|
|
return SharePointInstaller.SOUTH_COAST_INSULATION
|
|
elif installer.upper() == "SGEC":
|
|
return SharePointInstaller.JJC
|
|
else:
|
|
return None
|
|
|
|
|
|
class BaseModel(SQLModel):
|
|
id: uuid.UUID = Field(
|
|
default_factory=uuid.uuid4,
|
|
sa_column=Column(UUID(as_uuid=True), primary_key=True)
|
|
)
|
|
|
|
|
|
class SubmissionInfoFromDeal(BaseModel):
|
|
deal_id: str = Field(..., min_length=1)
|
|
deal_name: str = Field(..., min_length=1)
|
|
work_type: str = Field(..., min_length=1)
|
|
needs_trickle_ventilation: bool
|
|
post_sap_score: int
|
|
existing_wall_insulation: str = Field(..., min_length=1)
|
|
no_of_wet_rooms: int
|
|
installer: str = Field(..., min_length=1)
|
|
submission_folder_path: str = Field(..., min_length=1)
|
|
landlord_id: str = Field(..., min_length=1)
|
|
domna_id: str = Field(..., min_length=1)
|
|
uprn: str
|
|
|
|
@field_validator('post_sap_score', 'no_of_wet_rooms')
|
|
@classmethod
|
|
def must_be_non_negative(cls, v, info):
|
|
if v < 0:
|
|
raise ValueError(f"{info.field_name} must be non-negative")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def check_sharepoint_link_and_contents(self):
|
|
try:
|
|
path = get_sharepoint_path(self.submission_folder_path)
|
|
installer = string_to_installer(self.installer)
|
|
sp = SharePointScraper(installer)
|
|
except Exception as e:
|
|
raise ValueError(f"Error accessing SharePoint path: {self.submission_folder_path}. Error: {str(e)}")
|
|
|
|
try:
|
|
# Check if sharepoint link is reachable and has any contents
|
|
files = sp.get_folders_in_path(path)
|
|
if "value" in files and len(files["value"]) > 0:
|
|
pass
|
|
else:
|
|
raise ValueError(f"SharePoint folder is empty: {self.submission_folder_path}")
|
|
except Exception as e:
|
|
raise ValueError(str(e))
|
|
|
|
# download files in url and check files are there:
|
|
try:
|
|
|
|
files = sp.download_files_from_path(path)
|
|
print(files)
|
|
sdp = surveyedDataProcessor("fake address", files)
|
|
missing_items = []
|
|
|
|
if sdp.condition_report is None:
|
|
missing_items.append("Condition Report")
|
|
|
|
if sdp.energy_report is None:
|
|
missing_items.append("Energy Report PDF")
|
|
|
|
if sdp.rd_sap_xml is None:
|
|
missing_items.append("RDSAP XML")
|
|
|
|
if sdp.lig_sap_xml is None:
|
|
missing_items.append("LIG SAP XML")
|
|
|
|
if missing_items:
|
|
raise ValueError(f"Missing required items: {', '.join(missing_items)}")
|
|
|
|
except Exception as e:
|
|
raise ValueError(str(e))
|
|
|
|
return self
|
|
|
|
|
|
|
|
|
|
|