← back to Professional Directory
agents/ingest-agent/importers/osm-smallbiz.js
391 lines
#!/usr/bin/env node
/**
* OSM Overpass — small-business importer.
*
* Pulls LA County storefronts for the categories Steve sells themed websites to:
* • Cosmetology (hair, beauty, barbershop, nail salon)
* • Chiropractic offices
* • Chinese medicine / acupuncture / TCM
*
* OSM is the right surface (not licensee registries) because the customer is
* the BUSINESS owner — we want storefronts with addresses, phones, websites.
* Free, no key, polite-use rate limit per OSM Foundation operations policy.
*
* Each category writes a distinct organizations.type so the API can filter:
* salon | barbershop | nail_salon | beauty_supply
* chiropractor_office
* acupuncture_clinic | tcm_clinic
*
* Run:
* node agents/ingest-agent/importers/osm-smallbiz.js # all three
* CATEGORY=cosmetology node agents/ingest-agent/importers/osm-smallbiz.js
* CATEGORY=chiropractic ...
* CATEGORY=chinese_medicine ...
*/
const crypto = require('node:crypto');
const { fetch } = require('undici');
const { pool, query, withTx } = require('../../shared/db');
const SOURCE_NAME = 'OpenStreetMap Overpass API';
const OVERPASS = process.env.OVERPASS_URL || 'https://overpass-api.de/api/interpreter';
const USER_AGENT = process.env.USER_AGENT
|| 'ProfessionalDirectoryBot/0.1 (small-biz directory; contact: steve@designerwallcoverings.com)';
// ─── Overpass queries (LA County = Wikidata Q104994) ────────────────────────
const QUERIES = {
cosmetology: `
[out:json][timeout:90];
area["wikidata"="Q104994"]->.la;
(
nwr["shop"="hairdresser"](area.la);
nwr["shop"="beauty"](area.la);
nwr["shop"="cosmetics"](area.la);
nwr["shop"="nails"](area.la);
nwr["amenity"="beauty_salon"](area.la);
nwr["leisure"="spa"](area.la);
);
out center tags;
`.trim(),
dental: `
[out:json][timeout:90];
area["wikidata"="Q104994"]->.la;
(
nwr["healthcare"="dentist"](area.la);
nwr["amenity"="dentist"](area.la);
nwr["office"="dentist"](area.la);
);
out center tags;
`.trim(),
optometry: `
[out:json][timeout:90];
area["wikidata"="Q104994"]->.la;
(
nwr["healthcare"="optometrist"](area.la);
nwr["shop"="optician"](area.la);
nwr["amenity"="optometrist"](area.la);
);
out center tags;
`.trim(),
fitness: `
[out:json][timeout:90];
area["wikidata"="Q104994"]->.la;
(
nwr["leisure"="fitness_centre"](area.la);
nwr["sport"="yoga"](area.la);
nwr["leisure"="sports_centre"]["sport"~"yoga|pilates"](area.la);
nwr["amenity"="gym"](area.la);
);
out center tags;
`.trim(),
tattoo: `
[out:json][timeout:90];
area["wikidata"="Q104994"]->.la;
(
nwr["shop"="tattoo"](area.la);
nwr["shop"="piercing"](area.la);
);
out center tags;
`.trim(),
pets: `
[out:json][timeout:90];
area["wikidata"="Q104994"]->.la;
(
nwr["shop"="pet_grooming"](area.la);
nwr["shop"="pet"](area.la);
nwr["amenity"="veterinary"](area.la);
nwr["healthcare"="veterinary"](area.la);
);
out center tags;
`.trim(),
// Chiropractic — name regex catches the long tail; OSM tagging is inconsistent.
chiropractic: `
[out:json][timeout:90];
area["wikidata"="Q104994"]->.la;
nwr["name"~"chiropract",i](area.la);
out center tags;
`.trim(),
// Chinese medicine — Overpass union semantics misbehave when name-regex and
// structural tag queries combine, so we run pieces and merge client-side.
chinese_medicine: [
`[out:json][timeout:60];area["wikidata"="Q104994"]->.la;nwr["name"~"acupunctur",i](area.la);out center tags;`,
`[out:json][timeout:60];area["wikidata"="Q104994"]->.la;nwr["name"~"chinese medicin",i](area.la);out center tags;`,
`[out:json][timeout:60];area["wikidata"="Q104994"]->.la;nwr["name"~"oriental medicin",i](area.la);out center tags;`,
`[out:json][timeout:60];area["wikidata"="Q104994"]->.la;nwr["name"~"herbal",i](area.la);out center tags;`,
`[out:json][timeout:60];area["wikidata"="Q104994"]->.la;nwr["healthcare"="alternative"](area.la);out center tags;`,
],
};
// ─── Type mapping per element tags ──────────────────────────────────────────
function typeForTags(category, tags) {
if (category === 'cosmetology') {
if (tags.shop === 'nails') return 'nail_salon';
if (tags.shop === 'cosmetics') return 'beauty_supply';
if (tags.shop === 'hairdresser' && /barber/i.test(tags.name || '')) return 'barbershop';
if (tags.shop === 'hairdresser') return 'salon';
if (tags.shop === 'beauty') return 'salon';
if (tags.amenity === 'beauty_salon') return 'salon';
if (tags.leisure === 'spa') return 'spa';
return 'salon';
}
if (category === 'chiropractic') return 'chiropractor_office';
if (category === 'chinese_medicine') {
if (tags.alternative === 'acupuncture' || /acupunc/i.test(tags.name || '')) return 'acupuncture_clinic';
if (tags.shop === 'herbalist') return 'tcm_herbalist';
return 'tcm_clinic';
}
if (category === 'dental') return 'dental_office';
if (category === 'optometry') {
if (tags.shop === 'optician') return 'optical_shop';
return 'optometrist_office';
}
if (category === 'fitness') {
if (tags.sport === 'yoga' || /yoga/i.test(tags.name || '')) return 'yoga_studio';
if (/pilates/i.test(tags.name || '') || tags.sport === 'pilates') return 'pilates_studio';
return 'gym';
}
if (category === 'tattoo') {
if (tags.shop === 'piercing') return 'piercing_studio';
return 'tattoo_studio';
}
if (category === 'pets') {
if (tags.amenity === 'veterinary' || tags.healthcare === 'veterinary') return 'vet_clinic';
if (tags.shop === 'pet_grooming') return 'pet_grooming';
return 'pet_shop';
}
return 'small_biz';
}
// ─── Helpers ────────────────────────────────────────────────────────────────
function sha256(s) { return crypto.createHash('sha256').update(s).digest('hex'); }
function clean(s) {
if (s === undefined || s === null) return null;
const t = String(s).trim();
return t.length === 0 ? null : t;
}
function buildAddress(t) {
const street = [t['addr:housenumber'], t['addr:street']].filter(Boolean).join(' ');
const unit = t['addr:unit'] ? ` Suite ${t['addr:unit']}` : '';
const line1 = clean(street + unit);
const city = clean(t['addr:city']);
const state = clean(t['addr:state']) || 'CA';
const zip = clean(t['addr:postcode']);
const full = (line1 || city) ? [line1, city, state, zip].filter(Boolean).join(', ') : null;
return { line1, city, state, zip, full };
}
function normAddress(s) {
if (!s) return null;
return s.toLowerCase().replace(/[.,]/g, ' ').replace(/\b(suite|ste|unit|apt|#)\b/g, '')
.replace(/\s+/g, ' ').trim();
}
async function fetchOverpass(ql) {
const body = `data=${encodeURIComponent(ql)}`;
const r = await fetch(OVERPASS, {
method: 'POST',
headers: {
'User-Agent': USER_AGENT,
'Content-Type': 'application/x-www-form-urlencoded',
Accept: 'application/json',
},
body,
signal: AbortSignal.timeout(180000),
});
if (!r.ok) {
const txt = await r.text();
throw new Error(`Overpass ${r.status}: ${txt.slice(0, 400)}`);
}
const json = await r.json();
return json.elements || [];
}
async function getSourceId() {
const r = await query('SELECT id FROM sources WHERE source_name = $1', [SOURCE_NAME]);
if (r.rowCount === 0) throw new Error(`source not seeded: ${SOURCE_NAME}`);
return r.rows[0].id;
}
async function startJob(sourceId, label) {
const r = await query(
`INSERT INTO scrape_jobs (source_id, job_label, status, started_at)
VALUES ($1,$2,'running',NOW()) RETURNING id`,
[sourceId, label]
);
return r.rows[0].id;
}
async function finishJob(jobId, fields) {
const sets = []; const params = []; let i = 1;
for (const [k, v] of Object.entries(fields)) { sets.push(`${k} = $${i++}`); params.push(v); }
sets.push(`finished_at = NOW()`); params.push(jobId);
await query(`UPDATE scrape_jobs SET ${sets.join(', ')} WHERE id = $${i}`, params);
}
async function upsert(el, category, sourceId) {
const tags = el.tags || {};
const name = clean(tags.name);
if (!name) return null;
const lat = el.lat ?? el.center?.lat ?? null;
const lng = el.lon ?? el.center?.lon ?? null;
const addr = buildAddress(tags);
const phone = clean(tags.phone) || clean(tags['contact:phone']);
const website = clean(tags.website) || clean(tags['contact:website']);
const email = clean(tags.email) || clean(tags['contact:email']);
const sourceUrl = `https://www.openstreetmap.org/${el.type}/${el.id}`;
const type = typeForTags(category, tags);
return await withTx(async (client) => {
// Find existing by website first (stable), then name + address_norm + zip.
let orgId = null;
if (website) {
const r = await client.query(
`SELECT id FROM organizations WHERE LOWER(website) = LOWER($1) LIMIT 1`, [website]);
if (r.rowCount) orgId = r.rows[0].id;
}
if (!orgId && addr.full) {
const r = await client.query(
`SELECT id FROM organizations
WHERE LOWER(name) = LOWER($1)
AND zip IS NOT DISTINCT FROM $2
LIMIT 1`,
[name, addr.zip]);
if (r.rowCount) orgId = r.rows[0].id;
}
if (orgId) {
await client.query(`
UPDATE organizations SET
name = $2,
type = COALESCE(NULLIF($3,''), type),
address = COALESCE($4, address),
city = COALESCE($5, city),
zip = COALESCE($6, zip),
county = COALESCE($7, county),
phone = COALESCE($8, phone),
website = COALESCE($9, website),
lat = COALESCE($10::double precision, lat),
lng = COALESCE($11::double precision, lng),
geocoded_at = CASE WHEN $10::double precision IS NOT NULL
AND $11::double precision IS NOT NULL THEN NOW()
ELSE geocoded_at END,
updated_at = NOW()
WHERE id = $1
`, [orgId, name, type, addr.full, addr.city, addr.zip, 'Los Angeles',
phone, website, lat, lng]);
} else {
const r = await client.query(`
INSERT INTO organizations
(name, type, address, city, state, zip, county, phone, website,
lat, lng, geocoded_at)
VALUES ($1,$2,$3,$4,'CA',$5,'Los Angeles',$6,$7,
$8::double precision, $9::double precision,
CASE WHEN $8::double precision IS NOT NULL
AND $9::double precision IS NOT NULL THEN NOW() END)
RETURNING id
`, [name, type, addr.full, addr.city, addr.zip, phone, website, lat, lng]);
orgId = r.rows[0].id;
}
if (email) {
await client.query(`
INSERT INTO emails (organization_id, email, email_type, source_url, last_verified_at)
VALUES ($1,$2,'office',$3,NOW())
`, [orgId, email, sourceUrl]);
}
// Provenance.
const json = JSON.stringify({ ...el, tags });
const hash = sha256(json + '|organization|' + orgId);
await client.query(`
INSERT INTO raw_records (source_id, source_url, entity_type, entity_id, raw_json, hash)
VALUES ($1,$2,'organization',$3,$4::jsonb,$5)
ON CONFLICT (hash) DO NOTHING
`, [sourceId, sourceUrl, orgId, json, hash]);
return orgId;
});
}
async function runCategory(category, sourceId) {
const ql = QUERIES[category];
if (!ql) throw new Error(`unknown category: ${category}`);
console.log(`[osm:${category}] querying Overpass…`);
const jobId = await startJob(sourceId, `osm:smallbiz:${category}`);
// ql can be a single string or an array of strings (multi-query union).
const queries = Array.isArray(ql) ? ql : [ql];
let elements = [];
const seenKey = new Set();
try {
for (const q of queries) {
const els = await fetchOverpass(q);
for (const el of els) {
const k = `${el.type}/${el.id}`;
if (seenKey.has(k)) continue;
seenKey.add(k);
elements.push(el);
}
if (queries.length > 1) await new Promise(res => setTimeout(res, 1000));
}
} catch (e) {
await finishJob(jobId, { status: 'failed', error_message: e.message });
throw e;
}
console.log(`[osm:${category}] received ${elements.length} elements (deduped)`);
let inserted = 0, skipped = 0;
for (const el of elements) {
try {
const id = await upsert(el, category, sourceId);
if (id) inserted++; else skipped++;
} catch (e) {
console.error(`[osm:${category}] upsert err ${el.type}/${el.id}: ${e.message}`);
skipped++;
}
}
await finishJob(jobId, {
status: 'done',
records_found: elements.length,
records_inserted: inserted,
records_skipped: skipped,
});
console.log(`[osm:${category}] done. seen=${elements.length} kept=${inserted} skipped=${skipped}`);
return { category, seen: elements.length, kept: inserted, skipped };
}
async function main() {
const sourceId = await getSourceId();
const only = process.env.CATEGORY ? [process.env.CATEGORY] : Object.keys(QUERIES);
const results = [];
for (const cat of only) {
const r = await runCategory(cat, sourceId);
results.push(r);
// Polite: 2s between bulk Overpass queries.
await new Promise(res => setTimeout(res, 2000));
}
console.table(results);
await pool.end();
}
main().catch(async (err) => {
console.error('[osm] fatal:', err);
try { await pool.end(); } catch (_) {}
process.exit(1);
});