dockerize!
This commit is contained in:
Matteo Rosati
223
app.py
Normal file
223
app.py
Normal file
@@ -0,0 +1,223 @@
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from zoneinfo import ZoneInfo
|
||||
from flask import Flask, request, jsonify
|
||||
from ingress import IngressAPI
|
||||
from models import EventType, Plext
|
||||
|
||||
# Timezone configuration
|
||||
TIMEZONE = ZoneInfo("Europe/Rome")
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
|
||||
def parse_timestamp(value: str) -> int:
|
||||
"""
|
||||
Parse timestamp from either milliseconds (int) or ISO 8601 string.
|
||||
|
||||
Args:
|
||||
value: Either integer milliseconds or ISO 8601 string
|
||||
|
||||
Returns:
|
||||
Timestamp in milliseconds since epoch
|
||||
|
||||
Raises:
|
||||
ValueError: If format is invalid
|
||||
"""
|
||||
# Try parsing as integer (milliseconds)
|
||||
try:
|
||||
return int(value)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Try parsing as ISO 8601 datetime
|
||||
try:
|
||||
# Handle 'Z' suffix (UTC)
|
||||
iso_value = value.replace("Z", "+00:00")
|
||||
|
||||
# Handle timezone offset without colon (e.g., +0100 -> +01:00)
|
||||
# Match pattern like +0100 or -0100 at the end of the string
|
||||
import re
|
||||
|
||||
match = re.search(r"([+-])(\d{2})(\d{2})$", iso_value)
|
||||
if match:
|
||||
sign, hours, minutes = match.groups()
|
||||
iso_value = re.sub(
|
||||
r"([+-])(\d{2})(\d{2})$", f"{sign}{hours}:{minutes}", iso_value
|
||||
)
|
||||
|
||||
dt = datetime.fromisoformat(iso_value)
|
||||
return int(dt.timestamp() * 1000)
|
||||
except ValueError:
|
||||
raise ValueError(
|
||||
f"Invalid timestamp format: {value}. "
|
||||
"Use milliseconds (e.g., 1736659200000) or "
|
||||
"ISO 8601 format (e.g., '2026-01-12T10:00:00Z' or '2026-01-12T10:00:00+01:00')"
|
||||
)
|
||||
|
||||
|
||||
def plext_to_dict(plext: Plext) -> dict:
|
||||
"""
|
||||
Convert a Plext object to a dictionary for JSON serialization.
|
||||
|
||||
Args:
|
||||
plext: Plext object to convert
|
||||
|
||||
Returns:
|
||||
Dictionary representation of the plext
|
||||
"""
|
||||
coords = plext.get_event_coordinates()
|
||||
# Convert timestamp from milliseconds to datetime in Europe/Rome timezone
|
||||
dt = datetime.fromtimestamp(plext.timestamp / 1000.0, tz=TIMEZONE)
|
||||
return {
|
||||
"id": plext.id,
|
||||
"timestamp": plext.timestamp,
|
||||
"timestamp_formatted": dt.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"text": plext.text,
|
||||
"team": plext.team,
|
||||
"plext_type": plext.plext_type,
|
||||
"categories": plext.categories,
|
||||
"event_type": plext.get_event_type().name,
|
||||
"player_name": plext.get_player_name(),
|
||||
"portal_name": plext.get_portal_name(),
|
||||
"coordinates": {"lat": coords[0], "lng": coords[1]} if coords else None,
|
||||
"markup": [
|
||||
{
|
||||
"type": m.type,
|
||||
"plain": m.plain,
|
||||
"team": m.team,
|
||||
"name": m.name,
|
||||
"address": m.address,
|
||||
"latE6": m.latE6,
|
||||
"lngE6": m.lngE6,
|
||||
}
|
||||
for m in plext.markup
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@app.route("/plexts", methods=["GET"])
|
||||
def get_plexts():
|
||||
"""
|
||||
Get plexts from the Ingress API.
|
||||
|
||||
Query Parameters:
|
||||
event_types: List of event types to filter by (comma-separated)
|
||||
player_name: Player name to filter by
|
||||
min_lat: Minimum latitude (default: 45470259)
|
||||
min_lng: Minimum longitude (default: 12244155)
|
||||
max_lat: Maximum latitude (default: 45480370)
|
||||
max_lng: Maximum longitude (default: 12298207)
|
||||
min_timestamp: Minimum timestamp (milliseconds or ISO 8601 format, default: -1)
|
||||
max_timestamp: Maximum timestamp (milliseconds or ISO 8601 format, default: -1)
|
||||
|
||||
Returns:
|
||||
JSON response with list of plexts
|
||||
"""
|
||||
try:
|
||||
# Parse query parameters
|
||||
event_types_param = request.args.get("event_types")
|
||||
event_types = None
|
||||
if event_types_param:
|
||||
event_types_list = [et.strip() for et in event_types_param.split(",")]
|
||||
try:
|
||||
event_types = [EventType[et] for et in event_types_list]
|
||||
except KeyError as e:
|
||||
return jsonify({"error": f"Invalid event type: {e}"}), 400
|
||||
|
||||
player_name = request.args.get("player_name")
|
||||
|
||||
min_lat = int(request.args.get("min_lat", os.getenv("MIN_LAT")))
|
||||
min_lng = int(request.args.get("min_lng", os.getenv("MIN_LNG")))
|
||||
max_lat = int(request.args.get("max_lat", os.getenv("MAX_LAT")))
|
||||
max_lng = int(request.args.get("max_lng", os.getenv("MAX_LNG")))
|
||||
|
||||
min_timestamp_param = request.args.get("min_timestamp", "-1")
|
||||
max_timestamp_param = request.args.get("max_timestamp", "-1")
|
||||
|
||||
min_timestamp = parse_timestamp(min_timestamp_param)
|
||||
max_timestamp = parse_timestamp(max_timestamp_param)
|
||||
|
||||
# Initialize IngressAPI client
|
||||
cookie = os.getenv("INGRESS_COOKIE")
|
||||
|
||||
client = IngressAPI(
|
||||
version=os.getenv("V"),
|
||||
cookie=cookie,
|
||||
)
|
||||
|
||||
# Fetch plexts
|
||||
plexts = client.get_plexts(
|
||||
min_lat_e6=min_lat,
|
||||
min_lng_e6=min_lng,
|
||||
max_lat_e6=max_lat,
|
||||
max_lng_e6=max_lng,
|
||||
min_timestamp_ms=min_timestamp,
|
||||
max_timestamp_ms=max_timestamp,
|
||||
event_types=event_types,
|
||||
player_name=player_name,
|
||||
)
|
||||
|
||||
# Sort plexts by timestamp (ascending - oldest first)
|
||||
sorted_plexts = sorted(plexts, key=lambda p: p.timestamp)
|
||||
|
||||
# Convert to JSON-serializable format
|
||||
result = [plext_to_dict(p) for p in sorted_plexts]
|
||||
|
||||
return jsonify(
|
||||
{
|
||||
"count": len(result),
|
||||
"plexts": result,
|
||||
}
|
||||
)
|
||||
|
||||
except ValueError as e:
|
||||
logger.warning(f"Validation error: {e}")
|
||||
return jsonify({"error": str(e)}), 400
|
||||
except Exception as e:
|
||||
logger.exception("Error processing request")
|
||||
return jsonify({"error": f"An error occurred: {e}"}), 500
|
||||
|
||||
|
||||
@app.route("/", methods=["GET"])
|
||||
def index():
|
||||
"""Root endpoint providing API information.
|
||||
|
||||
Returns:
|
||||
JSON response containing API name, version, and available endpoints
|
||||
with their parameters and descriptions.
|
||||
"""
|
||||
return jsonify(
|
||||
{
|
||||
"name": "Ingress Intel API",
|
||||
"version": "1.0.0",
|
||||
"endpoints": {
|
||||
"/plexts": {
|
||||
"method": "GET",
|
||||
"description": "Get plexts from the Ingress API",
|
||||
"parameters": {
|
||||
"event_types": "List of event types to filter by (comma-separated)",
|
||||
"player_name": "Player name to filter by",
|
||||
"min_lat": "Minimum latitude (default: 45470259)",
|
||||
"min_lng": "Minimum longitude (default: 12244155)",
|
||||
"max_lat": "Maximum latitude (default: 45480370)",
|
||||
"max_lng": "Maximum longitude (default: 12298207)",
|
||||
"min_timestamp": "Minimum timestamp (milliseconds or ISO 8601 format, default: -1)",
|
||||
"max_timestamp": "Maximum timestamp (milliseconds or ISO 8601 format, default: -1)",
|
||||
},
|
||||
"event_types": [e.name for e in EventType],
|
||||
}
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(debug=True, host="0.0.0.0", port=os.getenv("PORT", 5000))
|
||||
Reference in New Issue
Block a user