"""Server/offline VIINA daily update. Never imported by the browser. ODbL derived data; upstream status is retained without recomputing votes. Only standard library + Shapely (see viina-requirements.txt). No paid service. """ import argparse import base64 import csv import datetime as dt import hashlib import http.client import ipaddress import io import json import math import os import socket from pathlib import Path import sys import tempfile import time import urllib.error import urllib.request from urllib.parse import urlsplit import zipfile from contextlib import contextmanager from email.utils import parsedate_to_datetime from shapely.geometry import shape, mapping, Point from shapely import union_all REPO = 'zhukovyuri/VIINA' PROJECT = 'https://github.com/' + REPO DAY = 86400 SOURCE_HOSTS = frozenset(('api.github.com', 'raw.githubusercontent.com', 'media.githubusercontent.com')) STATUSES = ('RU', 'UA', 'CONTESTED', 'UNKNOWN') FIELDS = ('geonameid', 'date', 'status_wiki', 'status_boost', 'status_dsm', 'status_isw', 'status', 'vcontrol_version') METHOD = ('Połączenie przez geonameid; zachowano opublikowane status bez przeliczania głosów. ' 'Suma geometryczna komórek tessellacji według statusu, bez upraszczania, buforowania ani zaokrąglania. ' 'Szacowane obszary wokół miejscowości, nie dokładna linia frontu. Braki i nieznane statusy = UNKNOWN.') NOTICE = '''Contains information from VIINA, made available under the Open Database License (ODbL) 1.0. Database: ODbL https://opendatacommons.org/licenses/odbl/1-0/ Individual contents: Database Contents License https://opendatacommons.org/licenses/dbcl/1-0/ VIINA source: https://github.com/zhukovyuri/VIINA Zhukov, Yuri and Natalie Ayers (2023). "VIINA 2.0: Violent Incident Information from News Articles on the 2022 Russian Invasion of Ukraine." Cambridge, MA: Harvard University. Zhukov, Yuri (2023). "Near-Real Time Analysis of War and Economic Activity during Russia’s Invasion of Ukraine." Journal of Comparative Economics 51 (4): 1232–1243. https://doi.org/10.1016/j.jce.2023.06.003 GeoNames: https://www.geonames.org/ — CC BY 4.0 https://creativecommons.org/licenses/by/4.0/ Input provenance: Wikipedia, VIINA events, DeepStateMap and ISW, as described by VIINA. Their rights are not automatically cleared by this notice. Public use with advertising: BLOCKED pending clarification of upstream rights; local review only. Adapted database: territory-viina.geojson (including RU, UA, CONTESTED and UNKNOWN), available without a separate charge alongside this notice. Complete transformation source: viina-transform.py; requirements: viina-requirements.txt. Application source code is separate from this database license. ''' def atomic_write(path, body): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) fd, tmp = tempfile.mkstemp(prefix='.' + path.name + '-', dir=path.parent) try: with os.fdopen(fd, 'wb') as f: f.write(body) f.flush() os.fsync(f.fileno()) # VIINA writer is isolated; the web/collector read group needs access. os.chmod(tmp, 0o640) os.replace(tmp, path) finally: if os.path.exists(tmp): os.unlink(tmp) def write_json(path, obj): atomic_write(path, json.dumps(obj, ensure_ascii=False, separators=(',', ':'), allow_nan=False).encode('utf-8')) def load_json(path, default=None): try: return json.loads(Path(path).read_text(encoding='utf-8')) except FileNotFoundError: return {} if default is None else default def sha256(path): with open(path, 'rb') as f: return hashlib.file_digest(f, 'sha256').hexdigest() def git_sha(path): h = hashlib.sha1(('blob ' + str(Path(path).stat().st_size) + '\0').encode()) with open(path, 'rb') as f: while chunk := f.read(1024 * 1024): h.update(chunk) return h.hexdigest() def identifier(value): if isinstance(value, bool): raise ValueError('Invalid geonameid') try: n = float(value) except (ValueError, TypeError): raise ValueError('Invalid geonameid') if not math.isfinite(n) or not n.is_integer() or not 0 < n < 2 ** 53: raise ValueError('Invalid geonameid') return int(n) def geometry_index(path): data = load_json(path) if data.get('type') != 'FeatureCollection' or not data.get('features'): raise ValueError('Invalid VIINA tessellation') crs = data.get('crs', {}).get('properties', {}).get('name', '') if crs and crs not in ('urn:ogc:def:crs:OGC:1.3:CRS84', 'urn:ogc:def:crs:EPSG::4326'): raise ValueError('Unexpected coordinate reference system') index = {} for f in data['features']: p = f.get('properties') or {} gid = identifier(p.get('geonameid')) if gid in index or p.get('country_code') != 'UA' or p.get('ADM0_ISO3') not in ('UKR', None): raise ValueError('Duplicate identifier or geometry outside Ukraine') g = f.get('geometry') or {} if g.get('type') not in ('Polygon', 'MultiPolygon'): raise ValueError('Unsupported tessellation geometry') obj = shape(g) if obj.is_empty or not obj.is_valid: raise ValueError('Invalid tessellation topology; no automatic repair') if not all(math.isfinite(v) for v in obj.bounds) or not (21 < obj.bounds[0] <= obj.bounds[2] < 41 and 43 < obj.bounds[1] <= obj.bounds[3] < 53): raise ValueError('Coordinates outside Ukraine including Crimea') index[gid] = obj return index def csv_rows(path, year): # No extraction to filesystem, no path traversal. zipfile checks CRC while streaming. with open(path, 'rb') as f: if f.read(4) != b'PK\x03\x04': raise ValueError('Not a ZIP archive (possibly a Git LFS pointer)') with zipfile.ZipFile(path) as z: names = z.infolist() if len(names) != 1 or names[0].filename != f'control_latest_{year}.csv' or names[0].file_size > 2_000_000_000: raise ValueError('Unexpected control archive contents or size') with z.open(names[0]) as stream: reader = csv.reader(io.TextIOWrapper(stream, encoding='utf-8-sig', newline='')) header = next(reader) if len(header) != len(set(header)) or any(k not in header for k in FIELDS): raise ValueError('VIINA control schema changed') positions = [header.index(k) for k in FIELDS] validated_dates = set() for row in reader: if len(row) != len(header): raise ValueError('Truncated control CSV') values = tuple(row[i] for i in positions) if values[1] not in validated_dates: date = dt.datetime.strptime(values[1], '%Y%m%d').date() if date.year != year or date > dt.datetime.now(dt.timezone.utc).date(): raise ValueError('Wrong-year or future control date') validated_dates.add(values[1]) yield identifier(values[0]), values[1], values def latest_complete(path, year, geometry): # About 9 MB of coverage bitmaps for a full year; never holds the annual CSV. ids = {gid: i for i, gid in enumerate(geometry)} coverage, counts, newest, latest = {}, {}, '', {} rows = 0 for gid, date, values in csv_rows(path, year): if gid not in ids: raise ValueError('Control identifier without tessellation') seen = coverage.get(date) if seen is None: seen = coverage[date] = bytearray(len(ids)) if seen[ids[gid]]: raise ValueError('Duplicate geonameid/date observation') seen[ids[gid]] = 1 counts[date] = counts.get(date, 0) + 1 rows += 1 if date > newest: newest, latest = date, {} if date == newest: latest[gid] = values complete = [date for date, count in counts.items() if count == len(ids)] if not complete: raise ValueError('No complete date: keep last good version') selected = max(complete) if selected != newest: latest = {gid: values for gid, date, values in csv_rows(path, year) if date == selected} versions = {v[-1] for v in latest.values()} if len(versions) != 1 or not all(len(v) >= 14 and v.isdigit() for v in versions): raise ValueError('Inconsistent release versions') return selected, latest, {'csvRows': rows, 'dates': len(counts), 'latestAvailableDate': newest, 'incompleteDates': sorted(d for d in counts if counts[d] != len(ids))} def build_layer(geometry, rows, date, fetched_at, checked_at, sources, diagnostics): groups = {s: [] for s in STATUSES} raw_unknown = set() for gid, obj in geometry.items(): raw = rows.get(gid, ('', '', '', '', '', '', '', ''))[6] status = raw if raw in STATUSES[:3] else 'UNKNOWN' if status == 'UNKNOWN': raw_unknown.add(raw) groups[status].append(obj) features = [] for status, cells in groups.items(): if not cells: continue merged = union_all(cells) if merged.is_empty or not merged.is_valid or merged.geom_type not in ('Polygon', 'MultiPolygon'): raise ValueError('Dissolved geometry failed validation') if abs(merged.area - sum(c.area for c in cells)) > 1e-7: raise ValueError('Overlapping cells or loss of area during union') features.append({'type': 'Feature', 'id': 'viina-' + status, 'properties': {'status': status, 'source': 'VIINA', 'cellCount': len(cells)}, 'geometry': mapping(merged)}) total = union_all(list(geometry.values())) if not total.covers(Point(34.108, 44.952)) or not total.covers(Point(24.0316, 49.8429)): raise ValueError('Coverage validation failed: Crimea or western Ukraine missing') # Every source cell is included once, even UNKNOWN and CONTESTED; no smoothing. reconstructed = union_all([shape(f['geometry']) for f in features]) if total.symmetric_difference(reconstructed).area > 1e-7: raise ValueError('Geometry reconstruction differs from source tessellation') meta = {'name': 'Szacowana kontrola terytorium — VIINA', 'source': PROJECT, 'dataDate': date, 'sourceTime': int(dt.datetime.strptime(date, '%Y%m%d').replace(tzinfo=dt.timezone.utc).timestamp()), 'fetchedAt': fetched_at, 'checkedAt': checked_at, 'viinaVersion': next(iter(rows.values()))[-1], 'method': METHOD, 'license': 'ODbL-1.0', 'contentsLicense': 'DbCL-1.0', 'geonamesLicense': 'CC-BY-4.0', 'commercialStatus': 'unresolved', 'publicUseBlocked': True, 'sources': sources, 'counts': {s: len(cells) for s, cells in groups.items()}, 'unknownValues': sorted(raw_unknown), 'diagnostics': diagnostics, 'totalAreaDegreesSquared': total.area, 'bounds': list(total.bounds), 'download': '/territory-viina.geojson', 'notice': '/data-licenses/viina.LICENSE.txt', 'updateIntervalSeconds': DAY} return {'type': 'FeatureCollection', 'features': features, 'territory': meta, 'sourceTime': meta['sourceTime'], 'fetchedAt': fetched_at, 'source': PROJECT, 'static': False} class SourceError(Exception): def __init__(self, code, retry_after=None): self.code, self.retry_after = code, retry_after super().__init__('Source HTTP ' + str(code)) def retry_at(value, now): try: delay = float(value) return now + max(DAY, delay) if math.isfinite(delay) else now + DAY except (ValueError, TypeError): try: return max(now + DAY, parsedate_to_datetime(value).timestamp()) except (ValueError, TypeError, OverflowError): return now + DAY def validate_source_url(url): # Check before urllib normalizes or discards whitespace/control characters. if not isinstance(url, str) or any(ord(c) <= 32 or ord(c) == 127 for c in url): raise ValueError('Invalid source URL') parsed = urlsplit(url) if (parsed.scheme != 'https' or parsed.hostname not in SOURCE_HOSTS or parsed.username is not None or parsed.password is not None or parsed.port not in (None, 443) or parsed.fragment): raise ValueError('Source must use an approved HTTPS host') def _connect_source(address, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None): host, port = address if host not in SOURCE_HOSTS or port != 443: raise ValueError('Invalid source connection') destinations = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP) if not destinations: raise OSError('Source DNS returned no addresses') # Reject the whole answer if it mixes public and internal destinations. # Connect to these numeric sockaddrs directly: no unchecked second DNS lookup. for family, _, _, _, sockaddr in destinations: if family not in (socket.AF_INET, socket.AF_INET6): raise ValueError('Invalid source address family') ip = ipaddress.ip_address(sockaddr[0]) if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped: ip = ip.ipv4_mapped if (not ip.is_global or ip.is_multicast or ip.is_reserved or getattr(ip, 'is_site_local', False)): raise ValueError('Source DNS destination is not public') last_error = None for family, socktype, protocol, _, sockaddr in destinations: connection = None try: connection = socket.socket(family, socktype, protocol) if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT: connection.settimeout(timeout) if source_address: connection.bind(source_address) connection.connect(sockaddr) return connection except OSError as error: last_error = error if connection is not None: connection.close() raise last_error class _SourceHTTPSConnection(http.client.HTTPSConnection): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # HTTPSConnection still verifies TLS and SNI using the original hostname. self._create_connection = _connect_source class _SourceHTTPSHandler(urllib.request.HTTPSHandler): def https_open(self, req): return self.do_open(_SourceHTTPSConnection, req, context=self._context) class _NoSourceRedirects(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError('Source redirects are forbidden') def request(url, headers=None): # No token required. Do not inherit environment proxies or a global opener. validate_source_url(url) req = urllib.request.Request(url, headers={'User-Agent': 'Horyzont-VIINA-daily-review/1.0', **(headers or {})}) opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _SourceHTTPSHandler(), _NoSourceRedirects()) try: return opener.open(req, timeout=90) except urllib.error.HTTPError as e: if e.code == 304: return e error = SourceError(e.code, e.headers.get('Retry-After')) e.close() raise error from None def metadata(cache, state, year, now): prior = state.get('listing', {}) with request(f'https://api.github.com/repos/{REPO}/contents/Data?ref=main', {'Accept': 'application/vnd.github+json', **({'If-None-Match': prior['etag']} if prior.get('etag') else {})}) as response: if response.status == 304: items = prior.get('items') if not items: raise ValueError('304 without verified listing') else: body = response.read(1_000_001) if len(body) > 1_000_000: raise ValueError('Oversized metadata') items = json.loads(body) if not isinstance(items, list): raise ValueError('Invalid GitHub listing') listing = {'etag': response.headers.get('ETag', prior.get('etag')), 'items': items} by_name = {e['name']: e for e in items} years = [y for y in range(2022, year + 1) if f'control_latest_{y}.zip' in by_name] if not years: raise ValueError('No current or previous yearly control file') selected_year = max(years) return selected_year, by_name[f'control_latest_{selected_year}.zip'], by_name['gn_UA_tess.geojson'], listing def download(url, path, maximum): tmp = path.with_suffix(path.suffix + '.part') try: with request(url) as response, open(tmp, 'wb') as f: length = 0 while chunk := response.read(1024 * 1024): length += len(chunk) if length > maximum: raise ValueError('Source exceeded download budget') f.write(chunk) return tmp except BaseException: tmp.unlink(missing_ok=True) raise @contextmanager def lock(cache): cache.mkdir(parents=True, exist_ok=True) with open(cache / 'update.lock', 'a+b') as f: if os.name == 'nt': import msvcrt if f.tell() == 0: f.write(b'0'); f.flush() f.seek(0) msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB) try: yield finally: if os.name == 'nt': f.seek(0); msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1) else: fcntl.flock(f, fcntl.LOCK_UN) def update(cache, output, now=None, year=None, local_archive=None, local_geometry=None): now = int(time.time()) if now is None else int(now) year = dt.datetime.fromtimestamp(now, dt.timezone.utc).year if year is None else year with lock(cache): state_path = cache / 'state.json' state = load_json(state_path) if state.get('nextAttemptAt', 0) > now: return {'state': 'cooldown', 'nextAttemptAt': state['nextAttemptAt'], 'error': state.get('error')} # Persist budget BEFORE network or CPU work. Crash cannot reset request allowance. state['nextAttemptAt'] = now + DAY write_json(state_path, state) try: if local_archive and local_geometry: archive, geom_path = Path(local_archive), Path(local_geometry) control = {'name': archive.name, 'sha': 'offline-' + sha256(archive)} geom = {'name': geom_path.name, 'sha': 'offline-' + sha256(geom_path)} archive_info = {'sha': control['sha'], 'sha256': sha256(archive), 'fetchedAt': now, 'url': f'https://media.githubusercontent.com/media/{REPO}/main/Data/control_latest_{year}.zip'} geom_info = {'sha': geom['sha'], 'sha256': sha256(geom_path), 'fetchedAt': now, 'url': f'https://raw.githubusercontent.com/{REPO}/main/Data/gn_UA_tess.geojson'} # Re-reading identical local input is not a new source download. for key, info in [('control', archive_info), ('geometry', geom_info)]: if state.get(key, {}).get('sha256') == info['sha256']: info['fetchedAt'] = state[key]['fetchedAt'] else: year, control, geom, listing = metadata(cache, state, year, now) state['listing'] = listing archive, geom_path = cache / f'control_latest_{year}.zip', cache / 'gn_UA_tess.geojson' archive_info, geom_info = state.get('control', {}), state.get('geometry', {}) if archive_info.get('sha') != control['sha'] or not archive.exists(): # A raw ZIP URL can be an LFS pointer. Fetch the pointer first and verify LFS payload hash. with request(f'https://api.github.com/repos/{REPO}/git/blobs/{control["sha"]}') as response: blob = json.loads(response.read(10000)) pointer = base64.b64decode(blob['content']).decode('ascii') lines = dict(line.split(' ', 1) for line in pointer.strip().splitlines()) if not lines.get('version', '').endswith('/spec/v1') or not lines.get('oid', '').startswith('sha256:'): raise ValueError('Expected verified Git LFS pointer') url = f'https://media.githubusercontent.com/media/{REPO}/main/Data/control_latest_{year}.zip' tmp = download(url, archive, 150_000_000) try: if tmp.stat().st_size != int(lines['size']) or sha256(tmp) != lines['oid'].split(':')[1]: raise ValueError('Git LFS payload hash mismatch') with zipfile.ZipFile(tmp) as z: if z.namelist() != [f'control_latest_{year}.csv']: raise ValueError('Unexpected archive') os.replace(tmp, archive) finally: tmp.unlink(missing_ok=True) archive_info = {'sha': control['sha'], 'sha256': sha256(archive), 'fetchedAt': now, 'url': url} if geom_info.get('sha') != geom['sha'] or not geom_path.exists(): url = f'https://raw.githubusercontent.com/{REPO}/main/Data/gn_UA_tess.geojson' tmp = download(url, geom_path, 100_000_000) try: if git_sha(tmp) != geom['sha']: raise ValueError('Geometry git blob hash mismatch') geometry_index(tmp) os.replace(tmp, geom_path) finally: tmp.unlink(missing_ok=True) geom_info = {'sha': geom['sha'], 'sha256': sha256(geom_path), 'fetchedAt': now, 'url': url} prior = load_json(output) unchanged = (prior.get('territory', {}).get('sources', {}).get('control', {}).get('sha256') == archive_info['sha256'] and prior.get('territory', {}).get('sources', {}).get('geometry', {}).get('sha256') == geom_info['sha256']) if unchanged: prior['territory']['checkedAt'] = now prior['territory']['sources'] = {'control': archive_info, 'geometry': geom_info} prior['fetchedAt'] = prior['territory']['fetchedAt'] = max(archive_info['fetchedAt'], geom_info['fetchedAt']) layer = prior else: geometry = geometry_index(geom_path) date, rows, diagnostics = latest_complete(archive, year, geometry) diagnostics['selectedFileYear'] = year diagnostics['currentYear'] = dt.datetime.fromtimestamp(now, dt.timezone.utc).year layer = build_layer(geometry, rows, date, max(archive_info['fetchedAt'], geom_info['fetchedAt']), now, {'control': archive_info, 'geometry': geom_info}, diagnostics) if prior.get('territory', {}).get('dataDate', '') > date: raise ValueError('Source regressed; keep newer last good layer') # One self-contained file is the commit point: geometry and dates always match. write_json(output, layer) state.update({'control': archive_info, 'geometry': geom_info, 'lastSuccessAt': now, 'error': None}) write_json(state_path, state) atomic_write(output.parent / 'data-licenses' / 'viina.LICENSE.txt', NOTICE.encode('utf-8')) atomic_write(output.parent / 'data-licenses' / 'viina-transform.py', Path(__file__).read_bytes()) atomic_write(output.parent / 'data-licenses' / 'viina-requirements.txt', Path(__file__).with_name('viina-requirements.txt').read_bytes()) return {'state': 'unchanged' if unchanged else 'updated', 'dataDate': layer['territory']['dataDate'], 'fetchedAt': layer['fetchedAt'], 'checkedAt': now, 'counts': layer['territory']['counts'], 'nextAttemptAt': now + DAY, 'bytes': output.stat().st_size} except Exception as e: code = e.code if isinstance(e, SourceError) else type(e).__name__ state['error'] = str(e) if isinstance(e, (SourceError, ValueError, zipfile.BadZipFile)) else 'Update failed: ' + str(code) state['nextAttemptAt'] = retry_at(e.retry_after, now) if isinstance(e, SourceError) else now + DAY write_json(state_path, state) return {'state': 'failed', 'error': state['error'], 'nextAttemptAt': state['nextAttemptAt'], 'lastGoodPreserved': output.exists()} if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--cache', type=Path, default=Path('var/viina-cache')) parser.add_argument('--output', type=Path, default=Path('public/territory-viina.geojson')) parser.add_argument('--local-archive', type=Path, help='Offline trial; never queries the network') parser.add_argument('--local-geometry', type=Path) parser.add_argument('--year', type=int, help='Only for an explicit offline input') args = parser.parse_args() if bool(args.local_archive) != bool(args.local_geometry) or args.year and not args.local_archive: parser.error('Offline input requires archive, geometry and optional year together') result = update(args.cache, args.output, year=args.year, local_archive=args.local_archive, local_geometry=args.local_geometry) print(json.dumps(result, ensure_ascii=True)) sys.exit(1 if result['state'] == 'failed' else 0)