refactor!: make addrkit a multi-city address normalizer

- remove the events/outages module, ArcGIS adapter, portal config and event statuses
- add a configurable city dictionary/parser and make City part of StreetKey
- seed: read city/city_name, namespace IDs by city, keep structures distinct
- fix reliability: schema v2 with fail-closed load, deterministic ordering,
  input limits, config validation, atomic DB publish, LLM response caps
- add read-only shadow mode comparing the old ES path with addrkit
- breaking: registry.db must be re-seeded; config has no portal section
This commit is contained in:
Maksim Totmin
2026-10-27 14:35:51 +07:00
parent a0b4cb2b14
commit 17ff5c591f
38 changed files with 2978 additions and 1164 deletions
+114 -15
View File
@@ -38,6 +38,8 @@ func main() {
os.Exit(runEval(os.Args[2:]))
case "selfcheck":
os.Exit(runSelfcheck(os.Args[2:]))
case "shadow":
os.Exit(runShadow(os.Args[2:]))
case "help", "-h", "--help":
usage()
os.Exit(0)
@@ -49,14 +51,15 @@ func main() {
}
func usage() {
fmt.Fprint(os.Stderr, `addrkit — address registry toolkit
fmt.Fprint(os.Stderr, `addrkit — multi-city address registry toolkit
Usage:
addrkit seed -in registry.ndjson -out registry.db [-version 2026-10-09]
addrkit seed -in registry.ndjson -out registry.db [-version 2026-10-09] [-city "Новосибирск"] [-cities cities.json]
addrkit verify -db registry.db
addrkit resolve -db registry.db "2-я Станционная улица 1/1 стр."
addrkit eval -db registry.db [-sample 500] [-seed 1]
addrkit resolve -db registry.db [-city "Новосибирск"] [-cities cities.json] "2-я Станционная улица 1/1 стр."
addrkit eval -db registry.db [-sample 500] [-seed 1] [-cities cities.json]
addrkit selfcheck -db registry.db
addrkit shadow -db registry.db -es https://es:9200 [-sample 300] [-log file.ndjson]
`)
}
@@ -65,6 +68,8 @@ func runSeed(args []string) int {
in := fs.String("in", "", "input registry dump (Bulk NDJSON or JSONL)")
out := fs.String("out", "", "output SQLite database")
version := fs.String("version", "", "data version (defaults to input file mtime)")
city := fs.String("city", "", "default city for rows without a city field")
citiesPath := fs.String("cities", "", "optional JSON file with the city dictionary")
_ = fs.Parse(args)
if *in == "" || *out == "" {
@@ -77,6 +82,12 @@ func runSeed(args []string) int {
}
}
cities, err := loadCities(*citiesPath)
if err != nil {
fmt.Fprintf(os.Stderr, "seed: %v\n", err)
return 1
}
f, err := os.Open(*in)
if err != nil {
fmt.Fprintf(os.Stderr, "seed: open input: %v\n", err)
@@ -89,7 +100,11 @@ func runSeed(args []string) int {
fmt.Fprintf(os.Stderr, "seed: read: %v\n", err)
return 1
}
entries, stats := seed.Build(sources)
parser := address.NewParser(address.NewDictionary(cities), *city)
entries, stats := seed.Build(sources, seed.BuildOptions{
Parser: parser,
DefaultCity: *city,
})
if stats.Parsed == 0 {
fmt.Fprintln(os.Stderr, "seed: no parseable entries; refusing to write empty index")
return 1
@@ -109,6 +124,15 @@ func runSeed(args []string) int {
return 0
}
// loadCities reads an optional city dictionary file. An empty path is valid and
// means "no city names beyond those in the address".
func loadCities(path string) ([]address.City, error) {
if path == "" {
return nil, nil
}
return config.LoadCitiesFile(path)
}
func runVerify(args []string) int {
fs := flag.NewFlagSet("verify", flag.ExitOnError)
db := fs.String("db", "", "SQLite database")
@@ -123,8 +147,10 @@ func runVerify(args []string) int {
return 1
}
streets := make(map[string]struct{})
cities := make(map[string]int)
for _, e := range entries {
streets[e.StreetKey] = struct{}{}
cities[e.City]++
}
enc := json.NewEncoder(os.Stdout)
enc.SetIndent("", " ")
@@ -132,6 +158,7 @@ func runVerify(args []string) int {
"version": version,
"entries": len(entries),
"distinct_streets": len(streets),
"cities": cities,
})
return 0
}
@@ -139,6 +166,8 @@ func runVerify(args []string) int {
func runResolve(args []string) int {
fs := flag.NewFlagSet("resolve", flag.ExitOnError)
db := fs.String("db", "", "SQLite database")
city := fs.String("city", "", "optional city hint")
citiesPath := fs.String("cities", "", "optional JSON file with the city dictionary")
llmURL := fs.String("llm-url", "", "enable the LLM rewriter with this base URL")
llmModel := fs.String("llm-model", "qwen3-14b-awq", "LLM model name")
llmToken := fs.String("llm-token", "", "LLM auth token")
@@ -148,6 +177,11 @@ func runResolve(args []string) int {
fmt.Fprintln(os.Stderr, `resolve: -db and an address are required`)
return 2
}
cities, err := loadCities(*citiesPath)
if err != nil {
fmt.Fprintf(os.Stderr, "resolve: %v\n", err)
return 1
}
version, entries, err := registrystore.Load(*db)
if err != nil {
fmt.Fprintf(os.Stderr, "resolve: %v\n", err)
@@ -160,8 +194,13 @@ func runResolve(args []string) int {
TimeoutSeconds: 5, MaxTokens: 512, DisableThinking: true,
})
}
resolver := match.NewResolver(match.NewIndex(version, entries, match.DefaultOptions()), rewriter)
res := resolver.Resolve(context.Background(), rest[0])
parser := address.NewParser(address.NewDictionary(cities), "")
resolver := match.NewResolverWithOptions(
match.NewIndex(version, entries, match.DefaultOptions()),
rewriter,
match.ResolverOptions{Parser: parser},
)
res := resolver.ResolveInCity(context.Background(), rest[0], *city)
enc := json.NewEncoder(os.Stdout)
enc.SetIndent("", " ")
_ = enc.Encode(res)
@@ -177,6 +216,7 @@ func runEval(args []string) int {
db := fs.String("db", "", "SQLite database")
sample := fs.Int("sample", 500, "number of registry entries to sample")
seedVal := fs.Int64("seed", 1, "random seed")
citiesPath := fs.String("cities", "", "optional JSON file with the city dictionary")
llmURL := fs.String("llm-url", "", "enable the LLM rewriter with this base URL")
llmModel := fs.String("llm-model", "qwen3-14b-awq", "LLM model name")
llmToken := fs.String("llm-token", "", "LLM auth token")
@@ -186,11 +226,19 @@ func runEval(args []string) int {
fmt.Fprintln(os.Stderr, "eval: -db is required")
return 2
}
cities, err := loadCities(*citiesPath)
if err != nil {
fmt.Fprintf(os.Stderr, "eval: %v\n", err)
return 1
}
_, entries, err := registrystore.Load(*db)
if err != nil {
fmt.Fprintf(os.Stderr, "eval: %v\n", err)
return 1
}
cities = mergeCities(cities, entries)
defaultCity := soleCity(entries)
parser := address.NewParser(address.NewDictionary(cities), defaultCity)
ix := match.NewIndex("eval", entries, match.DefaultOptions())
var rewriter match.Rewriter
@@ -208,7 +256,7 @@ func runEval(args []string) int {
rewriter = counter
fmt.Fprintf(os.Stderr, "llm rewriter enabled: %s (%s)\n", *llmURL, *llmModel)
}
resolver := match.NewResolver(ix, rewriter)
resolver := match.NewResolverWithOptions(ix, rewriter, match.ResolverOptions{Parser: parser})
rng := rand.New(rand.NewSource(*seedVal))
cases := eval.Generate(entries, *sample, rng)
@@ -286,14 +334,22 @@ func runSelfcheck(args []string) int {
fmt.Fprintf(os.Stderr, "selfcheck: %v\n", err)
return 1
}
resolver := match.NewResolver(match.NewIndex("selfcheck", entries, match.DefaultOptions()), nil)
parser := address.NewParser(
address.NewDictionary(mergeCities(nil, entries)),
soleCity(entries),
)
resolver := match.NewResolverWithOptions(
match.NewIndex("selfcheck", entries, match.DefaultOptions()),
nil,
match.ResolverOptions{Parser: parser},
)
exact := 0
byStatus := map[string]int{}
var failures []string
for _, e := range entries {
res := resolver.Resolve(context.Background(), e.Address)
if res.Status == address.StatusFound && sameEntry(res.Address, e) {
res := resolver.ResolveInCity(context.Background(), e.Address, e.City)
if res.Status == address.StatusFound && res.City == e.City && sameEntry(res.Address, e, parser) {
exact++
continue
}
@@ -316,10 +372,53 @@ func runSelfcheck(args []string) int {
}
// sameEntry reports whether a resolved address is the same street and house
// unit as the registry entry it came from.
func sameEntry(resolved string, e address.Entry) bool {
p := address.Parse(resolved)
// unit as the registry entry it came from. City is checked separately against
// Result.City by the caller; forcing it here only keeps the parse city-aware.
func sameEntry(resolved string, e address.Entry, parser *address.Parser) bool {
p := parser.ParseInCity(resolved, e.City)
return p.StreetKey == e.StreetKey &&
p.House == e.House && p.HouseFrac == e.HouseFrac && p.HouseLetter == e.HouseLetter &&
p.Building == e.Building && p.BuildingFrac == e.BuildingFrac && p.BuildingLetter == e.BuildingLetter
p.Building == e.Building && p.BuildingFrac == e.BuildingFrac && p.BuildingLetter == e.BuildingLetter &&
p.Structure == e.Structure
}
// mergeCities augments an explicit city dictionary with the canonical names
// found in the index. The names are enough for parsing addresses that include
// the city; aliases still come from the dictionary file.
func mergeCities(base []address.City, entries []address.Entry) []address.City {
seen := make(map[string]struct{}, len(base))
for _, c := range base {
seen[address.NormalizeText(c.Name)] = struct{}{}
}
out := append([]address.City(nil), base...)
for _, e := range entries {
if e.City == "" {
continue
}
key := address.NormalizeText(e.City)
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
out = append(out, address.City{Name: e.City})
}
return out
}
// soleCity returns the only city in the index, or "" when the index is
// city-less or multi-city.
func soleCity(entries []address.Entry) string {
names := make(map[string]string)
for _, e := range entries {
if e.City != "" {
names[e.CityKey] = e.City
}
}
if len(names) != 1 {
return ""
}
for _, name := range names {
return name
}
return ""
}
+685
View File
@@ -0,0 +1,685 @@
package main
// Shadow mode: run the old EДДС path (Elasticsearch fuzzy search) and addrkit
// on the same queries, without changing any production answer.
//
// The Elasticsearch side is strictly read-only: this file only issues
// `_search` requests. Nothing is indexed, updated or deleted.
//
// Two datasets are supported:
// - generated: realistic perturbations of real registry rows (eval harness);
// - log: real user utterances from the EДДС address log (file or ES index).
//
// The report compares the old top-1 guess with the new deterministic verdict.
// The key number is "old_wrong_new_safe": cases where the old path returned a
// different house and the new one refused to guess.
import (
"bytes"
"context"
"crypto/tls"
"encoding/json"
"flag"
"fmt"
"io"
"math/rand"
"net/http"
"os"
"sort"
"strings"
"sync"
"time"
"git.totmin.ru/en2zmax/addrkit/internal/adapters/registrystore"
"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
"git.totmin.ru/en2zmax/addrkit/internal/domain/match"
"git.totmin.ru/en2zmax/addrkit/internal/eval"
)
type shadowOptions struct {
db string
city string
citiesPath string
esURL string
esUser string
esPass string
esIndex string
insecure bool
timeout time.Duration
workers int
sample int
seed int64
logPath string
logField string
logESIndex string
logLimit int
maxExamples int
oldPreproc bool
out string
}
func runShadow(args []string) int {
fs := flag.NewFlagSet("shadow", flag.ExitOnError)
db := fs.String("db", "", "local registry.db (new path)")
city := fs.String("city", "", "default city for the new path")
citiesPath := fs.String("cities", "", "optional JSON city dictionary")
esURL := fs.String("es", envOr("ADDRKIT_ES_URL", ""), "Elasticsearch base URL (old path, read-only)")
esUser := fs.String("es-user", envOr("ADDRKIT_ES_USER", ""), "Elasticsearch user")
esPass := fs.String("es-pass", envOr("ADDRKIT_ES_PASSWORD", ""), "Elasticsearch password")
esIndex := fs.String("es-index", "address_registry", "Elasticsearch address index")
insecure := fs.Bool("insecure", false, "skip TLS verification for Elasticsearch")
timeout := fs.Duration("timeout", 10*time.Second, "per-request timeout")
workers := fs.Int("workers", 8, "parallel queries")
sample := fs.Int("sample", 300, "registry entries to perturb (each yields several queries)")
seed := fs.Int64("seed", 1, "random seed for generated queries")
logPath := fs.String("log", "", "address log bulk NDJSON (real utterances)")
logField := fs.String("log-field", "address", "log field to replay: address|request")
logESIndex := fs.String("log-es-index", "", "fetch real utterances from this ES index instead of a file")
logLimit := fs.Int("log-limit", 500, "documents to fetch from -log-es-index")
maxExamples := fs.Int("max-examples", 20, "examples per category in the report")
oldPreproc := fs.Bool("old-preprocess", true, "apply the robot's administrative-word stripping to the old ES query")
out := fs.String("out", "", "write the JSON report to this file (default stdout)")
_ = fs.Parse(args)
if *db == "" || *esURL == "" {
fmt.Fprintln(os.Stderr, "shadow: -db and -es (or ADDRKIT_ES_URL) are required")
return 2
}
if *workers < 1 {
*workers = 1
}
cities, err := loadCities(*citiesPath)
if err != nil {
fmt.Fprintf(os.Stderr, "shadow: %v\n", err)
return 1
}
_, entries, err := registrystore.Load(*db)
if err != nil {
fmt.Fprintf(os.Stderr, "shadow: load registry: %v\n", err)
return 1
}
parser := address.NewParser(address.NewDictionary(cities), *city)
resolver := match.NewResolverWithOptions(
match.NewIndex("shadow", entries, match.DefaultOptions()),
nil,
match.ResolverOptions{Parser: parser},
)
opts := shadowOptions{
db: *db, city: *city, citiesPath: *citiesPath,
esURL: *esURL, esUser: *esUser, esPass: *esPass, esIndex: *esIndex,
insecure: *insecure, timeout: *timeout, workers: *workers,
sample: *sample, seed: *seed, logPath: *logPath, logField: *logField,
logESIndex: *logESIndex, logLimit: *logLimit, maxExamples: *maxExamples,
oldPreproc: *oldPreproc, out: *out,
}
cases, mode, note, err := buildShadowCases(opts, entries, parser)
if err != nil {
fmt.Fprintf(os.Stderr, "shadow: %v\n", err)
return 1
}
if len(cases) == 0 {
fmt.Fprintln(os.Stderr, "shadow: no queries to replay")
return 1
}
client := newESClient(opts)
report := runShadowQueries(opts, cases, mode, note, parser, resolver, client)
data, err := json.MarshalIndent(report, "", " ")
if err != nil {
fmt.Fprintf(os.Stderr, "shadow: encode report: %v\n", err)
return 1
}
if opts.out != "" {
if err := os.WriteFile(opts.out, data, 0o600); err != nil {
fmt.Fprintf(os.Stderr, "shadow: write report: %v\n", err)
return 1
}
}
fmt.Println(string(data))
if report.New.Wrong > 0 {
return 1 // a wrong new answer is a blocker, make it visible to CI
}
return 0
}
func envOr(name, fallback string) string {
if v := strings.TrimSpace(os.Getenv(name)); v != "" {
return v
}
return fallback
}
type shadowCase struct {
Query string
Expect string
ExpectCity string
RequireExact bool
Category string
}
func buildShadowCases(opts shadowOptions, entries []address.Entry, parser *address.Parser) ([]shadowCase, string, string, error) {
switch {
case opts.logESIndex != "":
rows, err := fetchLogCases(opts)
if err != nil {
return nil, "", "", fmt.Errorf("fetch address log from ES: %w", err)
}
return logCases(rows, opts.logField), "log", "raw utterances are compared without the SEE conversion step", nil
case opts.logPath != "":
rows, err := readLogCases(opts.logPath)
if err != nil {
return nil, "", "", fmt.Errorf("read address log: %w", err)
}
return logCases(rows, opts.logField), "log", "raw utterances are compared without the SEE conversion step", nil
default:
rng := rand.New(rand.NewSource(opts.seed))
src := eval.Generate(entries, opts.sample, rng)
out := make([]shadowCase, 0, len(src))
for _, c := range src {
out = append(out, shadowCase{
Query: c.Query, Expect: c.Expect, ExpectCity: c.ExpectCity,
RequireExact: c.RequireExact, Category: c.Category,
})
}
return out, "generated", "real registry rows perturbed by the eval harness", nil
}
}
type logRow struct {
Utterance string
Request string
}
func logCases(rows []logRow, field string) []shadowCase {
out := make([]shadowCase, 0, len(rows))
seen := map[string]struct{}{}
for _, r := range rows {
q := r.Utterance
if field == "request" {
q = r.Request
}
q = strings.TrimSpace(q)
if q == "" {
continue
}
if _, ok := seen[q]; ok {
continue
}
seen[q] = struct{}{}
out = append(out, shadowCase{Query: q, Category: "log"})
}
return out
}
func readLogCases(path string) ([]logRow, error) {
f, err := os.Open(path)
if err != nil {
return nil, err
}
defer f.Close()
dec := json.NewDecoder(f)
var rows []logRow
for {
var raw json.RawMessage
if err := dec.Decode(&raw); err != nil {
if err == io.EOF {
break
}
return nil, err
}
var probe map[string]json.RawMessage
if json.Unmarshal(raw, &probe) != nil {
continue
}
if _, ok := probe["index"]; ok {
continue
}
var rec struct {
Address string `json:"address"`
Request string `json:"request"`
}
_ = json.Unmarshal(raw, &rec)
var req struct {
Address string `json:"address"`
}
_ = json.Unmarshal([]byte(rec.Request), &req)
rows = append(rows, logRow{Utterance: rec.Address, Request: req.Address})
}
return rows, nil
}
func fetchLogCases(opts shadowOptions) ([]logRow, error) {
body := map[string]any{
"size": opts.logLimit,
"sort": []any{map[string]any{"start_date": map[string]string{"order": "desc"}}},
"_source": []string{"address", "request"},
}
data, err := esRequest(context.Background(), newESClient(opts), opts.esURL, opts.logESIndex, "_search", body)
if err != nil {
return nil, err
}
var resp struct {
Hits struct {
Hits []struct {
Source struct {
Address string `json:"address"`
Request string `json:"request"`
} `json:"_source"`
} `json:"hits"`
} `json:"hits"`
}
if err := json.Unmarshal(data, &resp); err != nil {
return nil, err
}
rows := make([]logRow, 0, len(resp.Hits.Hits))
for _, h := range resp.Hits.Hits {
var req struct {
Address string `json:"address"`
}
_ = json.Unmarshal([]byte(h.Source.Request), &req)
rows = append(rows, logRow{Utterance: h.Source.Address, Request: req.Address})
}
return rows, nil
}
type esClient struct {
http *http.Client
user string
pass string
}
func newESClient(opts shadowOptions) *esClient {
tr := &http.Transport{}
if opts.insecure {
tr.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} // #nosec G402 -- self-signed on-prem ES
}
return &esClient{
http: &http.Client{Timeout: opts.timeout, Transport: tr},
user: opts.esUser,
pass: opts.esPass,
}
}
// esRequest performs a read-only search. It only builds `_search` URLs.
func esRequest(ctx context.Context, c *esClient, base, index, path string, body any) ([]byte, error) {
raw, err := json.Marshal(body)
if err != nil {
return nil, err
}
url := strings.TrimRight(base, "/") + "/" + index + "/" + path
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, bytes.NewReader(raw))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
if c.user != "" {
req.SetBasicAuth(c.user, c.pass)
}
resp, err := c.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
data, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("ES %s: HTTP %d: %s", path, resp.StatusCode, truncate(string(data), 200))
}
return data, nil
}
func truncate(s string, n int) string {
if len(s) <= n {
return s
}
return s[:n]
}
// oldDeleteWords mirrors the administrative words the EДДС robot removes
// before calling Elasticsearch (robot script: DELETE_WORDS). Matching is
// token-based here, so "новосибирская" is not mangled by removing the
// substring "новосибирск".
var oldDeleteWords = map[string]struct{}{
"дзержинский": {}, "железнодорожный": {}, "заельцовский": {},
"калининский": {}, "кировский": {}, "ленинский": {}, "октябрьский": {},
"первомайский": {}, "советский": {}, "центральный": {},
"район": {}, "город": {}, "номер": {}, "новосибирск": {}, "дом": {},
"г": {}, "обл": {}, "область": {},
}
func preprocessOld(q string) string {
fields := strings.Fields(strings.ToLower(q))
out := fields[:0]
for _, tok := range fields {
tok = strings.Trim(tok, ".,;:()")
if tok == "" {
continue
}
if _, ok := oldDeleteWords[tok]; ok {
continue
}
out = append(out, tok)
}
return strings.Join(out, " ")
}
// oldQuery mirrors the EДДС robot's normalize_address query exactly.
func oldQuery(q string) map[string]any {
return map[string]any{
"query": map[string]any{
"bool": map[string]any{
"must": []any{map[string]any{
"multi_match": map[string]any{
"query": q, "fields": []string{"address"},
"fuzziness": "AUTO", "operator": "and", "prefix_length": 2,
},
}},
"should": []any{map[string]any{
"match_phrase": map[string]any{
"address": map[string]any{"query": q, "slop": 3, "boost": 2},
},
}},
},
},
"size": 1,
"_source": []string{"address", "district_name"},
}
}
type esHit struct {
Address string `json:"address"`
District string `json:"district_name"`
}
func oldSearch(ctx context.Context, c *esClient, opts shadowOptions, q string) (esHit, error) {
data, err := esRequest(ctx, c, opts.esURL, opts.esIndex, "_search", oldQuery(q))
if err != nil {
return esHit{}, err
}
var resp struct {
Hits struct {
Hits []struct {
Source esHit `json:"_source"`
} `json:"hits"`
} `json:"hits"`
}
if err := json.Unmarshal(data, &resp); err != nil {
return esHit{}, err
}
if len(resp.Hits.Hits) == 0 {
return esHit{}, nil
}
return resp.Hits.Hits[0].Source, nil
}
type shadowReport struct {
Mode string `json:"mode"`
Note string `json:"note"`
Queries int `json:"queries"`
Workers int `json:"workers"`
ESIndex string `json:"es_index"`
OldPreprocess bool `json:"old_preprocess"`
Old sideReport `json:"old"`
New sideReport `json:"new"`
Compare compareReport `json:"compare"`
Examples []shadowExample `json:"examples,omitempty"`
Errors int `json:"errors"`
ElapsedMS int64 `json:"elapsed_ms"`
}
type sideReport struct {
Found int `json:"found"`
Exact int `json:"exact"`
Wrong int `json:"wrong"`
Lossy int `json:"lossy"`
Partial int `json:"partial,omitempty"`
Ambiguous int `json:"ambiguous,omitempty"`
NotFound int `json:"not_found,omitempty"`
Error int `json:"error,omitempty"`
P50MS float64 `json:"p50_ms"`
P95MS float64 `json:"p95_ms"`
P99MS float64 `json:"p99_ms"`
}
type compareReport struct {
BothCorrect int `json:"both_correct"`
OldCorrectNewNotExact int `json:"old_correct_new_not_exact"`
OldWrongNewSafe int `json:"old_wrong_new_safe"`
OldWrongNewCorrect int `json:"old_wrong_new_correct"`
OldMissNewCorrect int `json:"old_miss_new_correct,omitempty"`
OldWrongNewWrong int `json:"old_wrong_new_wrong"`
OldLossyNewSafe int `json:"old_lossy_new_safe,omitempty"`
OldLossyNewCorrect int `json:"old_lossy_new_correct,omitempty"`
BothNotExact int `json:"both_not_exact"`
Same int `json:"same,omitempty"`
OldOnly int `json:"old_only,omitempty"`
NewOnly int `json:"new_only,omitempty"`
Differ int `json:"differ,omitempty"`
BothMiss int `json:"both_miss,omitempty"`
}
type shadowExample struct {
Kind string `json:"kind"`
Query string `json:"query"`
Expected string `json:"expected,omitempty"`
Old string `json:"old,omitempty"`
New string `json:"new,omitempty"`
}
type shadowOutcome struct {
hit esHit
parsed address.ParsedAddress
has bool
err error
latency time.Duration
}
func runShadowQueries(opts shadowOptions, cases []shadowCase, mode, note string, parser *address.Parser, resolver *match.Resolver, client *esClient) shadowReport {
ctx := context.Background()
start := time.Now()
type job struct{ c shadowCase }
type result struct {
c shadowCase
old shadowOutcome
newRes match.Result
newLat time.Duration
newErr error
}
jobs := make(chan job)
results := make(chan result)
var wg sync.WaitGroup
for i := 0; i < opts.workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := range jobs {
oldStart := time.Now()
oldQ := j.c.Query
if opts.oldPreproc {
oldQ = preprocessOld(oldQ)
}
hit, err := oldSearch(ctx, client, opts, oldQ)
old := shadowOutcome{hit: hit, err: err, latency: time.Since(oldStart)}
if err == nil {
old.parsed = parser.ParseInCity(hit.Address, j.c.ExpectCity)
old.has = hit.Address != ""
if j.c.ExpectCity == "" && opts.city != "" {
old.parsed = parser.ParseInCity(hit.Address, opts.city)
}
}
newStart := time.Now()
res := resolver.Resolve(ctx, j.c.Query)
results <- result{c: j.c, old: old, newRes: res, newLat: time.Since(newStart)}
}
}()
}
go func() {
for _, c := range cases {
jobs <- job{c: c}
}
close(jobs)
wg.Wait()
close(results)
}()
var oldLat, newLat []time.Duration
rep := shadowReport{
Mode: mode, Note: note, Workers: opts.workers, ESIndex: opts.esIndex,
OldPreprocess: opts.oldPreproc, Queries: len(cases), Examples: []shadowExample{},
}
for r := range results {
oldLat = append(oldLat, r.old.latency)
newLat = append(newLat, r.newLat)
if r.old.err != nil {
rep.Old.Error++
rep.Errors++
}
if r.newErr != nil {
rep.New.Error++
rep.Errors++
}
if mode == "generated" {
classifyGenerated(&rep, opts, parser, r.c, r.old, r.newRes)
} else {
classifyLog(&rep, opts, parser, r.c, r.old, r.newRes)
}
}
rep.Old.P50MS, rep.Old.P95MS, rep.Old.P99MS = percentiles(oldLat)
rep.New.P50MS, rep.New.P95MS, rep.New.P99MS = percentiles(newLat)
rep.ElapsedMS = time.Since(start).Milliseconds()
return rep
}
func classifyGenerated(rep *shadowReport, opts shadowOptions, parser *address.Parser, c shadowCase, old shadowOutcome, newRes match.Result) {
want := parser.ParseInCity(c.Expect, c.ExpectCity)
oldExact := old.has && sameAddressUnit(old.parsed, want)
oldWrong := old.has && !oldExact && c.RequireExact
oldLossy := old.has && !oldExact && !c.RequireExact
newExact := newRes.Status == address.StatusFound && sameAddressUnit(parser.ParseInCity(newRes.Address, c.ExpectCity), want)
newWrong := newRes.Status == address.StatusFound && !newExact && c.RequireExact
newSafe := newRes.Status == address.StatusPartial ||
newRes.Status == address.StatusAmbiguous ||
newRes.Status == address.StatusNotFound
if old.has {
rep.Old.Found++
}
if oldExact {
rep.Old.Exact++
}
if oldWrong {
rep.Old.Wrong++
}
if oldLossy {
rep.Old.Lossy++
}
if newRes.Status == address.StatusFound {
rep.New.Found++
}
if newExact {
rep.New.Exact++
}
if newWrong {
rep.New.Wrong++
}
if newRes.Status == address.StatusFound && !newExact && !c.RequireExact {
rep.New.Lossy++
}
switch newRes.Status {
case address.StatusPartial:
rep.New.Partial++
case address.StatusAmbiguous:
rep.New.Ambiguous++
case address.StatusNotFound:
rep.New.NotFound++
}
switch {
case oldExact && newExact:
rep.Compare.BothCorrect++
case oldExact:
rep.Compare.OldCorrectNewNotExact++
addExample(rep, opts, shadowExample{Kind: "old_correct_new_not_exact", Query: c.Query, Expected: c.Expect, Old: old.hit.Address, New: newRes.Address})
case newExact && oldWrong:
rep.Compare.OldWrongNewCorrect++
addExample(rep, opts, shadowExample{Kind: "old_wrong_new_correct", Query: c.Query, Expected: c.Expect, Old: old.hit.Address, New: newRes.Address})
case newExact && oldLossy:
rep.Compare.OldLossyNewCorrect++
addExample(rep, opts, shadowExample{Kind: "old_lossy_new_correct", Query: c.Query, Expected: c.Expect, Old: old.hit.Address, New: newRes.Address})
case newExact:
rep.Compare.OldMissNewCorrect++
addExample(rep, opts, shadowExample{Kind: "old_miss_new_correct", Query: c.Query, Expected: c.Expect, Old: old.hit.Address, New: newRes.Address})
case oldWrong && newWrong:
rep.Compare.OldWrongNewWrong++
addExample(rep, opts, shadowExample{Kind: "old_wrong_new_wrong", Query: c.Query, Expected: c.Expect, Old: old.hit.Address, New: newRes.Address})
case oldWrong && newSafe:
rep.Compare.OldWrongNewSafe++
addExample(rep, opts, shadowExample{Kind: "old_wrong_new_safe", Query: c.Query, Expected: c.Expect, Old: old.hit.Address, New: newRes.Address})
case oldLossy && newSafe:
rep.Compare.OldLossyNewSafe++
addExample(rep, opts, shadowExample{Kind: "old_lossy_new_safe", Query: c.Query, Expected: c.Expect, Old: old.hit.Address, New: newRes.Address})
default:
rep.Compare.BothNotExact++
}
}
func classifyLog(rep *shadowReport, opts shadowOptions, parser *address.Parser, c shadowCase, old shadowOutcome, newRes match.Result) {
if old.has {
rep.Old.Found++
}
if newRes.Status == address.StatusFound {
rep.New.Found++
}
oldOK := old.has && old.parsed.HasHouse
newOK := newRes.Status == address.StatusFound
switch {
case oldOK && newOK && sameAddressUnit(old.parsed, parser.ParseInCity(newRes.Address, opts.city)):
rep.Compare.Same++
case oldOK && !newOK:
rep.Compare.OldOnly++
addExample(rep, opts, shadowExample{Kind: "old_only", Query: c.Query, Old: old.hit.Address, New: newRes.Address})
case !oldOK && newOK:
rep.Compare.NewOnly++
addExample(rep, opts, shadowExample{Kind: "new_only", Query: c.Query, Old: old.hit.Address, New: newRes.Address})
case oldOK && newOK:
rep.Compare.Differ++
addExample(rep, opts, shadowExample{Kind: "differ", Query: c.Query, Old: old.hit.Address, New: newRes.Address})
default:
rep.Compare.BothMiss++
}
}
func addExample(rep *shadowReport, opts shadowOptions, ex shadowExample) {
if len(rep.Examples) >= opts.maxExamples {
return
}
rep.Examples = append(rep.Examples, ex)
}
func sameAddressUnit(a, b address.ParsedAddress) bool {
return a.CityKey == b.CityKey &&
a.StreetKey == b.StreetKey &&
a.House == b.House && a.HouseFrac == b.HouseFrac && a.HouseLetter == b.HouseLetter &&
a.Building == b.Building && a.BuildingFrac == b.BuildingFrac && a.BuildingLetter == b.BuildingLetter &&
a.Structure == b.Structure
}
func percentiles(values []time.Duration) (p50, p95, p99 float64) {
if len(values) == 0 {
return 0, 0, 0
}
sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
at := func(q float64) float64 {
i := int(float64(len(values)-1) * q)
return float64(values[i].Microseconds()) / 1000
}
return at(0.50), at(0.95), at(0.99)
}
+32 -32
View File
@@ -1,9 +1,9 @@
// Command addrkitd runs the address resolution service.
//
// It loads the canonical registry into memory, wires the optional adapters
// (ArcGIS portal, LLM rewriter) and serves the versioned JSON API. The process
// fails closed: if the registry cannot be loaded it exits rather than serving
// an empty index.
// It loads the canonical registry into memory, wires the optional LLM rewriter
// and serves the versioned JSON API. The process fails closed: if the registry
// cannot be loaded or is empty it exits rather than serving misleading
// "not found" answers.
package main
import (
@@ -17,13 +17,12 @@ import (
"syscall"
"time"
"git.totmin.ru/en2zmax/addrkit/internal/adapters/arcgis"
"git.totmin.ru/en2zmax/addrkit/internal/adapters/httpapi"
"git.totmin.ru/en2zmax/addrkit/internal/adapters/llm"
"git.totmin.ru/en2zmax/addrkit/internal/adapters/registrystore"
"git.totmin.ru/en2zmax/addrkit/internal/config"
"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
"git.totmin.ru/en2zmax/addrkit/internal/domain/match"
"git.totmin.ru/en2zmax/addrkit/internal/domain/outages"
)
func main() {
@@ -40,10 +39,14 @@ func main() {
version, entries, err := registrystore.Load(cfg.DataPath)
if err != nil {
// Fail closed: no registry, no service.
// Fail closed: no usable registry, no service.
logger.Error("load registry", "error", err, "path", cfg.DataPath)
os.Exit(1)
}
if len(entries) == 0 {
logger.Error("registry is empty; refusing to serve", "path", cfg.DataPath)
os.Exit(1)
}
logger.Info("registry loaded", "version", version, "entries", len(entries))
index := match.NewIndex(version, entries, match.Options{
@@ -52,40 +55,47 @@ func main() {
MaxCandidates: cfg.Resolve.MaxCandidates,
})
parser := address.NewParser(address.NewDictionary(cfg.Cities), cfg.DefaultCity)
var rewriter match.Rewriter
if cfg.LLM.Enabled {
rewriter = llm.New(cfg.LLM)
logger.Info("llm rewriter enabled", "url", cfg.LLM.URL, "model", cfg.LLM.Model)
}
resolver := match.NewResolver(index, rewriter)
resolver := match.NewResolverWithOptions(index, rewriter, match.ResolverOptions{
Parser: parser,
OnRewriteError: func(err error) {
logger.Warn("llm rewriter failed; using deterministic result", "error", err)
},
})
var portal outages.Source
if cfg.Portal.URL != "" {
portal = arcgis.New(
cfg.Portal.URL,
time.Duration(cfg.Portal.TimeoutSeconds)*time.Second,
cfg.Portal.Retries,
)
logger.Info("outages module enabled", "url", cfg.Portal.URL)
} else {
// portal.url is optional: without it only /v1/resolve is served.
logger.Info("outages module disabled (portal.url not set)")
cities := resolver.Cities()
logger.Info("cities loaded", "count", len(cities), "cities", cities, "default", cfg.DefaultCity)
if cfg.DefaultCity != "" {
if len(cities) > 1 {
logger.Warn("default_city is set for a multi-city index; city-less queries will always use it", "default", cfg.DefaultCity)
} else if len(cities) == 1 && cities[0] != cfg.DefaultCity {
logger.Warn("default_city is not present in the index", "default", cfg.DefaultCity, "index_city", cities[0])
}
}
loc := operationalLocation()
srv := httpapi.New(resolver, portal, logger, loc)
srv := httpapi.New(resolver, logger)
httpServer := &http.Server{
Addr: cfg.Addr,
Handler: srv.Handler(),
ReadHeaderTimeout: 5 * time.Second,
ReadTimeout: 15 * time.Second,
WriteTimeout: 20 * time.Second,
IdleTimeout: 60 * time.Second,
MaxHeaderBytes: 64 << 10,
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
go func() {
logger.Info("listening", "addr", cfg.Addr, "tz", loc.String())
logger.Info("listening", "addr", cfg.Addr)
if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
logger.Error("serve", "error", err)
stop()
@@ -116,13 +126,3 @@ func newLogger(level string) *slog.Logger {
}
return slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: lvl}))
}
// operationalLocation returns Asia/Novosibirsk when the tz database is
// available and falls back to a fixed UTC+7 offset otherwise (containers
// without tzdata).
func operationalLocation() *time.Location {
if loc, err := time.LoadLocation("Asia/Novosibirsk"); err == nil {
return loc
}
return time.FixedZone("Asia/Novosibirsk", 7*3600)
}