Function get_url_content

apps/oss/serializers/file.py:219–265

(url, application_id: str)

def get_url_content(url, application_id: str):
    application = Application.objects.filter(id=application_id).first()
    if application is None:
        raise AppApiException(500, _('Application does not exist'))
    if not application.file_upload_enable:
        raise AppApiException(500, _('File upload is not enabled'))
    file_limit = 50 * 1024 * 1024
    if application.file_upload_setting and application.file_upload_setting.get('fileLimit'):
        file_limit = application.file_upload_setting.get('fileLimit') * 1024 * 1024
    parsed = validate_url(url)

    # 创建带有安全检查的 session
    session = requests.Session()
    safe_adapter = SafeHTTPAdapter()
    session.mount('http://', safe_adapter)
    session.mount('https://', safe_adapter)

    try:
        response = session.get(
            url,
            timeout=3,
            allow_redirects=False
        )
    finally:
        session.close()

    final_host = urlparse(response.url).hostname
    if is_private_ip(final_host):
        raise ValueError("Blocked unsafe redirect to internal host")
    # 判断文件大小
    if int(response.headers.get('Content-Length', 0)) > file_limit:
        raise AppApiException(500, _('File size exceeds limit'))
    # 返回状态码 响应内容大小  响应的contenttype 还有字节流
    content_type = response.headers.get('Content-Type', '')
    # 根据内容类型决定如何处理
    if 'text' in content_type or 'json' in content_type:
        content = response.text
    else:
        # 二进制内容使用Base64编码
        content = base64.b64encode(response.content).decode('utf-8')

    return {
        'status_code': response.status_code,
        'Content-Length': response.headers.get('Content-Length', 0),
        'Content-Type': content_type,
        'content': content,
    }


def is_private_ip(host: str) -> bool: