445 lines
24 KiB
Python
445 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline-first bridge with atomic server-side hardware registration.
|
|
Stdlib only; server assigns numbers and immutable reports stay unchanged.
|
|
"""
|
|
import argparse
|
|
import contextlib
|
|
import datetime as dt
|
|
import hashlib
|
|
import time
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sqlite3
|
|
import ssl
|
|
import socket
|
|
import uuid
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
def default_data_dir():
|
|
root = os.environ.get('XDG_DATA_HOME')
|
|
if root:
|
|
return Path(root) / 'csr-intake'
|
|
return Path.home() / '.local/share/csr-intake'
|
|
|
|
DEFAULT_DATA = default_data_dir()
|
|
|
|
def identity_value(value):
|
|
value = str(value or '').strip().casefold()
|
|
if value in ('', 'unknown', 'none', 'default string', 'to be filled by o.e.m.',
|
|
'not specified', 'system serial number', 'not applicable', 'n/a'):
|
|
return ''
|
|
if not value.strip('0f- '):
|
|
return ''
|
|
return value
|
|
|
|
def hardware_id(report):
|
|
raw_uuid = identity_value(report.get('system_uuid'))
|
|
try:
|
|
token = uuid.UUID(raw_uuid)
|
|
if token.int not in (0, (1 << 128) - 1):
|
|
return 'dmi-uuid:' + str(token)
|
|
except (ValueError, AttributeError):
|
|
pass
|
|
serial = identity_value(report.get('serial_number'))
|
|
vendor = identity_value(report.get('manufacturer'))
|
|
if serial and vendor:
|
|
payload = json.dumps([vendor, serial], ensure_ascii=True, separators=(',', ':'))
|
|
return 'dmi-serial-sha256:' + hashlib.sha256(payload.encode()).hexdigest()
|
|
raise ValueError('Keine zuverlässige Hardware-Kennung im Bericht. Mit der aktuellen Anwendung Hardware erneut erfassen; alte Berichte ohne Kennung werden nicht automatisch zugeordnet.')
|
|
|
|
def hardware_aliases(report):
|
|
primary = hardware_id(report)
|
|
aliases = [primary]
|
|
try:
|
|
serial = hardware_id({**report, 'system_uuid': ''})
|
|
if serial not in aliases:
|
|
aliases.append(serial)
|
|
except ValueError:
|
|
pass
|
|
return aliases
|
|
|
|
def device_kind(report):
|
|
chassis = str(report.get('chassis_type', ''))
|
|
if chassis == '13':
|
|
return 'AiO'
|
|
if chassis in ('8', '9', '10', '11', '14', '30', '31', '32') or report.get('batteries'):
|
|
return 'Laptop'
|
|
return 'Desktop'
|
|
|
|
def map_fields(report):
|
|
values = {}
|
|
for name in ('cpu', 'model'):
|
|
value = report.get(name)
|
|
if isinstance(value, str) and value.strip() and value.upper() != 'UNKNOWN':
|
|
values[name] = value.strip()[:200]
|
|
# Physically installed RAM, not Linux's usable memory (affected by iGPU reservations).
|
|
ram = report.get('installed_ram_gib')
|
|
if isinstance(ram, (int, float)) and not isinstance(ram, bool) and 1 <= ram <= 2048:
|
|
values['ram_gb'] = round(ram)
|
|
disks = report.get('disks') or []
|
|
ssd, hdd = [], []
|
|
for disk in disks:
|
|
if disk.get('removable') is not False:
|
|
continue
|
|
gib = disk.get('size_gib')
|
|
if not isinstance(gib, (int, float)) or isinstance(gib, bool) or not 1 <= gib <= 100000:
|
|
continue
|
|
# The kernel rotating flag is not reliable on USB drives; USB disks are excluded above.
|
|
(hdd if disk.get('rotational') is True else ssd if disk.get('rotational') is False else []).append(gib)
|
|
if ssd:
|
|
values['ssd_gb'] = round(sum(ssd) * (1024 ** 3) / 1e9)
|
|
if hdd:
|
|
values['hdd_gb'] = round(sum(hdd) * (1024 ** 3) / 1e9)
|
|
return values
|
|
|
|
def init_db(db):
|
|
db.execute('PRAGMA busy_timeout=10000')
|
|
db.execute('CREATE TABLE IF NOT EXISTS device_links (hwid TEXT PRIMARY KEY, inventory_number INTEGER NOT NULL)')
|
|
db.execute('CREATE TABLE IF NOT EXISTS inventory_queue (report_sha256 TEXT PRIMARY KEY, report_path TEXT NOT NULL, asset_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT \'PENDING\', attempts INTEGER NOT NULL DEFAULT 0, last_attempt INTEGER, next_attempt INTEGER NOT NULL DEFAULT 0, error TEXT NOT NULL DEFAULT \'\')')
|
|
db.execute('CREATE TABLE IF NOT EXISTS inventory_sync (report_sha256 TEXT PRIMARY KEY, asset_id TEXT NOT NULL, inventory_number INTEGER NOT NULL, status TEXT NOT NULL, last_attempt_utc TEXT, response TEXT)')
|
|
|
|
def api_request(base, key, method, suffix, payload=None, timeout=12, ca_file=None):
|
|
body = None if payload is None else json.dumps(payload).encode('utf-8')
|
|
url = base.rstrip('/') + '/api/v1/' + suffix.lstrip('/')
|
|
request = urllib.request.Request(url, data=body, method=method, headers={
|
|
'Authorization': 'Api-Key ' + key,
|
|
'Accept': 'application/json',
|
|
'Content-Type': 'application/json',
|
|
'User-Agent': 'csr-intake/2.6',
|
|
})
|
|
context = ssl.create_default_context()
|
|
if ca_file:
|
|
context.load_verify_locations(cafile=str(ca_file))
|
|
with urllib.request.urlopen(request, timeout=timeout, context=context) as res:
|
|
result = json.loads(res.read(1024 * 1024).decode('utf-8'))
|
|
if method == 'POST' and suffix.strip('/') == 'devices/resolve':
|
|
status = getattr(res, 'status', None)
|
|
if status in (200, 201):
|
|
result['_registration'] = 'NEW' if status == 201 else 'EXISTING'
|
|
return result
|
|
|
|
def credentials(config):
|
|
base = config.get('url', '')
|
|
if not base: raise ValueError('Inventory URL is not configured; reports remain offline')
|
|
parsed = urllib.parse.urlparse(base)
|
|
if (parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password
|
|
or parsed.path not in ('', '/') or parsed.params or parsed.query or parsed.fragment):
|
|
raise ValueError('Inventory URL must be an HTTPS origin (e.g. https://inventory.example.org)')
|
|
key = os.environ.get('CSR_API_KEY', '')
|
|
if not key:
|
|
secret = Path(config.get('key_file', DEFAULT_DATA / 'config/inventory-api-key'))
|
|
key = secret.read_text(encoding='utf-8').strip()
|
|
if not key:
|
|
raise ValueError('No API key configured')
|
|
return base.rstrip('/'), key
|
|
|
|
def connection_error(ex):
|
|
# Never include server responses, request headers, or raw URLs in errors.
|
|
if isinstance(ex, urllib.error.HTTPError):
|
|
hints = {401: 'API key missing, expired or invalid', 403: 'API key lacks permission',
|
|
404: 'Inventory API endpoint not found. Server update required for hardware registration', 409: 'Hardware identity or revision conflict. Check existing or archived records on the server'}
|
|
return f'HTTP {ex.code}: {hints.get(ex.code, "Inventory server rejected the request")}'
|
|
reason = ex.reason if isinstance(ex, urllib.error.URLError) else ex
|
|
if isinstance(reason, ssl.SSLCertVerificationError):
|
|
return 'TLS certificate verification failed. Check config/inventory-server.crt, server address and system clock.'
|
|
if isinstance(reason, ssl.SSLError):
|
|
return 'TLS connection failed. Check the server HTTPS configuration.'
|
|
if isinstance(reason, socket.gaierror):
|
|
return 'DNS lookup failed. Check Wi-Fi, DNS and inventory server address.'
|
|
if isinstance(reason, (TimeoutError, socket.timeout)):
|
|
return 'Connection timed out. Connect Wi-Fi and check the inventory server is reachable.'
|
|
if isinstance(reason, ConnectionRefusedError):
|
|
return 'Connection refused. Check the inventory server address, port and service.'
|
|
if isinstance(ex, urllib.error.URLError):
|
|
return 'Inventory server unreachable. Check Wi-Fi, network routing and server address.'
|
|
if isinstance(ex, FileNotFoundError):
|
|
return 'Inventory API key or certificate file is missing. Check the files in config/.'
|
|
if isinstance(ex, ValueError):
|
|
return str(ex)[:200]
|
|
return type(ex).__name__
|
|
|
|
def check_connection(config):
|
|
base, key = credentials(config)
|
|
tls = {'ca_file': config['ca_file']} if config.get('ca_file') else {}
|
|
api_request(base, key, 'GET', 'choices/', **tls)
|
|
return 'Inventory connection OK: HTTPS certificate and API key verified. No records uploaded.'
|
|
|
|
def sync_one(report_path, config, db, dry_run=False):
|
|
raw = report_path.read_bytes()
|
|
sha = hashlib.sha256(raw).hexdigest()
|
|
report = json.loads(raw)
|
|
asset = report['asset_id']
|
|
hwid = hardware_id(report)
|
|
values = map_fields(report)
|
|
done = db.execute('SELECT status FROM inventory_sync WHERE report_sha256=?', (sha,)).fetchone()
|
|
if done and done[0] == 'SYNCED':
|
|
row = db.execute('SELECT inventory_number FROM inventory_sync WHERE report_sha256=?', (sha,)).fetchone()
|
|
return 'ALREADY_SYNCED', row[0]
|
|
if not values:
|
|
return 'NO_VERIFIED_FIELDS', None
|
|
if dry_run:
|
|
print(json.dumps({'hwid': hwid, 'number': 'assigned by server', 'changes': values}, indent=2, ensure_ascii=False))
|
|
return 'DRY_RUN', None
|
|
base, key = credentials(config)
|
|
tls = {'ca_file': config['ca_file']} if config.get('ca_file') else {}
|
|
# Resolve atomically on the server; local asset labels never select a server record.
|
|
current = api_request(base, key, 'POST', 'devices/resolve/',
|
|
{'hwid': hwid, 'aliases': hardware_aliases(report), 'device_kind': device_kind(report), 'model': values.get('model', '')}, **tls)
|
|
number = current.get('number')
|
|
if not isinstance(number, int) or isinstance(number, bool) or number < 1:
|
|
raise ValueError('Server did not return a valid inventory number')
|
|
with db:
|
|
db.execute('INSERT INTO device_links(hwid,inventory_number) VALUES(?,?) ON CONFLICT(hwid) DO UPDATE SET inventory_number=excluded.inventory_number', (hwid, number))
|
|
revision = current.get('revision')
|
|
if not isinstance(revision, int) or revision < 1:
|
|
raise ValueError('Server did not return a valid revision')
|
|
# Reject suspicious identity collisions instead of overwriting another model.
|
|
existing_model = (current.get('model') or '').strip()
|
|
if existing_model and values.get('model') and existing_model.casefold() != values['model'].casefold():
|
|
raise ValueError(f'MODEL_MISMATCH: inventory #{number} is {existing_model!r}; scan is {values["model"]!r}')
|
|
# Avoid unnecessary revisions when all mapped values already match.
|
|
actual = {k: v for k, v in values.items() if current.get(k) != v}
|
|
if current.get('stage') not in (None, 'INTAKE'): actual = {}
|
|
if not actual:
|
|
status = 'SYNCED'
|
|
else:
|
|
updated = api_request(base, key, 'PATCH', f'devices/{number}/', {**actual, 'expected_revision': revision}, **tls)
|
|
if not isinstance(updated.get('revision'), int) or updated['revision'] <= revision:
|
|
raise ValueError('Server did not acknowledge the new revision; retry will recheck current values')
|
|
status = 'SYNCED'
|
|
now = dt.datetime.now(dt.timezone.utc).isoformat()
|
|
with db:
|
|
db.execute('INSERT INTO inventory_sync(report_sha256,asset_id,inventory_number,status,last_attempt_utc,response) VALUES(?,?,?,?,?,?) ON CONFLICT(report_sha256) DO UPDATE SET status=excluded.status,last_attempt_utc=excluded.last_attempt_utc,response=excluded.response', (sha, asset, number, status, now, 'OK'))
|
|
return status, number
|
|
|
|
def discover_queue(root, db):
|
|
"""Discover immutable reports even when no inventory connection is configured."""
|
|
for path in sorted((root / 'reports').glob('*/*/report.json')):
|
|
if not path.resolve().is_relative_to((root / 'reports').resolve()):
|
|
continue
|
|
try:
|
|
raw = path.read_bytes()
|
|
report = json.loads(raw)
|
|
asset = report['asset_id']
|
|
sha = hashlib.sha256(raw).hexdigest()
|
|
done = db.execute('SELECT status FROM inventory_sync WHERE report_sha256=?', (sha,)).fetchone()
|
|
with db:
|
|
db.execute('INSERT OR IGNORE INTO inventory_queue(report_sha256,report_path,asset_id,status) VALUES(?,?,?,?)', (sha, str(path.resolve()), asset, 'SYNCED' if done and done[0] == 'SYNCED' else 'PENDING'))
|
|
except (OSError, ValueError, KeyError):
|
|
continue
|
|
|
|
def process_queue(root, db, config, retry=False, selected=None, retry_seconds=60):
|
|
discover_queue(root, db)
|
|
failures = 0
|
|
rows = db.execute("SELECT report_sha256, report_path, attempts, next_attempt FROM inventory_queue WHERE status != 'SYNCED' ORDER BY report_path").fetchall()
|
|
for sha, raw_path, attempts, next_attempt in rows:
|
|
path = Path(raw_path)
|
|
if selected and path.resolve() != selected.resolve():
|
|
continue
|
|
if not retry and next_attempt > int(time.time()):
|
|
continue
|
|
stamp = int(time.time())
|
|
# Persist before the request; interrupted uploads can be retried with the same report hash.
|
|
with db:
|
|
db.execute("UPDATE inventory_queue SET status='UPLOADING', attempts=attempts+1, last_attempt=? WHERE report_sha256=?", (stamp, sha))
|
|
try:
|
|
if not path.resolve().is_relative_to((root / 'reports').resolve()):
|
|
raise ValueError('Report path outside reports directory')
|
|
if hashlib.sha256(path.read_bytes()).hexdigest() != sha:
|
|
raise ValueError('Queued report changed; original upload cancelled')
|
|
result, number = sync_one(path, config, db)
|
|
if result not in ('SYNCED', 'ALREADY_SYNCED'):
|
|
raise ValueError(result)
|
|
with db:
|
|
db.execute("UPDATE inventory_queue SET status='SYNCED', error='', next_attempt=0 WHERE report_sha256=?", (sha,))
|
|
if db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='scans'").fetchone():
|
|
db.execute("UPDATE scans SET sync_status='SYNCED', synced_inventory_id=? WHERE report_path=?", (str(number), str(path)))
|
|
print(f'{path.parent.parent.name}: SYNCED -> #{number}')
|
|
except (OSError, ValueError, KeyError, urllib.error.URLError) as ex:
|
|
failures += 1
|
|
# Error strings deliberately exclude URLs, API keys and raw server responses.
|
|
error = connection_error(ex)
|
|
if isinstance(ex, urllib.error.HTTPError):
|
|
ex.close()
|
|
delay = min(3600, retry_seconds * (2 ** min(attempts, 6)))
|
|
with db:
|
|
db.execute("UPDATE inventory_queue SET status='ERROR', error=?, next_attempt=? WHERE report_sha256=?", (error, stamp + delay, sha))
|
|
print(f'{path.parent.parent.name}: {error}', file=sys.stderr)
|
|
return failures
|
|
|
|
def station_identity():
|
|
def read(name):
|
|
for folder in ['/sys/class/dmi/id', '/run/csr-intake']:
|
|
try:
|
|
value = (Path(folder)/name).read_text().strip()
|
|
if value: return value
|
|
except OSError: pass
|
|
return ''
|
|
return {'system_uuid':read('product_uuid'),'serial_number':read('product_serial'),
|
|
'manufacturer':read('sys_vendor'),'model':read('product_name'),'chassis_type':read('chassis_type')}
|
|
|
|
def station_config(root):
|
|
config=json.loads((root/'config/inventory.json').read_text())
|
|
config.setdefault('key_file',str(root/'config/inventory-api-key'))
|
|
if (root/'config/inventory-server.crt').exists():config.setdefault('ca_file',str(root/'config/inventory-server.crt'))
|
|
return config
|
|
|
|
def station_api(config,method,suffix,payload=None):
|
|
base,key=credentials(config)
|
|
return api_request(base,key,method,suffix,payload,**({'ca_file':config['ca_file']} if config.get('ca_file') else {}))
|
|
|
|
def station_resolve(config,report):
|
|
return station_api(config,'POST','devices/resolve/',{'hwid':hardware_id(report),'aliases':hardware_aliases(report),
|
|
'device_kind':device_kind(report),'model':report.get('model','')})
|
|
|
|
def station_save(path,value):
|
|
path.parent.mkdir(parents=True,exist_ok=True)
|
|
tmp=path.with_suffix('.new')
|
|
with tmp.open('w') as f:
|
|
json.dump(value,f,ensure_ascii=False);f.flush();os.fsync(f.fileno())
|
|
os.replace(tmp,path)
|
|
|
|
def queue_inspection(root,path,phase):
|
|
inspection=json.loads(path.read_text())
|
|
ident=inspection['identity']
|
|
identity={'system_uuid':ident.get('uuid'),'serial_number':ident.get('serial'),
|
|
'manufacturer':ident.get('manufacturer'),'model':ident.get('model')}
|
|
# An immutable report provides the inspected computer's chassis information.
|
|
if inspection.get('report'): identity.update({k:inspection['report'].get(k,identity.get(k)) for k in ['system_uuid','serial_number','manufacturer','model','chassis_type','batteries']})
|
|
results={k:{'outcome':v['outcome'],'notes':v.get('notes',''),'history':v.get('history',[]),'evidence':v.get('evidence','')} for k,v in inspection['checks'].items()}
|
|
body={'hwid':hardware_id(identity),'phase':phase,'technician':inspection.get('technician',''),
|
|
'notes':inspection.get('notes',''),'results':results}
|
|
digest=hashlib.sha256(json.dumps(body,sort_keys=True).encode()).hexdigest()
|
|
body['request_id']=str(uuid.uuid5(uuid.NAMESPACE_URL,'csr-inspection:'+inspection['asset']+':'+str(inspection['revision'])+':'+str(inspection['updated'])+':'+digest))
|
|
file=root/'outbox/inspections'/f"{inspection['updated']:020}-{inspection['revision']:020}-{body['request_id']}.json"
|
|
receipt=root/'outbox/sent'/file.name
|
|
if not file.exists() and not receipt.exists():station_save(file,{'identity':identity,'body':body})
|
|
return file
|
|
|
|
def flush_inspections(root,config):
|
|
latest=None
|
|
for path in sorted((root/'outbox/inspections').glob('*.json')):
|
|
event=json.loads(path.read_text());device=station_resolve(config,event['identity'])
|
|
body={**event['body'],'expected_revision':device['revision']}
|
|
latest=station_api(config,'POST',f"devices/{device['number']}/inspection/",body)
|
|
done=root/'outbox/sent'/path.name;done.parent.mkdir(parents=True,exist_ok=True)
|
|
os.replace(path,done)
|
|
return latest
|
|
|
|
def run_station(root,action,report_path):
|
|
# Recover the last saved snapshot even if power was lost before a worker queued it.
|
|
# These files may belong to a different PC inspected earlier on this workstation.
|
|
if action in ['boot','flush']:
|
|
for saved in sorted((root/'inspections').glob('*.json')):
|
|
inspection=json.loads(saved.read_text())
|
|
if inspection.get('sync_phase'):
|
|
queue_inspection(root,saved,inspection['sync_phase'])
|
|
# Enqueue before reading config/network so interruption and offline use preserve results.
|
|
if action in ['start','progress','complete','queue']:
|
|
if not report_path:raise ValueError('Inspection file is required')
|
|
phase={'start':'STARTED','progress':'IN_PROGRESS','complete':'COMPLETE'}.get(action)
|
|
if phase is None: phase=json.loads(report_path.read_text()).get('sync_phase') or 'IN_PROGRESS'
|
|
queue_inspection(root,report_path,phase)
|
|
if action=='queue':
|
|
print(json.dumps({'queued':True}))
|
|
return 0
|
|
(root/'database').mkdir(parents=True,exist_ok=True)
|
|
config=station_config(root)
|
|
if action in ['start','progress','complete','flush']:
|
|
flush_inspections(root,config)
|
|
device=station_resolve(config,station_identity())
|
|
else:
|
|
report=json.loads(report_path.read_text()) if report_path else station_identity()
|
|
device=station_resolve(config,report)
|
|
registration=device.get('_registration')
|
|
if action=='intake' and device['stage']=='INTAKE':
|
|
device=station_api(config,'POST',f"devices/{device['number']}/intake/",{
|
|
'hwid':hardware_id(report),'expected_revision':device['revision'],'specifications':map_fields(report)})
|
|
if registration:
|
|
device['_registration']=registration
|
|
if action == 'intake' and report_path:
|
|
raw=report_path.read_bytes(); sha=hashlib.sha256(raw).hexdigest()
|
|
with contextlib.closing(sqlite3.connect(root/'database/intake.sqlite3')) as db, db:
|
|
init_db(db);discover_queue(root,db)
|
|
db.execute("INSERT OR REPLACE INTO inventory_sync(report_sha256,asset_id,inventory_number,status,last_attempt_utc,response) VALUES(?,?,?,'SYNCED',?,'OK')",(sha,report.get('asset_id',''),device['number'],dt.datetime.now(dt.timezone.utc).isoformat()))
|
|
db.execute("UPDATE inventory_queue SET status='SYNCED',error='',next_attempt=0 WHERE report_sha256=?",(sha,))
|
|
if db.execute("SELECT name FROM sqlite_master WHERE name='scans'").fetchone():
|
|
db.execute("UPDATE scans SET sync_status='SYNCED',synced_inventory_id=? WHERE report_path=?",(str(device['number']),str(report_path)))
|
|
result={'device':device,'pending':len(list((root/'outbox/inspections').glob('*.json')))}
|
|
station_save(root/'database/station-device.json',result)
|
|
print(json.dumps(result))
|
|
return 0
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Offline inventory queue with server-assigned device numbers')
|
|
parser.add_argument('--data', type=Path, default=DEFAULT_DATA)
|
|
parser.add_argument('--station', choices=['boot','intake','start','progress','complete','flush','queue'])
|
|
parser.add_argument('--identity-json', help=argparse.SUPPRESS)
|
|
parser.add_argument('--report', type=Path)
|
|
parser.add_argument('--dry-run', action='store_true')
|
|
parser.add_argument('--list', action='store_true', help='Discover reports and print queue as JSON; no network requests')
|
|
parser.add_argument('--check', action='store_true', help='Verify TLS and API authentication without uploading')
|
|
parser.add_argument('--retry', action='store_true', help='Retry errors immediately, ignoring backoff')
|
|
args = parser.parse_args()
|
|
if args.identity_json:
|
|
try:
|
|
hwid = hardware_id(json.loads(args.identity_json))
|
|
print(json.dumps({'hwid': hwid, 'local_id': 'HW-' + hashlib.sha256(hwid.encode()).hexdigest()[:24]}))
|
|
return 0
|
|
except ValueError as ex:
|
|
print(str(ex), file=sys.stderr)
|
|
return 2
|
|
root = args.data.resolve()
|
|
(root / 'database').mkdir(parents=True, exist_ok=True)
|
|
# Cross-process lock prevents GUI/CLI concurrent PATCH requests.
|
|
with (root / 'database' / 'inventory-sync.lock').open('a') as lock:
|
|
try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
print('Another upload worker is running', file=sys.stderr)
|
|
return 2
|
|
if args.station:
|
|
try: return run_station(root,args.station,args.report)
|
|
except (OSError,ValueError,KeyError,urllib.error.URLError) as ex:
|
|
print(connection_error(ex),file=sys.stderr)
|
|
if isinstance(ex, urllib.error.HTTPError):
|
|
ex.close()
|
|
return 2
|
|
with contextlib.closing(sqlite3.connect(root / 'database' / 'intake.sqlite3', timeout=20)) as db, db:
|
|
init_db(db)
|
|
discover_queue(root, db)
|
|
if args.list:
|
|
db.row_factory = sqlite3.Row
|
|
print(json.dumps([dict(row) for row in db.execute('SELECT q.*, s.inventory_number FROM inventory_queue q LEFT JOIN inventory_sync s USING(report_sha256) ORDER BY q.last_attempt DESC, q.report_path')]))
|
|
return 0
|
|
if args.dry_run:
|
|
paths = [args.report] if args.report else sorted((root / 'reports').glob('*/*/report.json'))
|
|
for path in paths: sync_one(path, {}, db, True)
|
|
return 0
|
|
config_path = root / 'config' / 'inventory.json'
|
|
try:
|
|
config = json.loads(config_path.read_text()) if config_path.exists() else {}
|
|
config.setdefault('key_file', str(root / 'config' / 'inventory-api-key'))
|
|
cert = root / 'config' / 'inventory-server.crt'
|
|
if cert.exists():
|
|
config.setdefault('ca_file', str(cert))
|
|
if args.check:
|
|
print(check_connection(config))
|
|
return 0
|
|
settings_path = root / 'config' / 'workshop.json'
|
|
settings = json.loads(settings_path.read_text()) if settings_path.exists() else {}
|
|
delay = max(10, min(3600, int(settings.get('retry_seconds', 60))))
|
|
return 1 if process_queue(root, db, config, args.retry, args.report, delay) else 0
|
|
except (OSError, ValueError, KeyError) as ex:
|
|
print(connection_error(ex), file=sys.stderr)
|
|
return 2
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|