houses: nightly Kamernet + Vesteda scraper, dedupe by address, owner-only /trackers/houses

- scripts/house-scrape.ts: polite nightly scrape (honest UA, 6-10 s apart, stops a source on 403/challenge), ingest + price history + removals, ntfy alerts on a separate topic (max 5 + summary, first run seeds)
- address_key + duplicate_of: copies on another source link to one listing, manual wins
- /projects/house-tracker -> /trackers/houses (308), owner only (404 otherwise); filters for min m2, rooms, source, Gouda region default; notes, add-by-hand form, total rent, also-on links
- scripts/house-db.sh: run the house-listing CLI as the limited house_cli role
- deploy/systemd/house-scrape.{service,timer}

src/lib/house-tracker/ingest.ts

@@ -0,0 +1,124 @@
+// Writes scraped listings into house.listings with plain postgres.js, for the nightly
+// scraper on stan-vm (scripts/house-scrape.ts). No #lib / $app imports.
+import type postgres from 'postgres';
+import { addressKey, addressKeyFromText, citySlug, DEDUPE_SQL } from './address';
+import type { ScrapedListing } from './scraped';
+
+type Sql = postgres.Sql;
+
+export type IngestResult = {
+	seeded: boolean;
+	inserted: { id: number; listing: ScrapedListing }[];
+	updated: number;
+	removed: number;
+};
+
+export async function ingestListings(
+	sql: Sql,
+	source: string,
+	listings: ScrapedListing[],
+	opts: { runStartedAt: Date; completed: boolean }
+): Promise<IngestResult> {
+	// The same listing can show up on two city pages; keep the first.
+	const unique = [...new Map(listings.map((l) => [l.sourceId, l])).values()];
+	return sql.begin(async (tx) => {
+		const [any] = await tx`SELECT 1 FROM house.listings WHERE source = ${source} LIMIT 1`;
+		const seeded = !any;
+		const ids = unique.map((l) => l.sourceId);
+		const existing = ids.length
+			? await tx<{ id: number; source_id: string; price_eur: number | null }[]>`
+				SELECT id, source_id, price_eur FROM house.listings
+				WHERE source = ${source} AND source_id IN ${tx(ids)}`
+			: [];
+		const known = new Map(existing.map((row) => [row.source_id, row]));
+		const inserted: IngestResult['inserted'] = [];
+		let updated = 0;
+
+		for (const l of unique) {
+			const columns = {
+				url: l.url,
+				title: l.title,
+				street: l.houseNumber ? `${l.street ?? ''} ${l.houseNumber}`.trim() : l.street,
+				postal_code: l.postalCode,
+				city: l.city,
+				city_slug: citySlug(l.city),
+				price_eur: l.priceEur,
+				total_price_eur: l.totalPriceEur,
+				area_m2: l.areaM2,
+				rooms: l.rooms,
+				bedrooms: l.bedrooms,
+				energy_label: l.energyLabel,
+				available_from: l.availableFrom,
+				interior: l.interior,
+				agent: l.agent,
+				image_url: l.imageUrl,
+				lat: l.lat,
+				lng: l.lng,
+				features: tx.json(l.features),
+				address_key: addressKey(l.street, l.houseNumber, l.city)
+			};
+			const previous = known.get(l.sourceId);
+			if (previous) {
+				await tx`UPDATE house.listings SET ${tx(columns)}, last_seen_at = now(), removed_at = NULL
+					WHERE id = ${previous.id}`;
+				if (l.priceEur != null && l.priceEur !== previous.price_eur)
+					await tx`INSERT INTO house.price_history (listing_id, price_eur) VALUES (${previous.id}, ${l.priceEur})`;
+				updated++;
+				continue;
+			}
+			const [row] = await tx<{ id: number }[]>`INSERT INTO house.listings ${tx({
+				...columns,
+				source,
+				source_id: l.sourceId,
+				offer: l.offer,
+				property_type: l.propertyType,
+				first_seen_lane: seeded ? 'seed' : 'nightly',
+				// Everything the source offers is already in the search result.
+				detail_fetched_at: new Date()
+			})} ON CONFLICT (source, source_id) DO NOTHING RETURNING id`;
+			if (!row) continue;
+			if (l.priceEur != null)
+				await tx`INSERT INTO house.price_history (listing_id, price_eur) VALUES (${row.id}, ${l.priceEur})`;
+			inserted.push({ id: row.id, listing: l });
+		}
+
+		let removed = 0;
+		if (opts.completed) {
+			const gone = await tx`UPDATE house.listings SET removed_at = now()
+				WHERE source = ${source} AND removed_at IS NULL AND last_seen_at < ${opts.runStartedAt}
+				RETURNING id`;
+			removed = gone.length;
+		}
+		return { seeded, inserted, updated, removed };
+	});
+}
+
+/**
+ * Fills address_key for rows written elsewhere (manual CLI, Pararius worker) from their
+ * street or title, then marks cross-source copies. Safe to run any time.
+ */
+export async function dedupe(sql: Sql): Promise<{ keyed: number; copies: number }> {
+	const rows = await sql<
+		{
+			id: number;
+			source: string;
+			street: string | null;
+			title: string | null;
+			city: string | null;
+		}[]
+	>`SELECT id, source, street, title, city FROM house.listings
+		WHERE address_key IS NULL AND (street IS NOT NULL OR source = 'manual')`;
+	let keyed = 0;
+	for (const row of rows) {
+		const key =
+			addressKeyFromText(row.street, row.city) ??
+			(row.source === 'manual' ? addressKeyFromText(row.title, row.city) : null);
+		if (!key) continue;
+		await sql`UPDATE house.listings SET address_key = ${key} WHERE id = ${row.id}`;
+		keyed++;
+	}
+	for (const statement of DEDUPE_SQL) await sql.unsafe(statement);
+	const [{ copies }] = await sql<{ copies: number }[]>`
+		SELECT count(*)::int AS copies FROM house.listings WHERE duplicate_of IS NOT NULL`;
+	return { keyed, copies };
+}