This guide demonstrates how to programmatically fetch MySQL query results, render them as styled HTML tables, convert those tables into high-fidelity images, upload them to Alibaba Cloud OSS, and deliver them via DingTalk group robots — all using Python.

1. Installing wkhtmltoimage on CentOS

The wkhtmltoimage utility (part of the wkhtmltopdf project) is essential for converting HTML content into raster images. It requires several system-level dependencies to function correctly, especially for Chinese character rendering.

1.1 Install Core Dependencies

yum install -y xorg-x11-fonts-75dpi xorg-x11-fonts-Type1 fontconfig libX11 libXext libXrender

1.2 Download and Install the RPM Package

For CentOS 7, use the official release:

wget https://github.com/wkhtmltopdf/packaging/releases/download/0.12.6-1/wkhtmltox-0.12.6-1.centos7.x86_64.rpm
sudo rpm -Uvh wkhtmltox-0.12.6-1.centos7.x86_64.rpm

Verify installation:

/usr/local/bin/wkhtmltoimage --version

1.3 Fix Chinese Font Rendering

If Chinese characters appear blank or as tofu, confirm font availability:

fc-list :lang=zh

If no output, install a CJK font (e.g., simsun.ttc) into /usr/share/fonts/, then refresh the font cache:

sudo mkfontscale && sudo mkfontdir && sudo fc-cache -fv

2. Querying MySQL Data

A lightweight database connector class handles connection pooling and safe SQL execution:

import pymysql
from contextlib import contextmanager

class MySqlConnector:
    def __init__(self, host, port, user, password, database):
        self.config = {
            'host': host,
            'port': port,
            'user': user,
            'password': password,
            'database': database,
            'charset': 'utf8mb4',
            'connect_timeout': 5
        }

    @contextmanager
    def get_cursor(self):
        conn = None
        cursor = None
        try:
            conn = pymysql.connect(**self.config)
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            yield cursor
        finally:
            if cursor:
                cursor.close()
            if conn:
                conn.close()

# Usage
db = MySqlConnector('192.168.0.101', 3306, 'db_user', 'db_passwd', 'testdb')
with db.get_cursor() as cur:
    cur.execute("SELECT * FROM daily_metrics WHERE date = CURDATE();")
    rows = cur.fetchall()

3. Rendering Tables as Images

We use HTMLTable to generate semantic, style-rich HTML and imgkit to rasterize it.

Install Required Packages

pip3 install html-table imgkit

Generate Styled Table Image

import imgkit
from HTMLTable import HTMLTable

def generate_table_image(
    title: str,
    output_path: str,
    headers: tuple,
    data_rows: tuple,
    width: str = "250px"
):
    table = HTMLTable(caption=title)
    table.append_header_rows(headers)
    table.append_data_rows(data_rows)

    # Apply consistent styling
    table.caption.set_style({'font-size': '22px', 'margin-bottom': '12px'})
    table.set_style({
        'border-collapse': 'collapse',
        'font-family': 'Arial, sans-serif',
        'font-size': '14px',
        'width': '100%',
        'max-width': '1200px'
    })
    table.set_cell_style({
        'border': '1px solid #333',
        'padding': '8px 12px',
        'text-align': 'center',
        'width': width
    })
    table.set_header_row_style({
        'background-color': '#2c3e50',
        'color': 'white',
        'font-weight': 'bold'
    })

    html_content = f"""
    <html>
      <head><meta charset="UTF-8"></head>
      <body style="margin: 20px;">
        {table.to_html()}
      </body>
    </html>
    """

    # Explicitly specify wkhtmltoimage path if not in $PATH
    config = imgkit.config(wkhtmltoimage='/usr/local/bin/wkhtmltoimage')
    imgkit.from_string(html_content, output_path, config=config)

4. Uploading to Alibaba Cloud OSS

After image generation, upload securely to OSS for public access.

import oss2

def upload_to_oss(
    local_path: str,
    oss_key: str,
    endpoint: str = 'https://oss-cn-hangzhou.aliyuncs.com',
    bucket_name: str = 'your-bucket-name',
    access_key_id: str = 'YOUR_ID',
    access_key_secret: str = 'YOUR_SECRET'
):
    auth = oss2.Auth(access_key_id, access_key_secret)
    bucket = oss2.Bucket(auth, endpoint, bucket_name)
    
    # Upload with public-read ACL
    bucket.put_object_from_file(
        key=oss_key,
        filename=local_path,
        headers={'x-oss-object-acl': 'public-read'}
    )
    return f"https://{bucket_name}.{endpoint.replace('https://', '').split('.')[0]}.aliyuncs.com/{oss_key}"

5. Sending to DingTalk via Webhook

DingTalk supports Markdown-formatted messages cnotaining embeddded images from public URLs.

import requests
import json

def send_to_dingtalk(
    webhook_url: str,
    title: str,
    image_url: str,
    timestamp: str
):
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": f"""### {title}
> **Automated Daily Report**  
> ![{title}]({image_url})  
> ###### Generated at {timestamp}"""
        }
    }
    response = requests.post(webhook_url, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()

6. Full Orchestration Example

Putting it all to gether in a production-ready flow:

from datetime import datetime
import os

def run_daily_report():
    # Configuration
    OUTPUT_DIR = "/var/reports/images"
    os.makedirs(OUTPUT_DIR, exist_ok=True)

    # Step 1: Fetch data
    with db.get_cursor() as cur:
        cur.execute("SELECT hour, new_users, retention_rate FROM hourly_summary ORDER BY hour;")
        result = cur.fetchall()

    # Step 2: Build table structure
    headers = (("Hour", "New Users", "Retention (%)"),)
    data = tuple((r["hour"], r["new_users"], f"{r['retention_rate']:.2f}") for r in result)

    # Step 3: Generate image
    now = datetime.now()
    filename = f"report_{now.strftime('%Y%m%d_%H%M%S')}.png"
    image_path = os.path.join(OUTPUT_DIR, filename)
    generate_table_image(
        title=f"📊 Hourly User Metrics — {now.strftime('%Y-%m-%d')}",
        output_path=image_path,
        headers=headers,
        data_rows=data
    )

    # Step 4: Upload to OSS
    oss_url = upload_to_oss(
        local_path=image_path,
        oss_key=f"reports/{filename}",
        bucket_name="prod-reports",
        access_key_id=os.getenv("OSS_KEY_ID"),
        access_key_secret=os.getenv("OSS_KEY_SECRET")
    )

    # Step 5: Notify DingTalk
    send_to_dingtalk(
        webhook_url=os.getenv("DINGTALK_WEBHOOK"),
        title="Daily Metrics Report",
        image_url=oss_url,
        timestamp=now.strftime("%Y-%m-%d %H:%M:%S")
    )

if __name__ == "__main__":
    run_daily_report()

Ensure environment variables (OSS_KEY_ID, OSS_KEY_SECRET, DINGTALK_WEBHOOK) are set securely before execution. For scheduled runs, integrate with cron or a workflow scheduler like Apache Airflow.