import os import logging from datetime import datetime from zoneinfo import ZoneInfo from functools import wraps from flask import Flask, request, jsonify, Response from flask_cors import CORS 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__) CORS(app) 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"]) @basic_auth_required def get_plexts_from_db(): """ Get plexts from MongoDB with optional filters. Returns plexts sorted by timestamp (most recent first). Query Parameters: player_name: Filter by player name (optional) timestamp_from: Minimum timestamp in milliseconds (optional) timestamp_to: Maximum timestamp in milliseconds (optional) limit: Maximum number of results to return (optional, default: 100) Returns: JSON response with list of plexts (without _id field), limited by the limit parameter """ 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") limit_param = request.args.get("limit") # 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 if timestamp_to is not None: try: timestamp_to = int(timestamp_to) except ValueError: return jsonify({"error": "timestamp_to must be an integer"}), 400 # Validate and convert limit parameter to integer if provided, otherwise default to 100 if limit_param is not None: try: limit = int(limit_param) if limit <= 0: return jsonify({"error": "limit must be a positive integer"}), 400 except ValueError: return jsonify({"error": "limit must be an integer"}), 400 else: limit = 100 # 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) and dynamic limit cursor = ( collection.find(filter=filter_query, projection=projection) .sort("timestamp", -1) .limit(limit) ) # Convert cursor to list plexts = list(cursor) return jsonify({"count": len(plexts), "plexts": plexts}), 200 except PyMongoError as e: logger.error(f"MongoDB error: {e}") return jsonify({"error": "Database error"}), 500 finally: client.close() except Exception as e: logger.exception("Unexpected error in get_plexts_from_db") return jsonify({"error": "An error occurred"}), 500 @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. Returns plexts sorted by timestamp (most recent first).", "parameters": { "player_name": "Filter by player name (optional)", "timestamp_from": "Minimum timestamp in milliseconds (optional)", "timestamp_to": "Maximum timestamp in milliseconds (optional)", "limit": "Maximum number of results to return (optional, default: 100)", }, "authentication": "Basic Auth required", }, }, } ) if __name__ == "__main__": app.run(debug=True, host="0.0.0.0", port=os.getenv("PORT", 5000))