Janady13/missionmatch / internal/platform/benchimport/structured_source.go

1package benchimport 2 3import ( 4 "bufio" 5 "context" 6 "encoding/csv" 7 "encoding/json" 8 "fmt" 9 "io" 10 "os" 11 "path/filepath" 12 "strconv" 13 "strings" 14) 15 16// structuredSource is the FileSource backed by a structured roster file, decoded with the standard library 17// only. It supports two formats picked by extension: CSV with a header row, and JSONL (one JSON object per 18// line). Each well formed row becomes a BenchRecord; a malformed row is skipped and counted, never aborts 19// the batch. The structured fields here are the same workforce tokens the resume path produces, so the 20// downstream ingest is identical regardless of source. 21type structuredSource struct{} 22 23// NewStructuredSource builds a structured roster FileSource. 24func NewStructuredSource() FileSource { return structuredSource{} } 25 26// Resumes on a structured source delegates to the resume reader is not meaningful; a structured source 27// has no resume directory. It returns an empty result so a caller that picked the structured source by 28// mistake gets nothing rather than a panic. 29func (structuredSource) Resumes(ctx context.Context, dir string, yield func(rec BenchRecord, file string) error) (BatchResult, error) { 30 _ = ctx 31 _ = dir 32 _ = yield 33 return BatchResult{}, nil 34} 35 36// Profiles decodes the roster at path and yields a BenchRecord per well formed row. The format is chosen 37// by extension: .csv uses the header mapped CSV decoder, .jsonl or .ndjson or .json uses the line 38// delimited JSON decoder. The batch identifier passed to yield is the file base name, recorded against 39// bench_import_log. 40func (s structuredSource) Profiles(ctx context.Context, path string, yield func(rec BenchRecord, batch string) error) (BatchResult, error) { 41 f, err := os.Open(path) 42 if err != nil { 43 return BatchResult{}, err 44 } 45 defer func() { _ = f.Close() }() 46 47 batch := filepath.Base(path) 48 ext := strings.ToLower(filepath.Ext(path)) 49 switch ext { 50 case ".csv": 51 return decodeCSV(ctx, f, batch, yield) 52 case ".jsonl", ".ndjson", ".json": 53 return decodeJSONL(ctx, f, batch, yield) 54 default: 55 return BatchResult{}, fmt.Errorf("benchimport: unsupported roster extension %q", ext) 56 } 57} 58 59// decodeCSV reads a header row, maps the standard bench columns by lowercased header name, and yields one 60// BenchRecord per data row. A row with the wrong field count or no usable signal is skipped and counted. 61// Columns are matched flexibly (name, location, skills, certs, agencies, years, clearance, email, phone, 62// external_id) with a few common aliases. 63func decodeCSV(ctx context.Context, r io.Reader, batch string, yield func(rec BenchRecord, batch string) error) (BatchResult, error) { 64 var res BatchResult 65 cr := csv.NewReader(r) 66 cr.FieldsPerRecord = -1 // a short or long row is handled per row, not as a fatal parse error 67 cr.TrimLeadingSpace = true 68 69 header, err := cr.Read() 70 if err == io.EOF { 71 return res, nil 72 } 73 if err != nil { 74 return res, fmt.Errorf("benchimport: read csv header: %w", err) 75 } 76 colIndex := map[string]int{} 77 for i, h := range header { 78 colIndex[canonHeader(h)] = i 79 } 80 81 for { 82 row, err := cr.Read() 83 if err == io.EOF { 84 break 85 } 86 if err != nil { 87 // A structurally broken line is skipped, not fatal, so one bad row never drops the batch. 88 res.Read++ 89 res.Skipped++ 90 continue 91 } 92 res.Read++ 93 get := func(key string) string { 94 i, ok := colIndex[key] 95 if !ok || i >= len(row) { 96 return "" 97 } 98 return strings.TrimSpace(row[i]) 99 } 100 rec := BenchRecord{ 101 FullName: firstNonEmpty(get("name"), get("full_name"), get("fullname")), 102 Location: firstNonEmpty(get("location"), get("city_state")), 103 JobTitle: firstNonEmpty(get("job_title"), get("title"), get("role")), 104 Skills: splitList(firstNonEmpty(get("skills"), get("skill"))), 105 Certs: splitList(firstNonEmpty(get("certs"), get("certifications"), get("certification"))), 106 Agencies: splitList(firstNonEmpty(get("agencies"), get("agency"), get("customers"))), 107 Years: parseYears(firstNonEmpty(get("years"), get("years_experience"), get("experience"))), 108 ClearanceCode: firstNonEmpty(get("clearance"), get("clearance_level")), 109 ExternalID: firstNonEmpty(get("external_id"), get("id")), 110 Email: strings.ToLower(firstNonEmpty(get("email"))), 111 Phone: firstNonEmpty(get("phone")), 112 } 113 if !hasSignal(rec) { 114 res.Skipped++ 115 continue 116 } 117 // Count a record as mapped only after yield accepts it, so a batch cap (yield returns a stop error 118 // before the record is processed) does not inflate the mapped count over new plus enriched. 119 if err := yield(rec, batch); err != nil { 120 return res, err 121 } 122 res.Mapped++ 123 } 124 _ = ctx 125 return res, nil 126} 127 128// jsonlRow is the decode shape for one JSONL object. The bench columns map to JSON keys with a few common 129// aliases handled in fromJSONLRow. 130type jsonlRow struct { 131 Name string `json:"name"` 132 FullName string `json:"full_name"` 133 Location string `json:"location"` 134 JobTitle string `json:"job_title"` 135 Title string `json:"title"` 136 Role string `json:"role"` 137 Skills []string `json:"skills"` 138 Certs []string `json:"certs"` 139 Certifications []string `json:"certifications"` 140 Agencies []string `json:"agencies"` 141 Years int `json:"years"` 142 YearsExp int `json:"years_experience"` 143 Clearance string `json:"clearance"` 144 ClearanceLevel string `json:"clearance_level"` 145 ExternalID string `json:"external_id"` 146 ID string `json:"id"` 147 Email string `json:"email"` 148 Phone string `json:"phone"` 149} 150 151// decodeJSONL reads one JSON object per line and yields one BenchRecord per well formed line. A blank line 152// is ignored; a line that does not parse, or that yields no signal, is skipped and counted. 153func decodeJSONL(ctx context.Context, r io.Reader, batch string, yield func(rec BenchRecord, batch string) error) (BatchResult, error) { 154 var res BatchResult 155 sc := bufio.NewScanner(r) 156 sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) // allow a fat profile line without splitting it 157 for sc.Scan() { 158 line := strings.TrimSpace(sc.Text()) 159 if line == "" { 160 continue 161 } 162 res.Read++ 163 var jr jsonlRow 164 if err := json.Unmarshal([]byte(line), &jr); err != nil { 165 res.Skipped++ 166 continue 167 } 168 rec := fromJSONLRow(jr) 169 if !hasSignal(rec) { 170 res.Skipped++ 171 continue 172 } 173 // Count a record as mapped only after yield accepts it, so a batch cap (yield returns a stop error 174 // before the record is processed) does not inflate the mapped count over new plus enriched. 175 if err := yield(rec, batch); err != nil { 176 return res, err 177 } 178 res.Mapped++ 179 } 180 if err := sc.Err(); err != nil { 181 return res, fmt.Errorf("benchimport: scan jsonl: %w", err) 182 } 183 _ = ctx 184 return res, nil 185} 186 187// fromJSONLRow folds the alias keys into one BenchRecord. 188func fromJSONLRow(jr jsonlRow) BenchRecord { 189 return BenchRecord{ 190 FullName: firstNonEmpty(strings.TrimSpace(jr.Name), strings.TrimSpace(jr.FullName)), 191 Location: strings.TrimSpace(jr.Location), 192 JobTitle: firstNonEmpty(strings.TrimSpace(jr.JobTitle), strings.TrimSpace(jr.Title), strings.TrimSpace(jr.Role)), 193 Skills: trimList(firstNonEmptyList(jr.Skills, nil)), 194 Certs: trimList(firstNonEmptyList(jr.Certs, jr.Certifications)), 195 Agencies: trimList(jr.Agencies), 196 Years: firstNonZero(jr.Years, jr.YearsExp), 197 ClearanceCode: firstNonEmpty(strings.TrimSpace(jr.Clearance), strings.TrimSpace(jr.ClearanceLevel)), 198 ExternalID: firstNonEmpty(strings.TrimSpace(jr.ExternalID), strings.TrimSpace(jr.ID)), 199 Email: strings.ToLower(strings.TrimSpace(jr.Email)), 200 Phone: strings.TrimSpace(jr.Phone), 201 } 202} 203 204// ---------- helpers ---------- 205 206// canonHeader normalizes a CSV header to a lowercased, underscore joined key so "Full Name", "full_name", 207// and "FULL NAME" all map to the same column. 208func canonHeader(h string) string { 209 h = strings.ToLower(strings.TrimSpace(h)) 210 h = strings.ReplaceAll(h, " ", "_") 211 return h 212} 213 214// splitList splits a delimited cell into trimmed tokens. The standard bench format uses a semicolon or a 215// comma between skills, so both are accepted; an empty cell yields no tokens. 216func splitList(s string) []string { 217 if strings.TrimSpace(s) == "" { 218 return nil 219 } 220 sep := "," 221 if strings.Contains(s, ";") { 222 sep = ";" 223 } 224 return trimList(strings.Split(s, sep)) 225} 226 227// trimList trims each element and drops the empties. 228func trimList(xs []string) []string { 229 out := make([]string, 0, len(xs)) 230 for _, x := range xs { 231 t := strings.TrimSpace(x) 232 if t != "" { 233 out = append(out, t) 234 } 235 } 236 if len(out) == 0 { 237 return nil 238 } 239 return out 240} 241 242func parseYears(s string) int { 243 s = strings.TrimSpace(s) 244 if s == "" { 245 return 0 246 } 247 n, err := strconv.Atoi(s) 248 if err != nil || n < 0 || n > 60 { 249 return 0 250 } 251 return n 252} 253 254func firstNonEmpty(xs ...string) string { 255 for _, x := range xs { 256 if strings.TrimSpace(x) != "" { 257 return strings.TrimSpace(x) 258 } 259 } 260 return "" 261} 262 263func firstNonEmptyList(a, b []string) []string { 264 if len(a) > 0 { 265 return a 266 } 267 return b 268} 269 270func firstNonZero(xs ...int) int { 271 for _, x := range xs { 272 if x != 0 { 273 return x 274 } 275 } 276 return 0 277} 278 279// hasSignal reports whether a row carries enough structured signal to be worth writing. A row with no 280// skills, no clearance, no certs, and no years is dropped so an empty or header only line is not imported. 281func hasSignal(r BenchRecord) bool { 282 return len(r.Skills) > 0 || r.ClearanceCode != "" || len(r.Certs) > 0 || r.Years > 0 283} 284