SPB Git forge

spb/websensor

Public
33commits 1branches 0releases
3.4 MBsize
maindefault branch
10 days agolast push
TypeScript 55.4% Python 43.2% SQL 1.2%
8.2 KB · 169 lines typescript
Raw Blame History
1import { existsSync, readdirSync, readFileSync, statSync } from "node:fs";2import { join } from "node:path";3import YAML from "yaml";4import { coverageKey, coverageSectorSchema, slugify, type CoverageMember, type CoverageSector } from "@websensor/core";5import { db, factorySeeds, sql, textArray } from "@websensor/db";6import { factoryConfig, log } from "../config";78/**9 * Factory seeds come from two places: the coverage universes (`config/coverage/*.yaml`, every member that is not10 * monitored yet — or that carries hints worth expanding) and ad-hoc seed files (`config/factory/seeds/*.yaml`,11 * `{ seeds: [{ name, domain, categories, country, … }] }`) or the admin API.12 */13export interface SeedInput {14  id?: string;15  name: string;16  domain: string;17  homepage?: string;18  categories?: string[];19  country?: string;20  language?: string;21  tier?: string;22  weight?: number;23  importance?: number;24  aliases?: string[];25  first_party?: boolean;26  sector?: string;27  universe?: string;28  hints?: Record<string, unknown>;29}3031export function loadCoverageSectors(dir = factoryConfig.coverageDir): { sectors: CoverageSector[]; issues: string[] } {32  const sectors: CoverageSector[] = [];33  const issues: string[] = [];34  if (!existsSync(dir) || !statSync(dir).isDirectory()) return { sectors, issues: [`coverage dir ${dir} not found`] };35  for (const f of readdirSync(dir).filter((x) => /\.ya?ml$/.test(x) && !x.startsWith("_")).sort()) {36    try {37      const parsed = coverageSectorSchema.safeParse(YAML.parse(readFileSync(join(dir, f), "utf8")));38      if (!parsed.success) {39        issues.push(`${f}: ${parsed.error.issues.slice(0, 3).map((i) => `${i.path.join(".")}: ${i.message}`).join("; ")}`);40        continue;41      }42      sectors.push(parsed.data);43    } catch (e) {44      issues.push(`${f}: ${(e as Error).message}`);45    }46  }47  return { sectors, issues };48}4950/** Merge files that share a sector key. */51export function mergeSectors(sectors: CoverageSector[]): CoverageSector[] {52  const byKey = new Map<string, CoverageSector>();53  for (const s of sectors) {54    const prev = byKey.get(s.sector);55    if (!prev) byKey.set(s.sector, { ...s, universes: [...s.universes] });56    else prev.universes.push(...s.universes);57  }58  return [...byKey.values()];59}6061export function memberToSeed(sector: CoverageSector, universe: string, m: CoverageMember): SeedInput {62  return {63    name: m.name,64    domain: m.domain,65    categories: m.categories ?? sector.categories,66    country: m.country,67    language: m.language,68    tier: m.tier ?? sector.tier,69    importance: m.importance,70    aliases: m.aliases,71    first_party: m.first_party,72    sector: sector.sector,73    universe,74    hints: m.hints as Record<string, unknown>,75  };76}7778function hasHints(h: Record<string, unknown> | undefined): boolean {79  if (!h) return false;80  return Boolean(h.cik || h.github_org || (Array.isArray(h.github_repos) && h.github_repos.length) || h.hf_author || h.status_url || (Array.isArray(h.urls) && h.urls.length) || (Array.isArray(h.hosts) && h.hosts.length));81}8283/** Registrable domains currently monitored (source domains + hosts of enabled non-shadow sensors). */84export async function monitoredDomainKeys(): Promise<Set<string>> {85  const rows = await db.execute<{ d: string }>(sql`86    select domain as d from sources where enabled and kind = 'registry'87    union88    select lower(split_part(split_part(url, '/', 3), ':', 1)) as d from sensors where enabled and status <> 'SHADOW' and url like 'http%'`);89  const out = new Set<string>();90  for (const r of rows.rows) if (r.d) out.add(coverageKey(r.d));91  return out;92}9394export async function upsertSeeds(inputs: SeedInput[], opts: { requeue?: boolean } = {}): Promise<{ inserted: number; updated: number; skipped: number }> {95  let inserted = 0;96  let updated = 0;97  let skipped = 0;98  const existingIds = new Set((await db.execute<{ id: string }>(sql`select id from sources`)).rows.map((r) => r.id));99  const existingSeedDomains = new Map((await db.execute<{ id: string; domain: string }>(sql`select id, domain from factory_seeds`)).rows.map((r) => [coverageKey(r.domain), r.id]));100  for (const s of inputs) {101    const domain = s.domain.toLowerCase().replace(/^https?:\/\//, "").replace(/^www\./, "").replace(/\/.*$/, "");102    const key = coverageKey(domain);103    let id = s.id ?? existingSeedDomains.get(key) ?? slugify(s.name).slice(0, 60);104    if (!id) id = slugify(domain);105    // A registry source with the same id but another organization → disambiguate by domain.106    if (!s.id && !existingSeedDomains.has(key) && existingIds.has(id)) {107      const src = (await db.execute<{ domain: string }>(sql`select domain from sources where id = ${id}`)).rows[0];108      if (src && coverageKey(src.domain) !== key) id = `${id}-${slugify(domain.split(".")[0] ?? domain)}`.slice(0, 70);109    }110    const r = await db.execute<{ inserted: boolean }>(sql`111      insert into factory_seeds (id, name, domain, homepage, categories, country, language, tier, weight, importance, aliases, first_party, sector, universe, hints, status)112      values (${id}, ${s.name}, ${domain}, ${s.homepage ?? null}, ${textArray(s.categories ?? [])}, ${s.country ?? null}, ${s.language ?? null}, ${s.tier ?? "B"}, ${s.weight ?? 1}, ${s.importance ?? 2}, ${textArray(s.aliases ?? [])}, ${s.first_party ?? true}, ${s.sector ?? null}, ${s.universe ?? null}, ${JSON.stringify(s.hints ?? {})}::jsonb, 'queued')113      on conflict (id) do update set name = excluded.name, domain = excluded.domain, categories = excluded.categories, country = coalesce(excluded.country, factory_seeds.country), language = coalesce(excluded.language, factory_seeds.language), tier = excluded.tier, importance = excluded.importance, aliases = excluded.aliases, sector = coalesce(excluded.sector, factory_seeds.sector), universe = coalesce(excluded.universe, factory_seeds.universe), hints = factory_seeds.hints || excluded.hints, updated_at = now()${opts.requeue ? sql`, status = 'queued'` : sql``}114      returning (xmax = 0) as inserted`);115    if (r.rows[0]?.inserted) inserted++;116    else if (r.rowCount) updated++;117    else skipped++;118    existingSeedDomains.set(key, id);119  }120  log.info({ inserted, updated, skipped }, "factory seeds upserted");121  return { inserted, updated, skipped };122}123124/**125 * Turn coverage universes into seeds. `mode`: `uncovered` = members without any monitored domain; `hinted` = also126 * covered members that carry expansion hints; `all` = every member (re-discover everything).127 */128export async function seedFromCoverage(opts: { mode?: "uncovered" | "hinted" | "all"; sectors?: string[]; requeue?: boolean } = {}): Promise<{ members: number; seeds: number; inserted: number; updated: number; issues: string[] }> {129  const mode = opts.mode ?? "hinted";130  const { sectors, issues } = loadCoverageSectors();131  const monitored = await monitoredDomainKeys();132  const inputs: SeedInput[] = [];133  const seen = new Set<string>();134  let members = 0;135  for (const sector of mergeSectors(sectors)) {136    if (opts.sectors && !opts.sectors.includes(sector.sector)) continue;137    for (const u of sector.universes) {138      for (const m of u.members) {139        members++;140        const key = coverageKey(m.domain);141        if (seen.has(key)) continue;142        const covered = monitored.has(key);143        if (mode === "uncovered" && covered) continue;144        if (mode === "hinted" && covered && !hasHints(m.hints as Record<string, unknown>)) continue;145        seen.add(key);146        inputs.push(memberToSeed(sector, u.key, m));147      }148    }149  }150  const r = await upsertSeeds(inputs, { requeue: opts.requeue });151  return { members, seeds: inputs.length, inserted: r.inserted, updated: r.updated, issues };152}153154/** `config/factory/seeds/*.yaml` → `{ seeds: [...] }` */155export async function seedFromFiles(dir = factoryConfig.seedsDir): Promise<{ files: number; seeds: number }> {156  if (!existsSync(dir)) return { files: 0, seeds: 0 };157  const inputs: SeedInput[] = [];158  let files = 0;159  for (const f of readdirSync(dir).filter((x) => /\.ya?ml$/.test(x)).sort()) {160    files++;161    const raw = YAML.parse(readFileSync(join(dir, f), "utf8")) as { seeds?: SeedInput[] };162    for (const s of raw.seeds ?? []) if (s?.name && s?.domain) inputs.push(s);163  }164  await upsertSeeds(inputs);165  return { files, seeds: inputs.length };166}167168export { factorySeeds };169