"""Alba Core orchestration pipeline.
This module wires the subsystem flow together:
``message -> extraction -> matching -> lead summary``.
It is the best first file for new contributors because it shows the full
request path without hiding business rules inside a framework or chat wrapper.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
from app.alba_core.extraction import RequirementExtractor
from app.alba_core.leads import build_lead_summary
from app.alba_core.matching import PropertyMatcher
from app.alba_core.models import RentalRequirements
from app.propertyme.cache import load_property_cache
[docs]
class AlbaCore:
"""Coordinates the full Alba Core search flow.
Read this file first when coding along. It shows the whole sequence while
keeping detailed business rules in extraction.py, matching.py, and leads.py.
"""
def __init__(self, property_cache_path: Path) -> None:
self.property_cache_path = property_cache_path
self.properties = load_property_cache(property_cache_path)
# Known locations come from real PropertyMe-backed records. This prevents
# the extractor from confidently accepting places that do not exist in
# Alba's current property cache.
known_locations = {
value
for property_record in self.properties
for value in [property_record.city, property_record.suburb, property_record.region]
if value
}
self.extractor = RequirementExtractor(known_locations)
self.matcher = PropertyMatcher(top_n=3)
[docs]
@classmethod
def from_environment(cls) -> "AlbaCore":
"""Create Alba Core from ALBA_PROPERTY_CACHE or the default real seed."""
path = Path(os.getenv("ALBA_PROPERTY_CACHE", "data/propertyme_property_seed_latest.json"))
return cls(path)
[docs]
def search(
self,
message: str,
current_requirements: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Run one customer message through the complete Alba Core flow."""
current = RentalRequirements(**current_requirements) if current_requirements else None
extraction = self.extractor.extract(message, current=current)
matching = self.matcher.match(extraction.requirements, self.properties)
lead_summary = build_lead_summary(extraction.requirements, matching.matches)
return {
"extraction": extraction.to_dict(),
"search_ready": extraction.requirements.is_search_ready,
"match_count": len(matching.matches),
"matches": [match.to_dict() for match in matching.matches],
"rejected_reasons": matching.rejected_reasons,
"lead_summary": lead_summary,
}