Janady13/missionmatch / internal/modules/ontology/adapters/onet.go

1package adapters 2 3import ( 4 "context" 5 "encoding/json" 6 "fmt" 7 "io" 8 "log" 9 "net/http" 10 "strings" 11 "time" 12 13 "github.com/missionworkforce/missionmatch/internal/modules/ontology" 14) 15 16// ONETAttribution is the required attribution string for O*NET Web Services content. 17// O*NET Web Services requires this notice on any product using their data. 18// The exact text must be reproduced verbatim including its internal punctuation. 19const ONETAttribution = "O*NET Web Services by the U.S. Department of Labor, Employment and Training Administration (USDOL/ETA). Used under the CC BY 4.0 license. O*NET is a trademark of USDOL/ETA." 20 21// onetBaseURL is the O*NET Web Services v2.0 base URL. It is the default used by NewONETSource 22// and can be overridden for tests by setting the baseURL field directly after construction. 23const onetBaseURL = "https://api-v2.onetcenter.org" 24 25// onetMaxRetries is the bounded number of retry attempts on a 429 response before giving up 26// and returning ontology.ErrThrottled. 27const onetMaxRetries = 3 28 29// onetRetryDelay is the back off duration applied between 429 retry attempts (about 200ms). 30const onetRetryDelay = 200 * time.Millisecond 31 32// onetMinInterval is the minimum time between requests, enforcing a ceiling of 5 requests 33// per second without requiring an external dependency. 34const onetMinInterval = 200 * time.Millisecond 35 36// ONETSource ingests the O*NET Web Services v2.0 occupation skills, knowledge, abilities, and 37// technology skills into the ontology store. Skills are yielded as KindSkill records; knowledge 38// and abilities are yielded as KindKSA records; technology skills (concrete software tools) are 39// yielded as KindSkill records. Each element kind is globally deduped: a skill element such as 40// Active Listening appears once no matter how many occupations list it, and a technology tool 41// such as Apache Kafka appears once regardless of how many occupations reference it. 42// 43// This adapter is DORMANT until an ONET_API_KEY is [credential removed] The CLI constructs it only when 44// the key is present. The unit tests cover the pure parser functions only and do not call the 45// live API. 46// 47// Attribution: this adapter incorporates information from O*NET Web Services by the U.S. 48// Department of Labor (see ONETAttribution for the full required notice). 49type ONETSource struct { 50 apiKey string 51 baseURL string 52 client *http.Client 53 lastCall time.Time 54} 55 56// NewONETSource builds the adapter. apiKey is the O*NET developer key; it is sent only in the 57// X-API-Key request header, never in the query string or body. client is the HTTP client; pass 58// nil for a sensible 30 second default. The base URL defaults to the O*NET v2.0 production 59// endpoint and can be overridden by setting the baseURL field after construction. 60func NewONETSource(apiKey string, client *http.Client) *ONETSource { 61 if client == nil { 62 client = &http.Client{Timeout: 30 * time.Second} 63 } 64 return &ONETSource{ 65 apiKey: [credential removed], 66 baseURL: onetBaseURL, 67 client: client, 68 } 69} 70 71// Name is the provenance label stamped on every record this source yields. 72func (s *ONETSource) Name() string { return "O*NET Web Services" } 73 74// Fetch lists all O*NET occupations, then for each occupation pulls skills, knowledge, abilities, 75// and technology skills, yielding one VocabRecord per distinct element. A global seen map ensures 76// each unique SourceRecordID is yielded exactly once: generic competencies (skills, knowledge, 77// abilities) and technology tools that appear across many occupations are stored as one row, not 78// one row per occupation. Requests are throttled to about 5 per second. A 429 that cannot be 79// recovered within onetMaxRetries attempts returns ontology.ErrThrottled. 80// 81// This method incorporates information from O*NET Web Services by the U.S. Department of Labor. 82func (s *ONETSource) Fetch(ctx context.Context, yield func(ontology.VocabRecord) error) error { 83 log.Print("onet: fetch started, incorporates information from O*NET Web Services by the U.S. Department of Labor") 84 85 codes, err := s.listOccupations(ctx) 86 if err != nil { 87 return fmt.Errorf("onet: list occupations: %w", err) 88 } 89 90 type elementSpec struct { 91 path string 92 category string 93 kind ontology.TermKind 94 isTech bool 95 } 96 97 specs := []elementSpec{ 98 {path: "skills", category: "skill", kind: ontology.KindSkill}, 99 {path: "knowledge", category: "knowledge", kind: ontology.KindKSA}, 100 {path: "abilities", category: "ability", kind: ontology.KindKSA}, 101 {path: "technology_skills", kind: ontology.KindSkill, isTech: true}, 102 } 103 104 // seen tracks SourceRecordIDs that have already been yielded. Any record whose key appears 105 // in seen is skipped so the store receives one upsert per distinct competency or tool, not 106 // one per occupation that mentions it (roughly 1016 repetitions for common elements). 107 seen := make(map[string]bool) 108 109 for _, code := range codes { 110 for _, spec := range specs { 111 url := fmt.Sprintf("%s/online/occupations/%s/summary/%s", s.baseURL, code, spec.path) 112 data, err := s.getWithRetry(ctx, url) 113 if err != nil { 114 return fmt.Errorf("onet: occupation %s %s: %w", code, spec.path, err) 115 } 116 117 var recs []ontology.VocabRecord 118 if spec.isTech { 119 recs, err = parseONETTechSkills(data, code) 120 } else { 121 recs, err = parseONETElement(data, code, spec.category, spec.kind) 122 } 123 if err != nil { 124 return fmt.Errorf("onet: parse occupation %s %s: %w", code, spec.path, err) 125 } 126 127 for _, r := range recs { 128 if seen[r.SourceRecordID] { 129 continue 130 } 131 seen[r.SourceRecordID] = true 132 if err := yield(r); err != nil { 133 return err 134 } 135 } 136 } 137 } 138 return nil 139} 140 141// listOccupations fetches all pages of /online/occupations/ and returns the O*NET SOC codes. 142func (s *ONETSource) listOccupations(ctx context.Context) ([]string, error) { 143 var codes []string 144 start := 1 145 for { 146 url := fmt.Sprintf("%s/online/occupations/?start=%d&end=%d", s.baseURL, start, start+99) 147 data, err := s.getWithRetry(ctx, url) 148 if err != nil { 149 return nil, err 150 } 151 page, err := parseONETOccupations(data) 152 if err != nil { 153 return nil, err 154 } 155 if len(page) == 0 { 156 break 157 } 158 codes = append(codes, page...) 159 160 var pag onetPagination 161 if err := json.Unmarshal(data, &pag); err != nil { 162 return nil, fmt.Errorf("onet: decode pagination: %w", err) 163 } 164 if pag.Next == "" || pag.End >= pag.Total { 165 break 166 } 167 start = pag.End + 1 168 } 169 return codes, nil 170} 171 172// throttle enforces a minimum interval between requests (about 5 per second) using a simple 173// time.Since check and time.Sleep. No external package is needed. It respects ctx cancellation. 174func (s *ONETSource) throttle(ctx context.Context) error { 175 elapsed := time.Since(s.lastCall) 176 if elapsed < onetMinInterval { 177 wait := onetMinInterval - elapsed 178 select { 179 case <-ctx.Done(): 180 return ctx.Err() 181 case <-time.After(wait): 182 } 183 } 184 s.lastCall = time.Now() 185 return nil 186} 187 188// getWithRetry performs a GET with the X-API-Key header, respects the rate limiter, and retries 189// up to onetMaxRetries times on a 429 response. If still throttled after all retries it returns 190// ontology.ErrThrottled. 191func (s *ONETSource) getWithRetry(ctx context.Context, url string) ([]byte, error) { 192 var lastStatus int 193 for attempt := 0; attempt <= onetMaxRetries; attempt++ { 194 if attempt > 0 { 195 select { 196 case <-ctx.Done(): 197 return nil, ctx.Err() 198 case <-time.After(onetRetryDelay): 199 } 200 } 201 202 if err := s.throttle(ctx); err != nil { 203 return nil, fmt.Errorf("onet: throttle: %w", err) 204 } 205 206 data, status, err := s.get(ctx, url) 207 if err != nil { 208 return nil, err 209 } 210 if status == http.StatusTooManyRequests { 211 lastStatus = status 212 continue 213 } 214 if status != http.StatusOK { 215 return nil, fmt.Errorf("onet: provider returned status %d for %s", status, url) 216 } 217 return data, nil 218 } 219 if lastStatus == http.StatusTooManyRequests { 220 return nil, ontology.ErrThrottled 221 } 222 return nil, fmt.Errorf("onet: exhausted retries for %s", url) 223} 224 225// get performs one HTTP GET with the X-API-Key header. The key is sent ONLY in the header, 226// never in the query string or body, as required by the O*NET Web Services terms of service. 227func (s *ONETSource) get(ctx context.Context, url string) ([]byte, int, error) { 228 req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) 229 if err != nil { 230 return nil, 0, fmt.Errorf("onet: new request: %w", err) 231 } 232 req.Header.Set("X-API-Key", s.apiKey) 233 234 resp, err := s.client.Do(req) 235 if err != nil { 236 return nil, 0, fmt.Errorf("onet: do request: %w", err) 237 } 238 defer resp.Body.Close() 239 240 body, err := io.ReadAll(resp.Body) 241 if err != nil { 242 return nil, resp.StatusCode, fmt.Errorf("onet: read body: %w", err) 243 } 244 return body, resp.StatusCode, nil 245} 246 247// onetPagination holds the pagination envelope shared across O*NET v2.0 list responses. 248type onetPagination struct { 249 Start int `json:"start"` 250 End int `json:"end"` 251 Total int `json:"total"` 252 Next string `json:"next"` 253} 254 255// onetOccupationsResponse mirrors the O*NET v2.0 occupations list response: 256// 257// GET /online/occupations/ 258// { occupation: [{ code, title, href, ... }], start, end, total, next } 259// 260// Confirmed from https://services.onetcenter.org/reference/online/browse/occupations. 261type onetOccupationsResponse struct { 262 onetPagination 263 Occupation []struct { 264 Code string `json:"code"` 265 Title string `json:"title"` 266 } `json:"occupation"` 267} 268 269// parseONETOccupations decodes an O*NET occupations list response and returns the SOC codes. 270// It is pure (no I/O) so it is directly unit testable. 271func parseONETOccupations(data []byte) ([]string, error) { 272 var doc onetOccupationsResponse 273 if err := json.Unmarshal(data, &doc); err != nil { 274 return nil, fmt.Errorf("onet: decode occupations: %w", err) 275 } 276 codes := make([]string, 0, len(doc.Occupation)) 277 for _, occ := range doc.Occupation { 278 c := strings.TrimSpace(occ.Code) 279 if c != "" { 280 codes = append(codes, c) 281 } 282 } 283 return codes, nil 284} 285 286// onetElementResponse mirrors the O*NET v2.0 per occupation skills, knowledge, and abilities 287// summary response shape: 288// 289// GET /online/occupations/{code}/summary/{skills|knowledge|abilities} 290// { element: [{ id, name, description, related }], start, end, total, next } 291// 292// Confirmed from https://services.onetcenter.org/reference/online/occupation/summary/skills, 293// https://services.onetcenter.org/reference/online/occupation/summary/knowledge, and 294// https://services.onetcenter.org/reference/online/occupation/summary/abilities. 295// The "id" field is the O*NET content model element id (e.g. "2.A.1.b" for Active Listening). 296// The "name" field is the human readable element name. 297type onetElementResponse struct { 298 onetPagination 299 Element []struct { 300 ID string `json:"id"` 301 Name string `json:"name"` 302 Description string `json:"description"` 303 } `json:"element"` 304} 305 306// parseONETElement decodes a per occupation skills, knowledge, or abilities response and returns 307// one VocabRecord per element. SourceRecordID uses the global "onet:<elementID>" scheme so the 308// same competency from different occupations carries the same key, enabling dedup in Fetch via the 309// seen map. category is the element type label ("skill", "knowledge", "ability"). kind is the 310// TermKind to stamp on each record. It is pure (no I/O) so it is directly unit testable. 311func parseONETElement(data []byte, socCode, category string, kind ontology.TermKind) ([]ontology.VocabRecord, error) { 312 var doc onetElementResponse 313 if err := json.Unmarshal(data, &doc); err != nil { 314 return nil, fmt.Errorf("onet: decode %s elements: %w", category, err) 315 } 316 out := make([]ontology.VocabRecord, 0, len(doc.Element)) 317 for _, el := range doc.Element { 318 id := strings.TrimSpace(el.ID) 319 name := strings.TrimSpace(el.Name) 320 if id == "" || name == "" { 321 continue 322 } 323 raw, _ := json.Marshal(map[string]string{ 324 "soc_code": socCode, 325 "element_id": id, 326 "name": name, 327 "description": el.Description, 328 "category": category, 329 }) 330 out = append(out, ontology.VocabRecord{ 331 SourceName: "O*NET Web Services", 332 SourceRecordID: "onet:" + id, 333 Kind: kind, 334 Term: name, 335 Code: id, 336 Category: category, 337 Description: strings.TrimSpace(el.Description), 338 Raw: raw, 339 }) 340 } 341 return out, nil 342} 343 344// onetTechSkillsResponse mirrors the O*NET v2.0 per occupation technology_skills summary response: 345// 346// GET /online/occupations/{code}/summary/technology_skills 347// { start, end, total, next, category: [{ code, title, example: [{ title, hot_technology }] }] } 348// 349// Confirmed from https://services.onetcenter.org/reference/online/occupation/summary/technology_skills. 350// Each category groups a class of software tools; each example is a concrete product title. 351// The response paginates via the shared "next" / "end" / "total" envelope. 352type onetTechSkillsResponse struct { 353 onetPagination 354 Category []struct { 355 Code int `json:"code"` 356 Title string `json:"title"` 357 Example []struct { 358 Title string `json:"title"` 359 HotTechnology bool `json:"hot_technology"` 360 } `json:"example"` 361 } `json:"category"` 362} 363 364// parseONETTechSkills decodes a per occupation technology_skills response and returns one KindSkill 365// VocabRecord per example title. SourceRecordID uses "onet-tech:" plus the lowercased, trimmed 366// example title so the same tool (e.g. "Apache Kafka") from different occupations carries the same 367// key and is deduped by the seen map in Fetch. Category is set to the tool category title (e.g. 368// "Development environment software"). Code is empty (tech skills have no O*NET content model id). 369// Raw is a compact JSON object carrying the SOC code, category title, and tool title. 370// It is pure (no I/O) so it is directly unit testable. 371func parseONETTechSkills(data []byte, socCode string) ([]ontology.VocabRecord, error) { 372 var doc onetTechSkillsResponse 373 if err := json.Unmarshal(data, &doc); err != nil { 374 return nil, fmt.Errorf("onet: decode technology_skills: %w", err) 375 } 376 var out []ontology.VocabRecord 377 for _, cat := range doc.Category { 378 catTitle := strings.TrimSpace(cat.Title) 379 for _, ex := range cat.Example { 380 title := strings.TrimSpace(ex.Title) 381 if title == "" { 382 continue 383 } 384 key := "onet-tech:" + strings.ToLower(title) 385 raw, _ := json.Marshal(map[string]string{ 386 "soc_code": socCode, 387 "category": catTitle, 388 "title": title, 389 }) 390 out = append(out, ontology.VocabRecord{ 391 SourceName: "O*NET Web Services", 392 SourceRecordID: key, 393 Kind: ontology.KindSkill, 394 Term: title, 395 Code: "", 396 Category: catTitle, 397 Description: "", 398 Raw: raw, 399 }) 400 } 401 } 402 return out, nil 403} 404