auto upload image

.github/workflows/check_mcversion.yml

@@ -20,9 +20,16 @@ jobs:
          run: |
             python -m pip install --upgrade pip
             pip install requests
+            pip install botocore
+            pip install boto3
+            pip install beautifulsoup4
        - name: 运行更新脚本
          id: run_script
-         run: python Actions/check_mcv.py >> "$GITHUB_ENV"  
+         run: python Actions/check_mcv.py >> "$GITHUB_ENV"
+         env:
+            R2_ACCOUNT_ID: ${{ secrets.R2_ACCOUNT_ID }}
+            R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+            R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
        - name: 显示版本信息
          run: |
             printf '● 最新 %s 版: %s %s\n' "$latest_release_type" "$latest_release_id"

Actions/check_mcv.py

@@ -3,6 +3,8 @@ import requests
 import re
 import os
 from pathlib import Path
+from minecraft_version import MCVM
+from upload_header_image import upload_header_image
 
 LAUNCHER_MANIFSET_URL = 'https://piston-meta.mojang.com/mc/game/version_manifest.json'
 
@@ -34,24 +36,28 @@ def generate_filepath(ver_type:str, ver_id:str) -> Path:
 
 def update_library(version_mainfest, version_libloc:str, ver_type:str, ver_id:str):
     #gen content
+    filepath = Path(version_libloc) / generate_filepath(ver_type, ver_id)
+    if os.path.exists(filepath):
+        #print(f'版本库内已存在 {ver_type} 版 {ver_id} 的文件，跳过添加')
+        return False
+    header_image_link = 'null'
+    try:
+        header_image_link = upload_header_image(version=ver_id)
+    finally:
+        pass
     content = '---\n'
-    content += 'version-image-link: null\n'
+    content += f'version-image-link: {header_image_link}\n'
     #content += 'version-image-link-fallback: null\n'
     content += 'wip: true\n'
     content += f'server-jar: {get_server_jar(version_mainfest, ver_id)}\n'
     content += 'translator: null\n'
     content += f"cats: ['{ver_id[:4]}']\n"
     content += '---\n'
-    #write file
-    filepath = Path(version_libloc) / generate_filepath(ver_type, ver_id)
-    if os.path.exists(filepath):
-        #print(f'版本库内已存在 {ver_type} 版 {ver_id} 的文件，跳过添加')
-        return False
-    else:
-        with open(filepath.absolute(), "w", encoding='UTF-8') as file:
-            file.write(content)
-        #print(f'已添加 {ver_type} 版 {ver_id} 的文件到版本库')
-        return True
+    upload_header_image(version=ver_id)
+    with open(filepath.absolute(), "w", encoding='UTF-8') as file:
+        file.write(content)
+    #print(f'已添加 {ver_type} 版 {ver_id} 的文件到版本库')
+    return True
 
 def get_server_jar(version_mainfest,version_id):
     versions = version_mainfest.get('versions')

Actions/minecraft_version.py

