feat(etl): 实现 OSS URL 转移工具

+import requests
+import oss2
+from urllib.parse import urlparse
+
+ARCHIVE_DEV_HOST = 'archive-dev.oss-cn-beijing.aliyuncs.com'
+
+
+def transfer_url(url: str | None, oss_key: str, bucket: oss2.Bucket, base_url: str) -> str:
+    """
+    将 url 指向的文件转移到 archive-dev OSS 的 oss_key 路径。
+    若 url 为空或已在 archive-dev，直接返回原 url（不上传）。
+    返回新的 archive-dev URL。
+    """
+    if not url:
+        return ''
+    if ARCHIVE_DEV_HOST in url:
+        return url
+    resp = requests.get(url, timeout=30)
+    resp.raise_for_status()
+    bucket.put_object(oss_key, resp.content)
+    return f"{base_url.rstrip('/')}/{oss_key}"
+
+
+def build_oss_key(platform: str, category: str, filename: str) -> str:
+    """
+    构造 archive-dev 内的存储路径。
+    platform: 'qq' | 'kugou' | 'netease'
+    category: 'audio' | 'cover' | 'singer' | 'album'
+    filename: 带扩展名的文件名
+    """
+    return f"crawler/{platform}/{category}/{filename}"

+from unittest.mock import MagicMock, patch
+from etl_to_crawler.oss import transfer_url
+
+ARCHIVE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com/some/path.mp3"
+OTHER_URL = "https://hikoon-data-platform.oss-cn-beijing.aliyuncs.com/qq-audio/abc.mp3"
+BASE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com"
+
+def test_already_archive_dev_returns_as_is():
+    bucket = MagicMock()
+    result = transfer_url(ARCHIVE_URL, "any/key.mp3", bucket, BASE_URL)
+    assert result == ARCHIVE_URL
+    bucket.put_object.assert_not_called()
+
+def test_empty_url_returns_empty():
+    bucket = MagicMock()
+    result = transfer_url('', "any/key.mp3", bucket, BASE_URL)
+    assert result == ''
+    bucket.put_object.assert_not_called()
+
+def test_none_url_returns_empty():
+    bucket = MagicMock()
+    result = transfer_url(None, "any/key.mp3", bucket, BASE_URL)
+    assert result == ''
+
+def test_external_url_downloads_and_uploads():
+    bucket = MagicMock()
+    fake_content = b"audio_bytes"
+    with patch('etl_to_crawler.oss.requests.get') as mock_get:
+        mock_get.return_value.content = fake_content
+        mock_get.return_value.raise_for_status = MagicMock()
+        result = transfer_url(OTHER_URL, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)
+    bucket.put_object.assert_called_once_with("crawler/qq/audio/abc.mp3", fake_content)
+    assert result == f"{BASE_URL}/crawler/qq/audio/abc.mp3"