Filter weak local campaign source units

apps/agent/agent/cli/local-campaign-worker.ts

@@ -0,0 +1,126 @@
+import { writeFile } from "node:fs/promises";
+import {
+	localCampaignCsv,
+	localCampaignJsonl,
+} from "../lib/local-campaign-export";
+import {
+	assertLocalCampaignOnly,
+	loadCampaignFile,
+	loadCampaignSeeds,
+	runLocalCampaign,
+} from "../lib/local-campaign-runner";
+import {
+	assertLocalCampaignPath,
+	defaultLocalCampaignStagingPath,
+	readStagedLeads,
+} from "../lib/local-campaign-staging";
+
+function argument(args: string[], flag: string): string | undefined {
+	const index = args.indexOf(flag);
+	return (
+		args
+			.find((value) => value.startsWith(`${flag}=`))
+			?.slice(flag.length + 1) ?? args[index + 1]
+	);
+}
+
+function numericArgument(args: string[], flag: string): number | undefined {
+	const value = argument(args, flag);
+	if (value === undefined) return undefined;
+	const parsed = Number(value);
+	if (!Number.isFinite(parsed) || parsed <= 0) {
+		throw new Error(`${flag} must be a positive number.`);
+	}
+	return Math.round(parsed);
+}
+
+async function main(): Promise<void> {
+	const args = process.argv.slice(2);
+	assertLocalCampaignOnly(args);
+	const command = args[0];
+	if (command === "validate") {
+		const campaign = await loadCampaignFile(argument(args, "--campaign") ?? "");
+		console.log(
+			JSON.stringify(
+				{ valid: true, campaign_id: campaign.campaign_id },
+				null,
+				2,
+			),
+		);
+		return;
+	}
+	if (command === "run") {
+		const baseCampaign = await loadCampaignFile(
+			argument(args, "--campaign") ?? "",
+		);
+		const campaign = {
+			...baseCampaign,
+			max_source_units_per_page:
+				numericArgument(args, "--max-source-units-per-page") ??
+				baseCampaign.max_source_units_per_page,
+			max_source_unit_chars:
+				numericArgument(args, "--max-source-unit-chars") ??
+				baseCampaign.max_source_unit_chars,
+			max_ollama_calls_per_seed:
+				numericArgument(args, "--max-ollama-calls-per-seed") ??
+				baseCampaign.max_ollama_calls_per_seed,
+			max_total_ollama_calls:
+				numericArgument(args, "--max-total-ollama-calls") ??
+				baseCampaign.max_total_ollama_calls,
+			per_ollama_call_timeout_ms:
+				numericArgument(args, "--per-ollama-call-timeout-ms") ??
+				baseCampaign.per_ollama_call_timeout_ms,
+		};
+		const seedFile = argument(args, "--seed-file");
+		const seeds = seedFile
+			? await loadCampaignSeeds(seedFile)
+			: campaign.seed_urls;
+		const progress = args.includes("--progress");
+		const result = await runLocalCampaign(campaign, seeds, {
+			stagingPath: argument(args, "--staging-path"),
+			seedStatusPath: argument(args, "--seed-status-path"),
+			perSeedTimeoutMs: numericArgument(args, "--per-seed-timeout-ms"),
+			onProgress: progress
+				? (event) => console.error(JSON.stringify(event))
+				: undefined,
+		});
+		console.log(JSON.stringify({ mode: "dry-run", ...result }, null, 2));
+		return;
+	}
+	if (command === "list") {
+		const path =
+			argument(args, "--staging-path") ?? defaultLocalCampaignStagingPath();
+		console.log(JSON.stringify(await readStagedLeads(path), null, 2));
+		return;
+	}
+	if (command === "export") {
+		const format = argument(args, "--format");
+		const path =
+			argument(args, "--staging-path") ?? defaultLocalCampaignStagingPath();
+		const leads = await readStagedLeads(path);
+		const output = argument(args, "--out");
+		if (!output) throw new Error("Export needs --out.");
+		const safeOutput = assertLocalCampaignPath(output);
+		if (format === "jsonl") {
+			await writeFile(safeOutput, localCampaignJsonl(leads), "utf8");
+		} else if (format === "csv") {
+			await writeFile(safeOutput, localCampaignCsv(leads), "utf8");
+		} else {
+			throw new Error("Export format must be csv or jsonl.");
+		}
+		console.log(
+			JSON.stringify(
+				{ exported: leads.length, output: safeOutput, format },
+				null,
+				2,
+			),
+		);
+		return;
+	}
+	throw new Error("Use validate, run, list, or export.");
+}
+
+main().catch((error) => {
+	console.error(error instanceof Error ? error.message : String(error));
+	process.exitCode = 1;
+});

apps/agent/agent/cli/local-lead-worker.ts

