98 lines
3.9 KiB
Python
98 lines
3.9 KiB
Python
import time
|
|
from datetime import datetime, timedelta, date
|
|
|
|
from flask import Flask, jsonify
|
|
from flask import request
|
|
import requests
|
|
import json
|
|
|
|
app = Flask(__name__)
|
|
app.config['JSON_AS_ASCII'] = False
|
|
|
|
@app.route("/")
|
|
def user_info():
|
|
url = "https://traewelling.de/api/v1/auth/user"
|
|
headers = {'Authorization': request.headers.get('Authorization')}
|
|
return requests.get(url, headers=headers).json()
|
|
|
|
@app.route("/self/followers")
|
|
def own_followers():
|
|
url = "https://traewelling.de/api/v1/user/self/followers"
|
|
headers = {'Authorization': request.headers.get('Authorization')}
|
|
req = requests.get(url, headers=headers).json()
|
|
data = req['data']
|
|
|
|
while req['links']['next'] is not None:
|
|
url = req['links']['next']
|
|
req = requests.get(url, headers=headers).json()
|
|
data += req['data']
|
|
return data
|
|
|
|
@app.route("/self/followings")
|
|
def own_followings():
|
|
url = "https://traewelling.de/api/v1/user/self/followings"
|
|
headers = {'Authorization': request.headers.get('Authorization')}
|
|
req = requests.get(url, headers=headers).json()
|
|
data = req['data']
|
|
|
|
while req['links']['next'] is not None:
|
|
url = req['links']['next']
|
|
req = requests.get(url, headers=headers).json()
|
|
data += req['data']
|
|
return data
|
|
|
|
@app.route("/self/stats")
|
|
def basic_stats():
|
|
try:
|
|
detailed = bool(request.args.get('detailed'))
|
|
except KeyError:
|
|
detailed = False
|
|
|
|
try:
|
|
poly = bool(request.args.get('polylines'))
|
|
except KeyError:
|
|
poly = False
|
|
|
|
try:
|
|
no_days = int(request.args.get('days'))
|
|
print(no_days)
|
|
except TypeError:
|
|
no_days = -1
|
|
|
|
if no_days < 0:
|
|
url = f"https://traewelling.de/api/v1/statistics?from={request.args.get('from')}&until={request.args.get('to')}"
|
|
else:
|
|
url = f"https://traewelling.de/api/v1/statistics?from={((date.today() - timedelta(days=no_days)) if no_days>0 else date.today()).strftime('%Y%m%dT%H%M%SZ')}&until={(date.today() + timedelta(days=1)).strftime('%Y%m%dT%H%M%SZ')}"
|
|
|
|
headers = {'Authorization': request.headers.get('Authorization')}
|
|
req = requests.get(url, headers=headers).json()
|
|
data = req['data']
|
|
|
|
if not detailed:
|
|
return json.dumps(data, ensure_ascii=False)
|
|
else:
|
|
detailed_data = []
|
|
total_distance = 0
|
|
total_duration = 0
|
|
total_points = 0
|
|
for day in data['time']:
|
|
if day['count'] > 0 or day['date'] == date.today().strftime('%Y-%m-%d'):
|
|
url = f"https://traewelling.de/api/v1/statistics/daily/{day['date']}?withPolylines={'true' if poly else 'false'}"
|
|
req = requests.get(url, headers=headers).json()
|
|
total_distance += req['data']['totalDistance']
|
|
total_duration += req['data']['totalDuration']
|
|
total_points += req['data']['totalPoints']
|
|
|
|
for status in req['data']['statuses']:
|
|
tags = {}
|
|
for tag in status['tags']:
|
|
tags[str(tag['key']).replace("trwl:", "")] = tag['value']
|
|
status['tags'] = tags
|
|
try:
|
|
status['train']['origin']['departureDelay'] = (datetime.fromisoformat(str(status['train']['origin']['departure'])) - datetime.fromisoformat(str(status['train']['origin']['departurePlanned']))).total_seconds()
|
|
status['train']['destination']['arrivalDelay'] = (datetime.fromisoformat(str(status['train']['origin']['arrival'])) - datetime.fromisoformat(str(status['train']['origin']['arrivalPlanned']))).total_seconds()
|
|
except KeyError:
|
|
continue
|
|
detailed_data.append(status)
|
|
ret_data = {'data': detailed_data, 'totalDistance': total_distance, 'totalDuration': total_duration, 'totalPoints': total_points}
|
|
return json.dumps(ret_data, ensure_ascii=False)
|