@@ -0,0 +1,214 @@
+import re
+import json
+import requests
+from enum import Enum
+from typing import Optional, TypedDict, Literal
+from logging import Logger
+
+# region 版本模式定义
+DEVELOP_VERSION_PATTERN = re.compile(r'^(?P<major>[\d.]+)-(?P<type>[a-zA-Z]+)-?(?P<index>\d)+$')
+"开发版本号模式"
+RELEASE_PATTERN = re.compile(r'^\d{1,2}\.\d+(\.\d+)?$')
+"正式版本号模式"
+OLD_SNAPSHOT_PATTERN = re.compile(r'^(1\d)|(2[0-5])[w|W]\d{2}[A-Fa-f]$')
+"旧版本快照模式，用于 1.21.11 及以前的版本"
+# endregion
+
+# region 常量定义
+OFFICIAL_CHANGELOG_URL_PREFIX = 'https://minecraft.net/article/minecraft'
+LAUNCHER_MANIFEST_URL = 'https://piston-meta.mojang.com/mc/game/version_manifest.json'
+# endregion
+
+# region 官网更新日志重载词典
+OFFICIAL_CHANGE_LOG_OVERRIDE_LINKS: dict[str, str] = {}
+
+# endregion
+
+logger = Logger('MCV')
+
+# region 版本信息类定义
+class MinecraftManifestVersionInfo(TypedDict):
+    id: str
+    type: Literal['snapshot', 'release', 'old_beta', 'old_alpha']
+    url: str
+    time: str
+    releaseTime: str
+
+
+class MinecraftManifestLatestVersionInfo(TypedDict):
+    snapshot: Optional[str]
+    release: Optional[str]
+
+
+class MinecraftManifest(TypedDict):
+    latest: MinecraftManifestLatestVersionInfo
+    versions: list[MinecraftManifestVersionInfo]
+
+
+# endregion
+
+# region 版本类定义
+class VersionType(Enum):
+    SNAPSHOT = 'snapshot'
+    PRE_RELEASE = 'pre'
+    RELEASE_CANDIDATE = 'rc'
+    RELEASE = 'release'
+    OTHER = 'other'
+
+
+class MinecraftVersion:
+    id: str
+    type: VersionType
+    url: str
+    time: str
+    release_time: str
+    use_old_id_format: bool = False
+    major_version: Optional[str] = None
+    release_index: Optional[str] = None
+
+    __server_jar_url: Optional[str] = None
+
+    @property
+    def server_jar_url(self) -> str:
+        if self.__server_jar_url:
+            return self.__server_jar_url
+        server_jar_url: Optional[str]
+        response = requests.get(self.url, timeout=10000)
+        server_jar_url = json.loads(response.content).get('downloads').get('server').get('url')
+        if not server_jar_url:
+            raise ValueError(f"版本 {self.id} 的服务器端下载链接不存在")
+        self.__server_jar_url = server_jar_url
+        return server_jar_url
+
+    __change_log_url: Optional[str] = None
+
+    @property
+    def change_log_url(self) -> str:
+        """获取版本更新日志链接"""
+        if self.__change_log_url:
+            return self.__change_log_url
+        if override_url := OFFICIAL_CHANGE_LOG_OVERRIDE_LINKS.get(self.id):
+            self.__change_log_url = override_url
+            return override_url
+        if self.type == VersionType.OTHER:
+            return ""
+        if self.use_old_id_format:
+            self.__change_log_url = OFFICIAL_CHANGELOG_URL_PREFIX + '-snapshot-' + self.id.lower()
+        elif self.type == VersionType.RELEASE:
+            self.__change_log_url = f"{OFFICIAL_CHANGELOG_URL_PREFIX}-java-edition-{self.id.replace('.', '-')}"
+        else:
+            assert self.major_version is not None
+            self.__change_log_url = f"{OFFICIAL_CHANGELOG_URL_PREFIX}-" \
+                                    f"{self.major_version.replace('.', '-')}-" \
+                                    f"{self.type.name.lower().replace('_', '-')}-{self.release_index}"
+        return self.__change_log_url
+
+    def __init__(self, version_id: str, url: str, time: str, release_time: str):
+        self.id = version_id
+        self.url = url
+        self.time = time
+        self.release_time = release_time
+        if m := re.match(DEVELOP_VERSION_PATTERN, version_id):
+            self.type = VersionType(m.group('type').lower())
+            self.major_version = m.group('major')
+            self.release_index = m.group('index')
+        elif re.match(OLD_SNAPSHOT_PATTERN, version_id):
+            self.type = VersionType.SNAPSHOT
+            self.use_old_id_format = True
+        elif re.match(RELEASE_PATTERN, version_id):
+            self.type = VersionType.RELEASE
+            self.major_version = version_id
+        else:
+            self.type = VersionType.OTHER
+
+
+class MinecraftVersionManager:
+    __latest_development_version_id: Optional[str]
+    __latest_release_version_id: Optional[str]
+    __manifest: MinecraftManifest
+    __manifest_versions: list[MinecraftManifestVersionInfo]
+    __versions: dict[str, MinecraftVersion] = {}
+    __version_ids: list[str] = []
+
+    def __init__(self):
+        self.refresh_manifest()
+        self.__latest_development_version_id = self.__manifest['latest']['snapshot']
+        self.__latest_release_version_id = self.__manifest['latest']['release']
+        self.__manifest_versions = self.__manifest.get('versions', [])
+        self.__version_ids = [version['id'] for version in self.__manifest_versions]
+
+    def get(self, version_id: str) -> Optional[MinecraftVersion]:
+        if version_id in self.__versions:
+            return self.__versions[version_id]
+        for version_info in self.__manifest_versions:
+            if version_info['id'] == version_id:
+                version = MinecraftVersion(version_id=version_info['id'],
+                                           url=version_info['url'],
+                                           time=version_info['time'],
+                                           release_time=version_info['releaseTime'])
+                self.__versions[version_id] = version
+                return version
+        return None
+
+    def get_version_id_list(self) -> list[str]:
+        return self.__version_ids
+
+    def get_latest_version(self, version_type: Literal['any', 'development', 'release'] = 'any'
+                           ) -> Optional[MinecraftVersion]:
+        if version_type == 'any':
+            if self.__latest_development_version_id:
+                return self.get(self.__latest_development_version_id)
+            elif self.__latest_release_version_id:
+                return self.get(self.__latest_release_version_id)
+            else:
+                return None
+        elif version_type == 'development' and self.__latest_development_version_id:
+            return self.get(self.__latest_development_version_id)
+        elif version_type == 'release' and self.__latest_release_version_id:
+            return self.get(self.__latest_release_version_id)
+        else:
+            return None
+    
+    @property
+    def versions(self):
+        for version_id in self.__version_ids:
+            version = self.get(version_id)
+            assert version is not None
+            yield version
+
+    def get_latest_development_versions(self) -> Optional[list[MinecraftVersion]]:
+        latest_version = self.get_latest_version()
+        if not latest_version or latest_version.type == VersionType.RELEASE:
+            return None
+        latest_development_versions = []
+        current_major_version_branches = set()
+        for version in self.versions:
+            if version.major_version in current_major_version_branches:
+                break
+            if version.type == VersionType.RELEASE:
+                break
+            if version.type == VersionType.OTHER:
+                continue
+            current_major_version_branches.add(version.major_version)
+            latest_development_versions.append(version)
+        return latest_development_versions
+
+    def refresh_manifest(self):
+        self.__manifest = self.fetch_manifest()
+
+    @classmethod
+    def fetch_manifest(cls) -> MinecraftManifest:
+        logger.info("正在加载 MC 版本列表")
+        try:
+            response = requests.get(LAUNCHER_MANIFEST_URL, timeout=10000)
+            return json.loads(response.content)
+        except Exception as e:
+            logger.warning("加载 MC 版本列表失败")
+            logger.exception(e)
+            return MinecraftManifest(latest=MinecraftManifestLatestVersionInfo(snapshot=None, release=None),
+                                     versions=[])
+
+
+# endregion
+
+MCVM = MinecraftVersionManager()

