MemeMeow/services/community_service.py

import os
import threading
import time
import json
import requests

import numpy as np
import pickle
import re
from typing import Optional, List, Dict

from config.settings import Config
from stpages.utils import ENDWITH_IMAGE

from services.embedding_service import EmbeddingService
from services.resource_pack import RESOURCE_PACK_SERVICE
from services.resource_pack_manager import RESOURCE_PACK_MANAGER
from services.utils import *
from services.llm_enhance import LLMEnhance
from services.cache_service import CacheService

"""manifest template:
{
"community_info": {
"resource_url": "https://github.com/MemeMeow-Studio/Memes-Community/raw/main",
"update_url": "https://github.com/MemeMeow-Studio/Memes-Community/raw/main/community_manifest.json",
"timestamp": 1747738810
},
"meme_libs": {
"4946f03b-11ae-4b03-81d5-db74dbf3f853": {
"name": "《这就是中国》大全",
"version": "1.0.1",
"author": "Official",
"description": "none",
"created_at": "2025-05-19",
"timestamp": 1747667458,
"tags": [],
"url": "https://raw.githubusercontent.com/MemeMeow-Studio/VVfromVideo/main/all_vvs",
"update_url": "https://raw.githubusercontent.com/MemeMeow-Studio/VVfromVideo/main/all_vvs/manifest.json",
"uuid": "4946f03b-11ae-4b03-81d5-db74dbf3f853"
},
"13e30784-6ebf-4c17-9139-aafe95719c61": {
"name": "精选v图",
"version": "1.0.1",
"author": "Official",
"description": "none",
"created_at": "2025-05-19",
"timestamp": 1747667458,
"tags": [],
"url": "https://raw.githubusercontent.com/MemeMeow-Studio/VVfromVideo/main/精选表情",
"update_url": "https://raw.githubusercontent.com/MemeMeow-Studio/VVfromVideo/main/精选表情/manifest.json",
"uuid": "13e30784-6ebf-4c17-9139-aafe95719c61"
}
}
}

"""

class CommunityService:
    def __init__(self):
        self.all_community_repos_info = []
        self.all_manifest_path = os.path.join(Config().temp_dir, "all_manifest.json")
        self.reload_community_info()


    def update_local_manifests(self):
        for community_info in self.all_community_repos_info:
            local_manifest = RESOURCE_PACK_MANAGER.enabled_packs.get(community_info.get("uuid"), None)
            # 如果本地没有manifest，或者本地的manifest比远程的旧，则下载远程的manifest
            if not local_manifest or community_info.get("timestamp") < local_manifest.get("timestamp"):
                RESOURCE_PACK_SERVICE.import_resource_pack_from_url(community_info.get("update_url"), uuid=community_info.get("uuid"))

    def reload_community_info(self):
        self.all_community_repos_info = []
        if not os.path.exists(self.all_manifest_path):
            self.download_and_compose_all_manifests()

        try:
            with open(self.all_manifest_path, 'r', encoding='utf-8') as f:
                manifest_data = json.load(f)
                meme_libs = manifest_data.get('meme_libs', {})
                self.all_community_repos_info = list(meme_libs.values())
        except Exception as e:
            print(f"Error loading manifest file: {str(e)}")
            self.all_community_repos_info = []

    def download_and_compose_all_manifests(self):
        composed_manifest = {}
        manifest_urls = Config().community.manifest_urls
        latest_timestamp = 0
        for url in manifest_urls:
            try:
                # 下载manifest文件
                response = requests.get(url)
                if response.status_code != 200:
                    print(f"Failed to download manifest from {url}, status code: {response.status_code}")
                    continue
                
                # 解析manifest内容
                manifest_data = response.json()
                
                if 'community_info' in manifest_data:
                    community_info = manifest_data['community_info']
                    # 更新最新的timestamp
                    if community_info.get('timestamp', 0) > latest_timestamp:
                        latest_timestamp = community_info['timestamp']
                    
                    # # 检查是否存在resource_url和update_url
                    # if 'resource_url' in community_info and 'update_url' in community_info:
                    #     # 更新或添加到composed_manifest
                    #     composed_manifest['community_info'] = {
                    #         "resource_url": community_info['resource_url'],
                    #         "update_url": community_info['update_url'],
                    #         "timestamp": latest_timestamp
                    #     }
                
                # 处理meme_libs部分
                if 'meme_libs' in manifest_data:
                    for uuid, lib_info in manifest_data['meme_libs'].items():
                        # 检查是否存在该UUID
                        if uuid not in composed_manifest.get('meme_libs', {}):
                            # UUID不存在，添加新条目
                            if 'meme_libs' not in composed_manifest:
                                composed_manifest['meme_libs'] = {}
                            composed_manifest['meme_libs'][uuid] = lib_info
                        else:
                            # UUID存在，比较timestamp
                            existing_timestamp = composed_manifest['meme_libs'][uuid].get('timestamp', 0)
                            new_timestamp = lib_info.get('timestamp', 0)
                            
                            if new_timestamp > existing_timestamp:
                                # 新条目timestamp更大，替换
                                composed_manifest['meme_libs'][uuid] = lib_info
            except Exception as e:
                print(f"Error processing manifest from {url}: {str(e)}")
                
        composed_manifest['community_info'] = {
            "timestamp": latest_timestamp
        }
        
        # 保存合并后的manifest到文件
        try:
            os.makedirs(os.path.dirname(self.all_manifest_path), exist_ok=True)
            with open(self.all_manifest_path, 'w', encoding='utf-8') as f:
                json.dump(composed_manifest, f, ensure_ascii=False, indent=2)
            print(f"Saved composed manifest to {self.all_manifest_path}")
        except Exception as e:
            print(f"Error saving composed manifest: {str(e)}")
        
        return composed_manifest