mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
735e83cef2
Behaviour-preserving lift-and-shift of the condition module out of legacy backend/condition/ into domain/condition, infrastructure/condition, infrastructure/postgres/condition_tables.py, repositories/condition, applications/condition, and tests/condition. Imports rewritten to the DDD paths; ConditionPostgres and the ORM models keep the legacy backend.app.db Base/db_session bridges so the existing suite proves behaviour is unchanged (SQLModel + session-DI conversion tracked as a follow-up in ADR-0064). 16 condition tests pass at the new location. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
from pathlib import Path
|
|
|
|
from applications.condition.condition_trigger_request import ConditionFileType
|
|
from infrastructure.condition.lookups.uprn_lookup_csv import UprnLookupLocal
|
|
from applications.condition.processor import process_file
|
|
|
|
|
|
def main() -> None:
|
|
try:
|
|
# Works in scripts / debugger / pytest
|
|
ROOT_DIR = Path(__file__).resolve().parents[1]
|
|
except NameError:
|
|
# __file__ is not defined in notebooks
|
|
ROOT_DIR = Path.cwd()
|
|
|
|
path: Path = ROOT_DIR / "condition" / "sample_data"
|
|
|
|
# TODO: get these from s3, maybe as part of devcontainer init
|
|
lbwf_path: Path = path / "lbwf" / "LBWF - Example Asset Data September 2025.xlsx"
|
|
peabody_path: Path = (
|
|
path
|
|
/ "peabody"
|
|
/ "2026_01_06 - Peabody - Stock Condition Data - Survey Records - D Lower.xlsx"
|
|
)
|
|
peabody_uprn_lookup_path: Path = (
|
|
path / "peabody" / "PeabodyPropertymatched_Dec25_propref_UPRN.csv"
|
|
)
|
|
# filepaths = [lbwf_path, peabody_path]
|
|
filepaths = [lbwf_path]
|
|
# filepaths = [peabody_path]
|
|
|
|
uprn_lookup = UprnLookupLocal(csv_path=peabody_uprn_lookup_path.as_posix())
|
|
|
|
def get_file_type(file_path: str) -> ConditionFileType:
|
|
if "peabody" in file_path:
|
|
return ConditionFileType.Peabody
|
|
if "lbwf" in file_path:
|
|
return ConditionFileType.LBWF
|
|
|
|
for fp in filepaths:
|
|
with fp.open("rb") as f:
|
|
process_file(
|
|
file_stream=f,
|
|
file_type=get_file_type(fp.as_posix()),
|
|
uprn_lookup=uprn_lookup,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|