Agent/examples/content_finder/db/open_aigc_pattern_connection.py

"""open_aigc_pattern 数据库连接封装（与 content_finder 默认 DB_* 不同）。"""

from __future__ import annotations

import os
from urllib.parse import unquote, urlparse

import pymysql


def get_open_aigc_pattern_connection():
    """
    获取 open_aigc_pattern 数据库连接。
    """


    host = os.getenv("OPEN_AIGC_PATTERN_DB_HOST", "").strip()
    port = int(os.getenv("OPEN_AIGC_PATTERN_DB_PORT", "3306"))
    user = os.getenv("OPEN_AIGC_PATTERN_DB_USER", "").strip()
    password = os.getenv("OPEN_AIGC_PATTERN_DB_PASSWORD", "")
    database = os.getenv("OPEN_AIGC_PATTERN_DB_NAME", "open_aigc_pattern").strip()
    if not all([host, user, database]):
        raise ValueError(
            "open_aigc_pattern 数据库未配置：请设置 OPEN_AIGC_PATTERN_DB_URL "
            "或 OPEN_AIGC_PATTERN_DB_HOST/USER/PASSWORD/NAME"
        )

    return pymysql.connect(
        host=host,
        port=port,
        user=user,
        password=password,
        database=database,
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
    )