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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
|
package main
import (
"bytes"
"database/sql"
"encoding/csv"
"errors"
"flag"
"fmt"
"io"
"net/http"
"net/url"
"os"
"sort"
"strconv"
"strings"
"time"
)
// The Wikidata Query Service endpoint and the User-Agent it expects. WDQS
// blocks requests with a generic or absent User-Agent, so this is not
// decoration: without it the queries fail with 403.
const (
wdqsEndpoint = "https://query.wikidata.org/sparql"
userAgent = "twin-towns/0.1 (https://profpatsch.de; twinned-cities map)"
)
// pairsQuery pulls every P190 ("twinned administrative body") statement.
//
// It deliberately does NOT filter to one direction. The obvious way to
// deduplicate a symmetric property in SPARQL is FILTER(STR(?a) < STR(?b)),
// and it is wrong here: P190 is *documented* as symmetric but is not
// symmetric in the data. 6537 statements exist in only one direction, and
// that filter drops the ones that happen to point "backwards" — 25980 pairs
// survive it versus 29212 actually present. So fetch both directions and
// normalise in Go, where the asymmetry is visible and handled.
const pairsQuery = `SELECT ?a ?b WHERE { ?a wdt:P190 ?b }`
// datesQuery pulls the start (P580) and end (P582) qualifiers of the P190
// statements that have either.
//
// This has to go through the full statement path (p:/ps:/pq:) rather than the
// truth-y wdt: shortcut, because qualifiers hang off the statement node and
// are invisible to wdt:. Restricting to statements that actually carry a date
// keeps the result to ~17.6k rows and the query to ~5 seconds.
//
// Dates are a minority: about 31% of statements record when the twinning
// began and under 1% when it ended.
const datesQuery = `SELECT ?a ?b ?start ?end WHERE {
?a p:P190 ?st .
?st ps:P190 ?b .
OPTIONAL { ?st pq:P580 ?start }
OPTIONAL { ?st pq:P582 ?end }
FILTER(BOUND(?start) || BOUND(?end))
}`
// placesQuery pulls the facts (but NOT the labels) for every entity taking
// part in a twinning, on either side. The inner DISTINCT subquery is what
// keeps this tractable: it collapses the ~52k statements to ~20k entities
// before the OPTIONALs fan out over them.
//
// Everything is OPTIONAL because Wikidata coverage is uneven: ~110 twinned
// entities have no coordinates and would vanish from the result entirely
// under an inner join, taking their partners' twin lists with them.
//
// Labels are deliberately absent here. Adding
// `SERVICE wikibase:label` to this query takes it from ~6 seconds to a hard
// 60-second WDQS timeout — and the failure is nastier than an error, because
// WDQS answers a timeout with a *truncated but 200 OK* CSV body, cut mid-row.
// Labels are fetched separately by labelsQuery instead.
const placesQuery = `SELECT ?e ?lat ?lon ?country ?pop WHERE {
{ SELECT DISTINCT ?e WHERE { { ?e wdt:P190 [] } UNION { [] wdt:P190 ?e } } }
OPTIONAL { ?e wdt:P625 ?coord . BIND(geof:latitude(?coord) AS ?lat) BIND(geof:longitude(?coord) AS ?lon) }
OPTIONAL { ?e wdt:P17 ?country }
OPTIONAL { ?e wdt:P1082 ?pop }
}`
// labelBatch is how many entities are named per label query. The label
// service is only affordable when it is handed an explicit list of entities
// rather than being asked to resolve the result of a scan: 3000 entities
// resolve in under three seconds this way, while the same service inlined
// into placesQuery times out at 60. Batching also bounds the query text,
// which is ~30 bytes per entity.
const labelBatch = 3000
// labelsQuery builds a query resolving the labels of an explicit set of
// entities. The label service binds "?eLabel" — derived from the "?e"
// variable name, NOT any name chosen in the projection. Selecting "?label"
// silently yields empty strings.
//
// The language list is a fallback chain: the first available wins, and
// wikibase:label falls back to the bare QID when an entity has none of them.
func labelsQuery(qids []string) string {
var b strings.Builder
b.WriteString(`SELECT ?e ?eLabel WHERE { VALUES ?e { `)
for _, q := range qids {
b.WriteString("wd:")
b.WriteString(q)
b.WriteByte(' ')
}
b.WriteString(`} SERVICE wikibase:label { bd:serviceParam wikibase:language "en,de,fr,es,it,nl,pl,pt,ru" } }`)
return b.String()
}
// runIngest fetches the current state of P190 from Wikidata and replaces the
// contents of the database with it.
func runIngest(args []string) error {
fs := flag.NewFlagSet("ingest", flag.ContinueOnError)
dbPath := fs.String("db", "", "path to the SQLite database (required)")
endpoint := fs.String("endpoint", wdqsEndpoint, "SPARQL endpoint")
timeout := fs.Duration("timeout", 5*time.Minute, "per-query HTTP timeout")
if err := fs.Parse(args); err != nil {
return err
}
if *dbPath == "" {
return errors.New("--db is required")
}
db, err := openDB(*dbPath, writeBusyTimeout)
if err != nil {
return err
}
defer db.Close()
client := &http.Client{Timeout: *timeout}
fmt.Fprintln(os.Stderr, "twin-towns: querying places…")
placeRows, err := sparqlCSV(client, *endpoint, placesQuery)
if err != nil {
return fmt.Errorf("query places: %w", err)
}
places, err := parsePlaces(placeRows)
if err != nil {
return err
}
fmt.Fprintf(os.Stderr, "twin-towns: %d places\n", len(places))
if err := fetchLabels(client, *endpoint, places); err != nil {
return fmt.Errorf("query labels: %w", err)
}
fmt.Fprintln(os.Stderr, "twin-towns: querying pairs…")
pairRows, err := sparqlCSV(client, *endpoint, pairsQuery)
if err != nil {
return fmt.Errorf("query pairs: %w", err)
}
pairs, skipped := parsePairs(pairRows, places)
fmt.Fprintf(os.Stderr, "twin-towns: %d unique pairs (%d statements skipped)\n", len(pairs), skipped)
if len(places) == 0 || len(pairs) == 0 {
return fmt.Errorf("refusing to write empty result (%d places, %d pairs)", len(places), len(pairs))
}
fmt.Fprintln(os.Stderr, "twin-towns: querying dates…")
dateRows, err := sparqlCSV(client, *endpoint, datesQuery)
if err != nil {
return fmt.Errorf("query dates: %w", err)
}
unparsable, err := parseDates(dateRows, pairs)
if err != nil {
return err
}
dated, ended := 0, 0
for _, info := range pairs {
if info.started != "" {
dated++
}
if info.ended != "" {
ended++
}
}
fmt.Fprintf(os.Stderr, "twin-towns: %d pairs dated, %d ended (%d unusable date values)\n",
dated, ended, unparsable)
return store(db, places, pairs)
}
// fetchLabels resolves the label of every place in batches, filling in the
// label field in place.
//
// Entities whose label the endpoint does not return keep the fallback set by
// parsePlaces (the bare QID), so a place is never nameless.
func fetchLabels(client *http.Client, endpoint string, places map[string]*place) error {
// Countries are entities too, and their labels come from the same
// service, so they ride along in the same batches rather than costing a
// second pass. There are only ~200 distinct ones across ~20k places.
want := make(map[string]struct{}, len(places))
for q, p := range places {
want[q] = struct{}{}
if p.country.Valid {
want[p.country.String] = struct{}{}
}
}
qids := make([]string, 0, len(want))
for q := range want {
qids = append(qids, q)
}
// Sorted so the batches — and therefore the queries — are identical
// between runs on unchanged data, which makes a failure reproducible.
sort.Strings(qids)
// Collected across all batches, then applied once at the end: a country's
// label may well arrive in a later batch than the places referring to it.
labels := make(map[string]string, len(qids))
for start := 0; start < len(qids); start += labelBatch {
end := min(start+labelBatch, len(qids))
batch := qids[start:end]
rows, err := sparqlCSV(client, endpoint, labelsQuery(batch))
if err != nil {
return fmt.Errorf("batch %d-%d: %w", start, end, err)
}
if len(rows) == 0 {
continue
}
col, err := columns(rows[0], "e", "eLabel")
if err != nil {
return err
}
for _, row := range rows[1:] {
q := qid(field(row, col["e"]))
label := field(row, col["eLabel"])
if q == "" || label == "" {
continue
}
labels[q] = label
}
fmt.Fprintf(os.Stderr, "twin-towns: labels %d/%d\n", end, len(qids))
}
for q, p := range places {
if l, ok := labels[q]; ok {
p.label = l
}
if p.country.Valid {
if l, ok := labels[p.country.String]; ok {
p.countryLabel = sql.NullString{String: l, Valid: true}
}
}
}
return nil
}
// place is one twinned entity.
type place struct {
qid string
label string
country sql.NullString
countryLabel sql.NullString
lat, lon sql.NullFloat64
pop sql.NullInt64
}
// pair is one unordered twinning, always normalised to a < b. It is the map
// key, so it stays a bare comparable struct.
type pair struct{ a, b string }
// pairInfo is what is known about a pair beyond its existence. Empty strings
// mean "not recorded", which is the common case: most twinnings have no date.
type pairInfo struct {
started string
ended string
}
// sparqlCSV runs a query and returns the parsed CSV rows (including the header
// row). CSV is used over JSON because the result is a flat table of literals
// and this avoids materialising ~29k rows of nested JSON.
func sparqlCSV(client *http.Client, endpoint, query string) ([][]string, error) {
// POST, not GET: the places query is long enough that some proxies
// truncate it in a URL.
req, err := http.NewRequest("POST", endpoint, strings.NewReader(url.Values{"query": {query}}.Encode()))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.Header.Set("Accept", "text/csv")
req.Header.Set("User-Agent", userAgent)
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2000))
return nil, fmt.Errorf("endpoint returned %s: %s", resp.Status, strings.TrimSpace(string(body)))
}
// Read the body fully before parsing so a truncated response can be
// detected as such. This matters more than it sounds: see below.
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
// WDQS answers a query that exceeds its 60-second limit with HTTP 200 and
// a body that is simply cut off — sometimes mid-row, sometimes with a
// Java stack trace appended. There is no status code, no trailer and no
// error field to check, so a truncated result is indistinguishable from a
// short one except that it does not end in a newline. Ignoring this means
// silently ingesting a partial world.
//
// A well-formed CSV response always ends with a line terminator; a cut
// one almost never does.
if len(body) > 0 && body[len(body)-1] != '\n' {
return nil, fmt.Errorf("truncated response (%d bytes, no trailing newline) — "+
"the endpoint most likely timed out; last bytes: %q",
len(body), lastBytes(body, 120))
}
// The stack trace case: a timeout that unwound far enough to print one
// still ends in a newline, so check for it explicitly.
if bytes.Contains(body, []byte("java.util.concurrent")) || bytes.Contains(body, []byte("QueryTimeoutException")) {
return nil, fmt.Errorf("endpoint returned a Java stack trace (query timed out); last bytes: %q",
lastBytes(body, 120))
}
r := csv.NewReader(bytes.NewReader(body))
// Rows are uniform, but let the reader accept variable field counts so a
// schema change upstream surfaces as a parse error at the row we care
// about rather than an opaque failure.
r.FieldsPerRecord = -1
return r.ReadAll()
}
// nullIfEmpty keeps "not recorded" as SQL NULL rather than an empty string,
// so the date columns can be counted and compared without special cases.
func nullIfEmpty(s string) any {
if s == "" {
return nil
}
return s
}
func lastBytes(b []byte, n int) string {
if len(b) > n {
b = b[len(b)-n:]
}
return string(b)
}
// qid extracts "Q1234" from a full entity URI. Values that are not entity
// URIs (or are empty) yield "".
func qid(uri string) string {
if uri == "" {
return ""
}
i := strings.LastIndexByte(uri, '/')
if i < 0 {
return ""
}
q := uri[i+1:]
if len(q) < 2 || q[0] != 'Q' {
return ""
}
return q
}
// parsePlaces turns the CSV result into places, keyed by QID.
//
// The query emits one row per combination of the OPTIONAL values, so an
// entity with two P17 countries or several censuses appears several times
// (28808 rows for 20141 entities). First row wins: they carry the same
// identity, and picking one arbitrary country beats inventing a merge rule.
func parsePlaces(rows [][]string) (map[string]*place, error) {
if len(rows) == 0 {
return nil, errors.New("empty place result")
}
col, err := columns(rows[0], "e", "lat", "lon", "country", "pop")
if err != nil {
return nil, err
}
places := make(map[string]*place, len(rows))
for _, row := range rows[1:] {
q := qid(field(row, col["e"]))
if q == "" {
continue
}
if _, seen := places[q]; seen {
continue
}
// Labels arrive later, from fetchLabels. Until then the QID stands in,
// and it remains as the fallback for any entity the label query does
// not answer for, so a place is never nameless.
p := &place{qid: q, label: q}
if lat, ok := parseFloat(field(row, col["lat"])); ok {
if lon, ok := parseFloat(field(row, col["lon"])); ok {
// Guard against nonsense coordinates; they would otherwise be
// projected off-canvas and be invisible but still clickable.
if lat >= -90 && lat <= 90 && lon >= -180 && lon <= 180 {
p.lat = sql.NullFloat64{Float64: lat, Valid: true}
p.lon = sql.NullFloat64{Float64: lon, Valid: true}
}
}
}
if c := qid(field(row, col["country"])); c != "" {
p.country = sql.NullString{String: c, Valid: true}
}
if pop, ok := parseFloat(field(row, col["pop"])); ok && pop >= 0 {
p.pop = sql.NullInt64{Int64: int64(pop), Valid: true}
}
places[q] = p
}
return places, nil
}
// parseWikidataDate turns a WDQS date literal into a plain ISO date
// ("1987-06-25"), or returns "" for anything it will not vouch for.
//
// The live data contains three kinds of value that are not dates, all of
// which reached this function during development:
//
// - Blank-node URIs like ".../.well-known/genid/5683a248…". These are how
// Wikidata serialises "unknown value": the statement asserts that a start
// date exists but not which. Storing the URI would put a genid string in
// a date column.
// - Years outside any plausible range, e.g. "51983-05-01" (a typo for 1983)
// and "0199-05-15". Keeping them would stretch any date axis to
// uselessness for the sake of two rows.
// - Empty strings, from the OPTIONAL not matching.
//
// The lower bound is deliberately generous rather than "modern": the oldest
// plausible start date in the data is 1561 (Greater London–Moscow, dating the
// relationship to the Muscovy Company rather than to any modern twinning),
// and the point of showing dates is partly to surface such outliers rather
// than quietly clip them. 700 is simply well below anything real, so it
// excludes only typos like the observed year 199.
func parseWikidataDate(s string) string {
if s == "" || strings.HasPrefix(s, "http") {
return ""
}
// WDQS emits xsd:dateTime, e.g. "1987-06-25T00:00:00Z". Take the date.
if i := strings.IndexByte(s, 'T'); i > 0 {
s = s[:i]
}
// Wikidata can express years before 1 CE with a leading "-"; those are not
// twinnings, and the rest of the pipeline assumes a positive year.
if len(s) != 10 || s[4] != '-' || s[7] != '-' {
return ""
}
year, err := strconv.Atoi(s[:4])
if err != nil || year < 700 || year > time.Now().Year()+1 {
return ""
}
// Wikidata records an unknown month or day as 00; keep the date but do
// not pretend to a precision it does not have.
if s[5:7] == "00" || s[8:10] == "00" {
return s[:4]
}
return s
}
// parseDates attaches start/end qualifiers to the pairs.
//
// Both directions of a symmetric statement can carry dates, and ~494 pairs
// disagree with themselves (typically 1996 vs 1997, or two adjacent days).
// There is no way to adjudicate, so the earliest start and the latest end
// win: the twinning demonstrably existed over at least that span.
func parseDates(rows [][]string, pairs map[pair]*pairInfo) (int, error) {
if len(rows) == 0 {
return 0, nil
}
col, err := columns(rows[0], "a", "b", "start", "end")
if err != nil {
return 0, err
}
dropped := 0
for _, row := range rows[1:] {
a, b := qid(field(row, col["a"])), qid(field(row, col["b"]))
if a == "" || b == "" || a == b {
continue
}
if a > b {
a, b = b, a
}
info, ok := pairs[pair{a, b}]
if !ok {
// The statement exists but the pair did not survive the pair
// query (e.g. one endpoint has no metadata row).
continue
}
start := parseWikidataDate(field(row, col["start"]))
end := parseWikidataDate(field(row, col["end"]))
if field(row, col["start"]) != "" && start == "" {
dropped++
}
if start != "" && (info.started == "" || start < info.started) {
info.started = start
}
if end != "" && end > info.ended {
info.ended = end
}
}
return dropped, nil
}
// parsePairs normalises the raw statements into unique unordered pairs.
//
// Three things get dropped, and the count of them is returned so an ingest
// that suddenly discards a lot is noticeable:
// - self-links (a place twinned with itself)
// - statements naming an entity absent from the place query (a race: the
// two queries are not a single snapshot, so a statement added between
// them can reference an entity we did not fetch). Keeping it would
// violate the foreign key.
// - the second copy of each already-seen pair.
func parsePairs(rows [][]string, places map[string]*place) (map[pair]*pairInfo, int) {
if len(rows) == 0 {
return nil, 0
}
col, err := columns(rows[0], "a", "b")
if err != nil {
return nil, len(rows) - 1
}
pairs := make(map[pair]*pairInfo, len(rows))
skipped := 0
for _, row := range rows[1:] {
a, b := qid(field(row, col["a"])), qid(field(row, col["b"]))
if a == "" || b == "" || a == b {
skipped++
continue
}
if _, ok := places[a]; !ok {
skipped++
continue
}
if _, ok := places[b]; !ok {
skipped++
continue
}
if a > b {
a, b = b, a
}
if _, seen := pairs[pair{a, b}]; !seen {
pairs[pair{a, b}] = &pairInfo{}
}
}
return pairs, skipped
}
// store replaces the database contents in a single transaction, so a failed
// or interrupted ingest leaves the previously-served data intact rather than
// a half-updated map.
func store(db *sql.DB, places map[string]*place, pairs map[pair]*pairInfo) error {
tx, err := db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
// twin references place, so it must go first on the way out and second on
// the way in.
if _, err := tx.Exec(`DELETE FROM twin`); err != nil {
return err
}
if _, err := tx.Exec(`DELETE FROM place`); err != nil {
return err
}
insPlace, err := tx.Prepare(`INSERT INTO place (qid, label, country, country_label, lat, lon, pop) VALUES (?, ?, ?, ?, ?, ?, ?)`)
if err != nil {
return err
}
defer insPlace.Close()
for _, p := range places {
if _, err := insPlace.Exec(p.qid, p.label, p.country, p.countryLabel, p.lat, p.lon, p.pop); err != nil {
return fmt.Errorf("insert place %s: %w", p.qid, err)
}
}
insTwin, err := tx.Prepare(`INSERT INTO twin (a, b, started, ended) VALUES (?, ?, ?, ?)`)
if err != nil {
return err
}
defer insTwin.Close()
for p, info := range pairs {
if _, err := insTwin.Exec(p.a, p.b, nullIfEmpty(info.started), nullIfEmpty(info.ended)); err != nil {
return fmt.Errorf("insert twin %s-%s: %w", p.a, p.b, err)
}
}
if _, err := tx.Exec(
`INSERT INTO meta (key, value) VALUES ('ingested_at', ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
time.Now().UTC().Format(time.RFC3339),
); err != nil {
return err
}
if err := tx.Commit(); err != nil {
return err
}
fmt.Fprintf(os.Stderr, "twin-towns: stored %d places, %d pairs\n", len(places), len(pairs))
return nil
}
// columns maps the wanted column names to their index in the CSV header,
// failing loudly if the endpoint returns a shape we did not expect.
func columns(header []string, want ...string) (map[string]int, error) {
idx := make(map[string]int, len(header))
for i, h := range header {
idx[strings.TrimPrefix(h, "?")] = i
}
col := make(map[string]int, len(want))
for _, w := range want {
i, ok := idx[w]
if !ok {
return nil, fmt.Errorf("missing column %q in result header %v", w, header)
}
col[w] = i
}
return col, nil
}
func field(row []string, i int) string {
if i < 0 || i >= len(row) {
return ""
}
return row[i]
}
// parseFloat accepts the scientific notation WDQS emits for coordinates
// ("4.7215277777E1").
func parseFloat(s string) (float64, bool) {
if s == "" {
return 0, false
}
f, err := strconv.ParseFloat(s, 64)
if err != nil {
return 0, false
}
return f, true
}
|