Actions/upload_header_image.py

@@ -0,0 +1,244 @@
+import os
+import sys
+
+import boto3
+import requests
+from bs4 import BeautifulSoup
+from botocore.config import Config
+
+from minecraft_version import MCVM, MinecraftVersion
+
+
+# ============================================================
+# 配置
+# ============================================================
+
+USER_AGENT = (
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) "
+    "Chrome/139.0 Safari/537.36"
+)
+
+
+# ============================================================
+# R2
+# ============================================================
+
+def create_r2_client():
+    account_id = os.environ["R2_ACCOUNT_ID"]
+    access_key = os.environ["R2_ACCESS_KEY_ID"]
+    secret_key = os.environ["R2_SECRET_ACCESS_KEY"]
+
+    return boto3.client(
+        "s3",
+        endpoint_url=f"https://{account_id}.r2.cloudflarestorage.com",
+        aws_access_key_id=access_key,
+        aws_secret_access_key=secret_key,
+        region_name="auto",
+        config=Config(
+            signature_version="s3v4"
+        ),
+    )
+
+
+# ============================================================
+# 获取更新日志 URL
+# ============================================================
+
+def get_change_log_url(version_id: str) -> str:
+    version = MCVM.get(version_id)
+
+    if version is None:
+        raise ValueError(
+            f"MC 版本 {version_id} 不存在"
+        )
+
+    url = version.change_log_url
+
+    if not url:
+        raise ValueError(
+            f"MC 版本 {version_id} 没有更新日志链接"
+        )
+
+    print(f"版本: {version.id}")
+    print(f"更新日志: {url}")
+
+    return url
+
+
+# ============================================================
+# 下载 HTML
+# ============================================================
+
+def download_html(url: str) -> str:
+    print("[1/4] 下载 HTML...")
+
+    headers = {
+        "User-Agent": USER_AGENT,
+        "Referer": "https://www.minecraft.net/",
+        "Accept": (
+            "text/html,application/xhtml+xml,"
+            "application/xml;q=0.9,*/*;q=0.8"
+        ),
+    }
+
+    response = requests.get(
+        url,
+        headers=headers,
+        timeout=30,
+    )
+
+    response.raise_for_status()
+
+    response.encoding = response.apparent_encoding
+
+    return response.text
+
+def find_hero_image(html: str, page_url: str) -> str:
+    print("[2/4] 获取图片路径...")
+
+    soup = BeautifulSoup(
+        html,
+        "html.parser",
+    )
+
+    poster = soup.select_one(
+        "div.MC_articleHeroA_poster"
+    )
+
+    if poster is None:
+        raise RuntimeError(
+            "找不到 div.MC_articleHeroA_poster"
+        )
+
+    img = poster.select_one(
+        "picture img"
+    )
+
+    if img is None:
+        raise RuntimeError(
+            "找不到 picture img"
+        )
+
+    src = img.get("src")
+
+    if not src:
+        raise RuntimeError(
+            "img 没有 src 属性"
+        )
+
+    from urllib.parse import urljoin
+
+    image_url = urljoin(
+        page_url,
+        src,
+    )
+
+    print(f"图片: {image_url}")
+
+    return image_url
+
+def download_image(
+    image_url: str,
+    page_url: str,
+) -> tuple[bytes, str]:
+
+    print("[3/4] 下载图片...")
+
+    headers = {
+        "User-Agent": USER_AGENT,
+        "Referer": page_url,
+        "Accept": (
+            "image/avif,image/webp,image/apng,"
+            "image/svg+xml,image/*,*/*;q=0.8"
+        ),
+    }
+
+    response = requests.get(
+        image_url,
+        headers=headers,
+        timeout=60,
+    )
+
+    response.raise_for_status()
+
+    content_type = response.headers.get(
+        "Content-Type",
+        "application/octet-stream",
+    )
+
+    print(
+        f"Content-Type: {content_type}"
+    )
+    print(
+        f"大小: {len(response.content):,} bytes"
+    )
+
+    return response.content, content_type
+
+def upload_to_r2(
+    data: bytes,
+    bucket: str,
+    object_key: str,
+    content_type: str,
+):
+    print("[4/4] 上传 Cloudflare R2...")
+
+    s3 = create_r2_client()
+    s3.put_object(
+        Bucket=bucket,
+        Key=object_key,
+        Body=data,
+        ContentType=content_type,
+    )
+
+def upload_header_image(version):
+    if isinstance(version, str):
+        version = MCVM.get(version)
+    if not isinstance(version, MinecraftVersion):
+        raise TypeError()
+    change_log_url = version.change_log_url
+    html = download_html(
+        change_log_url
+    )
+    image_url = find_hero_image(
+        html,
+        change_log_url,
+    )
+    image_data, content_type = download_image(
+        image_url,
+        change_log_url,
+    )
+    bucket = "images"
+    object_key = f"news/version/{version.major_version}/{version.id}.webp"
+    upload_to_r2(
+        image_data,
+        bucket,
+        object_key,
+        content_type,
+    )
+    return f"https://images.bugjump.net/{object_key}"
+    
+
+def main():
+    if len(sys.argv) != 2:
+        print(
+            f"用法: {sys.argv[0]} <Minecraft版本号>"
+        )
+        print()
+        print("例如:")
+        print(f"  {sys.argv[0]} 26.3")
+        print(f"  {sys.argv[0]} 26.3-rc-1")
+        sys.exit(1)
+
+    version_id = sys.argv[1]
+    version = MCVM.get(version_id)
+    if not version:
+        raise ValueError('错误：未找到版本')
+    url = upload_header_image(version)
+
+    print(f"已上传至{url}")
+
+
+if __name__ == "__main__":
+    main()