@@ -0,0 +1,72 @@
+import {
+	extractLocalLead,
+	fetchLocalLeadSource,
+} from "../lib/local-lead-extraction";
+import {
+	defaultLocalLeadStagingPath,
+	stageLocalLead,
+} from "../lib/local-lead-staging";
+
+function valueAfter(args: string[], flag: string): string | undefined {
+	const index = args.indexOf(flag);
+	return index >= 0 ? args[index + 1] : undefined;
+}
+
+function valueEquals(args: string[], flag: string): string | undefined {
+	const prefix = `${flag}=`;
+	return args.find((arg) => arg.startsWith(prefix))?.slice(prefix.length);
+}
+
+function argument(args: string[], flag: string): string | undefined {
+	return valueEquals(args, flag) ?? valueAfter(args, flag);
+}
+
+function assertDryRun(args: string[]): void {
+	const forbidden = args.filter((arg) =>
+		["--sync", "--sync-crm", "--production", "--neon"].includes(
+			arg.split("=", 1)[0] ?? "",
+		),
+	);
+	if (forbidden.length > 0) {
+		throw new Error(
+			`CRM synchronization is not implemented. Refused: ${forbidden.join(", ")}`,
+		);
+	}
+}
+
+async function main(): Promise<void> {
+	const args = process.argv.slice(2);
+	assertDryRun(args);
+
+	const text = argument(args, "--text");
+	const url = argument(args, "--url");
+	if ((text ? 1 : 0) + (url ? 1 : 0) !== 1) {
+		throw new Error("Provide exactly one of --text or --url.");
+	}
+
+	const source = url
+		? await fetchLocalLeadSource(url)
+		: {
+				sourceUrl:
+					argument(args, "--source-url") ?? "https://local.invalid/sample",
+				text: text as string,
+			};
+	const lead = await extractLocalLead(source.text, source.sourceUrl);
+	const staged = await stageLocalLead(
+		lead,
+		argument(args, "--staging-path") ?? defaultLocalLeadStagingPath(),
+	);
+
+	console.log(
+		JSON.stringify(
+			{ mode: "dry-run", extracted: lead, staging: staged },
+			null,
+			2,
+		),
+	);
+}
+
+main().catch((error) => {
+	console.error(error instanceof Error ? error.message : String(error));
+	process.exitCode = 1;
+});

apps/agent/agent/lib/local-campaign-cache.ts

@@ -0,0 +1,55 @@
+import { createHash } from "node:crypto";
+import { appendFile, mkdir, readFile } from "node:fs/promises";
+import { dirname } from "node:path";
+import { fetchLocalLeadSource } from "./local-lead-extraction";
+import { defaultLocalLeadStagingPath } from "./local-lead-staging";
+
+export type CachedPage = {
+	url: string;
+	content_hash: string;
+	text: string;
+	fetched_at: string;
+};
+
+const cachePath = () =>
+	defaultLocalLeadStagingPath().replace(/leads\.jsonl$/, "pages.jsonl");
+
+export async function readCachedPage(
+	url: string,
+	path = cachePath(),
+): Promise<CachedPage | null> {
+	let raw = "";
+	try {
+		raw = await readFile(path, "utf8");
+	} catch (error) {
+		if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
+		throw error;
+	}
+	for (const line of raw.split("\n")) {
+		if (!line.trim()) continue;
+		const row = JSON.parse(line) as CachedPage;
+		if (row.url === url) return row;
+	}
+	return null;
+}
+
+export async function fetchCachedPage(
+	url: string,
+	fetchImpl: typeof fetch = fetch,
+	path = cachePath(),
+): Promise<{ page: CachedPage; cached: boolean }> {
+	const existing = await readCachedPage(url, path);
+	if (existing) return { page: existing, cached: true };
+	const source = await fetchLocalLeadSource(url, fetchImpl);
+	const page: CachedPage = {
+		url,
+		content_hash: createHash("sha256")
+			.update(source.text, "utf8")
+			.digest("hex"),
+		text: source.text,
+		fetched_at: new Date().toISOString(),
+	};
+	await mkdir(dirname(path), { recursive: true });
+	await appendFile(path, `${JSON.stringify(page)}\n`, "utf8");
+	return { page, cached: false };
+}

apps/agent/agent/lib/local-campaign-export.ts

@@ -0,0 +1,23 @@
+import type { LocalStagedLead } from "./local-campaign-staging";
+
+type CsvValue = string | string[] | number | boolean | null | undefined;
+
+function csvValue(value: CsvValue): string {
+	const text = value === null || value === undefined ? "" : String(value);
+	return /[,"\n]/.test(text) ? `"${text.replaceAll('"', '""')}"` : text;
+}
+
+export function localCampaignJsonl(leads: LocalStagedLead[]): string {
+	return `${leads.map((lead) => JSON.stringify(lead)).join("\n")}\n`;
+}
+
+export function localCampaignCsv(leads: LocalStagedLead[]): string {
+	const headers = leads[0] ? Object.keys(leads[0]) : [];
+	const rows = [
+		headers,
+		...leads.map((lead) =>
+			headers.map((header) => lead[header as keyof typeof lead]),
+		),
+	];
+	return `${rows.map((row) => row.map(csvValue).join(",")).join("\n")}\n`;
+}