| 1 | import base64 |
| 2 | import os |
| 3 | import time |
| 4 | import traceback |
| 5 | from datetime import datetime |
| 6 | from datetime import timedelta |
| 7 | from pathlib import Path |
| 8 | from typing import List |
| 9 | |
| 10 | from fastapi import HTTPException |
| 11 | from jinja2 import Environment |
| 12 | from jinja2 import FileSystemLoader |
| 13 | from loguru import logger |
| 14 | from playwright.async_api import async_playwright |
| 15 | from sqlalchemy.ext.asyncio import AsyncSession |
| 16 | from sqlalchemy.future import select |
| 17 | |
| 18 | from app.connectors.grafana.schema.reporting import GenerateReportRequest |
| 19 | from app.connectors.grafana.schema.reporting import GenerateReportResponse |
| 20 | from app.connectors.grafana.schema.reporting import GrafanaDashboardDetails |
| 21 | from app.connectors.grafana.schema.reporting import GrafanaGenerateIframeLinksRequest |
| 22 | from app.connectors.grafana.schema.reporting import GrafanaLinksList |
| 23 | from app.connectors.grafana.schema.reporting import GrafanaOrganizationDashboards |
| 24 | from app.connectors.grafana.schema.reporting import GrafanaOrganizations |
| 25 | from app.connectors.grafana.schema.reporting import RequestPanel |
| 26 | from app.connectors.grafana.schema.reporting import TimeRange |
| 27 | from app.connectors.grafana.utils.universal import create_grafana_client |
| 28 | from app.connectors.models import Connectors |
| 29 | |
| 30 | # from app.utils import get_connector_attribute |
| 31 | |
| 32 | |
| 33 | async def get_grafana_url(session: AsyncSession): |
| 34 | connector = await session.execute(select(Connectors).where(Connectors.connector_name == "Grafana")) |
| 35 | connector = connector.scalars().first() |
| 36 | return connector.connector_url |
| 37 | |
| 38 | |
| 39 | async def get_granfana_user(session: AsyncSession): |
| 40 | connector = await session.execute(select(Connectors).where(Connectors.connector_name == "Grafana")) |
| 41 | connector = connector.scalars().first() |
| 42 | return connector.connector_username |
| 43 | |
| 44 | |
| 45 | async def get_granfana_password(session: AsyncSession): |
| 46 | connector = await session.execute(select(Connectors).where(Connectors.connector_name == "Grafana")) |
| 47 | connector = connector.scalars().first() |
| 48 | return connector.connector_password |
| 49 | |
| 50 | |
| 51 | def calculate_unix_timestamps(time_range: TimeRange): |
| 52 | now = datetime.now() |
| 53 | if time_range.unit == "m": |
| 54 | start_time = now - timedelta(minutes=time_range.value) |
| 55 | elif time_range.unit == "h": |
| 56 | start_time = now - timedelta(hours=time_range.value) |
| 57 | elif time_range.unit == "d": |
| 58 | start_time = now - timedelta(days=time_range.value) |
| 59 | |
| 60 | timestamp_from = int(time.mktime(start_time.timetuple())) * 1000 |
| 61 | timestamp_to = int(time.mktime(now.timetuple())) * 1000 |
| 62 | |
| 63 | return timestamp_from, timestamp_to |
| 64 | |
| 65 | |
| 66 | def generate_panel_urls( |
| 67 | grafana_url: str, |
| 68 | request: GrafanaGenerateIframeLinksRequest, |
| 69 | timestamp_from: int, |
| 70 | timestamp_to: int, |
| 71 | theme: str = "dark", |
| 72 | ): |
| 73 | panel_links: List[GrafanaLinksList] = [] |
| 74 | # for panel_id in request.panel_ids: |
| 75 | panel_url = ( |
| 76 | f"{grafana_url}/d-solo/{request.dashboard_uid}/{request.dashboard_title}" |
| 77 | f"?orgId={request.org_id}&from={timestamp_from}&to={timestamp_to}" |
| 78 | f"&panelId={request.panel_id}&theme={theme}" |
| 79 | ) |
| 80 | panel_links.append(GrafanaLinksList(panel_id=request.panel_id, panel_url=panel_url)) |
| 81 | return panel_links |
| 82 | |
| 83 | |
| 84 | async def get_orgs() -> List[GrafanaOrganizations]: |
| 85 | """ |
| 86 | Update a dashboard in Grafana. |
| 87 | |
| 88 | Args: |
| 89 | dashboard_json (dict): The updated dashboard JSON. |
| 90 | organization_id (int): The ID of the organization. |
| 91 | folder_id (int): The ID of the folder. |
| 92 | |
| 93 | Returns: |
| 94 | dict: The updated dashboard response. |
| 95 | |
| 96 | Raises: |
| 97 | HTTPException: If there is an error updating the dashboard. |
| 98 | """ |
| 99 | logger.info("Getting organizations from Grafana") |
| 100 | try: |
| 101 | grafana_client = await create_grafana_client("Grafana") |
| 102 | orgs = grafana_client.organizations.list_organization() |
| 103 | return orgs |
| 104 | except Exception as e: |
| 105 | logger.error(f"Failed to collect organizations: {e}") |
| 106 | raise HTTPException(status_code=500, detail=f"Failed to collect organizations: {e}") |
| 107 | |
| 108 | |
| 109 | async def get_dashboards(org_id: int) -> List[GrafanaOrganizationDashboards]: |
| 110 | """ |
| 111 | Get dashboards from Grafana. |
| 112 | |
| 113 | Returns: |
| 114 | dict: The response containing the dashboards collected from Grafana. |
| 115 | """ |
| 116 | logger.info("Getting dashboards from Grafana") |
| 117 | try: |
| 118 | grafana_client = await create_grafana_client("Grafana") |
| 119 | logger.info(f"Switching to organization {org_id}") |
| 120 | grafana_client.user.switch_actual_user_organisation(org_id) |
| 121 | dashboards = grafana_client.search.search_dashboards() |
| 122 | return dashboards |
| 123 | except Exception as e: |
| 124 | logger.error(f"Failed to collect dashboards: {e}") |
| 125 | raise HTTPException(status_code=500, detail=f"Failed to collect dashboards: {e}") |
| 126 | |
| 127 | |
| 128 | async def get_dashboard_details(dashboard_uid: str) -> GrafanaDashboardDetails: |
| 129 | """ |
| 130 | Get dashboard details from Grafana. |
| 131 | |
| 132 | Args: |
| 133 | dashboard_uid (str): The UID of the dashboard. |
| 134 | |
| 135 | Returns: |
| 136 | dict: The response containing the dashboard details collected from Grafana. |
| 137 | """ |
| 138 | logger.info("Getting dashboard details from Grafana") |
| 139 | try: |
| 140 | grafana_client = await create_grafana_client("Grafana") |
| 141 | dashboard_details = grafana_client.dashboard.get_dashboard(dashboard_uid) |
| 142 | return dashboard_details |
| 143 | except Exception as e: |
| 144 | logger.error(f"Failed to collect dashboard details: {e}") |
| 145 | raise HTTPException(status_code=500, detail=f"Failed to collect dashboard details: {e}") |
| 146 | |
| 147 | |
| 148 | async def login_to_page(page, session: AsyncSession): |
| 149 | try: |
| 150 | # Navigate to the login page |
| 151 | # await page.goto(f'{await get_connector_attribute(connector_id=8, column_name="connector_url", session=session)}/login') |
| 152 | await page.goto(f"{await get_grafana_url(session)}/login") |
| 153 | # Enter the username and password |
| 154 | # await page.fill( |
| 155 | # 'input[name="user"]', |
| 156 | # f'{await get_connector_attribute(connector_id=8, column_name="connector_username", session=session)}', |
| 157 | # ) |
| 158 | await page.fill( |
| 159 | 'input[name="user"]', |
| 160 | f"{await get_granfana_user(session)}", |
| 161 | ) |
| 162 | # await page.fill( |
| 163 | # 'input[name="password"]', |
| 164 | # f'{await get_connector_attribute(connector_id=8, column_name="connector_password", session=session)}', |
| 165 | # ) |
| 166 | await page.fill( |
| 167 | 'input[name="password"]', |
| 168 | f"{await get_granfana_password(session)}", |
| 169 | ) |
| 170 | # Click the login button |
| 171 | await page.click('button[data-testid="data-testid Login button"]') |
| 172 | # Wait for navigation to complete |
| 173 | await page.wait_for_load_state(state="networkidle") |
| 174 | except Exception as e: |
| 175 | logger.error(f"Failed to login to Grafana: {e}") |
| 176 | traceback.print_exc() |
| 177 | raise HTTPException(status_code=500, detail="Failed to login to Grafana") |
| 178 | |
| 179 | |
| 180 | async def check_login_success(page): |
| 181 | # Check if login was successful by checking for an element that is only visible when logged in |
| 182 | body_class = await page.evaluate("document.body.className") |
| 183 | logger.info(f"Body class: {body_class}") |
| 184 | # if 'app-grafana no-overlay-scrollbar page-dashboard' in body_class: |
| 185 | if "app-grafana" in body_class: |
| 186 | logger.info("Login to Grafana successful") |
| 187 | return True |
| 188 | else: |
| 189 | raise HTTPException(status_code=500, detail="Failed to login to Grafana") |
| 190 | |
| 191 | |
| 192 | async def capture_screenshots(page, panels: List[RequestPanel]) -> List[RequestPanel]: |
| 193 | logger.info("Capturing screenshots") |
| 194 | last_url = "" |
| 195 | for panel in panels: |
| 196 | try: |
| 197 | # Check if the panel's URL is different from the last to optimize navigation |
| 198 | if panel.panel_url != last_url: |
| 199 | await page.goto(panel.panel_url) |
| 200 | await page.wait_for_load_state(state="networkidle") |
| 201 | last_url = panel.panel_url |
| 202 | |
| 203 | # Assuming default dimensions are always used in this example |
| 204 | logger.info(f"Panel width: {panel.panel_width}, height: {panel.panel_height} for panel {panel.panel_id}") |
| 205 | width = panel.panel_width |
| 206 | height = panel.panel_height |
| 207 | await page.set_viewport_size({"width": width, "height": height}) |
| 208 | |
| 209 | screenshot = await page.screenshot(type="png") |
| 210 | base64_image = base64.b64encode(screenshot).decode("utf-8") |
| 211 | panel.panel_base64 = base64_image |
| 212 | except Exception as e: |
| 213 | print(f"Failed to capture screenshot for panel {panel.panel_id}: {e}") |
| 214 | # Optionally, set panel_base64 to None or a default value in case of failure |
| 215 | panel.panel_base64 = None |
| 216 | return panels |
| 217 | |
| 218 | |
| 219 | async def generate_grafana_iframe_links(request: GrafanaGenerateIframeLinksRequest, session: AsyncSession): |
| 220 | """ |
| 221 | Function to generate Grafana dashboard iframe links. |
| 222 | |
| 223 | Args: |
| 224 | request (GrafanaGenerateIframeLinksRequest): The request body containing the dashboard UID and organization ID. |
| 225 | |
| 226 | Returns: |
| 227 | GrafanaDashboardPanelsResponse: The response containing the result of the dashboard provisioning. |
| 228 | """ |
| 229 | # get the Grafana URL from the database |
| 230 | grafana_url = await get_grafana_url(session) |
| 231 | logger.info(f"Grafana URL: {grafana_url}") |
| 232 | |
| 233 | # calculate the Unix timestamps based on the current time and the provided time range |
| 234 | timestamp_from, timestamp_to = calculate_unix_timestamps(request.time_range) |
| 235 | |
| 236 | # build the URL string for each panel_id |
| 237 | panel_urls = generate_panel_urls(grafana_url, request, timestamp_from, timestamp_to, theme=request.theme) |
| 238 | |
| 239 | # return only the panel url |
| 240 | return panel_urls[0].panel_url |
| 241 | |
| 242 | |
| 243 | def generate_html(panels: List[RequestPanel], request: GenerateReportRequest) -> str: |
| 244 | # Load the template |
| 245 | logger.info(f"Rendering HTML with panels: {len(panels)} panels") |
| 246 | templates_dir = Path(__file__).parent / "../reporting" |
| 247 | env = Environment(loader=FileSystemLoader(templates_dir)) |
| 248 | logger.info(f"Templates dir: {templates_dir}") |
| 249 | template = env.get_template("report-template.html") |
| 250 | |
| 251 | panel_groups = {} |
| 252 | for panel in panels: |
| 253 | if panel.row_id not in panel_groups: |
| 254 | panel_groups[panel.row_id] = [panel] |
| 255 | else: |
| 256 | panel_groups[panel.row_id].append(panel) |
| 257 | |
| 258 | # Convert the dict to a list of panel groups for the template |
| 259 | panel_groups_list = list(panel_groups.values()) |
| 260 | |
| 261 | # Render the template with the grouped panels |
| 262 | html_content = template.render( |
| 263 | panel_groups=panel_groups_list, |
| 264 | company_name=request.company_name, |
| 265 | timerange_text=request.timerange_text, |
| 266 | logo_base64=request.logo_base64, |
| 267 | ) |
| 268 | return html_content |
| 269 | |
| 270 | |
| 271 | def parse_timerange(timerange: str) -> dict: |
| 272 | """Parse the timerange string into a dictionary with value and unit keys.""" |
| 273 | timerange_value, timerange_unit = int(timerange[:-1]), timerange[-1] |
| 274 | return {"value": timerange_value, "unit": timerange_unit} |
| 275 | |
| 276 | |
| 277 | async def generate_panel_urls_object(panel: RequestPanel, timerange: dict, session: AsyncSession) -> str: |
| 278 | """Generate the iframe links for a panel.""" |
| 279 | iframe_links_request = GrafanaGenerateIframeLinksRequest( |
| 280 | time_range=timerange, |
| 281 | dashboard_uid=panel.dashboard_uid, |
| 282 | dashboard_title=panel.dashboard_title, |
| 283 | org_id=panel.org_id, |
| 284 | panel_id=panel.panel_id, |
| 285 | theme=panel.theme, |
| 286 | ) |
| 287 | return await generate_grafana_iframe_links(iframe_links_request, session) |
| 288 | |
| 289 | |
| 290 | async def write_html_to_file(html_string: str, file_path: str): |
| 291 | """Write the given HTML string to a file.""" |
| 292 | with open(file_path, "w") as f: |
| 293 | f.write(html_string) |
| 294 | |
| 295 | |
| 296 | async def generate_pdf_from_html(html_file_path: str, pdf_file_path: str): |
| 297 | """Generate a PDF from the given HTML file using Playwright.""" |
| 298 | async with async_playwright() as p: |
| 299 | browser = await p.chromium.launch() |
| 300 | context = await browser.new_context() |
| 301 | page = await context.new_page() |
| 302 | await page.emulate_media(media="screen") |
| 303 | await page.goto(f"file://{os.getcwd()}/{html_file_path}") |
| 304 | await page.pdf(path=pdf_file_path) |
| 305 | await browser.close() |
| 306 | |
| 307 | |
| 308 | async def generate_report(request: GenerateReportRequest, session: AsyncSession): |
| 309 | logger.info("Generating report") |
| 310 | for row in request.rows: |
| 311 | for panel in row.panels: |
| 312 | panel.row_id = row.id |
| 313 | # Parse the timerange string |
| 314 | timerange = parse_timerange(request.timerange) |
| 315 | # Iterate over each row in the request |
| 316 | for row in request.rows: |
| 317 | # Iterate over each panel in the row |
| 318 | for panel in row.panels: |
| 319 | # Generate the iframe links for each panel |
| 320 | panel_urls = await generate_panel_urls_object(panel, timerange, session) |
| 321 | logger.info(f"Panel URLs: {panel_urls}") |
| 322 | # add the panel url to the request.panel_url |
| 323 | panel.panel_url = panel_urls |
| 324 | |
| 325 | async with async_playwright() as p: |
| 326 | browser = await p.chromium.launch(headless=True) |
| 327 | context = await browser.new_context(ignore_https_errors=True) |
| 328 | page = await context.new_page() |
| 329 | await login_to_page(page, session) |
| 330 | if not await check_login_success(page): |
| 331 | await browser.close() |
| 332 | return |
| 333 | # Flatten the list of panels |
| 334 | all_panels = [panel for row in request.rows for panel in row.panels] |
| 335 | panels = await capture_screenshots(page, all_panels) |
| 336 | await browser.close() |
| 337 | html_string = generate_html(panels, request) |
| 338 | await write_html_to_file(html_string, "report.html") |
| 339 | await generate_pdf_from_html("report.html", "report.pdf") |
| 340 | |
| 341 | # ! convert pdf to base64 and return |
| 342 | with open("report.pdf", "rb") as f: |
| 343 | pdf_base64 = base64.b64encode(f.read()).decode("utf-8") |
| 344 | |
| 345 | # ! Delete the report.html and report.pdf files |
| 346 | os.remove("report.html") |
| 347 | os.remove("report.pdf") |
| 348 | return GenerateReportResponse(base64_result=pdf_base64, message="Report generated successfully", success=True) |