import os import logging from datetime import datetime from zoneinfo import ZoneInfo from functools import wraps from flask import Flask, request, jsonify, Response from ingress import IngressAPI from models import EventType, Plext from pymongo import MongoClient from pymongo.errors import PyMongoError from dotenv import load_dotenv load_dotenv() # 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 check_basic_auth(username: str, password: str) -> bool: """ Check if the provided username and password match the configured credentials. Args: username: Username from the request password: Password from the request Returns: True if credentials match, False otherwise """ expected_username = os.getenv("BASIC_AUTH_USER") expected_password = os.getenv("BASIC_AUTH_PASSWORD") if not expected_username or not expected_password: logger.warning("BASIC_AUTH_USER or BASIC_AUTH_PASSWORD not configured") return False return username == expected_username and password == expected_password def basic_auth_required(f): """ Decorator to require basic authentication for an endpoint. Returns: 401 Unauthorized if authentication fails or is missing """ @wraps(f) def decorated(*args, **kwargs): auth = request.authorization if not auth or not check_basic_auth(auth.username, auth.password): return Response( "Could not verify your access level for that URL.\n" "You have to login with proper credentials", 401, {"WWW-Authenticate": 'Basic realm="Login Required"'}, ) return f(*args, **kwargs) return decorated 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/from-db", methods=["GET", "OPTIONS"]) @basic_auth_required def get_plexts_from_db(): """ Get plexts from MongoDB with optional filters. Query Parameters: player_name: Filter by player name (optional) timestamp_from: Minimum timestamp in milliseconds (optional) timestamp_to: Maximum timestamp in milliseconds (optional) Returns: JSON response with list of plexts (without _id field) """ # CORS headers cors_headers = { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "GET, OPTIONS", "Access-Control-Allow-Headers": "Content-Type, Authorization", } # Handle OPTIONS preflight request if request.method == "OPTIONS": return "", 200, cors_headers try: # Parse query parameters player_name = request.args.get("player_name") timestamp_from = request.args.get("timestamp_from") timestamp_to = request.args.get("timestamp_to") # Validate and convert timestamp parameters to integers if provided if timestamp_from is not None: try: timestamp_from = int(timestamp_from) except ValueError: return ( jsonify({"error": "timestamp_from must be an integer"}), 400, cors_headers, ) if timestamp_to is not None: try: timestamp_to = int(timestamp_to) except ValueError: return ( jsonify({"error": "timestamp_to must be an integer"}), 400, cors_headers, ) # Build MongoDB filter query filter_query = {} if player_name: filter_query["player_name"] = player_name if timestamp_from is not None: filter_query["timestamp"] = filter_query.get("timestamp", {}) filter_query["timestamp"]["$gte"] = timestamp_from if timestamp_to is not None: filter_query["timestamp"] = filter_query.get("timestamp", {}) filter_query["timestamp"]["$lte"] = timestamp_to # Connect to MongoDB mongo_uri = os.getenv("MONGO_URI") db_name = os.getenv("DB_NAME") collection_name = os.getenv("COLLECTION_NAME") client = MongoClient(mongo_uri) db = client[db_name] collection = db[collection_name] try: # Projection to exclude _id field projection = {"_id": 0} # Execute query with sorting (timestamp DESC - most recent first) cursor = collection.find(filter=filter_query, projection=projection).sort( "timestamp", -1 ) # Convert cursor to list plexts = list(cursor) return jsonify({"count": len(plexts), "plexts": plexts}), 200, cors_headers except PyMongoError as e: logger.error(f"MongoDB error: {e}") return jsonify({"error": "Database error"}), 500, cors_headers finally: client.close() except Exception as e: logger.exception("Unexpected error in get_plexts_from_db") return jsonify({"error": "An error occurred"}), 500, cors_headers @app.route("/plexts/from-api", methods=["GET"]) @basic_auth_required def get_plexts_from_api(): """ 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/from-api": { "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], }, "/plexts/from-db": { "method": "GET", "description": "Get plexts from MongoDB with optional filters", "parameters": { "player_name": "Filter by player name (optional)", "timestamp_from": "Minimum timestamp in milliseconds (optional)", "timestamp_to": "Maximum timestamp in milliseconds (optional)", }, "authentication": "Basic Auth required", }, }, } ) if __name__ == "__main__": app.run(debug=True, host="0.0.0.0", port=os.getenv("PORT", 5000))