Files
lingniu-vehicle-ingest/vehicle-data-platform/deploy/gzev-session-import.py
T
lingniu d6b957dfc8 feat: sync vehicle monitoring and accumulated platform updates
Add monitoring metrics, numeric sorting, report-time display, vehicle number labels, and map layout fixes. Include accumulated mileage, hydrogen, OneOS authentication, open platform, ingestion, deployment, and UI updates.

Exclude temporary files, generated reports, and credentials. Validate frontend 632 tests and production build, API packages, vehicle-map tests, and release installer tests. Gateway suite has one pre-existing date-sensitive identity throttling test failure reproduced on the parent commit.
2026-10-26 18:05:59 +08:00

58 lines
2.3 KiB
Python

#!/usr/bin/env python3
"""Import a browser session using hidden prompts, verify account, then replace atomically."""
import argparse
import getpass
import json
import os
from pathlib import Path
import tempfile
import urllib.error
import urllib.parse
import urllib.request
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--file', default='/var/lib/lingniu-gzev/session.json')
parser.add_argument('--account', default='guangdonglingniu')
args = parser.parse_args()
token = getpass.getpass('Chrome token Cookie (hidden): ').strip()
routing = getpass.getpass('Chrome bigs Cookie (hidden, may be empty): ').strip()
if not token or any(ch in token + routing for ch in '\r\n;'):
raise SystemExit('Invalid cookie value')
session = {'authorization': 'Bearer ' + token,
'cookies': [{'name': 'token', 'value': token}]}
if routing:
session['cookies'].append({'name': 'bigs', 'value': routing})
url = 'https://gzev.d.lnoneos.com/evdata-sso/api/base/getUserInfo?' + urllib.parse.urlencode({'token': token})
request = urllib.request.Request(url, headers={
'Authorization': session['authorization'],
'Cookie': '; '.join(c['name'] + '=' + c['value'] for c in session['cookies'])})
try:
with urllib.request.build_opener(NoRedirect()).open(request, timeout=30) as response:
result = json.load(response)
if str(result.get('code')) != '0' or result.get('data', {}).get('loginname') != args.account:
raise ValueError('identity mismatch')
except Exception:
raise SystemExit('Session validation failed; existing file unchanged') from None
path = Path(args.file)
path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
fd, name = tempfile.mkstemp(prefix='.session-', dir=str(path.parent))
try:
with os.fdopen(fd, 'w', encoding='utf-8') as stream:
json.dump(session, stream)
stream.flush()
os.fsync(stream.fileno())
os.replace(name, str(path))
finally:
if os.path.exists(name):
os.unlink(name)
print('Session validated and replaced; restart the mileage service to resume.')
if __name__ == '__main__':
main()