viktor/wrongmove
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

reformat with black; looks better

Browse source
This commit is contained in:
Viktor Barzin 2025-05-31 23:50:43 +00:00
parent 1122f5a96f
commit 0b9d50af47
No known key found for this signature in database
GPG key ID: 4056458DBDBF8863
11 changed files with 240 additions and 244 deletions
Download patch file Download diff file Expand all files Collapse all files

213
crawler/data_access.py
View file

@ -46,11 +46,13 @@ class Listing:
# data_dir is the first directory before the listing_path
data_dir = pathlib.Path(listing_path)
while str(d['identifier']) in str(data_dir.resolve().absolute()):
while str(d["identifier"]) in str(data_dir.resolve().absolute()):
data_dir = data_dir.parent
listing = Listing(d["identifier"], data_dir=data_dir)
if (listing.last_seen is not None
and listing.last_seen < seen_in_the_last_n_days):
if (
listing.last_seen is not None
and listing.last_seen < seen_in_the_last_n_days
):
identifiers.append(listing)
return identifiers
@ -107,16 +109,19 @@ class Listing:
def calculate_sqm_model(self):
objs = []
for floorplan_path in self.list_floorplans():
estimated_sqm, model_output, predictions = (
floorplan.calculate_model(floorplan_path))
objs.append({
"floorplan_path": str(floorplan_path),
"estimated_sqm": estimated_sqm,
"model_output": model_output,
"no_predictions": len(
predictions
), # cant serialize the predictions itself since its a tensor
})
estimated_sqm, model_output, predictions = floorplan.calculate_model(
floorplan_path
)
objs.append(
{
"floorplan_path": str(floorplan_path),
"estimated_sqm": estimated_sqm,
"model_output": model_output,
"no_predictions": len(
predictions
), # cant serialize the predictions itself since its a tensor
}
)
with open(self.path_floorplan_model_json(), "w") as f:
json.dump(objs, f)
@ -129,8 +134,9 @@ class Listing:
with open(self.path_floorplan_json()) as f:
objs = json.load(f)
max_sqm = max([o["estimated_sqm"] for o in objs
if o is None]) # filter out Nones
max_sqm = max(
[o["estimated_sqm"] for o in objs if o is None]
) # filter out Nones
return max_sqm
async def calculate_sqm_ocr(self, recalculate=True):
@ -143,12 +149,15 @@ class Listing:
for floorplan_path in self.list_floorplans():
estimated_sqm, model_output = await asyncio.to_thread(
floorplan.calculate_ocr, floorplan_path)
objs.append({
"floorplan_path": str(floorplan_path),
"estimated_sqm": estimated_sqm,
"text": model_output,
})
floorplan.calculate_ocr, floorplan_path
)
objs.append(
{
"floorplan_path": str(floorplan_path),
"estimated_sqm": estimated_sqm,
"text": model_output,
}
)
with open(self.path_floorplan_ocr_json(), "w") as f:
json.dump(objs, f)
@ -160,22 +169,20 @@ class Listing:
with open(self.path_floorplan_ocr_json()) as f:
objs = json.load(f)
sqms = [
o["estimated_sqm"] for o in objs if o["estimated_sqm"] is not None
]
sqms = [o["estimated_sqm"] for o in objs if o["estimated_sqm"] is not None]
if len(sqms) == 0:
return None
max_sqm = max(sqms)
return max_sqm
def calculate_route(self,
dest_address: str,
travel_mode: routing.TravelMode,
recalculate=False) -> dict[str, Any]:
def calculate_route(
self, dest_address: str, travel_mode: routing.TravelMode, recalculate=False
) -> dict[str, Any]:
routing_cache = self.__get_routing_cache()
cache_key = self.__routing_cache_key(dest_address, travel_mode)
if (route_cache :=
routing_cache.get(cache_key)) is not None and not recalculate:
if (
route_cache := routing_cache.get(cache_key)
) is not None and not recalculate:
return {cache_key: route_cache}
result = routing.transit_route(
@ -185,8 +192,12 @@ class Listing:
travel_mode,
)
if not result:
raise Exception((f"Error calculating route from {self.identifier} "
f"to '{dest_address}' by {travel_mode}"))
raise Exception(
(
f"Error calculating route from {self.identifier} "
f"to '{dest_address}' by {travel_mode}"
)
)
result = {**{cache_key: result}, **routing_cache}
with open(self.path_routing_json(), "w") as f:
json.dump(result, f)
@ -198,8 +209,7 @@ class Listing:
travel_mode: routing.TravelMode,
) -> list[dict[str, Any]]:
data = self.calculate_route(destination_address, travel_mode)
return self.__extract_travel_times(data, destination_address,
travel_mode)
return self.__extract_travel_times(data, destination_address, travel_mode)
@property
def url(self):
@ -246,8 +256,7 @@ class Listing:
@property
def leaseLeft(self) -> float | None:
ds = self.detailobject["property"].get("tenureInfo",
{}).get("content", [])
ds = self.detailobject["property"].get("tenureInfo", {}).get("content", [])
for d in ds:
if d["type"] == "lengthOfLease":
matches = re.findall(r"(\d+\.?\d*)", d["value"])
@ -267,15 +276,14 @@ class Listing:
if not self.path_last_seen_listing().exists():
return None
with open(self.path_last_seen_listing(), 'r') as f:
with open(self.path_last_seen_listing(), "r") as f:
datetime_str = json.load(f)
dt = datetime.datetime.fromisoformat(datetime_str)
return (datetime.datetime.now() - dt).days
@property
def serviceCharge(self) -> float | None:
ds = self.detailobject["property"].get("tenureInfo",
{}).get("content", [])
ds = self.detailobject["property"].get("tenureInfo", {}).get("content", [])
for d in ds:
if d["type"] == "annualServiceCharge":
matches = re.findall(r"([\d,.]+)", d["value"])
@ -300,25 +308,24 @@ class Listing:
@property
def status(self) -> str:
if self.isRemoved:
return 'removed'
return "removed"
status = self.detailobject["property"]["status"]
return status
@property
def agency(self) -> str:
return self.detailobject['property']["branch"]["brandName"]
return self.detailobject["property"]["branch"]["brandName"]
@property
def councilTaxBand(self) -> str:
return self.detailobject['property']["councilTaxInfo"]["content"][0][
"value"]
return self.detailobject["property"]["councilTaxInfo"]["content"][0]["value"]
@property
def photoThumbnail(self) -> str | None:
# options are: 'url', 'thumbnailUrl', 'maxSizeUrl'
photos = self.detailobject['property']['photos']
photos = self.detailobject["property"]["photos"]
if len(photos) > 0:
return photos[0]['url']
return photos[0]["url"]
return None
async def dict_nicely(self):
@ -328,57 +335,48 @@ class Listing:
with open(self.path_routing_json(), "r") as f:
travel_times = json.load(f)
for destination_mode in travel_times.keys():
destination_mode_clean = destination_mode.replace(" ",
"_").replace(
",", "_")
destination_mode_clean = destination_mode.replace(" ", "_").replace(
",", "_"
)
destination, travel_mode = self.__from_routing_cache_key(
destination_mode)
destination_mode
)
travel_time_fastest[destination_mode_clean] = self.travel_time(
destination, travel_mode)[0]['duration']
destination, travel_mode
)[0]["duration"]
travel_time_second[destination_mode_clean] = self.travel_time(
destination, travel_mode)[1]['duration']
destination, travel_mode
)[1]["duration"]
return {
"identifier":
self.identifier,
"sqm_ocr":
await self.sqm_ocr(),
"price":
self.price,
"price_per_sqm":
await self.price_per_sqm(),
"url":
self.url,
"bedrooms":
self.bedrooms,
"travel_time_fastest":
":".join(
sorted(f'{dest} in {travel_mode//60}min'
for dest, travel_mode in travel_time_fastest.items())),
"travel_time_second":
":".join(
sorted(f'{dest} in {travel_mode//60}min'
for dest, travel_mode in travel_time_second.items())),
"lease_left":
self.leaseLeft,
"service_charge":
self.serviceCharge,
"development":
self.development,
"tenure_type":
self.tenure_type,
"updated_days":
self.updateDaysAgo,
"status":
self.status,
"last_seen":
self.last_seen,
"agency":
self.agency,
"council_tax_band":
self.councilTaxBand,
"photo_thumbnail":
self.photoThumbnail,
"identifier": self.identifier,
"sqm_ocr": await self.sqm_ocr(),
"price": self.price,
"price_per_sqm": await self.price_per_sqm(),
"url": self.url,
"bedrooms": self.bedrooms,
"travel_time_fastest": ":".join(
sorted(
f"{dest} in {travel_mode//60}min"
for dest, travel_mode in travel_time_fastest.items()
)
),
"travel_time_second": ":".join(
sorted(
f"{dest} in {travel_mode//60}min"
for dest, travel_mode in travel_time_second.items()
)
),
"lease_left": self.leaseLeft,
"service_charge": self.serviceCharge,
"development": self.development,
"tenure_type": self.tenure_type,
"updated_days": self.updateDaysAgo,
"status": self.status,
"last_seen": self.last_seen,
"agency": self.agency,
"council_tax_band": self.councilTaxBand,
"photo_thumbnail": self.photoThumbnail,
}
def __routing_cache_key(
@ -420,35 +418,38 @@ class Listing:
for step in steps:
if not used_transit and step["travelMode"] == "WALK":
initial_walk_duration += int(
step["staticDuration"].strip("s"))
initial_walk_duration += int(step["staticDuration"].strip("s"))
else:
used_transit = True
duration_per_transit[step["travelMode"]] += int(
step["staticDuration"].strip("s"))
step["staticDuration"].strip("s")
)
distance_per_transit[step["travelMode"]] += step.get(
"distanceMeters", 0)
"distanceMeters", 0
)
if step["travelMode"] == "TRANSIT":
number_of_transit_stops += 1
res.append({
"duration": duration,
"distance": distance,
"duration_static": duration_static,
"initial_walk_duration": initial_walk_duration,
"duration_per_transit": dict(duration_per_transit),
"distance_per_transit": dict(distance_per_transit),
"number_of_transit_stops": number_of_transit_stops,
})
res.append(
{
"duration": duration,
"distance": distance,
"duration_static": duration_static,
"initial_walk_duration": initial_walk_duration,
"duration_per_transit": dict(duration_per_transit),
"distance_per_transit": dict(distance_per_transit),
"number_of_transit_stops": number_of_transit_stops,
}
)
return res[:limit]
def __get_routing_cache(self) -> dict[str, Any]:
try:
with open(self.path_routing_json(), 'x') as f:
with open(self.path_routing_json(), "x") as f:
json.dump({}, f)
return {}
except FileExistsError:
pass
with open(self.path_routing_json(), 'r') as f:
with open(self.path_routing_json(), "r") as f:
return json.load(f)