← back to Norma
app/api/ingest/govt-data/route.ts
738 lines
import { NextRequest, NextResponse } from 'next/server';
import { query } from '@/lib/db';
import { requireRole } from '@/lib/require-role';
import {
searchCFPBComplaints,
fetchCensusEducation,
fetchCensusZCTA,
fetchCollegeScorecard,
fetchBLSEarnings,
searchSocrataPortal,
searchEdGovDatasets,
fetchAllEducationGrants,
fetchAllFederalRegisterDocs,
fetchUSASpending,
fetchUSASpendingByAgency,
fetchAllStudentDebtNonprofits,
searchAllSocrataPortals,
} from '@/lib/govt-data';
/**
* POST /api/ingest/govt-data
*
* Master ingestion endpoint for all government data sources.
* Accepts optional `sources` array to limit which sources to ingest.
* Default: ingest all sources.
*/
export async function POST(request: NextRequest) {
const auth = requireRole(request, 'admin', 'staff');
if (auth instanceof NextResponse) return auth;
let body: { sources?: string[] } = {};
try {
body = await request.json();
} catch {
// no body = ingest all
}
const requestedSources = body.sources ?? [
'cfpb',
'census',
'census_zcta',
'scorecard',
'bls',
'socrata',
'edgov',
'grants_gov',
'federal_register',
'usaspending',
'propublica',
];
const stats: Record<string, { count: number; errors: string[] }> = {};
const startTime = Date.now();
// ── 1. CFPB Student Loan Complaints ─────────────────────────────────
if (requestedSources.includes('cfpb')) {
stats.cfpb = { count: 0, errors: [] };
try {
const complaints = await searchCFPBComplaints({ size: 200 });
for (const c of complaints) {
try {
await query(
`INSERT INTO cfpb_complaints (
complaint_id, date_received, product, sub_product, issue, sub_issue,
company, company_response, state, zip_code, consumer_consent,
submitted_via, timely_response, complaint_narrative, tags, raw_data
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)
ON CONFLICT (complaint_id) DO UPDATE SET
company_response = EXCLUDED.company_response,
timely_response = EXCLUDED.timely_response`,
[
c.complaint_id,
c.date_received || null,
c.product || 'Student loan',
c.sub_product || null,
c.issue || null,
c.sub_issue || null,
c.company || null,
c.company_response || null,
c.state || null,
c.zip_code || null,
c.consumer_consent || null,
c.submitted_via || null,
c.timely_response ?? null,
c.complaint_narrative || null,
c.tags ? [c.tags] : [],
JSON.stringify(c),
],
);
stats.cfpb.count++;
} catch (err) {
stats.cfpb.errors.push('insert: ' + (err as Error).message);
}
}
await updateSourceSync('cfpb', stats.cfpb.count, stats.cfpb.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.cfpb.errors.push('search: ' + (err as Error).message);
await updateSourceSync('cfpb', 0, 'failed');
}
}
// ── 2. Census ACS Education Data ────────────────────────────────────
if (requestedSources.includes('census')) {
stats.census = { count: 0, errors: [] };
try {
// Fetch all 50 states
const stateRows = await fetchCensusEducation(2023, 'state');
for (const row of stateRows) {
const gradPlus = (row.masters || 0) + (row.professional || 0) + (row.doctorate || 0);
const bPct = row.totalPop25Plus > 0 ? ((row.bachelors / row.totalPop25Plus) * 100) : null;
const gPct = row.totalPop25Plus > 0 ? ((gradPlus / row.totalPop25Plus) * 100) : null;
try {
await query(
`INSERT INTO census_education (
geo_level, geo_fips, geo_name, state_fips, year, dataset,
pop_25_plus, hs_diploma, associates, bachelors, graduate_plus,
bachelors_pct, graduate_pct, raw_data
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
ON CONFLICT (geo_fips, year, dataset) DO UPDATE SET
pop_25_plus = EXCLUDED.pop_25_plus,
bachelors = EXCLUDED.bachelors,
graduate_plus = EXCLUDED.graduate_plus,
bachelors_pct = EXCLUDED.bachelors_pct,
graduate_pct = EXCLUDED.graduate_pct`,
[
'state',
row.geoId,
row.geoName,
row.geoId,
2023,
'acs5',
row.totalPop25Plus,
row.highSchool,
row.associates,
row.bachelors,
gradPlus,
bPct,
gPct,
JSON.stringify(row),
],
);
stats.census.count++;
} catch (err) {
stats.census.errors.push('state_insert: ' + (err as Error).message);
}
}
// Fetch top 5 states by population (CA, TX, FL, NY, PA) at county level
const topStates = ['06', '48', '12', '36', '42'];
for (const fips of topStates) {
try {
const countyRows = await fetchCensusEducation(2023, 'county', fips);
for (const row of countyRows) {
const gradPlus = (row.masters || 0) + (row.professional || 0) + (row.doctorate || 0);
const bPct = row.totalPop25Plus > 0 ? ((row.bachelors / row.totalPop25Plus) * 100) : null;
try {
await query(
`INSERT INTO census_education (
geo_level, geo_fips, geo_name, state_fips, county_fips, year, dataset,
pop_25_plus, hs_diploma, associates, bachelors, graduate_plus,
bachelors_pct, raw_data
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
ON CONFLICT (geo_fips, year, dataset) DO UPDATE SET
pop_25_plus = EXCLUDED.pop_25_plus,
bachelors = EXCLUDED.bachelors,
bachelors_pct = EXCLUDED.bachelors_pct`,
[
'county',
row.geoId,
row.geoName,
fips,
row.geoId.slice(2),
2023,
'acs5',
row.totalPop25Plus,
row.highSchool,
row.associates,
row.bachelors,
gradPlus,
bPct,
JSON.stringify(row),
],
);
stats.census.count++;
} catch (err) {
stats.census.errors.push('county_insert: ' + (err as Error).message);
}
}
} catch (err) {
stats.census.errors.push(`county_fetch_${fips}: ` + (err as Error).message);
}
}
await updateSourceSync('census_acs', stats.census.count, stats.census.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.census.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('census_acs', 0, 'failed');
}
}
// ── 2b. Census ACS ZCTA (Zip Code) Education Data ───────────────────
if (requestedSources.includes('census_zcta')) {
stats.census_zcta = { count: 0, errors: [] };
try {
// Pull the unique, valid zip codes that appear in CFPB complaints
const { rows: zipRows } = await query(
`SELECT DISTINCT zip_code FROM cfpb_complaints WHERE zip_code IS NOT NULL AND zip_code != 'XXXXX'`,
);
const zips: string[] = zipRows.map((r: any) => r.zip_code as string);
if (zips.length === 0) {
console.warn('[census-zcta] No complaint zip codes found — skipping ZCTA fetch');
await updateSourceSync('census_zcta', 0, 'success');
} else {
// fetchCensusZCTA handles internal batching; we call it directly with the full list
const zctaRows = await fetchCensusZCTA(2023, undefined, zips);
for (const row of zctaRows) {
const gradPlus = (row.masters || 0) + (row.professional || 0) + (row.doctorate || 0);
const bPct =
row.totalPop25Plus > 0 ? (row.bachelors / row.totalPop25Plus) * 100 : null;
const gPct =
row.totalPop25Plus > 0 ? (gradPlus / row.totalPop25Plus) * 100 : null;
try {
await query(
`INSERT INTO census_education (
geo_level, geo_fips, geo_name, year, dataset,
pop_25_plus, hs_diploma, associates, bachelors, graduate_plus,
bachelors_pct, graduate_pct, raw_data
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
ON CONFLICT (geo_fips, year, dataset) DO UPDATE SET
pop_25_plus = EXCLUDED.pop_25_plus,
bachelors = EXCLUDED.bachelors,
graduate_plus = EXCLUDED.graduate_plus,
bachelors_pct = EXCLUDED.bachelors_pct,
graduate_pct = EXCLUDED.graduate_pct`,
[
'zcta',
row.geoId,
row.geoName,
2023,
'acs5',
row.totalPop25Plus,
row.highSchool,
row.associates,
row.bachelors,
gradPlus,
bPct,
gPct,
JSON.stringify(row),
],
);
stats.census_zcta.count++;
} catch (err) {
stats.census_zcta.errors.push('insert: ' + (err as Error).message);
}
}
await updateSourceSync(
'census_zcta',
stats.census_zcta.count,
stats.census_zcta.errors.length === 0 ? 'success' : 'partial',
);
}
} catch (err) {
stats.census_zcta.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('census_zcta', 0, 'failed');
}
}
// ── 3. College Scorecard — ALL institutions with ZIP + degree type ──
if (requestedSources.includes('scorecard')) {
stats.scorecard = { count: 0, errors: [] };
try {
// Fetch ALL schools: community colleges (type 2), then 4-year (type 3), then certs (type 1)
for (const degreeType of [2, 3, 1, 4]) {
for (let page = 0; page < 100; page++) {
const schools = await fetchCollegeScorecard({ perPage: 100, page, degreeType });
if (schools.length === 0) break;
if (page > 0) await new Promise((r) => setTimeout(r, 800));
for (const s of schools) {
try {
const ownershipLabel = s.ownership === 1 ? '1' : s.ownership === 2 ? '2' : s.ownership === 3 ? '3' : String(s.ownership ?? '');
const instType = s.degreesPredominant === 2 ? 'community_college'
: ownershipLabel === '1' && (s.degreesPredominant ?? 0) >= 3 ? '4year_public'
: ownershipLabel === '2' && (s.degreesPredominant ?? 0) >= 3 ? '4year_private'
: ownershipLabel === '3' ? 'for_profit' : 'other';
await query(
`INSERT INTO college_scorecard (
unit_id, ope_id, school_name, city, state, zip, school_url, ownership,
tuition_in_state, tuition_out_state, avg_net_price,
pell_grant_rate, federal_loan_rate,
median_debt, median_debt_completers, median_debt_noncompleters,
median_monthly_payment, completion_rate, retention_rate,
median_earnings_6yr, median_earnings_10yr,
enrollment, undergrad_enrollment,
institution_type, degrees_predominant,
raw_data, data_year
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27)
ON CONFLICT (unit_id) DO UPDATE SET
zip = COALESCE(EXCLUDED.zip, college_scorecard.zip),
tuition_in_state = EXCLUDED.tuition_in_state,
avg_net_price = EXCLUDED.avg_net_price,
median_debt = EXCLUDED.median_debt,
median_debt_completers = EXCLUDED.median_debt_completers,
median_debt_noncompleters = EXCLUDED.median_debt_noncompleters,
median_monthly_payment = EXCLUDED.median_monthly_payment,
pell_grant_rate = EXCLUDED.pell_grant_rate,
completion_rate = EXCLUDED.completion_rate,
retention_rate = EXCLUDED.retention_rate,
median_earnings_6yr = EXCLUDED.median_earnings_6yr,
median_earnings_10yr = EXCLUDED.median_earnings_10yr,
enrollment = EXCLUDED.enrollment,
undergrad_enrollment = EXCLUDED.undergrad_enrollment,
institution_type = EXCLUDED.institution_type,
degrees_predominant = EXCLUDED.degrees_predominant,
raw_data = EXCLUDED.raw_data,
updated_at = NOW()`,
[
s.id,
s.opeId || null,
s.name,
s.city || null,
s.state || null,
s.zip || null,
s.schoolUrl || null,
ownershipLabel || null,
s.tuitionInState ?? null,
s.tuitionOutOfState ?? null,
s.avgNetPrice ?? null,
s.pellGrantRate ?? null,
s.federalLoanRate ?? null,
s.medianDebt ?? null,
s.medianDebtCompleters ?? null,
s.medianDebtNoncompleters ?? null,
s.medianMonthlyPayment ?? null,
s.completionRate ?? null,
s.retentionRate ?? null,
s.medianEarnings6yr ?? null,
s.medianEarnings10yr ?? null,
s.enrollmentAll ?? null,
s.undergradEnrollment ?? null,
instType,
s.degreesPredominant ?? null,
JSON.stringify(s),
2024,
],
);
stats.scorecard.count++;
} catch (err) {
stats.scorecard.errors.push('insert: ' + (err as Error).message);
}
}
}
}
await updateSourceSync('college_scorecard', stats.scorecard.count, stats.scorecard.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.scorecard.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('college_scorecard', 0, 'failed');
}
}
// ── 4. BLS Education Earnings ───────────────────────────────────────
if (requestedSources.includes('bls')) {
stats.bls = { count: 0, errors: [] };
try {
const currentYear = new Date().getFullYear();
const data = await fetchBLSEarnings(currentYear - 5, currentYear);
// BLS adapter returns an object keyed by education level, each containing BLSDataPoint[]
const educLevelToSeriesId: Record<string, string> = {
highSchool: 'LEU0254530200',
someCollege: 'LEU0254530400',
associates: 'LEU0254530600',
bachelors: 'LEU0254530800',
masters: 'LEU0254531000',
professional: 'LEU0254531200',
doctoral: 'LEU0254531400',
};
for (const [educLevel, dataPoints] of Object.entries(data) as [string, { year: string; period: string; value: number }[]][]) {
const seriesId = educLevelToSeriesId[educLevel] || educLevel;
for (const dp of dataPoints) {
try {
await query(
`INSERT INTO bls_education_earnings (
series_id, education_level, year, period, value, metric, raw_data
) VALUES ($1,$2,$3,$4,$5,$6,$7)
ON CONFLICT (series_id, year, period) DO UPDATE SET
value = EXCLUDED.value`,
[
seriesId,
educLevel,
dp.year,
dp.period,
dp.value,
'median_weekly_earnings',
JSON.stringify(dp),
],
);
stats.bls.count++;
} catch (err) {
stats.bls.errors.push('insert: ' + (err as Error).message);
}
}
}
await updateSourceSync('bls', stats.bls.count, stats.bls.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.bls.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('bls', 0, 'failed');
}
}
// ── 5. Socrata Portal Discovery ─────────────────────────────────────
if (requestedSources.includes('socrata')) {
stats.socrata = { count: 0, errors: [] };
const portals = [
{ domain: 'data.ny.gov', level: 'state', name: 'New York' },
{ domain: 'data.cityofnewyork.us', level: 'city', name: 'New York City' },
{ domain: 'data.cityofchicago.org', level: 'city', name: 'Chicago' },
{ domain: 'data.lacity.org', level: 'city', name: 'Los Angeles' },
{ domain: 'data.sfgov.org', level: 'city', name: 'San Francisco' },
{ domain: 'data.texas.gov', level: 'state', name: 'Texas' },
{ domain: 'data.illinois.gov', level: 'state', name: 'Illinois' },
{ domain: 'data.wa.gov', level: 'state', name: 'Washington' },
{ domain: 'data.colorado.gov', level: 'state', name: 'Colorado' },
{ domain: 'opendata.maryland.gov', level: 'state', name: 'Maryland' },
];
for (const portal of portals) {
try {
const datasets = await searchSocrataPortal(portal.domain, 'education student');
for (const ds of datasets) {
try {
await query(
`INSERT INTO socrata_datasets (
portal_domain, dataset_id, portal_level, geo_name,
title, description, category, tags, row_count, api_url
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (portal_domain, dataset_id) DO UPDATE SET
title = EXCLUDED.title,
description = EXCLUDED.description`,
[
portal.domain,
ds.id,
portal.level,
portal.name,
ds.name || ds.id,
ds.description || null,
ds.type || null,
[],
null,
`https://${portal.domain}/resource/${ds.id}.json`,
],
);
stats.socrata.count++;
} catch (err) {
stats.socrata.errors.push(`${portal.domain}_insert: ` + (err as Error).message);
}
}
} catch (err) {
stats.socrata.errors.push(`${portal.domain}_search: ` + (err as Error).message);
}
}
await updateSourceSync('socrata_ny', stats.socrata.count, stats.socrata.errors.length === 0 ? 'success' : 'partial');
}
// ── 6. data.ed.gov Dataset Discovery ────────────────────────────────
if (requestedSources.includes('edgov')) {
stats.edgov = { count: 0, errors: [] };
try {
const searches = ['student aid', 'pell grant', 'student loan', 'fafsa', 'title iv'];
for (const q of searches) {
const datasets = await searchEdGovDatasets(q, 20);
stats.edgov.count += datasets.length;
}
await updateSourceSync('data_ed_gov', stats.edgov.count, 'success');
} catch (err) {
stats.edgov.errors.push('search: ' + (err as Error).message);
await updateSourceSync('data_ed_gov', 0, 'failed');
}
}
// ── 7. Grants.gov — Federal Grant Opportunities ────────────────────
if (requestedSources.includes('grants_gov')) {
stats.grants_gov = { count: 0, errors: [] };
try {
const grants = await fetchAllEducationGrants(50);
for (const g of grants) {
try {
await query(
`INSERT INTO federal_grants (
source_system, opportunity_id, opportunity_number, title, agency, sub_agency,
description, funding_category, funding_instrument,
award_floor, award_ceiling, estimated_typical_award,
posted_date, close_date, cfda_number, official_url,
is_education_related, raw_data
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18)
ON CONFLICT (opportunity_id) DO UPDATE SET
title = EXCLUDED.title,
description = EXCLUDED.description,
close_date = EXCLUDED.close_date,
award_ceiling = EXCLUDED.award_ceiling,
raw_data = EXCLUDED.raw_data,
updated_at = NOW()`,
[
'grants_gov',
g.opportunityId,
g.opportunityNumber,
g.title,
g.agency,
g.subAgency,
g.description,
g.fundingCategory,
g.fundingInstrument,
g.awardFloor,
g.awardCeiling,
g.estimatedAward,
g.postedDate,
g.closeDate,
g.cfdaNumber,
g.officialUrl,
true,
JSON.stringify(g.rawData),
],
);
stats.grants_gov.count++;
} catch (err) {
stats.grants_gov.errors.push('insert: ' + (err as Error).message);
}
}
await updateSourceSync('grants_gov', stats.grants_gov.count, stats.grants_gov.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.grants_gov.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('grants_gov', 0, 'failed');
}
}
// ── 8. Federal Register — Policy Documents ───────────────────────
if (requestedSources.includes('federal_register')) {
stats.federal_register = { count: 0, errors: [] };
try {
const docs = await fetchAllFederalRegisterDocs(20);
for (const d of docs) {
const textToCheck = ((d.title || '') + ' ' + (d.abstract || '')).toLowerCase();
const mentionsStudentDebt = /student (loan|debt|borrow|aid)|fafsa|pell grant|loan forgiv/i.test(textToCheck);
const mentionsEducation = /education|college|universit|higher ed|tuition/i.test(textToCheck);
const mentionsFinancialAid = /financial aid|grant|scholarship|work.study/i.test(textToCheck);
try {
await query(
`INSERT INTO meeting_minutes (
source_type, source_name, geo_level, geo_name,
meeting_date, title, url, full_text,
mentions_student_debt, mentions_education, mentions_financial_aid,
key_speakers, tags, raw_data
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
ON CONFLICT DO NOTHING`,
[
'federal_register',
d.agencies.join(', ') || 'Federal Register',
'federal',
'United States',
d.publicationDate || null,
d.title,
d.htmlUrl,
d.abstract,
mentionsStudentDebt,
mentionsEducation,
mentionsFinancialAid,
d.agencies,
[d.type || 'document', 'federal_register'],
JSON.stringify(d.rawData),
],
);
stats.federal_register.count++;
} catch (err) {
stats.federal_register.errors.push('insert: ' + (err as Error).message);
}
}
await updateSourceSync('federal_register', stats.federal_register.count, stats.federal_register.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.federal_register.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('federal_register', 0, 'failed');
}
}
// ── 9. USAspending.gov — Education Spending by CFDA ───────────────
if (requestedSources.includes('usaspending')) {
stats.usaspending = { count: 0, errors: [] };
try {
const INT_MAX = 2147483647;
const fy = new Date().getFullYear();
const records = await fetchUSASpending({ fiscalYear: fy, limit: 50 });
const agencyRecords = await fetchUSASpendingByAgency(fy);
const allRecords = [...records, ...agencyRecords];
for (const r of allRecords) {
const amount = r.amount || 0;
const amountMin = amount > 0 ? Math.min(Math.floor(amount * 0.8), INT_MAX) : null;
const amountMax = amount > 0 ? Math.min(Math.floor(amount * 1.2), INT_MAX) : null;
try {
await query(
`INSERT INTO grants (
title, funder, description, amount_min, amount_max,
focus_areas, status, priority, tags
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
ON CONFLICT DO NOTHING`,
[
`USAspending: ${r.name} (FY${fy})`,
`Federal Government - ${r.category}`,
`Federal spending record from USAspending.gov. Category: ${r.category}. Amount: $${Math.round(amount).toLocaleString()}. Fiscal Year: ${fy}.`,
amountMin,
amountMax,
['education', 'federal-spending', r.category],
'researching',
'low',
['usaspending', r.category, `fy${fy}`],
],
);
stats.usaspending.count++;
} catch (err) {
stats.usaspending.errors.push('insert: ' + (err as Error).message);
}
}
await updateSourceSync('usaspending', stats.usaspending.count, stats.usaspending.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.usaspending.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('usaspending', 0, 'failed');
}
}
// ── 10. ProPublica Nonprofits — Advocacy Organizations ────────────
if (requestedSources.includes('propublica')) {
stats.propublica = { count: 0, errors: [] };
try {
const orgs = await fetchAllStudentDebtNonprofits();
for (const o of orgs) {
const einFormatted = o.ein.length >= 9
? `${o.ein.slice(0, 2)}-${o.ein.slice(2)}`
: o.ein;
let category = 'education';
if (o.nteeCode) {
if (o.nteeCode.startsWith('B')) category = 'education';
else if (o.nteeCode.startsWith('W') || o.nteeCode.startsWith('P') || o.nteeCode.startsWith('R')) category = 'civil_rights';
else if (o.nteeCode.startsWith('I')) category = 'economic_justice';
}
try {
await query(
`INSERT INTO organizations (
name, ein, type, category, ntee_code,
city, state, annual_revenue, total_assets,
key_issues, source, source_id, source_url
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
ON CONFLICT ON CONSTRAINT org_unique_ein DO UPDATE SET
name = EXCLUDED.name,
city = COALESCE(EXCLUDED.city, organizations.city),
state = COALESCE(EXCLUDED.state, organizations.state),
annual_revenue = COALESCE(EXCLUDED.annual_revenue, organizations.annual_revenue),
total_assets = COALESCE(EXCLUDED.total_assets, organizations.total_assets),
ntee_code = COALESCE(EXCLUDED.ntee_code, organizations.ntee_code),
updated_at = NOW()`,
[
o.name,
einFormatted,
'nonprofit',
category,
o.nteeCode,
o.city,
o.state,
o.totalRevenue,
o.totalAssets,
['student_debt', 'education'],
'propublica',
o.ein,
o.filingUrl,
],
);
stats.propublica.count++;
} catch (err) {
stats.propublica.errors.push('insert: ' + (err as Error).message);
}
}
await updateSourceSync('propublica_nonprofits', stats.propublica.count, stats.propublica.errors.length === 0 ? 'success' : 'partial');
} catch (err) {
stats.propublica.errors.push('fetch: ' + (err as Error).message);
await updateSourceSync('propublica_nonprofits', 0, 'failed');
}
}
const elapsed = ((Date.now() - startTime) / 1000).toFixed(1);
return NextResponse.json({
success: true,
elapsed_seconds: elapsed,
stats,
message: Object.entries(stats)
.map(([k, v]) => `${k}: ${v.count} records`)
.join(', '),
});
}
/* ─── GET: Return data source sync status ──────────────────────────── */
export async function GET(request: NextRequest) {
const auth = requireRole(request, 'admin', 'staff');
if (auth instanceof NextResponse) return auth;
try {
const { rows } = await query(`
SELECT source_key, display_name, source_type, api_base_url,
is_active, last_sync_at, last_sync_status, last_sync_count,
sync_interval_hrs
FROM data_sources
ORDER BY source_type, display_name
`);
return NextResponse.json({ sources: rows });
} catch (err) {
console.error('[ingest/govt-data] GET error:', (err as Error).message);
return NextResponse.json({ error: 'Failed to fetch data sources' }, { status: 500 });
}
}
/* ─── Helper: Update data source sync timestamp ────────────────────── */
async function updateSourceSync(sourceKey: string, count: number, status: string) {
try {
await query(
`UPDATE data_sources SET last_sync_at = NOW(), last_sync_status = $1, last_sync_count = $2, updated_at = NOW() WHERE source_key = $3`,
[status, count, sourceKey],
);
} catch {
// Non-critical
}
}