feat: add energy, weather and consumption endpoints with models and error handling

2025-08-04 19:09:07 +03:00
parent f1b8097f7a
commit 8fba6cfa1e
7 changed files with 257 additions and 1 deletions
@@ -0,0 +1,115 @@
+from fastapi import APIRouter
+from httpx import AsyncClient
+from pydantic import BaseModel, Field
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+import json
+
+from lib.exceptions import FetchException, ParseException
+from models.energy import Consumption
+
+# from models.consumption import Consumption
+from lib import logger
+
+
+router = APIRouter(prefix="/consumption", tags=["Consumption"])
+
+
+class _CookieData(BaseModel):
+    Host_next_auth_csrf_token: str = Field(..., description="CSRF token")
+    Secure_next_auth_session_token_0: str = Field(..., description="Session token 0")
+    Secure_next_auth_session_token_1: str = Field(..., description="Session token 1")
+
+
+@router.post("/")
+async def get_consumption(
+    cookie_data: _CookieData,
+    to_date: datetime = datetime.now() - timedelta(days=1),
+    from_date: datetime = datetime.now() - timedelta(days=14),
+):
+    """Get consumption data for a given date range. Max 14 days.
+
+    Args:
+        cookie_data: Cookie data containing CSRF token and session tokens
+        start_date: Start date of the date range
+        end_date: End date of the date range
+    """
+
+    base_url = (
+        "https://www.fortum.com/fi/sahkoa/api/trpc/loggedIn.timeSeries.listTimeSeries"
+    )
+    print(from_date.astimezone(ZoneInfo("UTC")), to_date.astimezone(ZoneInfo("UTC")))
+    input_param = {
+        "0": {
+            "json": {
+                "meteringPointNo": ["6907998"],
+                "fromDate": from_date.astimezone(ZoneInfo("UTC"))
+                .isoformat()
+                .replace("+00:00", "Z"),
+                "toDate": to_date.astimezone(ZoneInfo("UTC"))
+                .isoformat()
+                .replace("+00:00", "Z"),
+                "resolution": "HOUR",
+                "type": "CONSUMPTION",
+            }
+        }
+    }
+    params = {"batch": 1, "input": json.dumps(input_param)}
+    cookies = {
+        "NEXT_LOCALE": "fi",
+        "__Host-next-auth.csrf-token": cookie_data.Host_next_auth_csrf_token,
+        "__Secure-next-auth.callback-url": "https://www.fortum.com/fi/sahkoa/kirjautunut/sahko",
+        "__Secure-next-auth.session-token.0": cookie_data.Secure_next_auth_session_token_0,
+        "__Secure-next-auth.session-token.1": cookie_data.Secure_next_auth_session_token_1,
+    }
+    headers = {
+        "accept": "application/json",
+        "accept-language": "en-US,en;q=0.7",
+        "cache-control": "no-cache",
+        "content-type": "application/json",
+        "pragma": "no-cache",
+        "priority": "u=1, i",
+        "referer": "https://www.fortum.com/fi/sahkoa/kirjautunut/sahko",
+        "sec-ch-ua": '"Not)A;Brand";v="8", "Chromium";v="138", "Brave";v="138"',
+        "sec-ch-ua-mobile": "?0",
+        "sec-ch-ua-platform": '"Linux"',
+        "sec-fetch-dest": "empty",
+        "sec-fetch-mode": "cors",
+        "sec-fetch-site": "same-origin",
+        "sec-gpc": "1",
+        "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36",
+    }
+    async with AsyncClient() as client:
+        response = await client.get(
+            base_url, params=params, cookies=cookies, headers=headers, timeout=30
+        )
+    if response.status_code != 200:
+        logger.error(
+            f"Failed to fetch data from {response.url}. Status code: {response.status_code}. Response text: {response.text}"
+        )
+        raise FetchException(response.url, response.status_code)
+    data = response.json()
+    if not data:
+        logger.error(
+            f"Failed to parse data from {response.url}. Status code: {response.status_code}. Response text: {response.text}"
+        )
+        raise FetchException(response.url, response.status_code)
+
+    series_data = data[0]["result"]["data"]["json"][0]["series"]
+    retval: list[Consumption] = []
+    for series in series_data:
+        if series["cost"] is not None:
+            try:
+                consumption = Consumption(
+                    timestamp=series["atUTC"],
+                    consumption=series["energy"][0]["value"],
+                    total_cost=series["price"]["total"],
+                    vat_percentage=series["price"]["vatPercentage"],
+                )
+                retval.append(consumption)
+            except Exception as e:
+                logger.error(
+                    f"Error parsing consumption: {str(e)}", extra={"data": series}
+                )
+                raise ParseException()
+    return retval