addrkit/internal/seed/seed.go

// Package seed turns a registry dump into canonical address entries. It is the
// only place that knows how the source data is shaped.
//
// The same parser used at query time is used here, so the index and the query
// never disagree about how an address decomposes.
package seed

import (
	"bufio"
	"encoding/json"
	"io"
	"sort"
	"strings"

	"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
)

// Source is one raw registry row.
type Source struct {
	ID       string
	Address  string
	District string
	// City is optional: when the dump is already split per city it may be
	// omitted and supplied as the default city to Build.
	City string
}

// Stats summarises a build and, crucially, surfaces rows the parser could not
// structure. A non-empty UnparsedCount is a data-quality signal worth alerting
// on, not an error.
type Stats struct {
	Total           int            `json:"total"`
	Parsed          int            `json:"parsed"`
	Unparsed        int            `json:"unparsed"`
	Duplicates      int            `json:"duplicates"`
	DistinctStreets int            `json:"distinct_streets"`
	Cities          map[string]int `json:"cities,omitempty"`
	UnparsedSample  []string       `json:"unparsed_sample,omitempty"`
}

// BuildOptions controls how source rows become canonical entries.
type BuildOptions struct {
	// Parser defaults to address.DefaultParser().
	Parser *address.Parser
	// DefaultCity is used for rows that neither carry a city field nor name a
	// city in the address. Leave it empty for multi-city dumps where every row
	// is expected to identify its city.
	DefaultCity string
}

// ReadBulkNDJSON reads an Elasticsearch Bulk NDJSON dump: pairs of an action
// line ({"index":{"_id":...}}) followed by a source line
// ({"address":...,"district_name":...,"city":...}). Bare source lines (no
// action line) are also accepted, which makes plain JSONL usable too.
func ReadBulkNDJSON(r io.Reader) ([]Source, error) {
	sc := bufio.NewScanner(r)
	// 1 MiB per line is far beyond any realistic address row.
	sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)

	var (
		sources   []Source
		pendingID string
	)
	for sc.Scan() {
		line := strings.TrimSpace(sc.Text())
		if line == "" {
			continue
		}
		var obj map[string]json.RawMessage
		if err := json.Unmarshal([]byte(line), &obj); err != nil {
			continue // skip malformed lines rather than aborting the whole seed
		}

		if action, ok := obj["index"]; ok {
			var a struct {
				ID string `json:"_id"`
			}
			_ = json.Unmarshal(action, &a)
			pendingID = a.ID
			continue
		}

		if addrRaw, ok := obj["address"]; ok {
			var addr string
			if err := json.Unmarshal(addrRaw, &addr); err != nil || addr == "" {
				continue
			}
			src := Source{ID: pendingID, Address: addr}
			if d, ok := obj["district_name"]; ok {
				_ = json.Unmarshal(d, &src.District)
			}
			src.City = firstString(obj, "city", "city_name")
			sources = append(sources, src)
			pendingID = ""
		}
	}
	if err := sc.Err(); err != nil {
		return nil, err
	}
	return sources, nil
}

// firstString returns the first non-empty string field among names.
func firstString(obj map[string]json.RawMessage, names ...string) string {
	for _, name := range names {
		raw, ok := obj[name]
		if !ok {
			continue
		}
		var s string
		if err := json.Unmarshal(raw, &s); err == nil && strings.TrimSpace(s) != "" {
			return s
		}
	}
	return ""
}

// Build parses every source row into a canonical Entry and reports statistics.
// Rows without a house number are still indexed for street lookups but counted
// as unparsed, since a registry row without a house is suspicious.
func Build(sources []Source, opts BuildOptions) ([]address.Entry, Stats) {
	parser := opts.Parser
	if parser == nil {
		parser = address.NewParser(address.DefaultDictionary(), opts.DefaultCity)
	}
	entries := make([]address.Entry, 0, len(sources))
	streets := make(map[string]struct{})
	seen := make(map[string]struct{})
	stats := Stats{Total: len(sources), Cities: make(map[string]int)}

	for _, src := range sources {
		// Precedence: an explicit source field wins over a city found in the
		// address text; the parser's default city is the last resort. Passing
		// an empty hint is what lets the parser apply that precedence.
		cityHint := strings.TrimSpace(src.City)
		p := parser.ParseInCity(src.Address, cityHint)
		if p.NameKey == "" {
			stats.Unparsed++
			appendSample(&stats, src.Address)
			continue
		}
		// Count parse quality on every source row, before de-duplication.
		if p.HasHouse {
			stats.Parsed++
		} else {
			stats.Unparsed++
			appendSample(&stats, src.Address)
		}

		entry := address.Entry{
			ID:             namespacedID(src.ID, p.CityKey),
			Address:        src.Address,
			District:       src.District,
			City:           p.City,
			CityKey:        p.CityKey,
			Ordinal:        p.Ordinal,
			NameKey:        p.NameKey,
			NameDisplay:    p.NameDisplay,
			StreetType:     p.StreetType,
			StreetKey:      p.StreetKey,
			House:          p.House,
			HouseFrac:      p.HouseFrac,
			HouseLetter:    p.HouseLetter,
			Building:       p.Building,
			BuildingFrac:   p.BuildingFrac,
			BuildingLetter: p.BuildingLetter,
			Structure:      p.Structure,
		}

		// The source registry contains rows that differ only by "стр."
		// (structure). They are distinct buildings, so structure is part of
		// the de-duplication key and an exact structure query can be honoured.
		if key := dedupKey(entry); key != "" {
			if _, ok := seen[key]; ok {
				stats.Duplicates++
				continue
			}
			seen[key] = struct{}{}
		}

		entries = append(entries, entry)
		streets[p.StreetKey] = struct{}{}
		stats.Cities[cityLabel(p.City)]++
	}
	stats.DistinctStreets = len(streets)
	sort.Strings(stats.UnparsedSample)
	return entries, stats
}

// namespacedID keeps source ids unique when several city registries are merged
// into one database. Source systems often number houses independently.
func namespacedID(id, cityKey string) string {
	if id == "" || cityKey == "" {
		return id
	}
	return cityKey + "/" + id
}

// dedupKey identifies the same house unit within the same city and street,
// including its structure and building components.
func dedupKey(e address.Entry) string {
	if e.House == "" {
		return ""
	}
	return strings.Join([]string{
		e.CityKey, e.StreetKey, e.House, e.HouseFrac, e.HouseLetter,
		e.Building, e.BuildingFrac, e.BuildingLetter, e.Structure,
	}, "|")
}

func cityLabel(city string) string {
	if city == "" {
		return "(none)"
	}
	return city
}

func appendSample(s *Stats, address string) {
	const maxSamples = 20
	if len(s.UnparsedSample) < maxSamples {
		s.UnparsedSample = append(s.UnparsedSample, address)
	}
}